2015-04-07 11:44:25 -07:00
|
|
|
package node
|
2014-07-08 00:02:04 -07:00
|
|
|
|
|
|
|
import (
|
2015-06-14 15:18:17 -07:00
|
|
|
"bytes"
|
2017-10-13 02:11:16 -07:00
|
|
|
"encoding/json"
|
2017-01-28 08:27:13 -08:00
|
|
|
"errors"
|
2017-09-20 15:29:36 -07:00
|
|
|
"fmt"
|
2015-04-20 15:29:01 -07:00
|
|
|
"net"
|
2015-04-10 02:12:17 -07:00
|
|
|
"net/http"
|
2015-05-12 17:40:29 -07:00
|
|
|
"strings"
|
2014-07-08 15:33:26 -07:00
|
|
|
|
2017-01-28 08:27:13 -08:00
|
|
|
abci "github.com/tendermint/abci/types"
|
2017-02-22 02:53:59 -08:00
|
|
|
crypto "github.com/tendermint/go-crypto"
|
|
|
|
wire "github.com/tendermint/go-wire"
|
2017-09-20 15:29:36 -07:00
|
|
|
cmn "github.com/tendermint/tmlibs/common"
|
|
|
|
dbm "github.com/tendermint/tmlibs/db"
|
|
|
|
"github.com/tendermint/tmlibs/log"
|
|
|
|
|
2015-11-01 11:34:08 -08:00
|
|
|
bc "github.com/tendermint/tendermint/blockchain"
|
2017-05-01 17:09:29 -07:00
|
|
|
cfg "github.com/tendermint/tendermint/config"
|
2018-01-21 08:45:04 -08:00
|
|
|
cs "github.com/tendermint/tendermint/consensus"
|
2017-11-19 20:22:25 -08:00
|
|
|
"github.com/tendermint/tendermint/evidence"
|
2015-04-01 17:30:16 -07:00
|
|
|
mempl "github.com/tendermint/tendermint/mempool"
|
2017-09-20 15:29:36 -07:00
|
|
|
"github.com/tendermint/tendermint/p2p"
|
2018-01-20 18:44:09 -08:00
|
|
|
"github.com/tendermint/tendermint/p2p/pex"
|
2017-10-30 12:34:49 -07:00
|
|
|
"github.com/tendermint/tendermint/p2p/trust"
|
2015-12-01 20:12:01 -08:00
|
|
|
"github.com/tendermint/tendermint/proxy"
|
2017-04-26 16:57:33 -07:00
|
|
|
rpccore "github.com/tendermint/tendermint/rpc/core"
|
|
|
|
grpccore "github.com/tendermint/tendermint/rpc/grpc"
|
|
|
|
rpc "github.com/tendermint/tendermint/rpc/lib"
|
|
|
|
rpcserver "github.com/tendermint/tendermint/rpc/lib/server"
|
2015-04-01 17:30:16 -07:00
|
|
|
sm "github.com/tendermint/tendermint/state"
|
2017-04-18 16:56:41 -07:00
|
|
|
"github.com/tendermint/tendermint/state/txindex"
|
|
|
|
"github.com/tendermint/tendermint/state/txindex/kv"
|
|
|
|
"github.com/tendermint/tendermint/state/txindex/null"
|
2015-04-20 15:29:01 -07:00
|
|
|
"github.com/tendermint/tendermint/types"
|
2018-02-19 10:20:01 -08:00
|
|
|
priv_val "github.com/tendermint/tendermint/types/priv_validator"
|
2016-01-20 10:12:42 -08:00
|
|
|
"github.com/tendermint/tendermint/version"
|
2014-07-08 00:02:04 -07:00
|
|
|
|
2017-01-28 08:27:13 -08:00
|
|
|
_ "net/http/pprof"
|
|
|
|
)
|
2015-04-27 07:25:11 -07:00
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
//------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
// DBContext specifies config information for loading a new DB.
|
|
|
|
type DBContext struct {
|
|
|
|
ID string
|
|
|
|
Config *cfg.Config
|
|
|
|
}
|
|
|
|
|
|
|
|
// DBProvider takes a DBContext and returns an instantiated DB.
|
|
|
|
type DBProvider func(*DBContext) (dbm.DB, error)
|
|
|
|
|
|
|
|
// DefaultDBProvider returns a database using the DBBackend and DBDir
|
|
|
|
// specified in the ctx.Config.
|
|
|
|
func DefaultDBProvider(ctx *DBContext) (dbm.DB, error) {
|
2018-02-03 00:50:59 -08:00
|
|
|
dbType := dbm.DBBackendType(ctx.Config.DBBackend)
|
|
|
|
return dbm.NewDB(ctx.ID, dbType, ctx.Config.DBDir()), nil
|
2017-09-20 15:29:36 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
// GenesisDocProvider returns a GenesisDoc.
|
|
|
|
// It allows the GenesisDoc to be pulled from sources other than the
|
|
|
|
// filesystem, for instance from a distributed key-value store cluster.
|
|
|
|
type GenesisDocProvider func() (*types.GenesisDoc, error)
|
|
|
|
|
|
|
|
// DefaultGenesisDocProviderFunc returns a GenesisDocProvider that loads
|
|
|
|
// the GenesisDoc from the config.GenesisFile() on the filesystem.
|
|
|
|
func DefaultGenesisDocProviderFunc(config *cfg.Config) GenesisDocProvider {
|
|
|
|
return func() (*types.GenesisDoc, error) {
|
|
|
|
return types.GenesisDocFromFile(config.GenesisFile())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-09-21 14:08:17 -07:00
|
|
|
// NodeProvider takes a config and a logger and returns a ready to go Node.
|
|
|
|
type NodeProvider func(*cfg.Config, log.Logger) (*Node, error)
|
|
|
|
|
|
|
|
// DefaultNewNode returns a Tendermint node with default settings for the
|
|
|
|
// PrivValidator, ClientCreator, GenesisDoc, and DBProvider.
|
|
|
|
// It implements NodeProvider.
|
|
|
|
func DefaultNewNode(config *cfg.Config, logger log.Logger) (*Node, error) {
|
|
|
|
return NewNode(config,
|
|
|
|
types.LoadOrGenPrivValidatorFS(config.PrivValidatorFile()),
|
|
|
|
proxy.DefaultClientCreator(config.ProxyApp, config.ABCI, config.DBDir()),
|
|
|
|
DefaultGenesisDocProviderFunc(config),
|
|
|
|
DefaultDBProvider,
|
2018-02-07 09:35:01 -08:00
|
|
|
logger,
|
|
|
|
)
|
2017-09-21 14:08:17 -07:00
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
//------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
// Node is the highest level interface to a full Tendermint node.
|
|
|
|
// It includes all configuration information and running services.
|
2014-07-09 18:33:44 -07:00
|
|
|
type Node struct {
|
2017-01-15 16:59:10 -08:00
|
|
|
cmn.BaseService
|
|
|
|
|
2017-03-04 20:25:12 -08:00
|
|
|
// config
|
2017-05-04 11:07:08 -07:00
|
|
|
config *cfg.Config
|
2017-09-18 15:12:31 -07:00
|
|
|
genesisDoc *types.GenesisDoc // initial validator set
|
|
|
|
privValidator types.PrivValidator // local node's validator key
|
2017-03-04 20:25:12 -08:00
|
|
|
|
|
|
|
// network
|
2017-10-30 15:45:54 -07:00
|
|
|
sw *p2p.Switch // p2p connections
|
2018-01-20 18:44:09 -08:00
|
|
|
addrBook pex.AddrBook // known peers
|
2017-10-30 15:45:54 -07:00
|
|
|
trustMetricStore *trust.TrustMetricStore // trust metrics for all peers
|
2017-03-04 20:25:12 -08:00
|
|
|
|
|
|
|
// services
|
2017-12-27 19:09:48 -08:00
|
|
|
eventBus *types.EventBus // pub/sub for services
|
|
|
|
stateDB dbm.DB
|
2018-01-21 08:45:04 -08:00
|
|
|
blockStore *bc.BlockStore // store the blockchain to disk
|
|
|
|
bcReactor *bc.BlockchainReactor // for fast-syncing
|
|
|
|
mempoolReactor *mempl.MempoolReactor // for gossipping transactions
|
|
|
|
consensusState *cs.ConsensusState // latest consensus state
|
|
|
|
consensusReactor *cs.ConsensusReactor // for participating in the consensus
|
|
|
|
evidencePool *evidence.EvidencePool // tracking evidence
|
|
|
|
proxyApp proxy.AppConns // connection to the application
|
|
|
|
rpcListeners []net.Listener // rpc servers
|
2017-04-18 16:56:41 -07:00
|
|
|
txIndexer txindex.TxIndexer
|
2017-11-28 19:24:37 -08:00
|
|
|
indexerService *txindex.IndexerService
|
2014-07-09 18:33:44 -07:00
|
|
|
}
|
2014-07-08 00:02:04 -07:00
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// NewNode returns a new, ready to go, Tendermint Node.
|
|
|
|
func NewNode(config *cfg.Config,
|
|
|
|
privValidator types.PrivValidator,
|
|
|
|
clientCreator proxy.ClientCreator,
|
2017-09-21 18:44:36 -07:00
|
|
|
genesisDocProvider GenesisDocProvider,
|
2017-09-20 15:29:36 -07:00
|
|
|
dbProvider DBProvider,
|
|
|
|
logger log.Logger) (*Node, error) {
|
|
|
|
|
2014-10-22 17:20:44 -07:00
|
|
|
// Get BlockStore
|
2017-09-20 15:29:36 -07:00
|
|
|
blockStoreDB, err := dbProvider(&DBContext{"blockstore", config})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2015-03-25 00:15:18 -07:00
|
|
|
blockStore := bc.NewBlockStore(blockStoreDB)
|
2014-10-22 17:20:44 -07:00
|
|
|
|
|
|
|
// Get State
|
2017-09-20 15:29:36 -07:00
|
|
|
stateDB, err := dbProvider(&DBContext{"state", config})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2017-10-05 05:50:05 -07:00
|
|
|
|
2017-10-13 02:11:16 -07:00
|
|
|
// Get genesis doc
|
2017-12-28 15:26:13 -08:00
|
|
|
// TODO: move to state package?
|
2017-10-13 02:11:16 -07:00
|
|
|
genDoc, err := loadGenesisDoc(stateDB)
|
2017-10-05 05:50:05 -07:00
|
|
|
if err != nil {
|
2017-10-13 02:11:16 -07:00
|
|
|
genDoc, err = genesisDocProvider()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
// save genesis doc to prevent a certain class of user errors (e.g. when it
|
|
|
|
// was changed, accidentally or not). Also good for audit trail.
|
2017-10-17 02:33:57 -07:00
|
|
|
saveGenesisDoc(stateDB, genDoc)
|
2017-10-05 05:50:05 -07:00
|
|
|
}
|
|
|
|
|
2017-12-28 15:26:13 -08:00
|
|
|
state, err := sm.LoadStateFromDBOrGenesisDoc(stateDB, genDoc)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2017-09-20 15:29:36 -07:00
|
|
|
}
|
2017-05-12 14:07:53 -07:00
|
|
|
|
2016-08-23 18:44:07 -07:00
|
|
|
// Create the proxyApp, which manages connections (consensus, mempool, query)
|
2017-12-27 19:09:48 -08:00
|
|
|
// and sync tendermint and the app by performing a handshake
|
|
|
|
// and replaying any necessary blocks
|
2017-11-18 18:02:58 -08:00
|
|
|
consensusLogger := logger.With("module", "consensus")
|
2018-01-21 08:45:04 -08:00
|
|
|
handshaker := cs.NewHandshaker(stateDB, state, blockStore)
|
2017-05-12 14:07:53 -07:00
|
|
|
handshaker.SetLogger(consensusLogger)
|
|
|
|
proxyApp := proxy.NewAppConns(clientCreator, handshaker)
|
|
|
|
proxyApp.SetLogger(logger.With("module", "proxy"))
|
2017-11-06 10:20:39 -08:00
|
|
|
if err := proxyApp.Start(); err != nil {
|
2017-09-20 15:29:36 -07:00
|
|
|
return nil, fmt.Errorf("Error starting proxy app connections: %v", err)
|
2016-09-11 13:02:29 -07:00
|
|
|
}
|
2015-12-01 20:12:01 -08:00
|
|
|
|
2017-02-17 07:51:05 -08:00
|
|
|
// reload the state (it may have been updated by the handshake)
|
2017-02-22 02:53:59 -08:00
|
|
|
state = sm.LoadState(stateDB)
|
|
|
|
|
2018-02-22 15:40:29 -08:00
|
|
|
// Connect to external signing process, if an address is provided.
|
2018-02-07 09:35:01 -08:00
|
|
|
if config.PrivValidatorAddr != "" {
|
2018-02-19 10:20:01 -08:00
|
|
|
var (
|
|
|
|
privKey = crypto.GenPrivKeyEd25519()
|
2018-02-22 15:40:29 -08:00
|
|
|
pvsc = priv_val.NewSocketClient(
|
2018-02-19 10:20:01 -08:00
|
|
|
logger.With("module", "priv_val"),
|
|
|
|
config.PrivValidatorAddr,
|
|
|
|
&privKey,
|
|
|
|
)
|
2018-02-07 09:35:01 -08:00
|
|
|
)
|
2018-02-19 10:20:01 -08:00
|
|
|
|
2018-02-22 15:40:29 -08:00
|
|
|
if err := pvsc.Start(); err != nil {
|
|
|
|
return nil, fmt.Errorf("Error starting private validator client: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
privValidator = pvsc
|
2018-02-07 09:35:01 -08:00
|
|
|
}
|
|
|
|
|
2016-03-24 18:08:18 -07:00
|
|
|
// Decide whether to fast-sync or not
|
|
|
|
// We don't fast-sync when the only validator is us.
|
2017-05-01 17:09:29 -07:00
|
|
|
fastSync := config.FastSync
|
2016-03-24 18:08:18 -07:00
|
|
|
if state.Validators.Size() == 1 {
|
|
|
|
addr, _ := state.Validators.GetByIndex(0)
|
2017-09-21 13:32:02 -07:00
|
|
|
if bytes.Equal(privValidator.GetAddress(), addr) {
|
2016-03-24 18:08:18 -07:00
|
|
|
fastSync = false
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-05-15 02:02:40 -07:00
|
|
|
// Log whether this node is a validator or an observer
|
2017-09-21 13:32:02 -07:00
|
|
|
if state.Validators.HasAddress(privValidator.GetAddress()) {
|
2018-01-04 10:52:41 -08:00
|
|
|
consensusLogger.Info("This node is a validator", "addr", privValidator.GetAddress(), "pubKey", privValidator.GetPubKey())
|
2017-05-15 02:02:40 -07:00
|
|
|
} else {
|
2018-01-04 10:52:41 -08:00
|
|
|
consensusLogger.Info("This node is not a validator", "addr", privValidator.GetAddress(), "pubKey", privValidator.GetPubKey())
|
2017-05-15 02:02:40 -07:00
|
|
|
}
|
|
|
|
|
2015-07-20 14:40:41 -07:00
|
|
|
// Make MempoolReactor
|
2017-05-15 10:33:45 -07:00
|
|
|
mempoolLogger := logger.With("module", "mempool")
|
2017-07-25 10:57:11 -07:00
|
|
|
mempool := mempl.NewMempool(config.Mempool, proxyApp.Mempool(), state.LastBlockHeight)
|
2018-01-18 21:57:00 -08:00
|
|
|
mempool.InitWAL() // no need to have the mempool wal during tests
|
2017-05-12 14:07:53 -07:00
|
|
|
mempool.SetLogger(mempoolLogger)
|
2017-05-01 17:09:29 -07:00
|
|
|
mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool)
|
2017-05-12 14:07:53 -07:00
|
|
|
mempoolReactor.SetLogger(mempoolLogger)
|
2014-10-22 17:20:44 -07:00
|
|
|
|
2017-08-04 18:46:17 -07:00
|
|
|
if config.Consensus.WaitForTxs() {
|
2017-07-25 10:57:11 -07:00
|
|
|
mempool.EnableTxsAvailable()
|
2017-07-13 10:19:44 -07:00
|
|
|
}
|
|
|
|
|
2017-11-18 18:02:58 -08:00
|
|
|
// Make Evidence Reactor
|
|
|
|
evidenceDB, err := dbProvider(&DBContext{"evidence", config})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
evidenceLogger := logger.With("module", "evidence")
|
|
|
|
evidenceStore := evidence.NewEvidenceStore(evidenceDB)
|
2017-12-28 18:08:39 -08:00
|
|
|
evidencePool := evidence.NewEvidencePool(stateDB, evidenceStore)
|
2017-11-18 18:02:58 -08:00
|
|
|
evidencePool.SetLogger(evidenceLogger)
|
2017-12-26 17:34:57 -08:00
|
|
|
evidenceReactor := evidence.NewEvidenceReactor(evidencePool)
|
2017-11-18 18:02:58 -08:00
|
|
|
evidenceReactor.SetLogger(evidenceLogger)
|
|
|
|
|
2017-12-27 19:09:48 -08:00
|
|
|
blockExecLogger := logger.With("module", "state")
|
|
|
|
// make block executor for consensus and blockchain reactors to execute blocks
|
2017-12-28 15:26:13 -08:00
|
|
|
blockExec := sm.NewBlockExecutor(stateDB, blockExecLogger, proxyApp.Consensus(), mempool, evidencePool)
|
2017-12-27 19:09:48 -08:00
|
|
|
|
|
|
|
// Make BlockchainReactor
|
|
|
|
bcReactor := bc.NewBlockchainReactor(state.Copy(), blockExec, blockStore, fastSync)
|
|
|
|
bcReactor.SetLogger(logger.With("module", "blockchain"))
|
|
|
|
|
2015-07-20 14:40:41 -07:00
|
|
|
// Make ConsensusReactor
|
2018-01-21 08:45:04 -08:00
|
|
|
consensusState := cs.NewConsensusState(config.Consensus, state.Copy(),
|
2017-12-27 19:09:48 -08:00
|
|
|
blockExec, blockStore, mempool, evidencePool)
|
2017-05-12 14:07:53 -07:00
|
|
|
consensusState.SetLogger(consensusLogger)
|
2014-10-24 14:37:12 -07:00
|
|
|
if privValidator != nil {
|
2016-11-16 17:52:08 -08:00
|
|
|
consensusState.SetPrivValidator(privValidator)
|
2014-10-24 14:37:12 -07:00
|
|
|
}
|
2018-01-21 08:45:04 -08:00
|
|
|
consensusReactor := cs.NewConsensusReactor(consensusState, fastSync)
|
2017-05-12 14:07:53 -07:00
|
|
|
consensusReactor.SetLogger(consensusLogger)
|
|
|
|
|
|
|
|
p2pLogger := logger.With("module", "p2p")
|
2014-10-22 17:20:44 -07:00
|
|
|
|
2017-05-01 19:14:37 -07:00
|
|
|
sw := p2p.NewSwitch(config.P2P)
|
2017-05-12 14:07:53 -07:00
|
|
|
sw.SetLogger(p2pLogger)
|
2015-03-30 22:28:34 -07:00
|
|
|
sw.AddReactor("MEMPOOL", mempoolReactor)
|
|
|
|
sw.AddReactor("BLOCKCHAIN", bcReactor)
|
|
|
|
sw.AddReactor("CONSENSUS", consensusReactor)
|
2017-11-18 18:02:58 -08:00
|
|
|
sw.AddReactor("EVIDENCE", evidenceReactor)
|
2014-07-08 00:02:04 -07:00
|
|
|
|
2016-11-30 19:58:47 -08:00
|
|
|
// Optionally, start the pex reactor
|
2018-01-20 18:44:09 -08:00
|
|
|
var addrBook pex.AddrBook
|
2017-10-30 15:45:54 -07:00
|
|
|
var trustMetricStore *trust.TrustMetricStore
|
2017-05-01 17:09:29 -07:00
|
|
|
if config.P2P.PexReactor {
|
2018-01-20 18:44:09 -08:00
|
|
|
addrBook = pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
|
2017-05-12 14:07:53 -07:00
|
|
|
addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
|
2017-10-30 15:45:54 -07:00
|
|
|
|
|
|
|
// Get the trust metric history data
|
|
|
|
trustHistoryDB, err := dbProvider(&DBContext{"trusthistory", config})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
trustMetricStore = trust.NewTrustMetricStore(trustHistoryDB, trust.DefaultConfig())
|
|
|
|
trustMetricStore.SetLogger(p2pLogger)
|
|
|
|
|
2018-01-14 10:24:43 -08:00
|
|
|
var seeds []string
|
|
|
|
if config.P2P.Seeds != "" {
|
|
|
|
seeds = strings.Split(config.P2P.Seeds, ",")
|
|
|
|
}
|
2018-01-20 18:44:09 -08:00
|
|
|
pexReactor := pex.NewPEXReactor(addrBook,
|
2018-02-08 05:20:55 -08:00
|
|
|
&pex.PEXReactorConfig{Seeds: seeds, SeedMode: config.P2P.SeedMode})
|
2017-05-12 14:07:53 -07:00
|
|
|
pexReactor.SetLogger(p2pLogger)
|
2016-11-30 19:58:47 -08:00
|
|
|
sw.AddReactor("PEX", pexReactor)
|
|
|
|
}
|
|
|
|
|
2017-01-28 08:27:13 -08:00
|
|
|
// Filter peers by addr or pubkey with an ABCI query.
|
|
|
|
// If the query return code is OK, add peer.
|
|
|
|
// XXX: Query format subject to change
|
2017-05-01 17:09:29 -07:00
|
|
|
if config.FilterPeers {
|
2016-09-10 12:16:23 -07:00
|
|
|
// NOTE: addr is ip:port
|
2016-09-08 15:56:02 -07:00
|
|
|
sw.SetAddrFilter(func(addr net.Addr) error {
|
2017-01-28 08:27:13 -08:00
|
|
|
resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/addr/%s", addr.String())})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2017-12-01 22:47:55 -08:00
|
|
|
if resQuery.IsErr() {
|
2018-01-05 22:26:51 -08:00
|
|
|
return fmt.Errorf("Error querying abci app: %v", resQuery)
|
2016-09-08 15:56:02 -07:00
|
|
|
}
|
2017-12-01 22:47:55 -08:00
|
|
|
return nil
|
2016-09-08 15:56:02 -07:00
|
|
|
})
|
p2p: introduce peerConn to simplify peer creation (#1226)
* expose AuthEnc in the P2P config
if AuthEnc is true, dialed peers must have a node ID in the address and
it must match the persistent pubkey from the secret handshake.
Refs #1157
* fixes after my own review
* fix docs
* fix build failure
```
p2p/pex/pex_reactor_test.go:288:88: cannot use seed.NodeInfo().NetAddress() (type *p2p.NetAddress) as type string in array or slice literal
```
* p2p: introduce peerConn to simplify peer creation
* Introduce `peerConn` containing the known fields of `peer`
* `peer` only created in `sw.addPeer` once handshake is complete and NodeInfo is checked
* Eliminates some mutable variables and makes the code flow better
* Simplifies the `newXxxPeer` funcs
* Use ID instead of PubKey where possible.
* SetPubKeyFilter -> SetIDFilter
* nodeInfo.Validate takes ID
* remove peer.PubKey()
* persistent node ids
* fixes from review
* test: use ip_plus_id.sh more
* fix invalid memory panic during fast_sync test
```
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: panic: runtime error: invalid memory address or nil pointer dereference
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x98dd3e]
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]:
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: goroutine 3432 [running]:
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.newOutboundPeerConn(0xc423fd1380, 0xc420933e00, 0x1, 0x1239a60, 0
xc420128c40, 0x2, 0x42caf6, 0xc42001f300, 0xc422831d98, 0xc4227951c0, ...)
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/peer.go:123 +0x31e
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).addOutboundPeerWithConfig(0xc4200ad040, 0xc423fd1380, 0
xc420933e00, 0xc423f48801, 0x28, 0x2)
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:455 +0x12b
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).DialPeerWithAddress(0xc4200ad040, 0xc423fd1380, 0x1, 0x
0, 0x0)
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:371 +0xdc
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).reconnectToPeer(0xc4200ad040, 0x123e000, 0xc42007bb00)
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:290 +0x25f
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: created by github.com/tendermint/tendermint/p2p.(*Switch).StopPeerForError
2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:256 +0x1b7
```
2018-02-27 03:54:40 -08:00
|
|
|
sw.SetIDFilter(func(id p2p.ID) error {
|
|
|
|
resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/pubkey/%s", id)})
|
2017-01-28 08:27:13 -08:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2017-12-01 22:47:55 -08:00
|
|
|
if resQuery.IsErr() {
|
2018-01-05 22:26:51 -08:00
|
|
|
return fmt.Errorf("Error querying abci app: %v", resQuery)
|
2016-09-08 15:56:02 -07:00
|
|
|
}
|
2017-12-01 22:47:55 -08:00
|
|
|
return nil
|
2016-09-08 15:56:02 -07:00
|
|
|
})
|
|
|
|
}
|
2016-08-25 11:17:15 -07:00
|
|
|
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
eventBus := types.NewEventBus()
|
|
|
|
eventBus.SetLogger(logger.With("module", "events"))
|
|
|
|
|
|
|
|
// services which will be publishing and/or subscribing for messages (events)
|
2017-12-28 16:35:56 -08:00
|
|
|
// consensusReactor will set it on consensusState and blockExecutor
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
consensusReactor.SetEventBus(eventBus)
|
2015-04-02 17:35:27 -07:00
|
|
|
|
2017-11-15 13:07:08 -08:00
|
|
|
// Transaction indexing
|
|
|
|
var txIndexer txindex.TxIndexer
|
|
|
|
switch config.TxIndex.Indexer {
|
|
|
|
case "kv":
|
|
|
|
store, err := dbProvider(&DBContext{"tx_index", config})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2017-11-30 18:02:39 -08:00
|
|
|
if config.TxIndex.IndexTags != "" {
|
|
|
|
txIndexer = kv.NewTxIndex(store, kv.IndexTags(strings.Split(config.TxIndex.IndexTags, ",")))
|
|
|
|
} else if config.TxIndex.IndexAllTags {
|
|
|
|
txIndexer = kv.NewTxIndex(store, kv.IndexAllTags())
|
|
|
|
} else {
|
|
|
|
txIndexer = kv.NewTxIndex(store)
|
|
|
|
}
|
2017-11-15 13:07:08 -08:00
|
|
|
default:
|
|
|
|
txIndexer = &null.TxIndex{}
|
|
|
|
}
|
|
|
|
|
2017-11-28 19:24:37 -08:00
|
|
|
indexerService := txindex.NewIndexerService(txIndexer, eventBus)
|
2017-11-15 13:07:08 -08:00
|
|
|
|
2015-09-24 12:48:44 -07:00
|
|
|
// run the profile server
|
2017-05-01 17:09:29 -07:00
|
|
|
profileHost := config.ProfListenAddress
|
2015-09-24 12:48:44 -07:00
|
|
|
if profileHost != "" {
|
|
|
|
go func() {
|
2017-06-14 01:50:49 -07:00
|
|
|
logger.Error("Profile server", "err", http.ListenAndServe(profileHost, nil))
|
2015-09-24 12:48:44 -07:00
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
2017-01-15 16:59:10 -08:00
|
|
|
node := &Node{
|
2017-03-04 20:25:12 -08:00
|
|
|
config: config,
|
2017-10-05 05:50:05 -07:00
|
|
|
genesisDoc: genDoc,
|
2017-03-04 20:25:12 -08:00
|
|
|
privValidator: privValidator,
|
|
|
|
|
2017-10-30 15:45:54 -07:00
|
|
|
sw: sw,
|
|
|
|
addrBook: addrBook,
|
|
|
|
trustMetricStore: trustMetricStore,
|
2017-03-04 20:25:12 -08:00
|
|
|
|
2017-12-27 19:09:48 -08:00
|
|
|
stateDB: stateDB,
|
2015-01-06 15:51:41 -08:00
|
|
|
blockStore: blockStore,
|
2015-03-25 00:15:18 -07:00
|
|
|
bcReactor: bcReactor,
|
2014-10-22 17:20:44 -07:00
|
|
|
mempoolReactor: mempoolReactor,
|
2015-01-11 14:27:46 -08:00
|
|
|
consensusState: consensusState,
|
2014-10-22 17:20:44 -07:00
|
|
|
consensusReactor: consensusReactor,
|
2017-11-18 18:02:58 -08:00
|
|
|
evidencePool: evidencePool,
|
2016-08-22 13:00:08 -07:00
|
|
|
proxyApp: proxyApp,
|
2017-02-22 02:53:59 -08:00
|
|
|
txIndexer: txIndexer,
|
2017-11-28 19:24:37 -08:00
|
|
|
indexerService: indexerService,
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
eventBus: eventBus,
|
2014-07-09 18:33:44 -07:00
|
|
|
}
|
2017-05-02 00:53:32 -07:00
|
|
|
node.BaseService = *cmn.NewBaseService(logger, "Node", node)
|
2017-09-20 15:29:36 -07:00
|
|
|
return node, nil
|
2014-07-09 18:33:44 -07:00
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// OnStart starts the Node. It implements cmn.Service.
|
2017-01-15 16:59:10 -08:00
|
|
|
func (n *Node) OnStart() error {
|
2017-11-06 10:20:39 -08:00
|
|
|
err := n.eventBus.Start()
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2017-10-23 07:03:54 -07:00
|
|
|
// Run the RPC server first
|
|
|
|
// so we can eg. receive txs for the first block
|
|
|
|
if n.config.RPC.ListenAddress != "" {
|
|
|
|
listeners, err := n.startRPC()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
n.rpcListeners = listeners
|
|
|
|
}
|
|
|
|
|
2017-01-15 16:59:10 -08:00
|
|
|
// Create & add listener
|
2017-07-19 12:03:36 -07:00
|
|
|
protocol, address := cmn.ProtocolAndAddress(n.config.P2P.ListenAddress)
|
2017-05-02 00:53:32 -07:00
|
|
|
l := p2p.NewDefaultListener(protocol, address, n.config.P2P.SkipUPNP, n.Logger.With("module", "p2p"))
|
2017-01-15 16:59:10 -08:00
|
|
|
n.sw.AddListener(l)
|
|
|
|
|
2017-12-28 22:53:41 -08:00
|
|
|
// Generate node PrivKey
|
2018-01-13 13:14:28 -08:00
|
|
|
// TODO: pass in like priv_val
|
2018-01-20 21:33:53 -08:00
|
|
|
nodeKey, err := p2p.LoadOrGenNodeKey(n.config.NodeKeyFile())
|
2018-01-01 17:21:42 -08:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
n.Logger.Info("P2P Node ID", "ID", nodeKey.ID(), "file", n.config.NodeKeyFile())
|
2017-12-28 22:53:41 -08:00
|
|
|
|
2017-01-15 16:59:10 -08:00
|
|
|
// Start the switch
|
2018-01-01 17:21:42 -08:00
|
|
|
n.sw.SetNodeInfo(n.makeNodeInfo(nodeKey.PubKey()))
|
|
|
|
n.sw.SetNodeKey(nodeKey)
|
2017-11-06 10:20:39 -08:00
|
|
|
err = n.sw.Start()
|
2017-01-15 16:59:10 -08:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2018-01-09 14:18:05 -08:00
|
|
|
// Always connect to persistent peers
|
|
|
|
if n.config.P2P.PersistentPeers != "" {
|
|
|
|
err = n.sw.DialPeersAsync(n.addrBook, strings.Split(n.config.P2P.PersistentPeers, ","), true)
|
2017-12-28 10:54:39 -08:00
|
|
|
if err != nil {
|
2017-03-04 20:25:12 -08:00
|
|
|
return err
|
|
|
|
}
|
2017-01-15 16:59:10 -08:00
|
|
|
}
|
|
|
|
|
2017-11-28 19:24:37 -08:00
|
|
|
// start tx indexer
|
2017-12-10 09:44:22 -08:00
|
|
|
return n.indexerService.Start()
|
2014-07-09 18:33:44 -07:00
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// OnStop stops the Node. It implements cmn.Service.
|
2017-01-15 16:59:10 -08:00
|
|
|
func (n *Node) OnStop() {
|
|
|
|
n.BaseService.OnStop()
|
|
|
|
|
2017-05-02 00:53:32 -07:00
|
|
|
n.Logger.Info("Stopping Node")
|
2014-07-16 21:13:02 -07:00
|
|
|
// TODO: gracefully disconnect from peers.
|
|
|
|
n.sw.Stop()
|
2017-03-03 16:28:17 -08:00
|
|
|
|
|
|
|
for _, l := range n.rpcListeners {
|
2017-05-02 00:53:32 -07:00
|
|
|
n.Logger.Info("Closing rpc listener", "listener", l)
|
2017-03-03 16:28:17 -08:00
|
|
|
if err := l.Close(); err != nil {
|
2017-06-14 01:50:49 -07:00
|
|
|
n.Logger.Error("Error closing listener", "listener", l, "err", err)
|
2017-03-03 16:28:17 -08:00
|
|
|
}
|
|
|
|
}
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
|
|
|
|
n.eventBus.Stop()
|
2017-11-28 19:24:37 -08:00
|
|
|
|
|
|
|
n.indexerService.Stop()
|
2014-07-16 21:13:02 -07:00
|
|
|
}
|
|
|
|
|
2017-10-03 16:11:55 -07:00
|
|
|
// RunForever waits for an interrupt signal and stops the node.
|
2017-03-05 20:13:34 -08:00
|
|
|
func (n *Node) RunForever() {
|
|
|
|
// Sleep forever and then...
|
|
|
|
cmn.TrapSignal(func() {
|
|
|
|
n.Stop()
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// AddListener adds a listener to accept inbound peer connections.
|
|
|
|
// It should be called before starting the Node.
|
2015-04-20 15:29:01 -07:00
|
|
|
// The first listener is the primary listener (in NodeInfo)
|
2014-07-10 22:14:23 -07:00
|
|
|
func (n *Node) AddListener(l p2p.Listener) {
|
2015-04-01 14:52:25 -07:00
|
|
|
n.sw.AddListener(l)
|
2014-07-09 18:33:44 -07:00
|
|
|
}
|
2014-07-08 00:02:04 -07:00
|
|
|
|
2017-02-22 05:41:10 -08:00
|
|
|
// ConfigureRPC sets all variables in rpccore so they will serve
|
|
|
|
// rpc calls from this node
|
|
|
|
func (n *Node) ConfigureRPC() {
|
2017-12-27 19:09:48 -08:00
|
|
|
rpccore.SetStateDB(n.stateDB)
|
2016-01-20 10:12:42 -08:00
|
|
|
rpccore.SetBlockStore(n.blockStore)
|
|
|
|
rpccore.SetConsensusState(n.consensusState)
|
2016-10-14 18:36:42 -07:00
|
|
|
rpccore.SetMempool(n.mempoolReactor.Mempool)
|
2017-11-18 18:02:58 -08:00
|
|
|
rpccore.SetEvidencePool(n.evidencePool)
|
2016-01-20 10:12:42 -08:00
|
|
|
rpccore.SetSwitch(n.sw)
|
2017-09-21 13:32:02 -07:00
|
|
|
rpccore.SetPubKey(n.privValidator.GetPubKey())
|
2016-01-20 10:12:42 -08:00
|
|
|
rpccore.SetGenesisDoc(n.genesisDoc)
|
2017-03-05 20:13:34 -08:00
|
|
|
rpccore.SetAddrBook(n.addrBook)
|
2016-08-22 13:00:48 -07:00
|
|
|
rpccore.SetProxyAppQuery(n.proxyApp.Query())
|
2017-02-22 02:53:59 -08:00
|
|
|
rpccore.SetTxIndexer(n.txIndexer)
|
2017-07-16 23:44:23 -07:00
|
|
|
rpccore.SetConsensusReactor(n.consensusReactor)
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
rpccore.SetEventBus(n.eventBus)
|
2017-05-02 00:53:32 -07:00
|
|
|
rpccore.SetLogger(n.Logger.With("module", "rpc"))
|
2017-02-22 05:41:10 -08:00
|
|
|
}
|
2015-04-10 02:12:17 -07:00
|
|
|
|
2017-02-22 05:41:10 -08:00
|
|
|
func (n *Node) startRPC() ([]net.Listener, error) {
|
|
|
|
n.ConfigureRPC()
|
2017-05-24 10:56:12 -07:00
|
|
|
listenAddrs := strings.Split(n.config.RPC.ListenAddress, ",")
|
2016-02-18 16:21:02 -08:00
|
|
|
|
2017-05-26 11:20:23 -07:00
|
|
|
if n.config.RPC.Unsafe {
|
|
|
|
rpccore.AddUnsafeRoutes()
|
|
|
|
}
|
|
|
|
|
2016-02-18 16:21:02 -08:00
|
|
|
// we may expose the rpc over both a unix and tcp socket
|
|
|
|
listeners := make([]net.Listener, len(listenAddrs))
|
|
|
|
for i, listenAddr := range listenAddrs {
|
|
|
|
mux := http.NewServeMux()
|
2017-05-15 10:33:45 -07:00
|
|
|
rpcLogger := n.Logger.With("module", "rpc-server")
|
2017-12-09 21:40:51 -08:00
|
|
|
wm := rpcserver.NewWebsocketManager(rpccore.Routes, rpcserver.EventSubscriber(n.eventBus))
|
2017-08-10 14:39:38 -07:00
|
|
|
wm.SetLogger(rpcLogger.With("protocol", "websocket"))
|
2016-02-18 16:21:02 -08:00
|
|
|
mux.HandleFunc("/websocket", wm.WebsocketHandler)
|
2017-05-02 00:53:32 -07:00
|
|
|
rpcserver.RegisterRPCFuncs(mux, rpccore.Routes, rpcLogger)
|
|
|
|
listener, err := rpcserver.StartHTTPServer(listenAddr, mux, rpcLogger)
|
2016-02-18 16:21:02 -08:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
listeners[i] = listener
|
|
|
|
}
|
2016-06-21 10:19:49 -07:00
|
|
|
|
|
|
|
// we expose a simplified api over grpc for convenience to app devs
|
2017-05-24 10:56:12 -07:00
|
|
|
grpcListenAddr := n.config.RPC.GRPCListenAddress
|
2016-06-21 10:19:49 -07:00
|
|
|
if grpcListenAddr != "" {
|
|
|
|
listener, err := grpccore.StartGRPCServer(grpcListenAddr)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
listeners = append(listeners, listener)
|
|
|
|
}
|
|
|
|
|
2016-02-18 16:21:02 -08:00
|
|
|
return listeners, nil
|
2015-03-19 02:08:07 -07:00
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// Switch returns the Node's Switch.
|
2015-03-25 18:06:57 -07:00
|
|
|
func (n *Node) Switch() *p2p.Switch {
|
|
|
|
return n.sw
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// BlockStore returns the Node's BlockStore.
|
2015-05-15 07:27:22 -07:00
|
|
|
func (n *Node) BlockStore() *bc.BlockStore {
|
|
|
|
return n.blockStore
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// ConsensusState returns the Node's ConsensusState.
|
2018-01-21 08:45:04 -08:00
|
|
|
func (n *Node) ConsensusState() *cs.ConsensusState {
|
2015-03-19 02:08:07 -07:00
|
|
|
return n.consensusState
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// ConsensusReactor returns the Node's ConsensusReactor.
|
2018-01-21 08:45:04 -08:00
|
|
|
func (n *Node) ConsensusReactor() *cs.ConsensusReactor {
|
2016-05-11 11:45:20 -07:00
|
|
|
return n.consensusReactor
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// MempoolReactor returns the Node's MempoolReactor.
|
2015-03-19 02:08:07 -07:00
|
|
|
func (n *Node) MempoolReactor() *mempl.MempoolReactor {
|
|
|
|
return n.mempoolReactor
|
|
|
|
}
|
|
|
|
|
2017-11-18 18:02:58 -08:00
|
|
|
// EvidencePool returns the Node's EvidencePool.
|
|
|
|
func (n *Node) EvidencePool() *evidence.EvidencePool {
|
|
|
|
return n.evidencePool
|
|
|
|
}
|
|
|
|
|
new pubsub package
comment out failing consensus tests for now
rewrite rpc httpclient to use new pubsub package
import pubsub as tmpubsub, query as tmquery
make event IDs constants
EventKey -> EventTypeKey
rename EventsPubsub to PubSub
mempool does not use pubsub
rename eventsSub to pubsub
new subscribe API
fix channel size issues and consensus tests bugs
refactor rpc client
add missing discardFromChan method
add mutex
rename pubsub to eventBus
remove IsRunning from WSRPCConnection interface (not needed)
add a comment in broadcastNewRoundStepsAndVotes
rename registerEventCallbacks to broadcastNewRoundStepsAndVotes
See https://dave.cheney.net/2014/03/19/channel-axioms
stop eventBuses after reactor tests
remove unnecessary Unsubscribe
return subscribe helper function
move discardFromChan to where it is used
subscribe now returns an err
this gives us ability to refuse to subscribe if pubsub is at its max
capacity.
use context for control overflow
cache queries
handle err when subscribing in replay_test
rename testClientID to testSubscriber
extract var
set channel buffer capacity to 1 in replay_file
fix byzantine_test
unsubscribe from single event, not all events
refactor httpclient to return events to appropriate channels
return failing testReplayCrashBeforeWriteVote test
fix TestValidatorSetChanges
refactor code a bit
fix testReplayCrashBeforeWriteVote
add comment
fix TestValidatorSetChanges
fixes from Bucky's review
update comment [ci skip]
test TxEventBuffer
update changelog
fix TestValidatorSetChanges (2nd attempt)
only do wg.Done when no errors
benchmark event bus
create pubsub server inside NewEventBus
only expose config params (later if needed)
set buffer capacity to 0 so we are not testing cache
new tx event format: key = "Tx" plus a tag {"tx.hash": XYZ}
This should allow to subscribe to all transactions! or a specific one
using a query: "tm.events.type = Tx and tx.hash = '013ABF99434...'"
use TimeoutCommit instead of afterPublishEventNewBlockTimeout
TimeoutCommit is the time a node waits after committing a block, before
it goes into the next height. So it will finish everything from the last
block, but then wait a bit. The idea is this gives it time to hear more
votes from other validators, to strengthen the commit it includes in the
next block. But it also gives it time to hear about new transactions.
waitForBlockWithUpdatedVals
rewrite WAL crash tests
Task:
test that we can recover from any WAL crash.
Solution:
the old tests were relying on event hub being run in the same thread (we
were injecting the private validator's last signature).
when considering a rewrite, we considered two possible solutions: write
a "fuzzy" testing system where WAL is crashing upon receiving a new
message, or inject failures and trigger them in tests using something
like https://github.com/coreos/gofail.
remove sleep
no cs.Lock around wal.Save
test different cases (empty block, non-empty block, ...)
comments
add comments
test 4 cases: empty block, non-empty block, non-empty block with smaller part size, many blocks
fixes as per Bucky's last review
reset subscriptions on UnsubscribeAll
use a simple counter to track message for which we panicked
also, set a smaller part size for all test cases
2017-06-26 08:00:30 -07:00
|
|
|
// EventBus returns the Node's EventBus.
|
|
|
|
func (n *Node) EventBus() *types.EventBus {
|
|
|
|
return n.eventBus
|
2015-04-17 13:18:50 -07:00
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// PrivValidator returns the Node's PrivValidator.
|
|
|
|
// XXX: for convenience only!
|
2017-09-18 15:12:31 -07:00
|
|
|
func (n *Node) PrivValidator() types.PrivValidator {
|
2016-05-11 11:45:20 -07:00
|
|
|
return n.privValidator
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// GenesisDoc returns the Node's GenesisDoc.
|
2016-10-14 18:36:42 -07:00
|
|
|
func (n *Node) GenesisDoc() *types.GenesisDoc {
|
|
|
|
return n.genesisDoc
|
|
|
|
}
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// ProxyApp returns the Node's AppConns, representing its connections to the ABCI application.
|
2016-10-14 18:36:42 -07:00
|
|
|
func (n *Node) ProxyApp() proxy.AppConns {
|
|
|
|
return n.proxyApp
|
|
|
|
}
|
|
|
|
|
2018-01-13 21:10:29 -08:00
|
|
|
func (n *Node) makeNodeInfo(pubKey crypto.PubKey) p2p.NodeInfo {
|
2017-04-18 17:10:02 -07:00
|
|
|
txIndexerStatus := "on"
|
|
|
|
if _, ok := n.txIndexer.(*null.TxIndex); ok {
|
|
|
|
txIndexerStatus = "off"
|
|
|
|
}
|
2018-01-13 21:10:29 -08:00
|
|
|
nodeInfo := p2p.NodeInfo{
|
2017-12-28 22:53:41 -08:00
|
|
|
PubKey: pubKey,
|
2017-10-05 05:50:05 -07:00
|
|
|
Network: n.genesisDoc.ChainID,
|
2016-01-20 10:12:42 -08:00
|
|
|
Version: version.Version,
|
2018-01-21 08:45:04 -08:00
|
|
|
Channels: []byte{
|
|
|
|
bc.BlockchainChannel,
|
|
|
|
cs.StateChannel, cs.DataChannel, cs.VoteChannel, cs.VoteSetBitsChannel,
|
|
|
|
mempl.MempoolChannel,
|
|
|
|
evidence.EvidenceChannel,
|
|
|
|
},
|
2018-01-13 20:48:41 -08:00
|
|
|
Moniker: n.config.Moniker,
|
2015-11-01 11:34:08 -08:00
|
|
|
Other: []string{
|
2017-01-15 16:59:10 -08:00
|
|
|
cmn.Fmt("wire_version=%v", wire.Version),
|
|
|
|
cmn.Fmt("p2p_version=%v", p2p.Version),
|
2018-01-21 08:45:04 -08:00
|
|
|
cmn.Fmt("consensus_version=%v", cs.Version),
|
2017-01-15 16:59:10 -08:00
|
|
|
cmn.Fmt("rpc_version=%v/%v", rpc.Version, rpccore.Version),
|
2017-04-18 17:55:40 -07:00
|
|
|
cmn.Fmt("tx_index=%v", txIndexerStatus),
|
2015-09-15 16:11:45 -07:00
|
|
|
},
|
2015-04-20 15:29:01 -07:00
|
|
|
}
|
2015-07-10 08:50:58 -07:00
|
|
|
|
2018-01-21 08:45:04 -08:00
|
|
|
if n.config.P2P.PexReactor {
|
|
|
|
nodeInfo.Channels = append(nodeInfo.Channels, pex.PexChannel)
|
|
|
|
}
|
|
|
|
|
2017-11-16 17:20:15 -08:00
|
|
|
rpcListenAddr := n.config.RPC.ListenAddress
|
|
|
|
nodeInfo.Other = append(nodeInfo.Other, cmn.Fmt("rpc_addr=%v", rpcListenAddr))
|
2015-07-10 08:50:58 -07:00
|
|
|
|
2017-04-18 17:10:02 -07:00
|
|
|
if !n.sw.IsListening() {
|
2015-04-20 15:29:01 -07:00
|
|
|
return nodeInfo
|
|
|
|
}
|
2015-07-10 08:50:58 -07:00
|
|
|
|
2017-04-18 17:10:02 -07:00
|
|
|
p2pListener := n.sw.Listeners()[0]
|
2015-04-20 15:29:01 -07:00
|
|
|
p2pHost := p2pListener.ExternalAddress().IP.String()
|
|
|
|
p2pPort := p2pListener.ExternalAddress().Port
|
2017-01-15 16:59:10 -08:00
|
|
|
nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
|
2017-11-16 17:20:15 -08:00
|
|
|
|
2015-04-20 15:29:01 -07:00
|
|
|
return nodeInfo
|
|
|
|
}
|
|
|
|
|
2017-01-13 12:47:13 -08:00
|
|
|
//------------------------------------------------------------------------------
|
|
|
|
|
2017-09-20 15:29:36 -07:00
|
|
|
// NodeInfo returns the Node's Info from the Switch.
|
2018-01-13 21:10:29 -08:00
|
|
|
func (n *Node) NodeInfo() p2p.NodeInfo {
|
2016-04-13 15:23:25 -07:00
|
|
|
return n.sw.NodeInfo()
|
|
|
|
}
|
|
|
|
|
2017-04-28 20:59:02 -07:00
|
|
|
//------------------------------------------------------------------------------
|
2017-10-13 02:11:16 -07:00
|
|
|
|
|
|
|
var (
|
|
|
|
genesisDocKey = []byte("genesisDoc")
|
|
|
|
)
|
|
|
|
|
2017-10-17 02:33:57 -07:00
|
|
|
// panics if failed to unmarshal bytes
|
2017-10-13 02:11:16 -07:00
|
|
|
func loadGenesisDoc(db dbm.DB) (*types.GenesisDoc, error) {
|
|
|
|
bytes := db.Get(genesisDocKey)
|
|
|
|
if len(bytes) == 0 {
|
|
|
|
return nil, errors.New("Genesis doc not found")
|
|
|
|
} else {
|
|
|
|
var genDoc *types.GenesisDoc
|
|
|
|
err := json.Unmarshal(bytes, &genDoc)
|
2017-10-17 02:33:57 -07:00
|
|
|
if err != nil {
|
|
|
|
cmn.PanicCrisis(fmt.Sprintf("Failed to load genesis doc due to unmarshaling error: %v (bytes: %X)", err, bytes))
|
|
|
|
}
|
|
|
|
return genDoc, nil
|
2017-10-13 02:11:16 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-10-17 02:33:57 -07:00
|
|
|
// panics if failed to marshal the given genesis document
|
|
|
|
func saveGenesisDoc(db dbm.DB, genDoc *types.GenesisDoc) {
|
2017-10-13 02:11:16 -07:00
|
|
|
bytes, err := json.Marshal(genDoc)
|
|
|
|
if err != nil {
|
2017-10-17 02:33:57 -07:00
|
|
|
cmn.PanicCrisis(fmt.Sprintf("Failed to save genesis doc due to marshaling error: %v", err))
|
2017-10-13 02:11:16 -07:00
|
|
|
}
|
|
|
|
db.SetSync(genesisDocKey, bytes)
|
|
|
|
}
|