diff --git a/config/config.go b/config/config.go index 7d19616aa..6350449d7 100644 --- a/config/config.go +++ b/config/config.go @@ -13,7 +13,7 @@ import ( tmjson "github.com/tendermint/tendermint/libs/json" "github.com/tendermint/tendermint/libs/log" tmos "github.com/tendermint/tendermint/libs/os" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) const ( @@ -288,23 +288,23 @@ func (cfg BaseConfig) NodeKeyFile() string { } // LoadNodeKey loads NodeKey located in filePath. -func (cfg BaseConfig) LoadNodeKeyID() (types.NodeID, error) { +func (cfg BaseConfig) LoadNodeKeyID() (p2p.NodeID, error) { jsonBytes, err := ioutil.ReadFile(cfg.NodeKeyFile()) if err != nil { return "", err } - nodeKey := types.NodeKey{} + nodeKey := p2p.NodeKey{} err = tmjson.Unmarshal(jsonBytes, &nodeKey) if err != nil { return "", err } - nodeKey.ID = types.NodeIDFromPubKey(nodeKey.PubKey()) + nodeKey.ID = p2p.NodeIDFromPubKey(nodeKey.PubKey()) return nodeKey.ID, nil } // LoadOrGenNodeKey attempts to load the NodeKey from the given filePath. If // the file does not exist, it generates and saves a new NodeKey. -func (cfg BaseConfig) LoadOrGenNodeKeyID() (types.NodeID, error) { +func (cfg BaseConfig) LoadOrGenNodeKeyID() (p2p.NodeID, error) { if tmos.FileExists(cfg.NodeKeyFile()) { nodeKey, err := cfg.LoadNodeKeyID() if err != nil { @@ -313,7 +313,7 @@ func (cfg BaseConfig) LoadOrGenNodeKeyID() (types.NodeID, error) { return nodeKey, nil } - nodeKey := types.GenNodeKey() + nodeKey := p2p.GenNodeKey() if err := nodeKey.SaveAs(cfg.NodeKeyFile()); err != nil { return "", err diff --git a/internal/blocksync/msgs.go b/internal/blocksync/msgs.go index caad44b7b..72696f44f 100644 --- a/internal/blocksync/msgs.go +++ b/internal/blocksync/msgs.go @@ -1,12 +1,12 @@ package blocksync import ( + "github.com/tendermint/tendermint/pkg/meta" bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync" - "github.com/tendermint/tendermint/types" ) const ( - MaxMsgSize = types.MaxBlockSizeBytes + + MaxMsgSize = meta.MaxBlockSizeBytes + bcproto.BlockResponseMessagePrefixSize + bcproto.BlockResponseMessageFieldKeySize ) diff --git a/internal/blocksync/v2/internal/behavior/peer_behaviour.go b/internal/blocksync/v2/internal/behavior/peer_behaviour.go index 90948d888..0f3b2d16b 100644 --- a/internal/blocksync/v2/internal/behavior/peer_behaviour.go +++ b/internal/blocksync/v2/internal/behavior/peer_behaviour.go @@ -1,12 +1,12 @@ package behavior -import "github.com/tendermint/tendermint/types" +import "github.com/tendermint/tendermint/internal/p2p" // PeerBehavior is a struct describing a behavior a peer performed. // `peerID` identifies the peer and reason characterizes the specific // behavior performed by the peer. type PeerBehavior struct { - peerID types.NodeID + peerID p2p.NodeID reason interface{} } @@ -15,7 +15,7 @@ type badMessage struct { } // BadMessage returns a badMessage PeerBehavior. -func BadMessage(peerID types.NodeID, explanation string) PeerBehavior { +func BadMessage(peerID p2p.NodeID, explanation string) PeerBehavior { return PeerBehavior{peerID: peerID, reason: badMessage{explanation}} } @@ -24,7 +24,7 @@ type messageOutOfOrder struct { } // MessageOutOfOrder returns a messagOutOfOrder PeerBehavior. -func MessageOutOfOrder(peerID types.NodeID, explanation string) PeerBehavior { +func MessageOutOfOrder(peerID p2p.NodeID, explanation string) PeerBehavior { return PeerBehavior{peerID: peerID, reason: messageOutOfOrder{explanation}} } @@ -33,7 +33,7 @@ type consensusVote struct { } // ConsensusVote returns a consensusVote PeerBehavior. -func ConsensusVote(peerID types.NodeID, explanation string) PeerBehavior { +func ConsensusVote(peerID p2p.NodeID, explanation string) PeerBehavior { return PeerBehavior{peerID: peerID, reason: consensusVote{explanation}} } @@ -42,6 +42,6 @@ type blockPart struct { } // BlockPart returns blockPart PeerBehavior. -func BlockPart(peerID types.NodeID, explanation string) PeerBehavior { +func BlockPart(peerID p2p.NodeID, explanation string) PeerBehavior { return PeerBehavior{peerID: peerID, reason: blockPart{explanation}} } diff --git a/internal/blocksync/v2/internal/behavior/reporter.go b/internal/blocksync/v2/internal/behavior/reporter.go index c150a98d5..a4455ad51 100644 --- a/internal/blocksync/v2/internal/behavior/reporter.go +++ b/internal/blocksync/v2/internal/behavior/reporter.go @@ -5,7 +5,6 @@ import ( tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/internal/p2p" - "github.com/tendermint/tendermint/types" ) // Reporter provides an interface for reactors to report the behavior @@ -52,14 +51,14 @@ func (spbr *SwitchReporter) Report(behavior PeerBehavior) error { // behavior in manufactured scenarios. type MockReporter struct { mtx tmsync.RWMutex - pb map[types.NodeID][]PeerBehavior + pb map[p2p.NodeID][]PeerBehavior } // NewMockReporter returns a Reporter which records all reported // behaviors in memory. func NewMockReporter() *MockReporter { return &MockReporter{ - pb: map[types.NodeID][]PeerBehavior{}, + pb: map[p2p.NodeID][]PeerBehavior{}, } } @@ -73,7 +72,7 @@ func (mpbr *MockReporter) Report(behavior PeerBehavior) error { } // GetBehaviors returns all behaviors reported on the peer identified by peerID. -func (mpbr *MockReporter) GetBehaviors(peerID types.NodeID) []PeerBehavior { +func (mpbr *MockReporter) GetBehaviors(peerID p2p.NodeID) []PeerBehavior { mpbr.mtx.RLock() defer mpbr.mtx.RUnlock() if items, ok := mpbr.pb[peerID]; ok { diff --git a/internal/consensus/types/height_vote_set.go b/internal/consensus/types/height_vote_set.go index 86b3e2c4f..3f08e8acb 100644 --- a/internal/consensus/types/height_vote_set.go +++ b/internal/consensus/types/height_vote_set.go @@ -8,13 +8,16 @@ import ( tmjson "github.com/tendermint/tendermint/libs/json" tmmath "github.com/tendermint/tendermint/libs/math" + "github.com/tendermint/tendermint/pkg/consensus" + "github.com/tendermint/tendermint/pkg/meta" + "github.com/tendermint/tendermint/pkg/p2p" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" "github.com/tendermint/tendermint/types" ) type RoundVoteSet struct { - Prevotes *types.VoteSet - Precommits *types.VoteSet + Prevotes *consensus.VoteSet + Precommits *consensus.VoteSet } var ( @@ -40,15 +43,15 @@ One for their LastCommit round, and another for the official commit round. type HeightVoteSet struct { chainID string height int64 - valSet *types.ValidatorSet + valSet *consensus.ValidatorSet mtx sync.Mutex - round int32 // max tracked round - roundVoteSets map[int32]RoundVoteSet // keys: [0...round] - peerCatchupRounds map[types.NodeID][]int32 // keys: peer.ID; values: at most 2 rounds + round int32 // max tracked round + roundVoteSets map[int32]RoundVoteSet // keys: [0...round] + peerCatchupRounds map[p2p.NodeID][]int32 // keys: peer.ID; values: at most 2 rounds } -func NewHeightVoteSet(chainID string, height int64, valSet *types.ValidatorSet) *HeightVoteSet { +func NewHeightVoteSet(chainID string, height int64, valSet *consensus.ValidatorSet) *HeightVoteSet { hvs := &HeightVoteSet{ chainID: chainID, } @@ -56,7 +59,7 @@ func NewHeightVoteSet(chainID string, height int64, valSet *types.ValidatorSet) return hvs } -func (hvs *HeightVoteSet) Reset(height int64, valSet *types.ValidatorSet) { +func (hvs *HeightVoteSet) Reset(height int64, valSet *consensus.ValidatorSet) { hvs.mtx.Lock() defer hvs.mtx.Unlock() @@ -103,8 +106,8 @@ func (hvs *HeightVoteSet) addRound(round int32) { panic("addRound() for an existing round") } // log.Debug("addRound(round)", "round", round) - prevotes := types.NewVoteSet(hvs.chainID, hvs.height, round, tmproto.PrevoteType, hvs.valSet) - precommits := types.NewVoteSet(hvs.chainID, hvs.height, round, tmproto.PrecommitType, hvs.valSet) + prevotes := consensus.NewVoteSet(hvs.chainID, hvs.height, round, tmproto.PrevoteType, hvs.valSet) + precommits := consensus.NewVoteSet(hvs.chainID, hvs.height, round, tmproto.PrecommitType, hvs.valSet) hvs.roundVoteSets[round] = RoundVoteSet{ Prevotes: prevotes, Precommits: precommits, @@ -113,10 +116,10 @@ func (hvs *HeightVoteSet) addRound(round int32) { // Duplicate votes return added=false, err=nil. // By convention, peerID is "" if origin is self. -func (hvs *HeightVoteSet) AddVote(vote *types.Vote, peerID types.NodeID) (added bool, err error) { +func (hvs *HeightVoteSet) AddVote(vote *consensus.Vote, peerID p2p.NodeID) (added bool, err error) { hvs.mtx.Lock() defer hvs.mtx.Unlock() - if !types.IsVoteTypeValid(vote.Type) { + if !consensus.IsVoteTypeValid(vote.Type) { return } voteSet := hvs.getVoteSet(vote.Round, vote.Type) @@ -135,13 +138,13 @@ func (hvs *HeightVoteSet) AddVote(vote *types.Vote, peerID types.NodeID) (added return } -func (hvs *HeightVoteSet) Prevotes(round int32) *types.VoteSet { +func (hvs *HeightVoteSet) Prevotes(round int32) *consensus.VoteSet { hvs.mtx.Lock() defer hvs.mtx.Unlock() return hvs.getVoteSet(round, tmproto.PrevoteType) } -func (hvs *HeightVoteSet) Precommits(round int32) *types.VoteSet { +func (hvs *HeightVoteSet) Precommits(round int32) *consensus.VoteSet { hvs.mtx.Lock() defer hvs.mtx.Unlock() return hvs.getVoteSet(round, tmproto.PrecommitType) @@ -149,7 +152,7 @@ func (hvs *HeightVoteSet) Precommits(round int32) *types.VoteSet { // Last round and blockID that has +2/3 prevotes for a particular block or nil. // Returns -1 if no such round exists. -func (hvs *HeightVoteSet) POLInfo() (polRound int32, polBlockID types.BlockID) { +func (hvs *HeightVoteSet) POLInfo() (polRound int32, polBlockID meta.BlockID) { hvs.mtx.Lock() defer hvs.mtx.Unlock() for r := hvs.round; r >= 0; r-- { @@ -162,7 +165,7 @@ func (hvs *HeightVoteSet) POLInfo() (polRound int32, polBlockID types.BlockID) { return -1, types.BlockID{} } -func (hvs *HeightVoteSet) getVoteSet(round int32, voteType tmproto.SignedMsgType) *types.VoteSet { +func (hvs *HeightVoteSet) getVoteSet(round int32, voteType tmproto.SignedMsgType) *consensus.VoteSet { rvs, ok := hvs.roundVoteSets[round] if !ok { return nil diff --git a/internal/mempool/cache.go b/internal/mempool/cache.go index 43174f106..33722f003 100644 --- a/internal/mempool/cache.go +++ b/internal/mempool/cache.go @@ -4,7 +4,7 @@ import ( "container/list" tmsync "github.com/tendermint/tendermint/internal/libs/sync" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/mempool" ) // TxCache defines an interface for raw transaction caching in a mempool. @@ -18,10 +18,10 @@ type TxCache interface { // Push adds the given raw transaction to the cache and returns true if it was // newly added. Otherwise, it returns false. - Push(tx types.Tx) bool + Push(tx mempool.Tx) bool // Remove removes the given raw transaction from the cache. - Remove(tx types.Tx) + Remove(tx mempool.Tx) } var _ TxCache = (*LRUTxCache)(nil) @@ -57,7 +57,7 @@ func (c *LRUTxCache) Reset() { c.list.Init() } -func (c *LRUTxCache) Push(tx types.Tx) bool { +func (c *LRUTxCache) Push(tx mempool.Tx) bool { c.mtx.Lock() defer c.mtx.Unlock() @@ -84,7 +84,7 @@ func (c *LRUTxCache) Push(tx types.Tx) bool { return true } -func (c *LRUTxCache) Remove(tx types.Tx) { +func (c *LRUTxCache) Remove(tx mempool.Tx) { c.mtx.Lock() defer c.mtx.Unlock() @@ -102,6 +102,6 @@ type NopTxCache struct{} var _ TxCache = (*NopTxCache)(nil) -func (NopTxCache) Reset() {} -func (NopTxCache) Push(types.Tx) bool { return true } -func (NopTxCache) Remove(types.Tx) {} +func (NopTxCache) Reset() {} +func (NopTxCache) Push(mempool.Tx) bool { return true } +func (NopTxCache) Remove(mempool.Tx) {} diff --git a/internal/mempool/ids.go b/internal/mempool/ids.go index 49a9ac607..fbb80da24 100644 --- a/internal/mempool/ids.go +++ b/internal/mempool/ids.go @@ -4,21 +4,21 @@ import ( "fmt" tmsync "github.com/tendermint/tendermint/internal/libs/sync" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) // nolint: golint // TODO: Rename type. type MempoolIDs struct { mtx tmsync.RWMutex - peerMap map[types.NodeID]uint16 + peerMap map[p2p.NodeID]uint16 nextID uint16 // assumes that a node will never have over 65536 active peers activeIDs map[uint16]struct{} // used to check if a given peerID key is used } func NewMempoolIDs() *MempoolIDs { return &MempoolIDs{ - peerMap: make(map[types.NodeID]uint16), + peerMap: make(map[p2p.NodeID]uint16), // reserve UnknownPeerID for mempoolReactor.BroadcastTx activeIDs: map[uint16]struct{}{UnknownPeerID: {}}, @@ -28,7 +28,7 @@ func NewMempoolIDs() *MempoolIDs { // ReserveForPeer searches for the next unused ID and assigns it to the provided // peer. -func (ids *MempoolIDs) ReserveForPeer(peerID types.NodeID) { +func (ids *MempoolIDs) ReserveForPeer(peerID p2p.NodeID) { ids.mtx.Lock() defer ids.mtx.Unlock() @@ -38,7 +38,7 @@ func (ids *MempoolIDs) ReserveForPeer(peerID types.NodeID) { } // Reclaim returns the ID reserved for the peer back to unused pool. -func (ids *MempoolIDs) Reclaim(peerID types.NodeID) { +func (ids *MempoolIDs) Reclaim(peerID p2p.NodeID) { ids.mtx.Lock() defer ids.mtx.Unlock() @@ -50,7 +50,7 @@ func (ids *MempoolIDs) Reclaim(peerID types.NodeID) { } // GetForPeer returns an ID reserved for the peer. -func (ids *MempoolIDs) GetForPeer(peerID types.NodeID) uint16 { +func (ids *MempoolIDs) GetForPeer(peerID p2p.NodeID) uint16 { ids.mtx.RLock() defer ids.mtx.RUnlock() diff --git a/internal/mempool/mempool.go b/internal/mempool/mempool.go index d679b3506..998a78c7b 100644 --- a/internal/mempool/mempool.go +++ b/internal/mempool/mempool.go @@ -7,7 +7,8 @@ import ( abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/internal/p2p" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/block" + "github.com/tendermint/tendermint/pkg/mempool" ) const ( @@ -30,7 +31,7 @@ const ( type Mempool interface { // CheckTx executes a new transaction against the application to determine // its validity and whether it should be added to the mempool. - CheckTx(ctx context.Context, tx types.Tx, callback func(*abci.Response), txInfo TxInfo) error + CheckTx(ctx context.Context, tx mempool.Tx, callback func(*abci.Response), txInfo TxInfo) error // ReapMaxBytesMaxGas reaps transactions from the mempool up to maxBytes // bytes total with the condition that the total gasWanted must be less than @@ -38,12 +39,12 @@ type Mempool interface { // // If both maxes are negative, there is no cap on the size of all returned // transactions (~ all available transactions). - ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs + ReapMaxBytesMaxGas(maxBytes, maxGas int64) mempool.Txs // ReapMaxTxs reaps up to max transactions from the mempool. If max is // negative, there is no cap on the size of all returned transactions // (~ all available transactions). - ReapMaxTxs(max int) types.Txs + ReapMaxTxs(max int) mempool.Txs // Lock locks the mempool. The consensus must be able to hold lock to safely // update. @@ -60,7 +61,7 @@ type Mempool interface { // 2. Lock/Unlock must be managed by the caller. Update( blockHeight int64, - blockTxs types.Txs, + blockTxs mempool.Txs, deliverTxResponses []*abci.ResponseDeliverTx, newPreFn PreCheckFunc, newPostFn PostCheckFunc, @@ -97,18 +98,18 @@ type Mempool interface { // PreCheckFunc is an optional filter executed before CheckTx and rejects // transaction if false is returned. An example would be to ensure that a // transaction doesn't exceeded the block size. -type PreCheckFunc func(types.Tx) error +type PreCheckFunc func(mempool.Tx) error // PostCheckFunc is an optional filter executed after CheckTx and rejects // transaction if false is returned. An example would be to ensure a // transaction doesn't require more gas than available for the block. -type PostCheckFunc func(types.Tx, *abci.ResponseCheckTx) error +type PostCheckFunc func(mempool.Tx, *abci.ResponseCheckTx) error // PreCheckMaxBytes checks that the size of the transaction is smaller or equal // to the expected maxBytes. func PreCheckMaxBytes(maxBytes int64) PreCheckFunc { - return func(tx types.Tx) error { - txSize := types.ComputeProtoSizeForTxs([]types.Tx{tx}) + return func(tx mempool.Tx) error { + txSize := block.ComputeProtoSizeForTxs([]mempool.Tx{tx}) if txSize > maxBytes { return fmt.Errorf("tx size is too big: %d, max: %d", txSize, maxBytes) @@ -121,7 +122,7 @@ func PreCheckMaxBytes(maxBytes int64) PreCheckFunc { // PostCheckMaxGas checks that the wanted gas is smaller or equal to the passed // maxGas. Returns nil if maxGas is -1. func PostCheckMaxGas(maxGas int64) PostCheckFunc { - return func(tx types.Tx, res *abci.ResponseCheckTx) error { + return func(tx mempool.Tx, res *abci.ResponseCheckTx) error { if maxGas == -1 { return nil } diff --git a/internal/mempool/tx.go b/internal/mempool/tx.go index 860d3d3b4..7aebd32bf 100644 --- a/internal/mempool/tx.go +++ b/internal/mempool/tx.go @@ -3,20 +3,21 @@ package mempool import ( "crypto/sha256" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/mempool" + "github.com/tendermint/tendermint/pkg/p2p" ) // TxKeySize defines the size of the transaction's key used for indexing. const TxKeySize = sha256.Size // TxKey is the fixed length array key used as an index. -func TxKey(tx types.Tx) [TxKeySize]byte { +func TxKey(tx mempool.Tx) [TxKeySize]byte { return sha256.Sum256(tx) } // TxHashFromBytes returns the hash of a transaction from raw bytes. func TxHashFromBytes(tx []byte) []byte { - return types.Tx(tx).Hash() + return mempool.Tx(tx).Hash() } // TxInfo are parameters that get passed when attempting to add a tx to the @@ -28,5 +29,5 @@ type TxInfo struct { SenderID uint16 // SenderNodeID is the actual types.NodeID of the sender. - SenderNodeID types.NodeID + SenderNodeID p2p.NodeID } diff --git a/internal/mempool/v0/clist_mempool.go b/internal/mempool/v0/clist_mempool.go index 40e93cc13..95e6fb224 100644 --- a/internal/mempool/v0/clist_mempool.go +++ b/internal/mempool/v0/clist_mempool.go @@ -14,9 +14,11 @@ import ( "github.com/tendermint/tendermint/internal/mempool" "github.com/tendermint/tendermint/libs/log" tmmath "github.com/tendermint/tendermint/libs/math" + "github.com/tendermint/tendermint/pkg/block" + pkgmempool "github.com/tendermint/tendermint/pkg/mempool" pubmempool "github.com/tendermint/tendermint/pkg/mempool" + "github.com/tendermint/tendermint/pkg/p2p" "github.com/tendermint/tendermint/proxy" - "github.com/tendermint/tendermint/types" ) // CListMempool is an ordered in-memory pool for transactions before they are @@ -201,7 +203,7 @@ func (mem *CListMempool) TxsWaitChan() <-chan struct{} { // Safe for concurrent use by multiple goroutines. func (mem *CListMempool) CheckTx( ctx context.Context, - tx types.Tx, + tx pkgmempool.Tx, cb func(*abci.Response), txInfo mempool.TxInfo, ) error { @@ -303,7 +305,7 @@ func (mem *CListMempool) globalCb(req *abci.Request, res *abci.Response) { func (mem *CListMempool) reqResCb( tx []byte, peerID uint16, - peerP2PID types.NodeID, + peerP2PID p2p.NodeID, externalCb func(*abci.Response), ) func(res *abci.Response) { return func(res *abci.Response) { @@ -336,7 +338,7 @@ func (mem *CListMempool) addTx(memTx *mempoolTx) { // Called from: // - Update (lock held) if tx was committed // - resCbRecheck (lock not held) if tx was invalidated -func (mem *CListMempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromCache bool) { +func (mem *CListMempool) removeTx(tx pkgmempool.Tx, elem *clist.CElement, removeFromCache bool) { mem.txs.Remove(elem) elem.DetachPrev() mem.txsMap.Delete(mempool.TxKey(tx)) @@ -382,7 +384,7 @@ func (mem *CListMempool) isFull(txSize int) error { func (mem *CListMempool) resCbFirstTime( tx []byte, peerID uint16, - peerP2PID types.NodeID, + peerP2PID p2p.NodeID, res *abci.Response, ) { switch r := res.Value.(type) { @@ -504,7 +506,7 @@ func (mem *CListMempool) notifyTxsAvailable() { } // Safe for concurrent use by multiple goroutines. -func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { +func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) pkgmempool.Txs { mem.updateMtx.RLock() defer mem.updateMtx.RUnlock() @@ -515,14 +517,14 @@ func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { // TODO: we will get a performance boost if we have a good estimate of avg // size per tx, and set the initial capacity based off of that. - // txs := make([]types.Tx, 0, tmmath.MinInt(mem.txs.Len(), max/mem.avgTxSize)) - txs := make([]types.Tx, 0, mem.txs.Len()) + // txs := make([]pkgmempool.Tx, 0, tmmath.MinInt(mem.txs.Len(), max/mem.avgTxSize)) + txs := make([]pkgmempool.Tx, 0, mem.txs.Len()) for e := mem.txs.Front(); e != nil; e = e.Next() { memTx := e.Value.(*mempoolTx) txs = append(txs, memTx.tx) - dataSize := types.ComputeProtoSizeForTxs([]types.Tx{memTx.tx}) + dataSize := block.ComputeProtoSizeForTxs([]pkgmempool.Tx{memTx.tx}) // Check total size requirement if maxBytes > -1 && runningSize+dataSize > maxBytes { @@ -545,7 +547,7 @@ func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { } // Safe for concurrent use by multiple goroutines. -func (mem *CListMempool) ReapMaxTxs(max int) types.Txs { +func (mem *CListMempool) ReapMaxTxs(max int) pkgmempool.Txs { mem.updateMtx.RLock() defer mem.updateMtx.RUnlock() @@ -553,7 +555,7 @@ func (mem *CListMempool) ReapMaxTxs(max int) types.Txs { max = mem.txs.Len() } - txs := make([]types.Tx, 0, tmmath.MinInt(mem.txs.Len(), max)) + txs := make([]pkgmempool.Tx, 0, tmmath.MinInt(mem.txs.Len(), max)) for e := mem.txs.Front(); e != nil && len(txs) <= max; e = e.Next() { memTx := e.Value.(*mempoolTx) txs = append(txs, memTx.tx) @@ -564,7 +566,7 @@ func (mem *CListMempool) ReapMaxTxs(max int) types.Txs { // Lock() must be help by the caller during execution. func (mem *CListMempool) Update( height int64, - txs types.Txs, + txs pkgmempool.Txs, deliverTxResponses []*abci.ResponseDeliverTx, preCheck mempool.PreCheckFunc, postCheck mempool.PostCheckFunc, @@ -658,9 +660,9 @@ func (mem *CListMempool) recheckTxs() { // mempoolTx is a transaction that successfully ran type mempoolTx struct { - height int64 // height that this tx had been validated in - gasWanted int64 // amount of gas this tx states it will require - tx types.Tx // + height int64 // height that this tx had been validated in + gasWanted int64 // amount of gas this tx states it will require + tx pkgmempool.Tx // // ids of peers who've sent us this tx (as a map for quick lookups). // senders: PeerID -> bool diff --git a/internal/mempool/v0/reactor.go b/internal/mempool/v0/reactor.go index 29dec5833..fb320230b 100644 --- a/internal/mempool/v0/reactor.go +++ b/internal/mempool/v0/reactor.go @@ -15,8 +15,9 @@ import ( "github.com/tendermint/tendermint/internal/p2p" "github.com/tendermint/tendermint/libs/log" "github.com/tendermint/tendermint/libs/service" + pkgmempool "github.com/tendermint/tendermint/pkg/mempool" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" protomem "github.com/tendermint/tendermint/proto/tendermint/mempool" - "github.com/tendermint/tendermint/types" ) var ( @@ -28,7 +29,7 @@ var ( // peer information. This should eventually be replaced with a message-oriented // approach utilizing the p2p stack. type PeerManager interface { - GetHeight(types.NodeID) int64 + GetHeight(pkgp2p.NodeID) int64 } // Reactor implements a service that contains mempool of txs that are broadcasted @@ -55,7 +56,7 @@ type Reactor struct { peerWG sync.WaitGroup mtx tmsync.Mutex - peerRoutines map[types.NodeID]*tmsync.Closer + peerRoutines map[pkgp2p.NodeID]*tmsync.Closer } // NewReactor returns a reference to a new reactor. @@ -76,7 +77,7 @@ func NewReactor( mempoolCh: mempoolCh, peerUpdates: peerUpdates, closeCh: make(chan struct{}), - peerRoutines: make(map[types.NodeID]*tmsync.Closer), + peerRoutines: make(map[pkgp2p.NodeID]*tmsync.Closer), } r.BaseService = *service.NewBaseService(logger, "Mempool", r) @@ -170,7 +171,7 @@ func (r *Reactor) handleMempoolMessage(envelope p2p.Envelope) error { } for _, tx := range protoTxs { - if err := r.mempool.CheckTx(context.Background(), types.Tx(tx), nil, txInfo); err != nil { + if err := r.mempool.CheckTx(context.Background(), pkgmempool.Tx(tx), nil, txInfo); err != nil { logger.Error("checktx failed for tx", "tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(tx)), "err", err) } } @@ -305,7 +306,7 @@ func (r *Reactor) processPeerUpdates() { } } -func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) { +func (r *Reactor) broadcastTxRoutine(peerID pkgp2p.NodeID, closer *tmsync.Closer) { peerMempoolID := r.ids.GetForPeer(peerID) var next *clist.CElement diff --git a/internal/mempool/v1/mempool.go b/internal/mempool/v1/mempool.go index 850600697..2cecc4678 100644 --- a/internal/mempool/v1/mempool.go +++ b/internal/mempool/v1/mempool.go @@ -14,9 +14,9 @@ import ( "github.com/tendermint/tendermint/internal/mempool" "github.com/tendermint/tendermint/libs/log" tmmath "github.com/tendermint/tendermint/libs/math" - pubmempool "github.com/tendermint/tendermint/pkg/mempool" + "github.com/tendermint/tendermint/pkg/block" + pkgmempool "github.com/tendermint/tendermint/pkg/mempool" "github.com/tendermint/tendermint/proxy" - "github.com/tendermint/tendermint/types" ) var _ mempool.Mempool = (*TxMempool)(nil) @@ -229,7 +229,7 @@ func (txmp *TxMempool) TxsAvailable() <-chan struct{} { // - The caller is not to explicitly require any locks for executing CheckTx. func (txmp *TxMempool) CheckTx( ctx context.Context, - tx types.Tx, + tx pkgmempool.Tx, cb func(*abci.Response), txInfo mempool.TxInfo, ) error { @@ -239,7 +239,7 @@ func (txmp *TxMempool) CheckTx( txSize := len(tx) if txSize > txmp.config.MaxTxBytes { - return pubmempool.ErrTxTooLarge{ + return pkgmempool.ErrTxTooLarge{ Max: txmp.config.MaxTxBytes, Actual: txSize, } @@ -247,7 +247,7 @@ func (txmp *TxMempool) CheckTx( if txmp.preCheck != nil { if err := txmp.preCheck(tx); err != nil { - return pubmempool.ErrPreCheck{ + return pkgmempool.ErrPreCheck{ Reason: err, } } @@ -267,7 +267,7 @@ func (txmp *TxMempool) CheckTx( if wtx != nil && ok { // We already have the transaction stored and the we've already seen this // transaction from txInfo.SenderID. - return pubmempool.ErrTxInCache + return pkgmempool.ErrTxInCache } txmp.logger.Debug("tx exists already in cache", "tx_hash", tx.Hash()) @@ -333,7 +333,7 @@ func (txmp *TxMempool) Flush() { // - A read-lock is acquired. // - Transactions returned are not actually removed from the mempool transaction // store or indexes. -func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { +func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) pkgmempool.Txs { txmp.mtx.RLock() defer txmp.mtx.RUnlock() @@ -351,12 +351,12 @@ func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { } }() - txs := make([]types.Tx, 0, txmp.priorityIndex.NumTxs()) + txs := make([]pkgmempool.Tx, 0, txmp.priorityIndex.NumTxs()) for txmp.priorityIndex.NumTxs() > 0 { wtx := txmp.priorityIndex.PopTx() txs = append(txs, wtx.tx) wTxs = append(wTxs, wtx) - size := types.ComputeProtoSizeForTxs([]types.Tx{wtx.tx}) + size := block.ComputeProtoSizeForTxs([]pkgmempool.Tx{wtx.tx}) // Ensure we have capacity for the transaction with respect to the // transaction size. @@ -385,7 +385,7 @@ func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs { // - A read-lock is acquired. // - Transactions returned are not actually removed from the mempool transaction // store or indexes. -func (txmp *TxMempool) ReapMaxTxs(max int) types.Txs { +func (txmp *TxMempool) ReapMaxTxs(max int) pkgmempool.Txs { txmp.mtx.RLock() defer txmp.mtx.RUnlock() @@ -405,7 +405,7 @@ func (txmp *TxMempool) ReapMaxTxs(max int) types.Txs { } }() - txs := make([]types.Tx, 0, cap) + txs := make([]pkgmempool.Tx, 0, cap) for txmp.priorityIndex.NumTxs() > 0 && len(txs) < max { wtx := txmp.priorityIndex.PopTx() txs = append(txs, wtx.tx) @@ -426,7 +426,7 @@ func (txmp *TxMempool) ReapMaxTxs(max int) types.Txs { // - The caller must explicitly acquire a write-lock via Lock(). func (txmp *TxMempool) Update( blockHeight int64, - blockTxs types.Txs, + blockTxs pkgmempool.Txs, deliverTxResponses []*abci.ResponseDeliverTx, newPreFn mempool.PreCheckFunc, newPostFn mempool.PostCheckFunc, @@ -728,7 +728,7 @@ func (txmp *TxMempool) canAddTx(wtx *WrappedTx) error { ) if numTxs >= txmp.config.Size || int64(wtx.Size())+sizeBytes > txmp.config.MaxTxsBytes { - return pubmempool.ErrMempoolIsFull{ + return pkgmempool.ErrMempoolIsFull{ NumTxs: numTxs, MaxTxs: txmp.config.Size, TxsBytes: sizeBytes, diff --git a/internal/mempool/v1/reactor.go b/internal/mempool/v1/reactor.go index 3014e0519..4978fab52 100644 --- a/internal/mempool/v1/reactor.go +++ b/internal/mempool/v1/reactor.go @@ -15,8 +15,9 @@ import ( "github.com/tendermint/tendermint/internal/p2p" "github.com/tendermint/tendermint/libs/log" "github.com/tendermint/tendermint/libs/service" + pkgmempool "github.com/tendermint/tendermint/pkg/mempool" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" protomem "github.com/tendermint/tendermint/proto/tendermint/mempool" - "github.com/tendermint/tendermint/types" ) var ( @@ -28,7 +29,7 @@ var ( // peer information. This should eventually be replaced with a message-oriented // approach utilizing the p2p stack. type PeerManager interface { - GetHeight(types.NodeID) int64 + GetHeight(pkgp2p.NodeID) int64 } // Reactor implements a service that contains mempool of txs that are broadcasted @@ -59,7 +60,7 @@ type Reactor struct { observePanic func(interface{}) mtx tmsync.Mutex - peerRoutines map[types.NodeID]*tmsync.Closer + peerRoutines map[pkgp2p.NodeID]*tmsync.Closer } // NewReactor returns a reference to a new reactor. @@ -80,7 +81,7 @@ func NewReactor( mempoolCh: mempoolCh, peerUpdates: peerUpdates, closeCh: make(chan struct{}), - peerRoutines: make(map[types.NodeID]*tmsync.Closer), + peerRoutines: make(map[pkgp2p.NodeID]*tmsync.Closer), observePanic: defaultObservePanic, } @@ -177,7 +178,7 @@ func (r *Reactor) handleMempoolMessage(envelope p2p.Envelope) error { } for _, tx := range protoTxs { - if err := r.mempool.CheckTx(context.Background(), types.Tx(tx), nil, txInfo); err != nil { + if err := r.mempool.CheckTx(context.Background(), pkgmempool.Tx(tx), nil, txInfo); err != nil { logger.Error("checktx failed for tx", "tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(tx)), "err", err) } } @@ -313,7 +314,7 @@ func (r *Reactor) processPeerUpdates() { } } -func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) { +func (r *Reactor) broadcastTxRoutine(peerID pkgp2p.NodeID, closer *tmsync.Closer) { peerMempoolID := r.ids.GetForPeer(peerID) var nextGossipTx *clist.CElement diff --git a/internal/mempool/v1/tx.go b/internal/mempool/v1/tx.go index 15173b91f..3c7d4cbd6 100644 --- a/internal/mempool/v1/tx.go +++ b/internal/mempool/v1/tx.go @@ -7,14 +7,14 @@ import ( "github.com/tendermint/tendermint/internal/libs/clist" tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/internal/mempool" - "github.com/tendermint/tendermint/types" + pkgmempool "github.com/tendermint/tendermint/pkg/mempool" ) // WrappedTx defines a wrapper around a raw transaction with additional metadata // that is used for indexing. type WrappedTx struct { // tx represents the raw binary transaction data - tx types.Tx + tx pkgmempool.Tx // hash defines the transaction hash and the primary key used in the mempool hash [mempool.TxKeySize]byte diff --git a/internal/p2p/address.go b/internal/p2p/address.go index 7c084216e..48572efa5 100644 --- a/internal/p2p/address.go +++ b/internal/p2p/address.go @@ -10,7 +10,7 @@ import ( "strconv" "strings" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) var ( @@ -31,7 +31,7 @@ var ( // If the URL is opaque, i.e. of the form "scheme:opaque", then the opaque part // is expected to contain a node ID. type NodeAddress struct { - NodeID types.NodeID + NodeID p2p.NodeID Protocol Protocol Hostname string Port uint16 @@ -58,13 +58,13 @@ func ParseNodeAddress(urlString string) (NodeAddress, error) { // Opaque URLs are expected to contain only a node ID. if url.Opaque != "" { - address.NodeID = types.NodeID(url.Opaque) + address.NodeID = p2p.NodeID(url.Opaque) return address, address.Validate() } // Otherwise, just parse a normal networked URL. if url.User != nil { - address.NodeID = types.NodeID(strings.ToLower(url.User.Username())) + address.NodeID = p2p.NodeID(strings.ToLower(url.User.Username())) } address.Hostname = strings.ToLower(url.Hostname()) diff --git a/internal/p2p/errors.go b/internal/p2p/errors.go index 648f2cb3a..051473611 100644 --- a/internal/p2p/errors.go +++ b/internal/p2p/errors.go @@ -4,7 +4,7 @@ import ( "fmt" "net" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) // ErrFilterTimeout indicates that a filter operation timed out. @@ -20,7 +20,7 @@ type ErrRejected struct { addr NetAddress conn net.Conn err error - id types.NodeID + id p2p.NodeID isAuthFailure bool isDuplicate bool isFiltered bool @@ -101,7 +101,7 @@ func (e ErrRejected) IsSelf() bool { return e.isSelf } // ErrSwitchDuplicatePeerID to be raised when a peer is connecting with a known // ID. type ErrSwitchDuplicatePeerID struct { - ID types.NodeID + ID p2p.NodeID } func (e ErrSwitchDuplicatePeerID) Error() string { @@ -129,7 +129,7 @@ func (e ErrSwitchConnectToSelf) Error() string { type ErrSwitchAuthenticationFailure struct { Dialed *NetAddress - Got types.NodeID + Got p2p.NodeID } func (e ErrSwitchAuthenticationFailure) Error() string { diff --git a/internal/p2p/netaddress.go b/internal/p2p/netaddress.go index 6fce3a769..973d006cd 100644 --- a/internal/p2p/netaddress.go +++ b/internal/p2p/netaddress.go @@ -4,8 +4,6 @@ package p2p -import ( - "github.com/tendermint/tendermint/types" -) +import "github.com/tendermint/tendermint/pkg/p2p" -type NetAddress = types.NetAddress +type NetAddress = p2p.NetAddress diff --git a/internal/p2p/peer.go b/internal/p2p/peer.go index 709a1294a..1591725fa 100644 --- a/internal/p2p/peer.go +++ b/internal/p2p/peer.go @@ -11,7 +11,7 @@ import ( "github.com/tendermint/tendermint/libs/cmap" "github.com/tendermint/tendermint/libs/log" "github.com/tendermint/tendermint/libs/service" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) //go:generate ../../scripts/mockery_generate.sh Peer @@ -23,7 +23,7 @@ type Peer interface { service.Service FlushStop() - ID() types.NodeID // peer's cryptographic ID + ID() p2p.NodeID // peer's cryptographic ID RemoteIP() net.IP // remote IP of the connection RemoteAddr() net.Addr // remote address of the connection @@ -32,7 +32,7 @@ type Peer interface { CloseConn() error // close original connection - NodeInfo() types.NodeInfo // peer's info + NodeInfo() p2p.NodeInfo // peer's info Status() tmconn.ConnectionStatus SocketAddr() *NetAddress // actual address of the socket @@ -81,7 +81,7 @@ type peer struct { // peer's node info and the channel it knows about // channels = nodeInfo.Channels // cached to avoid copying nodeInfo in hasChannel - nodeInfo types.NodeInfo + nodeInfo p2p.NodeInfo channels []byte reactors map[byte]Reactor onPeerError func(Peer, interface{}) @@ -96,7 +96,7 @@ type peer struct { type PeerOption func(*peer) func newPeer( - nodeInfo types.NodeInfo, + nodeInfo p2p.NodeInfo, pc peerConn, reactorsByCh map[byte]Reactor, onPeerError func(Peer, interface{}), @@ -203,7 +203,7 @@ func (p *peer) OnStop() { // Implements Peer // ID returns the peer's ID - the hex encoded hash of its pubkey. -func (p *peer) ID() types.NodeID { +func (p *peer) ID() p2p.NodeID { return p.nodeInfo.ID() } @@ -218,7 +218,7 @@ func (p *peer) IsPersistent() bool { } // NodeInfo returns a copy of the peer's NodeInfo. -func (p *peer) NodeInfo() types.NodeInfo { +func (p *peer) NodeInfo() p2p.NodeInfo { return p.nodeInfo } diff --git a/internal/p2p/peer_set.go b/internal/p2p/peer_set.go index 8d4ad4939..29f74820c 100644 --- a/internal/p2p/peer_set.go +++ b/internal/p2p/peer_set.go @@ -4,14 +4,14 @@ import ( "net" tmsync "github.com/tendermint/tendermint/internal/libs/sync" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) // IPeerSet has a (immutable) subset of the methods of PeerSet. type IPeerSet interface { - Has(key types.NodeID) bool + Has(key p2p.NodeID) bool HasIP(ip net.IP) bool - Get(key types.NodeID) Peer + Get(key p2p.NodeID) Peer List() []Peer Size() int } @@ -22,7 +22,7 @@ type IPeerSet interface { // Iteration over the peers is super fast and thread-safe. type PeerSet struct { mtx tmsync.Mutex - lookup map[types.NodeID]*peerSetItem + lookup map[p2p.NodeID]*peerSetItem list []Peer } @@ -34,7 +34,7 @@ type peerSetItem struct { // NewPeerSet creates a new peerSet with a list of initial capacity of 256 items. func NewPeerSet() *PeerSet { return &PeerSet{ - lookup: make(map[types.NodeID]*peerSetItem), + lookup: make(map[p2p.NodeID]*peerSetItem), list: make([]Peer, 0, 256), } } @@ -59,7 +59,7 @@ func (ps *PeerSet) Add(peer Peer) error { // Has returns true if the set contains the peer referred to by this // peerKey, otherwise false. -func (ps *PeerSet) Has(peerKey types.NodeID) bool { +func (ps *PeerSet) Has(peerKey p2p.NodeID) bool { ps.mtx.Lock() _, ok := ps.lookup[peerKey] ps.mtx.Unlock() @@ -89,7 +89,7 @@ func (ps *PeerSet) hasIP(peerIP net.IP) bool { // Get looks up a peer by the provided peerKey. Returns nil if peer is not // found. -func (ps *PeerSet) Get(peerKey types.NodeID) Peer { +func (ps *PeerSet) Get(peerKey p2p.NodeID) Peer { ps.mtx.Lock() defer ps.mtx.Unlock() item, ok := ps.lookup[peerKey] diff --git a/internal/p2p/peermanager.go b/internal/p2p/peermanager.go index 1e9afb38b..7f6950405 100644 --- a/internal/p2p/peermanager.go +++ b/internal/p2p/peermanager.go @@ -15,8 +15,8 @@ import ( dbm "github.com/tendermint/tm-db" tmsync "github.com/tendermint/tendermint/internal/libs/sync" + "github.com/tendermint/tendermint/pkg/p2p" p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p" - "github.com/tendermint/tendermint/types" ) const ( @@ -47,7 +47,7 @@ const ( // PeerUpdate is a peer update event sent via PeerUpdates. type PeerUpdate struct { - NodeID types.NodeID + NodeID p2p.NodeID Status PeerStatus } @@ -106,7 +106,7 @@ type PeerManagerOptions struct { // to. These will be scored higher than other peers, and if // MaxConnectedUpgrade is non-zero any lower-scored peers will be evicted if // necessary to make room for these. - PersistentPeers []types.NodeID + PersistentPeers []p2p.NodeID // MaxPeers is the maximum number of peers to track information about, i.e. // store in the peer store. When exceeded, the lowest-scored unconnected peers @@ -148,15 +148,15 @@ type PeerManagerOptions struct { // PeerScores sets fixed scores for specific peers. It is mainly used // for testing. A score of 0 is ignored. - PeerScores map[types.NodeID]PeerScore + PeerScores map[p2p.NodeID]PeerScore // PrivatePeerIDs defines a set of NodeID objects which the PEX reactor will // consider private and never gossip. - PrivatePeers map[types.NodeID]struct{} + PrivatePeers map[p2p.NodeID]struct{} // persistentPeers provides fast PersistentPeers lookups. It is built // by optimize(). - persistentPeers map[types.NodeID]bool + persistentPeers map[p2p.NodeID]bool } // Validate validates the options. @@ -210,7 +210,7 @@ func (o *PeerManagerOptions) Validate() error { // isPersistentPeer checks if a peer is in PersistentPeers. It will panic // if called before optimize(). -func (o *PeerManagerOptions) isPersistent(id types.NodeID) bool { +func (o *PeerManagerOptions) isPersistent(id p2p.NodeID) bool { if o.persistentPeers == nil { panic("isPersistentPeer() called before optimize()") } @@ -221,7 +221,7 @@ func (o *PeerManagerOptions) isPersistent(id types.NodeID) bool { // separate method instead of memoizing during calls to avoid dealing with // concurrency and mutex overhead. func (o *PeerManagerOptions) optimize() { - o.persistentPeers = make(map[types.NodeID]bool, len(o.PersistentPeers)) + o.persistentPeers = make(map[p2p.NodeID]bool, len(o.PersistentPeers)) for _, p := range o.PersistentPeers { o.persistentPeers[p] = true } @@ -271,7 +271,7 @@ func (o *PeerManagerOptions) optimize() { // - EvictNext: pick peer from evict, mark as evicting. // - Disconnected: unmark connected, upgrading[from]=to, evict, evicting. type PeerManager struct { - selfID types.NodeID + selfID p2p.NodeID options PeerManagerOptions rand *rand.Rand dialWaker *tmsync.Waker // wakes up DialNext() on relevant peer changes @@ -282,16 +282,16 @@ type PeerManager struct { mtx sync.Mutex store *peerStore subscriptions map[*PeerUpdates]*PeerUpdates // keyed by struct identity (address) - dialing map[types.NodeID]bool // peers being dialed (DialNext → Dialed/DialFail) - upgrading map[types.NodeID]types.NodeID // peers claimed for upgrade (DialNext → Dialed/DialFail) - connected map[types.NodeID]bool // connected peers (Dialed/Accepted → Disconnected) - ready map[types.NodeID]bool // ready peers (Ready → Disconnected) - evict map[types.NodeID]bool // peers scheduled for eviction (Connected → EvictNext) - evicting map[types.NodeID]bool // peers being evicted (EvictNext → Disconnected) + dialing map[p2p.NodeID]bool // peers being dialed (DialNext → Dialed/DialFail) + upgrading map[p2p.NodeID]p2p.NodeID // peers claimed for upgrade (DialNext → Dialed/DialFail) + connected map[p2p.NodeID]bool // connected peers (Dialed/Accepted → Disconnected) + ready map[p2p.NodeID]bool // ready peers (Ready → Disconnected) + evict map[p2p.NodeID]bool // peers scheduled for eviction (Connected → EvictNext) + evicting map[p2p.NodeID]bool // peers being evicted (EvictNext → Disconnected) } // NewPeerManager creates a new peer manager. -func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptions) (*PeerManager, error) { +func NewPeerManager(selfID p2p.NodeID, peerDB dbm.DB, options PeerManagerOptions) (*PeerManager, error) { if selfID == "" { return nil, errors.New("self ID not given") } @@ -315,12 +315,12 @@ func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptio closeCh: make(chan struct{}), store: store, - dialing: map[types.NodeID]bool{}, - upgrading: map[types.NodeID]types.NodeID{}, - connected: map[types.NodeID]bool{}, - ready: map[types.NodeID]bool{}, - evict: map[types.NodeID]bool{}, - evicting: map[types.NodeID]bool{}, + dialing: map[p2p.NodeID]bool{}, + upgrading: map[p2p.NodeID]p2p.NodeID{}, + connected: map[p2p.NodeID]bool{}, + ready: map[p2p.NodeID]bool{}, + evict: map[p2p.NodeID]bool{}, + evicting: map[p2p.NodeID]bool{}, subscriptions: map[*PeerUpdates]*PeerUpdates{}, } if err = peerManager.configurePeers(); err != nil { @@ -340,7 +340,7 @@ func (m *PeerManager) configurePeers() error { return err } - configure := map[types.NodeID]bool{} + configure := map[p2p.NodeID]bool{} for _, id := range m.options.PersistentPeers { configure[id] = true } @@ -365,7 +365,7 @@ func (m *PeerManager) configurePeer(peer peerInfo) peerInfo { } // newPeerInfo creates a peerInfo for a new peer. -func (m *PeerManager) newPeerInfo(id types.NodeID) peerInfo { +func (m *PeerManager) newPeerInfo(id p2p.NodeID) peerInfo { peerInfo := peerInfo{ ID: id, AddressInfo: map[NodeAddress]*peerAddressInfo{}, @@ -569,7 +569,7 @@ func (m *PeerManager) Dialed(address NodeAddress) error { delete(m.dialing, address.NodeID) - var upgradeFromPeer types.NodeID + var upgradeFromPeer p2p.NodeID for from, to := range m.upgrading { if to == address.NodeID { delete(m.upgrading, from) @@ -640,7 +640,7 @@ func (m *PeerManager) Dialed(address NodeAddress) error { // that, we'll need to get the remote address after all, but as noted above that // can't be the remote endpoint since that will usually have the wrong port // number. -func (m *PeerManager) Accepted(peerID types.NodeID) error { +func (m *PeerManager) Accepted(peerID p2p.NodeID) error { m.mtx.Lock() defer m.mtx.Unlock() @@ -663,7 +663,7 @@ func (m *PeerManager) Accepted(peerID types.NodeID) error { // If all connections slots are full, but we allow upgrades (and we checked // above that we have upgrade capacity), then we can look for a lower-scored // peer to replace and if found accept the connection anyway and evict it. - var upgradeFromPeer types.NodeID + var upgradeFromPeer p2p.NodeID if m.options.MaxConnected > 0 && len(m.connected) >= int(m.options.MaxConnected) { upgradeFromPeer = m.findUpgradeCandidate(peer.ID, peer.Score()) if upgradeFromPeer == "" { @@ -688,7 +688,7 @@ func (m *PeerManager) Accepted(peerID types.NodeID) error { // peer must already be marked as connected. This is separate from Dialed() and // Accepted() to allow the router to set up its internal queues before reactors // start sending messages. -func (m *PeerManager) Ready(peerID types.NodeID) { +func (m *PeerManager) Ready(peerID p2p.NodeID) { m.mtx.Lock() defer m.mtx.Unlock() @@ -703,7 +703,7 @@ func (m *PeerManager) Ready(peerID types.NodeID) { // EvictNext returns the next peer to evict (i.e. disconnect). If no evictable // peers are found, the call will block until one becomes available. -func (m *PeerManager) EvictNext(ctx context.Context) (types.NodeID, error) { +func (m *PeerManager) EvictNext(ctx context.Context) (p2p.NodeID, error) { for { id, err := m.TryEvictNext() if err != nil || id != "" { @@ -719,7 +719,7 @@ func (m *PeerManager) EvictNext(ctx context.Context) (types.NodeID, error) { // TryEvictNext is equivalent to EvictNext, but immediately returns an empty // node ID if no evictable peers are found. -func (m *PeerManager) TryEvictNext() (types.NodeID, error) { +func (m *PeerManager) TryEvictNext() (p2p.NodeID, error) { m.mtx.Lock() defer m.mtx.Unlock() @@ -755,7 +755,7 @@ func (m *PeerManager) TryEvictNext() (types.NodeID, error) { // Disconnected unmarks a peer as connected, allowing it to be dialed or // accepted again as appropriate. -func (m *PeerManager) Disconnected(peerID types.NodeID) { +func (m *PeerManager) Disconnected(peerID p2p.NodeID) { m.mtx.Lock() defer m.mtx.Unlock() @@ -785,7 +785,7 @@ func (m *PeerManager) Disconnected(peerID types.NodeID) { // // FIXME: This will cause the peer manager to immediately try to reconnect to // the peer, which is probably not always what we want. -func (m *PeerManager) Errored(peerID types.NodeID, err error) { +func (m *PeerManager) Errored(peerID p2p.NodeID, err error) { m.mtx.Lock() defer m.mtx.Unlock() @@ -800,7 +800,7 @@ func (m *PeerManager) Errored(peerID types.NodeID, err error) { // // FIXME: This is fairly naïve and only returns the addresses of the // highest-ranked peers. -func (m *PeerManager) Advertise(peerID types.NodeID, limit uint16) []NodeAddress { +func (m *PeerManager) Advertise(peerID p2p.NodeID, limit uint16) []NodeAddress { m.mtx.Lock() defer m.mtx.Unlock() @@ -925,7 +925,7 @@ func (m *PeerManager) Close() { // Addresses returns all known addresses for a peer, primarily for testing. // The order is arbitrary. -func (m *PeerManager) Addresses(peerID types.NodeID) []NodeAddress { +func (m *PeerManager) Addresses(peerID p2p.NodeID) []NodeAddress { m.mtx.Lock() defer m.mtx.Unlock() @@ -939,11 +939,11 @@ func (m *PeerManager) Addresses(peerID types.NodeID) []NodeAddress { } // Peers returns all known peers, primarily for testing. The order is arbitrary. -func (m *PeerManager) Peers() []types.NodeID { +func (m *PeerManager) Peers() []p2p.NodeID { m.mtx.Lock() defer m.mtx.Unlock() - peers := []types.NodeID{} + peers := []p2p.NodeID{} for _, peer := range m.store.Ranked() { peers = append(peers, peer.ID) } @@ -951,11 +951,11 @@ func (m *PeerManager) Peers() []types.NodeID { } // Scores returns the peer scores for all known peers, primarily for testing. -func (m *PeerManager) Scores() map[types.NodeID]PeerScore { +func (m *PeerManager) Scores() map[p2p.NodeID]PeerScore { m.mtx.Lock() defer m.mtx.Unlock() - scores := map[types.NodeID]PeerScore{} + scores := map[p2p.NodeID]PeerScore{} for _, peer := range m.store.Ranked() { scores[peer.ID] = peer.Score() } @@ -963,7 +963,7 @@ func (m *PeerManager) Scores() map[types.NodeID]PeerScore { } // Status returns the status for a peer, primarily for testing. -func (m *PeerManager) Status(id types.NodeID) PeerStatus { +func (m *PeerManager) Status(id p2p.NodeID) PeerStatus { m.mtx.Lock() defer m.mtx.Unlock() switch { @@ -978,7 +978,7 @@ func (m *PeerManager) Status(id types.NodeID) PeerStatus { // to make room for the given peer. Returns an empty ID if none is found. // If the peer is already being upgraded to, we return that same upgrade. // The caller must hold the mutex lock. -func (m *PeerManager) findUpgradeCandidate(id types.NodeID, score PeerScore) types.NodeID { +func (m *PeerManager) findUpgradeCandidate(id p2p.NodeID, score PeerScore) p2p.NodeID { for from, to := range m.upgrading { if to == id { return from @@ -1034,7 +1034,7 @@ func (m *PeerManager) retryDelay(failures uint32, persistent bool) time.Duration // FIXME: This is a temporary workaround to share state between the consensus // and mempool reactors, carried over from the legacy P2P stack. Reactors should // not have dependencies on each other, instead tracking this themselves. -func (m *PeerManager) GetHeight(peerID types.NodeID) int64 { +func (m *PeerManager) GetHeight(peerID p2p.NodeID) int64 { m.mtx.Lock() defer m.mtx.Unlock() @@ -1047,7 +1047,7 @@ func (m *PeerManager) GetHeight(peerID types.NodeID) int64 { // FIXME: This is a temporary workaround to share state between the consensus // and mempool reactors, carried over from the legacy P2P stack. Reactors should // not have dependencies on each other, instead tracking this themselves. -func (m *PeerManager) SetHeight(peerID types.NodeID, height int64) error { +func (m *PeerManager) SetHeight(peerID p2p.NodeID, height int64) error { m.mtx.Lock() defer m.mtx.Unlock() @@ -1068,7 +1068,7 @@ func (m *PeerManager) SetHeight(peerID types.NodeID, height int64) error { // (without fsync, since we can afford to lose recent writes). type peerStore struct { db dbm.DB - peers map[types.NodeID]*peerInfo + peers map[p2p.NodeID]*peerInfo ranked []*peerInfo // cache for Ranked(), nil invalidates cache } @@ -1087,7 +1087,7 @@ func newPeerStore(db dbm.DB) (*peerStore, error) { // loadPeers loads all peers from the database into memory. func (s *peerStore) loadPeers() error { - peers := map[types.NodeID]*peerInfo{} + peers := map[p2p.NodeID]*peerInfo{} start, end := keyPeerInfoRange() iter, err := s.db.Iterator(start, end) @@ -1118,7 +1118,7 @@ func (s *peerStore) loadPeers() error { // Get fetches a peer. The boolean indicates whether the peer existed or not. // The returned peer info is a copy, and can be mutated at will. -func (s *peerStore) Get(id types.NodeID) (peerInfo, bool) { +func (s *peerStore) Get(id p2p.NodeID) (peerInfo, bool) { peer, ok := s.peers[id] return peer.Copy(), ok } @@ -1156,7 +1156,7 @@ func (s *peerStore) Set(peer peerInfo) error { } // Delete deletes a peer, or does nothing if it does not exist. -func (s *peerStore) Delete(id types.NodeID) error { +func (s *peerStore) Delete(id p2p.NodeID) error { if _, ok := s.peers[id]; !ok { return nil } @@ -1214,7 +1214,7 @@ func (s *peerStore) Size() int { // peerInfo contains peer information stored in a peerStore. type peerInfo struct { - ID types.NodeID + ID p2p.NodeID AddressInfo map[NodeAddress]*peerAddressInfo LastConnected time.Time @@ -1230,7 +1230,7 @@ type peerInfo struct { // erroring if the data is invalid. func peerInfoFromProto(msg *p2pproto.PeerInfo) (*peerInfo, error) { p := &peerInfo{ - ID: types.NodeID(msg.ID), + ID: p2p.NodeID(msg.ID), AddressInfo: map[NodeAddress]*peerAddressInfo{}, } if msg.LastConnected != nil { @@ -1367,7 +1367,7 @@ const ( ) // keyPeerInfo generates a peerInfo database key. -func keyPeerInfo(id types.NodeID) []byte { +func keyPeerInfo(id p2p.NodeID) []byte { key, err := orderedcode.Append(nil, prefixPeerInfo, string(id)) if err != nil { panic(err) diff --git a/internal/p2p/pex/addrbook.go b/internal/p2p/pex/addrbook.go index 6c5f78663..1f3ef73fd 100644 --- a/internal/p2p/pex/addrbook.go +++ b/internal/p2p/pex/addrbook.go @@ -21,7 +21,7 @@ import ( tmmath "github.com/tendermint/tendermint/libs/math" tmrand "github.com/tendermint/tendermint/libs/rand" "github.com/tendermint/tendermint/libs/service" - "github.com/tendermint/tendermint/types" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" ) const ( @@ -60,7 +60,7 @@ type AddrBook interface { PickAddress(biasTowardsNewAddrs int) *p2p.NetAddress // Mark address - MarkGood(types.NodeID) + MarkGood(pkgp2p.NodeID) MarkAttempt(*p2p.NetAddress) MarkBad(*p2p.NetAddress, time.Duration) // Move peer to bad peers list // Add bad peers back to addrBook @@ -90,9 +90,9 @@ type addrBook struct { // accessed concurrently mtx tmsync.Mutex ourAddrs map[string]struct{} - privateIDs map[types.NodeID]struct{} - addrLookup map[types.NodeID]*knownAddress // new & old - badPeers map[types.NodeID]*knownAddress // blacklisted peers + privateIDs map[pkgp2p.NodeID]struct{} + addrLookup map[pkgp2p.NodeID]*knownAddress // new & old + badPeers map[pkgp2p.NodeID]*knownAddress // blacklisted peers bucketsOld []map[string]*knownAddress bucketsNew []map[string]*knownAddress nOld int @@ -121,9 +121,9 @@ func mustNewHasher() hash.Hash64 { func NewAddrBook(filePath string, routabilityStrict bool) AddrBook { am := &addrBook{ ourAddrs: make(map[string]struct{}), - privateIDs: make(map[types.NodeID]struct{}), - addrLookup: make(map[types.NodeID]*knownAddress), - badPeers: make(map[types.NodeID]*knownAddress), + privateIDs: make(map[pkgp2p.NodeID]struct{}), + addrLookup: make(map[pkgp2p.NodeID]*knownAddress), + badPeers: make(map[pkgp2p.NodeID]*knownAddress), filePath: filePath, routabilityStrict: routabilityStrict, } @@ -202,7 +202,7 @@ func (a *addrBook) AddPrivateIDs(ids []string) { defer a.mtx.Unlock() for _, id := range ids { - a.privateIDs[types.NodeID(id)] = struct{}{} + a.privateIDs[pkgp2p.NodeID(id)] = struct{}{} } } @@ -320,7 +320,7 @@ func (a *addrBook) PickAddress(biasTowardsNewAddrs int) *p2p.NetAddress { // MarkGood implements AddrBook - it marks the peer as good and // moves it into an "old" bucket. -func (a *addrBook) MarkGood(id types.NodeID) { +func (a *addrBook) MarkGood(id pkgp2p.NodeID) { a.mtx.Lock() defer a.mtx.Unlock() diff --git a/internal/p2p/pex/known_address.go b/internal/p2p/pex/known_address.go index 2a2ebe038..179d329dc 100644 --- a/internal/p2p/pex/known_address.go +++ b/internal/p2p/pex/known_address.go @@ -4,20 +4,20 @@ import ( "time" "github.com/tendermint/tendermint/internal/p2p" - "github.com/tendermint/tendermint/types" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" ) // knownAddress tracks information about a known network address // that is used to determine how viable an address is. type knownAddress struct { - Addr *p2p.NetAddress `json:"addr"` - Src *p2p.NetAddress `json:"src"` - Buckets []int `json:"buckets"` - Attempts int32 `json:"attempts"` - BucketType byte `json:"bucket_type"` - LastAttempt time.Time `json:"last_attempt"` - LastSuccess time.Time `json:"last_success"` - LastBanTime time.Time `json:"last_ban_time"` + Addr *pkgp2p.NetAddress `json:"addr"` + Src *pkgp2p.NetAddress `json:"src"` + Buckets []int `json:"buckets"` + Attempts int32 `json:"attempts"` + BucketType byte `json:"bucket_type"` + LastAttempt time.Time `json:"last_attempt"` + LastSuccess time.Time `json:"last_success"` + LastBanTime time.Time `json:"last_ban_time"` } func newKnownAddress(addr *p2p.NetAddress, src *p2p.NetAddress) *knownAddress { @@ -31,7 +31,7 @@ func newKnownAddress(addr *p2p.NetAddress, src *p2p.NetAddress) *knownAddress { } } -func (ka *knownAddress) ID() types.NodeID { +func (ka *knownAddress) ID() pkgp2p.NodeID { return ka.Addr.ID } diff --git a/internal/p2p/pex/pex_reactor.go b/internal/p2p/pex/pex_reactor.go index 049dbd9f1..cb9efd2d3 100644 --- a/internal/p2p/pex/pex_reactor.go +++ b/internal/p2p/pex/pex_reactor.go @@ -15,8 +15,8 @@ import ( tmmath "github.com/tendermint/tendermint/libs/math" tmrand "github.com/tendermint/tendermint/libs/rand" "github.com/tendermint/tendermint/libs/service" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" tmp2p "github.com/tendermint/tendermint/proto/tendermint/p2p" - "github.com/tendermint/tendermint/types" ) type Peer = p2p.Peer @@ -94,12 +94,12 @@ type Reactor struct { requestsSent *cmap.CMap // ID->struct{}: unanswered send requests lastReceivedRequests *cmap.CMap // ID->time.Time: last time peer requested from us - seedAddrs []*p2p.NetAddress + seedAddrs []*pkgp2p.NetAddress attemptsToDial sync.Map // address (string) -> {number of attempts (int), last time dialed (time.Time)} // seed/crawled mode fields - crawlPeerInfos map[types.NodeID]crawlPeerInfo + crawlPeerInfos map[pkgp2p.NodeID]crawlPeerInfo } func (r *Reactor) minReceiveRequestInterval() time.Duration { @@ -139,7 +139,7 @@ func NewReactor(b AddrBook, config *ReactorConfig) *Reactor { ensurePeersPeriod: defaultEnsurePeersPeriod, requestsSent: cmap.NewCMap(), lastReceivedRequests: cmap.NewCMap(), - crawlPeerInfos: make(map[types.NodeID]crawlPeerInfo), + crawlPeerInfos: make(map[pkgp2p.NodeID]crawlPeerInfo), } r.BaseReactor = *p2p.NewBaseReactor("PEX", r) return r @@ -479,7 +479,7 @@ func (r *Reactor) ensurePeers() { // NOTE: range here is [10, 90]. Too high ? newBias := tmmath.MinInt(out, 8)*10 + 10 - toDial := make(map[types.NodeID]*p2p.NetAddress) + toDial := make(map[pkgp2p.NodeID]*pkgp2p.NetAddress) // Try maxAttempts times to pick numToDial addresses to dial maxAttempts := numToDial * 3 @@ -617,7 +617,7 @@ func (r *Reactor) checkSeeds() (numOnline int, netAddrs []*p2p.NetAddress, err e numOnline = lSeeds - len(errs) for _, err := range errs { switch e := err.(type) { - case types.ErrNetAddressLookup: + case pkgp2p.ErrNetAddressLookup: r.Logger.Error("Connecting to seed failed", "err", e) default: return 0, nil, fmt.Errorf("seed node configuration has error: %w", e) @@ -819,7 +819,7 @@ func decodeMsg(bz []byte) (proto.Message, error) { // address converters // NetAddressFromProto converts a Protobuf PexAddress into a native struct. -func NetAddressFromProto(pb tmp2p.PexAddress) (*types.NetAddress, error) { +func NetAddressFromProto(pb tmp2p.PexAddress) (*p2p.NetAddress, error) { ip := net.ParseIP(pb.IP) if ip == nil { return nil, fmt.Errorf("invalid IP address %v", pb.IP) @@ -827,16 +827,16 @@ func NetAddressFromProto(pb tmp2p.PexAddress) (*types.NetAddress, error) { if pb.Port >= 1<<16 { return nil, fmt.Errorf("invalid port number %v", pb.Port) } - return &types.NetAddress{ - ID: types.NodeID(pb.ID), + return &p2p.NetAddress{ + ID: pkgp2p.NodeID(pb.ID), IP: ip, Port: uint16(pb.Port), }, nil } // NetAddressesFromProto converts a slice of Protobuf PexAddresses into a native slice. -func NetAddressesFromProto(pbs []tmp2p.PexAddress) ([]*types.NetAddress, error) { - nas := make([]*types.NetAddress, 0, len(pbs)) +func NetAddressesFromProto(pbs []tmp2p.PexAddress) ([]*p2p.NetAddress, error) { + nas := make([]*p2p.NetAddress, 0, len(pbs)) for _, pb := range pbs { na, err := NetAddressFromProto(pb) if err != nil { @@ -848,7 +848,7 @@ func NetAddressesFromProto(pbs []tmp2p.PexAddress) ([]*types.NetAddress, error) } // NetAddressesToProto converts a slice of NetAddresses into a Protobuf PexAddress slice. -func NetAddressesToProto(nas []*types.NetAddress) []tmp2p.PexAddress { +func NetAddressesToProto(nas []*p2p.NetAddress) []tmp2p.PexAddress { pbs := make([]tmp2p.PexAddress, 0, len(nas)) for _, na := range nas { if na != nil { diff --git a/internal/p2p/pex/reactor.go b/internal/p2p/pex/reactor.go index 8cff2f95b..2a7850885 100644 --- a/internal/p2p/pex/reactor.go +++ b/internal/p2p/pex/reactor.go @@ -12,8 +12,8 @@ import ( "github.com/tendermint/tendermint/libs/log" tmmath "github.com/tendermint/tendermint/libs/math" "github.com/tendermint/tendermint/libs/service" + pkgp2p "github.com/tendermint/tendermint/pkg/p2p" protop2p "github.com/tendermint/tendermint/proto/tendermint/p2p" - "github.com/tendermint/tendermint/types" ) var ( @@ -79,7 +79,7 @@ type ReactorV2 struct { closeCh chan struct{} // list of available peers to loop through and send peer requests to - availablePeers map[types.NodeID]struct{} + availablePeers map[pkgp2p.NodeID]struct{} mtx sync.RWMutex @@ -87,12 +87,12 @@ type ReactorV2 struct { // to. This prevents the sending of spurious responses. // NOTE: If a node never responds, they will remain in this map until a // peer down status update is sent - requestsSent map[types.NodeID]struct{} + requestsSent map[pkgp2p.NodeID]struct{} // lastReceivedRequests keeps track of when peers send a request to prevent // peers from sending requests too often (as defined by // minReceiveRequestInterval). - lastReceivedRequests map[types.NodeID]time.Time + lastReceivedRequests map[pkgp2p.NodeID]time.Time // the time when another request will be sent nextRequestTime time.Time @@ -121,9 +121,9 @@ func NewReactorV2( pexCh: pexCh, peerUpdates: peerUpdates, closeCh: make(chan struct{}), - availablePeers: make(map[types.NodeID]struct{}), - requestsSent: make(map[types.NodeID]struct{}), - lastReceivedRequests: make(map[types.NodeID]time.Time), + availablePeers: make(map[pkgp2p.NodeID]struct{}), + requestsSent: make(map[pkgp2p.NodeID]struct{}), + lastReceivedRequests: make(map[pkgp2p.NodeID]time.Time), } r.BaseService = *service.NewBaseService(logger, "PEX", r) @@ -426,7 +426,7 @@ func (r *ReactorV2) sendRequestForPeers() { return } - var peerID types.NodeID + var peerID pkgp2p.NodeID // use range to get a random peer. for peerID = range r.availablePeers { @@ -500,7 +500,7 @@ func (r *ReactorV2) calculateNextRequestTime() { r.nextRequestTime = time.Now().Add(baseTime * time.Duration(r.discoveryRatio)) } -func (r *ReactorV2) markPeerRequest(peer types.NodeID) error { +func (r *ReactorV2) markPeerRequest(peer pkgp2p.NodeID) error { r.mtx.Lock() defer r.mtx.Unlock() if lastRequestTime, ok := r.lastReceivedRequests[peer]; ok { @@ -513,7 +513,7 @@ func (r *ReactorV2) markPeerRequest(peer types.NodeID) error { return nil } -func (r *ReactorV2) markPeerResponse(peer types.NodeID) error { +func (r *ReactorV2) markPeerResponse(peer pkgp2p.NodeID) error { r.mtx.Lock() defer r.mtx.Unlock() // check if a request to this peer was sent @@ -530,7 +530,7 @@ func (r *ReactorV2) markPeerResponse(peer types.NodeID) error { // all addresses must use a MCONN protocol for the peer to be considered part of the // legacy p2p pex system -func (r *ReactorV2) isLegacyPeer(peer types.NodeID) bool { +func (r *ReactorV2) isLegacyPeer(peer pkgp2p.NodeID) bool { for _, addr := range r.peerManager.Addresses(peer) { if addr.Protocol != p2p.MConnProtocol { return false diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 1171566d1..4e5403687 100644 --- a/internal/p2p/router.go +++ b/internal/p2p/router.go @@ -16,7 +16,7 @@ import ( "github.com/tendermint/tendermint/crypto" "github.com/tendermint/tendermint/libs/log" "github.com/tendermint/tendermint/libs/service" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) const queueBufferDefault = 32 @@ -26,8 +26,8 @@ type ChannelID uint16 // Envelope contains a message with sender/receiver routing info. type Envelope struct { - From types.NodeID // sender (empty if outbound) - To types.NodeID // receiver (empty if inbound) + From p2p.NodeID // sender (empty if outbound) + To p2p.NodeID // receiver (empty if inbound) Broadcast bool // send to all connected peers (ignores To) Message proto.Message // message payload @@ -52,7 +52,7 @@ type Envelope struct { // It should possibly also allow reactors to request explicit actions, e.g. // disconnection or banning, in addition to doing this based on aggregates. type PeerError struct { - NodeID types.NodeID + NodeID p2p.NodeID Err error } @@ -157,7 +157,7 @@ type RouterOptions struct { // but this occurs after the handshake is complete. Filter by // IP address to filter before the handshake. Functions should // return an error to reject the peer. - FilterPeerByID func(context.Context, types.NodeID) error + FilterPeerByID func(context.Context, p2p.NodeID) error // DialSleep controls the amount of time that the router // sleeps between dialing peers. If not set, a default value @@ -248,7 +248,7 @@ type Router struct { logger log.Logger metrics *Metrics options RouterOptions - nodeInfo types.NodeInfo + nodeInfo p2p.NodeInfo privKey crypto.PrivKey peerManager *PeerManager chDescs []ChannelDescriptor @@ -258,9 +258,9 @@ type Router struct { stopCh chan struct{} // signals Router shutdown peerMtx sync.RWMutex - peerQueues map[types.NodeID]queue // outbound messages per peer for all channels + peerQueues map[p2p.NodeID]queue // outbound messages per peer for all channels // the channels that the peer queue has open - peerChannels map[types.NodeID]channelIDs + peerChannels map[p2p.NodeID]channelIDs queueFactory func(int) queue // FIXME: We don't strictly need to use a mutex for this if we seal the @@ -277,7 +277,7 @@ type Router struct { func NewRouter( logger log.Logger, metrics *Metrics, - nodeInfo types.NodeInfo, + nodeInfo p2p.NodeInfo, privKey crypto.PrivKey, peerManager *PeerManager, transports []Transport, @@ -305,8 +305,8 @@ func NewRouter( stopCh: make(chan struct{}), channelQueues: map[ChannelID]queue{}, channelMessages: map[ChannelID]proto.Message{}, - peerQueues: map[types.NodeID]queue{}, - peerChannels: make(map[types.NodeID]channelIDs), + peerQueues: map[p2p.NodeID]queue{}, + peerChannels: make(map[p2p.NodeID]channelIDs), } router.BaseService = service.NewBaseService(logger, "router", router) @@ -533,7 +533,7 @@ func (r *Router) filterPeersIP(ctx context.Context, ip net.IP, port uint16) erro return r.options.FilterPeerByIP(ctx, ip, port) } -func (r *Router) filterPeersID(ctx context.Context, id types.NodeID) error { +func (r *Router) filterPeersID(ctx context.Context, id p2p.NodeID) error { if r.options.FilterPeerByID == nil { return nil } @@ -739,7 +739,7 @@ func (r *Router) connectPeer(ctx context.Context, address NodeAddress) { go r.routePeer(address.NodeID, conn, toChannelIDs(peerInfo.Channels)) } -func (r *Router) getOrMakeQueue(peerID types.NodeID, channels channelIDs) queue { +func (r *Router) getOrMakeQueue(peerID p2p.NodeID, channels channelIDs) queue { r.peerMtx.Lock() defer r.peerMtx.Unlock() @@ -808,8 +808,8 @@ func (r *Router) dialPeer(ctx context.Context, address NodeAddress) (Connection, func (r *Router) handshakePeer( ctx context.Context, conn Connection, - expectID types.NodeID, -) (types.NodeInfo, crypto.PubKey, error) { + expectID p2p.NodeID, +) (p2p.NodeInfo, crypto.PubKey, error) { if r.options.HandshakeTimeout > 0 { var cancel context.CancelFunc @@ -824,9 +824,9 @@ func (r *Router) handshakePeer( if err = peerInfo.Validate(); err != nil { return peerInfo, peerKey, fmt.Errorf("invalid handshake NodeInfo: %w", err) } - if types.NodeIDFromPubKey(peerKey) != peerInfo.NodeID { + if p2p.NodeIDFromPubKey(peerKey) != peerInfo.NodeID { return peerInfo, peerKey, fmt.Errorf("peer's public key did not match its node ID %q (expected %q)", - peerInfo.NodeID, types.NodeIDFromPubKey(peerKey)) + peerInfo.NodeID, p2p.NodeIDFromPubKey(peerKey)) } if expectID != "" && expectID != peerInfo.NodeID { return peerInfo, peerKey, fmt.Errorf("expected to connect with peer %q, got %q", @@ -851,7 +851,7 @@ func (r *Router) runWithPeerMutex(fn func() error) error { // routePeer routes inbound and outbound messages between a peer and the reactor // channels. It will close the given connection and send queue when done, or if // they are closed elsewhere it will cause this method to shut down and return. -func (r *Router) routePeer(peerID types.NodeID, conn Connection, channels channelIDs) { +func (r *Router) routePeer(peerID p2p.NodeID, conn Connection, channels channelIDs) { r.metrics.Peers.Add(1) r.peerManager.Ready(peerID) @@ -901,7 +901,7 @@ func (r *Router) routePeer(peerID types.NodeID, conn Connection, channels channe // receivePeer receives inbound messages from a peer, deserializes them and // passes them on to the appropriate channel. -func (r *Router) receivePeer(peerID types.NodeID, conn Connection) error { +func (r *Router) receivePeer(peerID p2p.NodeID, conn Connection) error { for { chID, bz, err := conn.ReceiveMessage() if err != nil { @@ -952,7 +952,7 @@ func (r *Router) receivePeer(peerID types.NodeID, conn Connection) error { } // sendPeer sends queued messages to a peer. -func (r *Router) sendPeer(peerID types.NodeID, conn Connection, peerQueue queue) error { +func (r *Router) sendPeer(peerID p2p.NodeID, conn Connection, peerQueue queue) error { for { start := time.Now().UTC() @@ -1017,7 +1017,7 @@ func (r *Router) evictPeers() { } // NodeInfo returns a copy of the current NodeInfo. Used for testing. -func (r *Router) NodeInfo() types.NodeInfo { +func (r *Router) NodeInfo() p2p.NodeInfo { return r.nodeInfo.Copy() } diff --git a/internal/p2p/switch.go b/internal/p2p/switch.go index eeb93a994..7bad0f1fc 100644 --- a/internal/p2p/switch.go +++ b/internal/p2p/switch.go @@ -16,7 +16,7 @@ import ( "github.com/tendermint/tendermint/libs/cmap" tmrand "github.com/tendermint/tendermint/libs/rand" "github.com/tendermint/tendermint/libs/service" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) const ( @@ -57,7 +57,7 @@ type AddrBook interface { AddPrivateIDs([]string) AddOurAddress(*NetAddress) OurAddress(*NetAddress) bool - MarkGood(types.NodeID) + MarkGood(p2p.NodeID) RemoveAddress(*NetAddress) HasAddress(*NetAddress) bool Save() @@ -103,12 +103,12 @@ type Switch struct { peers *PeerSet dialing *cmap.CMap reconnecting *cmap.CMap - nodeInfo types.NodeInfo // our node info - nodeKey types.NodeKey // our node privkey + nodeInfo p2p.NodeInfo // our node info + nodeKey p2p.NodeKey // our node privkey addrBook AddrBook // peers addresses with whom we'll maintain constant connection persistentPeersAddrs []*NetAddress - unconditionalPeerIDs map[types.NodeID]struct{} + unconditionalPeerIDs map[p2p.NodeID]struct{} transport Transport @@ -154,7 +154,7 @@ func NewSwitch( metrics: NopMetrics(), transport: transport, persistentPeersAddrs: make([]*NetAddress, 0), - unconditionalPeerIDs: make(map[types.NodeID]struct{}), + unconditionalPeerIDs: make(map[p2p.NodeID]struct{}), filterTimeout: defaultFilterTimeout, conns: NewConnSet(), } @@ -242,19 +242,19 @@ func (sw *Switch) Reactor(name string) Reactor { // SetNodeInfo sets the switch's NodeInfo for checking compatibility and handshaking with other nodes. // NOTE: Not goroutine safe. -func (sw *Switch) SetNodeInfo(nodeInfo types.NodeInfo) { +func (sw *Switch) SetNodeInfo(nodeInfo p2p.NodeInfo) { sw.nodeInfo = nodeInfo } // NodeInfo returns the switch's NodeInfo. // NOTE: Not goroutine safe. -func (sw *Switch) NodeInfo() types.NodeInfo { +func (sw *Switch) NodeInfo() p2p.NodeInfo { return sw.nodeInfo } // SetNodeKey sets the switch's private key for authenticated encryption. // NOTE: Not goroutine safe. -func (sw *Switch) SetNodeKey(nodeKey types.NodeKey) { +func (sw *Switch) SetNodeKey(nodeKey p2p.NodeKey) { sw.nodeKey = nodeKey } @@ -353,7 +353,7 @@ func (sw *Switch) NumPeers() (outbound, inbound, dialing int) { return } -func (sw *Switch) IsPeerUnconditional(id types.NodeID) bool { +func (sw *Switch) IsPeerUnconditional(id p2p.NodeID) bool { _, ok := sw.unconditionalPeerIDs[id] return ok } @@ -518,7 +518,7 @@ func (sw *Switch) DialPeersAsync(peers []string) error { } // return first non-ErrNetAddressLookup error for _, err := range errs { - if _, ok := err.(types.ErrNetAddressLookup); ok { + if _, ok := err.(p2p.ErrNetAddressLookup); ok { continue } return err @@ -622,7 +622,7 @@ func (sw *Switch) AddPersistentPeers(addrs []string) error { } // return first non-ErrNetAddressLookup error for _, err := range errs { - if _, ok := err.(types.ErrNetAddressLookup); ok { + if _, ok := err.(p2p.ErrNetAddressLookup); ok { continue } return err @@ -634,11 +634,11 @@ func (sw *Switch) AddPersistentPeers(addrs []string) error { func (sw *Switch) AddUnconditionalPeerIDs(ids []string) error { sw.Logger.Info("Adding unconditional peer ids", "ids", ids) for i, id := range ids { - err := types.NodeID(id).Validate() + err := p2p.NodeID(id).Validate() if err != nil { return fmt.Errorf("wrong ID #%d: %w", i, err) } - sw.unconditionalPeerIDs[types.NodeID(id)] = struct{}{} + sw.unconditionalPeerIDs[p2p.NodeID(id)] = struct{}{} } return nil } @@ -646,7 +646,7 @@ func (sw *Switch) AddUnconditionalPeerIDs(ids []string) error { func (sw *Switch) AddPrivatePeerIDs(ids []string) error { validIDs := make([]string, 0, len(ids)) for i, id := range ids { - err := types.NodeID(id).Validate() + err := p2p.NodeID(id).Validate() if err != nil { return fmt.Errorf("wrong ID #%d: %w", i, err) } @@ -669,7 +669,7 @@ func (sw *Switch) IsPeerPersistent(na *NetAddress) bool { func (sw *Switch) acceptRoutine() { for { - var peerNodeInfo types.NodeInfo + var peerNodeInfo p2p.NodeInfo c, err := sw.transport.Accept() if err == nil { // NOTE: The legacy MConn transport did handshaking in Accept(), @@ -803,7 +803,7 @@ func (sw *Switch) addOutboundPeerWithConfig( ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - var peerNodeInfo types.NodeInfo + var peerNodeInfo p2p.NodeInfo c, err := sw.transport.Dial(ctx, Endpoint{ Protocol: MConnProtocol, IP: addr.IP, @@ -864,8 +864,8 @@ func (sw *Switch) addOutboundPeerWithConfig( func (sw *Switch) handshakePeer( c Connection, - expectPeerID types.NodeID, -) (types.NodeInfo, crypto.PubKey, error) { + expectPeerID p2p.NodeID, +) (p2p.NodeInfo, crypto.PubKey, error) { // Moved from transport and hardcoded until legacy P2P stack removal. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -889,7 +889,7 @@ func (sw *Switch) handshakePeer( // For outgoing conns, ensure connection key matches dialed key. if expectPeerID != "" { - peerID := types.NodeIDFromPubKey(peerKey) + peerID := p2p.NodeIDFromPubKey(peerKey) if expectPeerID != peerID { return peerInfo, peerKey, ErrRejected{ conn: c.(*mConnConnection).conn, @@ -906,7 +906,7 @@ func (sw *Switch) handshakePeer( if sw.nodeInfo.ID() == peerInfo.ID() { return peerInfo, peerKey, ErrRejected{ - addr: *types.NewNetAddress(peerInfo.ID(), c.(*mConnConnection).conn.RemoteAddr()), + addr: *p2p.NewNetAddress(peerInfo.ID(), c.(*mConnConnection).conn.RemoteAddr()), conn: c.(*mConnConnection).conn, id: peerInfo.ID(), isSelf: true, @@ -1054,7 +1054,7 @@ func NewNetAddressStrings(addrs []string) ([]*NetAddress, []error) { netAddrs := make([]*NetAddress, 0) errs := make([]error, 0) for _, addr := range addrs { - netAddr, err := types.NewNetAddressString(addr) + netAddr, err := p2p.NewNetAddressString(addr) if err != nil { errs = append(errs, err) } else { diff --git a/internal/p2p/test_util.go b/internal/p2p/test_util.go index b2851646d..e257ec1c1 100644 --- a/internal/p2p/test_util.go +++ b/internal/p2p/test_util.go @@ -9,7 +9,7 @@ import ( "github.com/tendermint/tendermint/libs/log" tmnet "github.com/tendermint/tendermint/libs/net" tmrand "github.com/tendermint/tendermint/libs/rand" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" "github.com/tendermint/tendermint/config" "github.com/tendermint/tendermint/internal/p2p/conn" @@ -27,7 +27,7 @@ func CreateRandomPeer(outbound bool) Peer { addr, netAddr := CreateRoutableAddr() p := &peer{ peerConn: peerConn{outbound: outbound}, - nodeInfo: types.NodeInfo{ + nodeInfo: p2p.NodeInfo{ NodeID: netAddr.ID, ListenAddr: netAddr.DialString(), }, @@ -47,7 +47,7 @@ func CreateRoutableAddr() (addr string, netAddr *NetAddress) { mrand.Int()%256, mrand.Int()%256, mrand.Int()%256) - netAddr, err = types.NewNetAddressString(addr) + netAddr, err = p2p.NewNetAddressString(addr) if err != nil { panic(err) } @@ -170,9 +170,9 @@ func MakeSwitch( opts ...SwitchOption, ) *Switch { - nodeKey := types.GenNodeKey() + nodeKey := p2p.GenNodeKey() nodeInfo := testNodeInfo(nodeKey.ID, fmt.Sprintf("node%d", i)) - addr, err := types.NewNetAddressString( + addr, err := p2p.NewNetAddressString( nodeKey.ID.AddressString(nodeInfo.ListenAddr), ) if err != nil { @@ -227,12 +227,12 @@ func testPeerConn( //---------------------------------------------------------------- // rand node info -func testNodeInfo(id types.NodeID, name string) types.NodeInfo { +func testNodeInfo(id p2p.NodeID, name string) p2p.NodeInfo { return testNodeInfoWithNetwork(id, name, "testing") } -func testNodeInfoWithNetwork(id types.NodeID, name, network string) types.NodeInfo { - return types.NodeInfo{ +func testNodeInfoWithNetwork(id p2p.NodeID, name, network string) p2p.NodeInfo { + return p2p.NodeInfo{ ProtocolVersion: defaultProtocolVersion, NodeID: id, ListenAddr: fmt.Sprintf("127.0.0.1:%d", getFreePort()), @@ -240,7 +240,7 @@ func testNodeInfoWithNetwork(id types.NodeID, name, network string) types.NodeIn Version: "1.2.3-rc0-deadbeef", Channels: []byte{testCh}, Moniker: name, - Other: types.NodeInfoOther{ + Other: p2p.NodeInfoOther{ TxIndex: "on", RPCAddress: fmt.Sprintf("127.0.0.1:%d", getFreePort()), }, @@ -272,7 +272,7 @@ func (book *AddrBookMock) OurAddress(addr *NetAddress) bool { _, ok := book.OurAddrs[addr.String()] return ok } -func (book *AddrBookMock) MarkGood(types.NodeID) {} +func (book *AddrBookMock) MarkGood(p2p.NodeID) {} func (book *AddrBookMock) HasAddress(addr *NetAddress) bool { _, ok := book.Addrs[addr.String()] return ok diff --git a/internal/p2p/transport.go b/internal/p2p/transport.go index a3245dfc8..fe1f779ab 100644 --- a/internal/p2p/transport.go +++ b/internal/p2p/transport.go @@ -8,7 +8,7 @@ import ( "github.com/tendermint/tendermint/crypto" "github.com/tendermint/tendermint/internal/p2p/conn" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" "github.com/tendermint/tendermint/version" ) @@ -22,7 +22,7 @@ const ( // defaultProtocolVersion populates the Block and P2P versions using // the global values, but not the App. -var defaultProtocolVersion = types.ProtocolVersion{ +var defaultProtocolVersion = p2p.ProtocolVersion{ P2P: version.P2PProtocol, Block: version.BlockProtocol, App: 0, @@ -84,7 +84,7 @@ type Connection interface { // FIXME: The handshake should really be the Router's responsibility, but // that requires the connection interface to be byte-oriented rather than // message-oriented (see comment above). - Handshake(context.Context, types.NodeInfo, crypto.PrivKey) (types.NodeInfo, crypto.PubKey, error) + Handshake(context.Context, p2p.NodeInfo, crypto.PrivKey) (p2p.NodeInfo, crypto.PubKey, error) // ReceiveMessage returns the next message received on the connection, // blocking until one is available. Returns io.EOF if closed. @@ -156,7 +156,7 @@ type Endpoint struct { } // NewEndpoint constructs an Endpoint from a types.NetAddress structure. -func NewEndpoint(na *types.NetAddress) Endpoint { +func NewEndpoint(na *p2p.NetAddress) Endpoint { return Endpoint{ Protocol: MConnProtocol, IP: na.IP, @@ -165,7 +165,7 @@ func NewEndpoint(na *types.NetAddress) Endpoint { } // NodeAddress converts the endpoint into a NodeAddress for the given node ID. -func (e Endpoint) NodeAddress(nodeID types.NodeID) NodeAddress { +func (e Endpoint) NodeAddress(nodeID p2p.NodeID) NodeAddress { address := NodeAddress{ NodeID: nodeID, Protocol: e.Protocol, @@ -184,7 +184,7 @@ func (e Endpoint) String() string { // assume that path is a node ID (to handle opaque URLs of the form // scheme:id). if e.IP == nil { - if nodeID, err := types.NewNodeID(e.Path); err == nil { + if nodeID, err := p2p.NewNodeID(e.Path); err == nil { return e.NodeAddress(nodeID).String() } } diff --git a/internal/p2p/transport_mconn.go b/internal/p2p/transport_mconn.go index eca261476..7162a8627 100644 --- a/internal/p2p/transport_mconn.go +++ b/internal/p2p/transport_mconn.go @@ -16,8 +16,8 @@ import ( "github.com/tendermint/tendermint/internal/libs/protoio" "github.com/tendermint/tendermint/internal/p2p/conn" "github.com/tendermint/tendermint/libs/log" + "github.com/tendermint/tendermint/pkg/p2p" p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p" - "github.com/tendermint/tendermint/types" ) const ( @@ -255,12 +255,12 @@ func newMConnConnection( // Handshake implements Connection. func (c *mConnConnection) Handshake( ctx context.Context, - nodeInfo types.NodeInfo, + nodeInfo p2p.NodeInfo, privKey crypto.PrivKey, -) (types.NodeInfo, crypto.PubKey, error) { +) (p2p.NodeInfo, crypto.PubKey, error) { var ( mconn *conn.MConnection - peerInfo types.NodeInfo + peerInfo p2p.NodeInfo peerKey crypto.PubKey errCh = make(chan error, 1) ) @@ -283,16 +283,16 @@ func (c *mConnConnection) Handshake( select { case <-ctx.Done(): _ = c.Close() - return types.NodeInfo{}, nil, ctx.Err() + return p2p.NodeInfo{}, nil, ctx.Err() case err := <-errCh: if err != nil { - return types.NodeInfo{}, nil, err + return p2p.NodeInfo{}, nil, err } c.mconn = mconn c.logger = mconn.Logger if err = c.mconn.Start(); err != nil { - return types.NodeInfo{}, nil, err + return p2p.NodeInfo{}, nil, err } return peerInfo, peerKey, nil } @@ -303,16 +303,16 @@ func (c *mConnConnection) Handshake( // unstarted but handshaked MConnection, to avoid concurrent field writes. func (c *mConnConnection) handshake( ctx context.Context, - nodeInfo types.NodeInfo, + nodeInfo p2p.NodeInfo, privKey crypto.PrivKey, -) (*conn.MConnection, types.NodeInfo, crypto.PubKey, error) { +) (*conn.MConnection, p2p.NodeInfo, crypto.PubKey, error) { if c.mconn != nil { - return nil, types.NodeInfo{}, nil, errors.New("connection is already handshaked") + return nil, p2p.NodeInfo{}, nil, errors.New("connection is already handshaked") } secretConn, err := conn.MakeSecretConnection(c.conn, privKey) if err != nil { - return nil, types.NodeInfo{}, nil, err + return nil, p2p.NodeInfo{}, nil, err } var pbPeerInfo p2pproto.NodeInfo @@ -322,17 +322,17 @@ func (c *mConnConnection) handshake( errCh <- err }() go func() { - _, err := protoio.NewDelimitedReader(secretConn, types.MaxNodeInfoSize()).ReadMsg(&pbPeerInfo) + _, err := protoio.NewDelimitedReader(secretConn, p2p.MaxNodeInfoSize()).ReadMsg(&pbPeerInfo) errCh <- err }() for i := 0; i < cap(errCh); i++ { if err = <-errCh; err != nil { - return nil, types.NodeInfo{}, nil, err + return nil, p2p.NodeInfo{}, nil, err } } - peerInfo, err := types.NodeInfoFromProto(&pbPeerInfo) + peerInfo, err := p2p.NodeInfoFromProto(&pbPeerInfo) if err != nil { - return nil, types.NodeInfo{}, nil, err + return nil, p2p.NodeInfo{}, nil, err } mconn := conn.NewMConnectionWithConfig( diff --git a/internal/p2p/transport_memory.go b/internal/p2p/transport_memory.go index 09a387254..c6a9fb66e 100644 --- a/internal/p2p/transport_memory.go +++ b/internal/p2p/transport_memory.go @@ -12,7 +12,7 @@ import ( tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/internal/p2p/conn" "github.com/tendermint/tendermint/libs/log" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/p2p" ) const ( @@ -28,7 +28,7 @@ type MemoryNetwork struct { logger log.Logger mtx sync.RWMutex - transports map[types.NodeID]*MemoryTransport + transports map[p2p.NodeID]*MemoryTransport bufferSize int } @@ -37,14 +37,14 @@ func NewMemoryNetwork(logger log.Logger, bufferSize int) *MemoryNetwork { return &MemoryNetwork{ bufferSize: bufferSize, logger: logger, - transports: map[types.NodeID]*MemoryTransport{}, + transports: map[p2p.NodeID]*MemoryTransport{}, } } // CreateTransport creates a new memory transport endpoint with the given node // ID and immediately begins listening on the address "memory:". It panics // if the node ID is already in use (which is fine, since this is for tests). -func (n *MemoryNetwork) CreateTransport(nodeID types.NodeID) *MemoryTransport { +func (n *MemoryNetwork) CreateTransport(nodeID p2p.NodeID) *MemoryTransport { t := newMemoryTransport(n, nodeID) n.mtx.Lock() @@ -57,14 +57,14 @@ func (n *MemoryNetwork) CreateTransport(nodeID types.NodeID) *MemoryTransport { } // GetTransport looks up a transport in the network, returning nil if not found. -func (n *MemoryNetwork) GetTransport(id types.NodeID) *MemoryTransport { +func (n *MemoryNetwork) GetTransport(id p2p.NodeID) *MemoryTransport { n.mtx.RLock() defer n.mtx.RUnlock() return n.transports[id] } // RemoveTransport removes a transport from the network and closes it. -func (n *MemoryNetwork) RemoveTransport(id types.NodeID) { +func (n *MemoryNetwork) RemoveTransport(id p2p.NodeID) { n.mtx.Lock() t, ok := n.transports[id] delete(n.transports, id) @@ -92,7 +92,7 @@ func (n *MemoryNetwork) Size() int { type MemoryTransport struct { logger log.Logger network *MemoryNetwork - nodeID types.NodeID + nodeID p2p.NodeID bufferSize int acceptCh chan *MemoryConnection @@ -102,7 +102,7 @@ type MemoryTransport struct { // newMemoryTransport creates a new MemoryTransport. This is for internal use by // MemoryNetwork, use MemoryNetwork.CreateTransport() instead. -func newMemoryTransport(network *MemoryNetwork, nodeID types.NodeID) *MemoryTransport { +func newMemoryTransport(network *MemoryNetwork, nodeID p2p.NodeID) *MemoryTransport { return &MemoryTransport{ logger: network.logger.With("local", nodeID), network: network, @@ -163,7 +163,7 @@ func (t *MemoryTransport) Dial(ctx context.Context, endpoint Endpoint) (Connecti return nil, err } - nodeID, err := types.NewNodeID(endpoint.Path) + nodeID, err := p2p.NewNodeID(endpoint.Path) if err != nil { return nil, err } @@ -204,8 +204,8 @@ func (t *MemoryTransport) Close() error { // MemoryConnection is an in-memory connection between two transport endpoints. type MemoryConnection struct { logger log.Logger - localID types.NodeID - remoteID types.NodeID + localID p2p.NodeID + remoteID p2p.NodeID receiveCh <-chan memoryMessage sendCh chan<- memoryMessage @@ -218,15 +218,15 @@ type memoryMessage struct { message []byte // For handshakes. - nodeInfo *types.NodeInfo + nodeInfo *p2p.NodeInfo pubKey crypto.PubKey } // newMemoryConnection creates a new MemoryConnection. func newMemoryConnection( logger log.Logger, - localID types.NodeID, - remoteID types.NodeID, + localID p2p.NodeID, + remoteID p2p.NodeID, receiveCh <-chan memoryMessage, sendCh chan<- memoryMessage, closer *tmsync.Closer, @@ -270,29 +270,29 @@ func (c *MemoryConnection) Status() conn.ConnectionStatus { // Handshake implements Connection. func (c *MemoryConnection) Handshake( ctx context.Context, - nodeInfo types.NodeInfo, + nodeInfo p2p.NodeInfo, privKey crypto.PrivKey, -) (types.NodeInfo, crypto.PubKey, error) { +) (p2p.NodeInfo, crypto.PubKey, error) { select { case c.sendCh <- memoryMessage{nodeInfo: &nodeInfo, pubKey: privKey.PubKey()}: c.logger.Debug("sent handshake", "nodeInfo", nodeInfo) case <-c.closer.Done(): - return types.NodeInfo{}, nil, io.EOF + return p2p.NodeInfo{}, nil, io.EOF case <-ctx.Done(): - return types.NodeInfo{}, nil, ctx.Err() + return p2p.NodeInfo{}, nil, ctx.Err() } select { case msg := <-c.receiveCh: if msg.nodeInfo == nil { - return types.NodeInfo{}, nil, errors.New("no NodeInfo in handshake") + return p2p.NodeInfo{}, nil, errors.New("no NodeInfo in handshake") } c.logger.Debug("received handshake", "peerInfo", msg.nodeInfo) return *msg.nodeInfo, msg.pubKey, nil case <-c.closer.Done(): - return types.NodeInfo{}, nil, io.EOF + return p2p.NodeInfo{}, nil, io.EOF case <-ctx.Done(): - return types.NodeInfo{}, nil, ctx.Err() + return p2p.NodeInfo{}, nil, ctx.Err() } } diff --git a/light/provider/provider.go b/light/provider/provider.go index 7f15d5c75..ff19b84be 100644 --- a/light/provider/provider.go +++ b/light/provider/provider.go @@ -3,7 +3,8 @@ package provider import ( "context" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/evidence" + "github.com/tendermint/tendermint/pkg/light" ) //go:generate ../../scripts/mockery_generate.sh Provider @@ -21,8 +22,8 @@ type Provider interface { // issues, an error will be returned. // If there's no LightBlock for the given height, ErrLightBlockNotFound // error is returned. - LightBlock(ctx context.Context, height int64) (*types.LightBlock, error) + LightBlock(ctx context.Context, height int64) (*light.LightBlock, error) // ReportEvidence reports an evidence of misbehavior. - ReportEvidence(context.Context, types.Evidence) error + ReportEvidence(context.Context, evidence.Evidence) error } diff --git a/light/store/db/db.go b/light/store/db/db.go index acfda1f79..ed50d9a34 100644 --- a/light/store/db/db.go +++ b/light/store/db/db.go @@ -9,8 +9,8 @@ import ( tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/light/store" + "github.com/tendermint/tendermint/pkg/light" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) const ( @@ -45,7 +45,7 @@ func New(db dbm.DB) store.Store { // SaveLightBlock persists LightBlock to the db. // // Safe for concurrent use by multiple goroutines. -func (s *dbs) SaveLightBlock(lb *types.LightBlock) error { +func (s *dbs) SaveLightBlock(lb *light.LightBlock) error { if lb.Height <= 0 { panic("negative or zero height") } @@ -110,7 +110,7 @@ func (s *dbs) DeleteLightBlock(height int64) error { // LightBlock retrieves the LightBlock at the given height. // // Safe for concurrent use by multiple goroutines. -func (s *dbs) LightBlock(height int64) (*types.LightBlock, error) { +func (s *dbs) LightBlock(height int64) (*light.LightBlock, error) { if height <= 0 { panic("negative or zero height") } @@ -129,7 +129,7 @@ func (s *dbs) LightBlock(height int64) (*types.LightBlock, error) { return nil, fmt.Errorf("unmarshal error: %w", err) } - lightBlock, err := types.LightBlockFromProto(&lbpb) + lightBlock, err := light.LightBlockFromProto(&lbpb) if err != nil { return nil, fmt.Errorf("proto conversion error: %w", err) } @@ -181,7 +181,7 @@ func (s *dbs) FirstLightBlockHeight() (int64, error) { // the given height. It returns ErrLightBlockNotFound if no such block exists. // // Safe for concurrent use by multiple goroutines. -func (s *dbs) LightBlockBefore(height int64) (*types.LightBlock, error) { +func (s *dbs) LightBlockBefore(height int64) (*light.LightBlock, error) { if height <= 0 { panic("negative or zero height") } @@ -202,7 +202,7 @@ func (s *dbs) LightBlockBefore(height int64) (*types.LightBlock, error) { return nil, fmt.Errorf("unmarshal error: %w", err) } - lightBlock, err := types.LightBlockFromProto(&lbpb) + lightBlock, err := light.LightBlockFromProto(&lbpb) if err != nil { return nil, fmt.Errorf("proto conversion error: %w", err) } diff --git a/light/store/store.go b/light/store/store.go index 7c29f233d..3a6aad405 100644 --- a/light/store/store.go +++ b/light/store/store.go @@ -1,6 +1,8 @@ package store -import "github.com/tendermint/tendermint/types" +import ( + "github.com/tendermint/tendermint/pkg/light" +) // Store is anything that can persistently store headers. type Store interface { @@ -8,7 +10,7 @@ type Store interface { // ValidatorSet (h: sh.Height). // // height must be > 0. - SaveLightBlock(lb *types.LightBlock) error + SaveLightBlock(lb *light.LightBlock) error // DeleteSignedHeaderAndValidatorSet deletes SignedHeader (h: height) and // ValidatorSet (h: height). @@ -22,7 +24,7 @@ type Store interface { // height must be > 0. // // If LightBlock is not found, ErrLightBlockNotFound is returned. - LightBlock(height int64) (*types.LightBlock, error) + LightBlock(height int64) (*light.LightBlock, error) // LastLightBlockHeight returns the last (newest) LightBlock height. // @@ -37,7 +39,7 @@ type Store interface { // LightBlockBefore returns the LightBlock before a certain height. // // height must be > 0 && <= LastLightBlockHeight. - LightBlockBefore(height int64) (*types.LightBlock, error) + LightBlockBefore(height int64) (*light.LightBlock, error) // Prune removes headers & the associated validator sets when Store reaches a // defined size (number of header & validator set pairs). diff --git a/privval/file.go b/privval/file.go index 4ec918c70..27a9169e3 100644 --- a/privval/file.go +++ b/privval/file.go @@ -19,8 +19,8 @@ import ( tmjson "github.com/tendermint/tendermint/libs/json" tmos "github.com/tendermint/tendermint/libs/os" tmtime "github.com/tendermint/tendermint/libs/time" + "github.com/tendermint/tendermint/pkg/consensus" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) // TODO: type ? @@ -47,7 +47,7 @@ func voteToStep(vote *tmproto.Vote) int8 { // FilePVKey stores the immutable part of PrivValidator. type FilePVKey struct { - Address types.Address `json:"address"` + Address crypto.Address `json:"address"` PubKey crypto.PubKey `json:"pub_key"` PrivKey crypto.PrivKey `json:"priv_key"` @@ -154,7 +154,7 @@ type FilePV struct { LastSignState FilePVLastSignState } -var _ types.PrivValidator = (*FilePV)(nil) +var _ consensus.PrivValidator = (*FilePV)(nil) // NewFilePV generates a new validator from the given key and paths. func NewFilePV(privKey crypto.PrivKey, keyFilePath, stateFilePath string) *FilePV { @@ -176,9 +176,9 @@ func NewFilePV(privKey crypto.PrivKey, keyFilePath, stateFilePath string) *FileP // and sets the filePaths, but does not call Save(). func GenFilePV(keyFilePath, stateFilePath, keyType string) (*FilePV, error) { switch keyType { - case types.ABCIPubKeyTypeSecp256k1: + case consensus.ABCIPubKeyTypeSecp256k1: return NewFilePV(secp256k1.GenPrivKey(), keyFilePath, stateFilePath), nil - case "", types.ABCIPubKeyTypeEd25519: + case "", consensus.ABCIPubKeyTypeEd25519: return NewFilePV(ed25519.GenPrivKey(), keyFilePath, stateFilePath), nil default: return nil, fmt.Errorf("key type: %s is not supported", keyType) @@ -254,7 +254,7 @@ func LoadOrGenFilePV(keyFilePath, stateFilePath string) (*FilePV, error) { // GetAddress returns the address of the validator. // Implements PrivValidator. -func (pv *FilePV) GetAddress() types.Address { +func (pv *FilePV) GetAddress() crypto.Address { return pv.Key.Address } @@ -326,7 +326,7 @@ func (pv *FilePV) signVote(chainID string, vote *tmproto.Vote) error { return err } - signBytes := types.VoteSignBytes(chainID, vote) + signBytes := consensus.VoteSignBytes(chainID, vote) // We might crash before writing to the wal, // causing us to try to re-sign for the same HRS. @@ -368,7 +368,7 @@ func (pv *FilePV) signProposal(chainID string, proposal *tmproto.Proposal) error return err } - signBytes := types.ProposalSignBytes(chainID, proposal) + signBytes := consensus.ProposalSignBytes(chainID, proposal) // We might crash before writing to the wal, // causing us to try to re-sign for the same HRS. diff --git a/privval/grpc/client.go b/privval/grpc/client.go index 77f3930aa..17c223b1a 100644 --- a/privval/grpc/client.go +++ b/privval/grpc/client.go @@ -9,9 +9,9 @@ import ( "github.com/tendermint/tendermint/crypto" cryptoenc "github.com/tendermint/tendermint/crypto/encoding" "github.com/tendermint/tendermint/libs/log" + "github.com/tendermint/tendermint/pkg/consensus" privvalproto "github.com/tendermint/tendermint/proto/tendermint/privval" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) // SignerClient implements PrivValidator. @@ -24,7 +24,7 @@ type SignerClient struct { chainID string } -var _ types.PrivValidator = (*SignerClient)(nil) +var _ consensus.PrivValidator = (*SignerClient)(nil) // NewSignerClient returns an instance of SignerClient. // it will start the endpoint (if not already started) diff --git a/privval/grpc/server.go b/privval/grpc/server.go index f5c434b1b..76bd0e07f 100644 --- a/privval/grpc/server.go +++ b/privval/grpc/server.go @@ -9,8 +9,8 @@ import ( "github.com/tendermint/tendermint/crypto" cryptoenc "github.com/tendermint/tendermint/crypto/encoding" "github.com/tendermint/tendermint/libs/log" + "github.com/tendermint/tendermint/pkg/consensus" privvalproto "github.com/tendermint/tendermint/proto/tendermint/privval" - "github.com/tendermint/tendermint/types" ) // SignerServer implements PrivValidatorAPIServer 9generated via protobuf services) @@ -18,11 +18,11 @@ import ( type SignerServer struct { logger log.Logger chainID string - privVal types.PrivValidator + privVal consensus.PrivValidator } func NewSignerServer(chainID string, - privVal types.PrivValidator, log log.Logger) *SignerServer { + privVal consensus.PrivValidator, log log.Logger) *SignerServer { return &SignerServer{ logger: log, diff --git a/privval/retry_signer_client.go b/privval/retry_signer_client.go index ccd9834e4..ebd34078b 100644 --- a/privval/retry_signer_client.go +++ b/privval/retry_signer_client.go @@ -6,8 +6,8 @@ import ( "time" "github.com/tendermint/tendermint/crypto" + "github.com/tendermint/tendermint/pkg/consensus" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) // RetrySignerClient wraps SignerClient adding retry for each operation (except @@ -24,7 +24,7 @@ func NewRetrySignerClient(sc *SignerClient, retries int, timeout time.Duration) return &RetrySignerClient{sc, retries, timeout} } -var _ types.PrivValidator = (*RetrySignerClient)(nil) +var _ consensus.PrivValidator = (*RetrySignerClient)(nil) func (sc *RetrySignerClient) Close() error { return sc.next.Close() diff --git a/privval/signer_client.go b/privval/signer_client.go index d25584c8f..2b22be202 100644 --- a/privval/signer_client.go +++ b/privval/signer_client.go @@ -7,9 +7,9 @@ import ( "github.com/tendermint/tendermint/crypto" cryptoenc "github.com/tendermint/tendermint/crypto/encoding" + "github.com/tendermint/tendermint/pkg/consensus" privvalproto "github.com/tendermint/tendermint/proto/tendermint/privval" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) // SignerClient implements PrivValidator. @@ -19,7 +19,7 @@ type SignerClient struct { chainID string } -var _ types.PrivValidator = (*SignerClient)(nil) +var _ consensus.PrivValidator = (*SignerClient)(nil) // NewSignerClient returns an instance of SignerClient. // it will start the endpoint (if not already started) diff --git a/privval/signer_requestHandler.go b/privval/signer_requestHandler.go index 18ad8a996..d900be297 100644 --- a/privval/signer_requestHandler.go +++ b/privval/signer_requestHandler.go @@ -6,15 +6,15 @@ import ( "github.com/tendermint/tendermint/crypto" cryptoenc "github.com/tendermint/tendermint/crypto/encoding" + "github.com/tendermint/tendermint/pkg/consensus" cryptoproto "github.com/tendermint/tendermint/proto/tendermint/crypto" privvalproto "github.com/tendermint/tendermint/proto/tendermint/privval" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) func DefaultValidationRequestHandler( ctx context.Context, - privVal types.PrivValidator, + privVal consensus.PrivValidator, req privvalproto.Message, chainID string, ) (privvalproto.Message, error) { diff --git a/privval/signer_server.go b/privval/signer_server.go index 24bf67cc5..96d64c83f 100644 --- a/privval/signer_server.go +++ b/privval/signer_server.go @@ -6,14 +6,14 @@ import ( tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/libs/service" + "github.com/tendermint/tendermint/pkg/consensus" privvalproto "github.com/tendermint/tendermint/proto/tendermint/privval" - "github.com/tendermint/tendermint/types" ) // ValidationRequestHandlerFunc handles different remoteSigner requests type ValidationRequestHandlerFunc func( ctx context.Context, - privVal types.PrivValidator, + privVal consensus.PrivValidator, requestMessage privvalproto.Message, chainID string) (privvalproto.Message, error) @@ -22,13 +22,13 @@ type SignerServer struct { endpoint *SignerDialerEndpoint chainID string - privVal types.PrivValidator + privVal consensus.PrivValidator handlerMtx tmsync.Mutex validationRequestHandler ValidationRequestHandlerFunc } -func NewSignerServer(endpoint *SignerDialerEndpoint, chainID string, privVal types.PrivValidator) *SignerServer { +func NewSignerServer(endpoint *SignerDialerEndpoint, chainID string, privVal consensus.PrivValidator) *SignerServer { ss := &SignerServer{ endpoint: endpoint, chainID: chainID, diff --git a/state/indexer/eventsink.go b/state/indexer/eventsink.go index d7c9d10e0..36a624979 100644 --- a/state/indexer/eventsink.go +++ b/state/indexer/eventsink.go @@ -5,7 +5,7 @@ import ( abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/libs/pubsub/query" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/events" ) type EventSinkType string @@ -26,7 +26,7 @@ const ( type EventSink interface { // IndexBlockEvents indexes the blockheader. - IndexBlockEvents(types.EventDataNewBlockHeader) error + IndexBlockEvents(events.EventDataNewBlockHeader) error // IndexTxEvents indexes the given result of transactions. To call it with multi transactions, // must guarantee the index of given transactions are in order. diff --git a/state/indexer/indexer.go b/state/indexer/indexer.go index 24dc62d70..f0665000e 100644 --- a/state/indexer/indexer.go +++ b/state/indexer/indexer.go @@ -6,7 +6,7 @@ import ( abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/libs/pubsub/query" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/events" ) // TxIndexer interface defines methods to index and search transactions. @@ -31,7 +31,7 @@ type BlockIndexer interface { Has(height int64) (bool, error) // Index indexes BeginBlock and EndBlock events for a given block by its height. - Index(types.EventDataNewBlockHeader) error + Index(events.EventDataNewBlockHeader) error // Search performs a query for block heights that match a given BeginBlock // and Endblock event search criteria. diff --git a/state/indexer/indexer_service.go b/state/indexer/indexer_service.go index a429b66a0..042168df7 100644 --- a/state/indexer/indexer_service.go +++ b/state/indexer/indexer_service.go @@ -4,7 +4,7 @@ import ( "context" "github.com/tendermint/tendermint/libs/service" - "github.com/tendermint/tendermint/types" + "github.com/tendermint/tendermint/pkg/events" ) // XXX/TODO: These types should be moved to the indexer package. @@ -19,12 +19,11 @@ type Service struct { service.BaseService eventSinks []EventSink - eventBus *types.EventBus + eventBus *events.EventBus } // NewIndexerService returns a new service instance. -func NewIndexerService(es []EventSink, eventBus *types.EventBus) *Service { - +func NewIndexerService(es []EventSink, eventBus *events.EventBus) *Service { is := &Service{eventSinks: es, eventBus: eventBus} is.BaseService = *service.NewBaseService(nil, "IndexerService", is) return is @@ -39,12 +38,12 @@ func (is *Service) OnStart() error { blockHeadersSub, err := is.eventBus.SubscribeUnbuffered( context.Background(), subscriber, - types.EventQueryNewBlockHeader) + events.EventQueryNewBlockHeader) if err != nil { return err } - txsSub, err := is.eventBus.SubscribeUnbuffered(context.Background(), subscriber, types.EventQueryTx) + txsSub, err := is.eventBus.SubscribeUnbuffered(context.Background(), subscriber, events.EventQueryTx) if err != nil { return err } @@ -53,13 +52,13 @@ func (is *Service) OnStart() error { for { msg := <-blockHeadersSub.Out() - eventDataHeader := msg.Data().(types.EventDataNewBlockHeader) + eventDataHeader := msg.Data().(events.EventDataNewBlockHeader) height := eventDataHeader.Header.Height batch := NewBatch(eventDataHeader.NumTxs) for i := int64(0); i < eventDataHeader.NumTxs; i++ { msg2 := <-txsSub.Out() - txResult := msg2.Data().(types.EventDataTx).TxResult + txResult := msg2.Data().(events.EventDataTx).TxResult if err = batch.Add(&txResult); err != nil { is.Logger.Error( diff --git a/state/indexer/sink/null/null.go b/state/indexer/sink/null/null.go index b5ad93ab4..965a89dbc 100644 --- a/state/indexer/sink/null/null.go +++ b/state/indexer/sink/null/null.go @@ -5,8 +5,8 @@ import ( abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/libs/pubsub/query" + "github.com/tendermint/tendermint/pkg/events" "github.com/tendermint/tendermint/state/indexer" - "github.com/tendermint/tendermint/types" ) var _ indexer.EventSink = (*EventSink)(nil) @@ -22,7 +22,7 @@ func (nes *EventSink) Type() indexer.EventSinkType { return indexer.NULL } -func (nes *EventSink) IndexBlockEvents(bh types.EventDataNewBlockHeader) error { +func (nes *EventSink) IndexBlockEvents(bh events.EventDataNewBlockHeader) error { return nil } diff --git a/state/indexer/tx/kv/kv.go b/state/indexer/tx/kv/kv.go index 080dbce2c..019d857c1 100644 --- a/state/indexer/tx/kv/kv.go +++ b/state/indexer/tx/kv/kv.go @@ -13,6 +13,7 @@ import ( abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/libs/pubsub/query" + "github.com/tendermint/tendermint/pkg/mempool" indexer "github.com/tendermint/tendermint/state/indexer" "github.com/tendermint/tendermint/types" ) @@ -67,7 +68,7 @@ func (txi *TxIndex) Index(results []*abci.TxResult) error { defer b.Close() for _, result := range results { - hash := types.Tx(result.Tx).Hash() + hash := mempool.Tx(result.Tx).Hash() // index tx by events err := txi.indexEvents(result, hash, b) diff --git a/store/store.go b/store/store.go index 8848b76d9..719860467 100644 --- a/store/store.go +++ b/store/store.go @@ -9,8 +9,10 @@ import ( "github.com/google/orderedcode" dbm "github.com/tendermint/tm-db" + "github.com/tendermint/tendermint/pkg/block" + "github.com/tendermint/tendermint/pkg/light" + "github.com/tendermint/tendermint/pkg/meta" tmproto "github.com/tendermint/tendermint/proto/tendermint/types" - "github.com/tendermint/tendermint/types" ) /* @@ -98,7 +100,7 @@ func (bs *BlockStore) Size() int64 { } // LoadBase atomically loads the base block meta, or returns nil if no base is found. -func (bs *BlockStore) LoadBaseMeta() *types.BlockMeta { +func (bs *BlockStore) LoadBaseMeta() *block.BlockMeta { iter, err := bs.db.Iterator( blockMetaKey(1), blockMetaKey(1<<63-1), @@ -115,7 +117,7 @@ func (bs *BlockStore) LoadBaseMeta() *types.BlockMeta { panic(fmt.Errorf("unmarshal to tmproto.BlockMeta: %w", err)) } - blockMeta, err := types.BlockMetaFromProto(pbbm) + blockMeta, err := block.BlockMetaFromProto(pbbm) if err != nil { panic(fmt.Errorf("error from proto blockMeta: %w", err)) } @@ -128,7 +130,7 @@ func (bs *BlockStore) LoadBaseMeta() *types.BlockMeta { // LoadBlock returns the block with the given height. // If no block is found for that height, it returns nil. -func (bs *BlockStore) LoadBlock(height int64) *types.Block { +func (bs *BlockStore) LoadBlock(height int64) *block.Block { var blockMeta = bs.LoadBlockMeta(height) if blockMeta == nil { return nil @@ -152,7 +154,7 @@ func (bs *BlockStore) LoadBlock(height int64) *types.Block { panic(fmt.Sprintf("Error reading block: %v", err)) } - block, err := types.BlockFromProto(pbb) + block, err := block.BlockFromProto(pbb) if err != nil { panic(fmt.Errorf("error from proto block: %w", err)) } @@ -163,7 +165,7 @@ func (bs *BlockStore) LoadBlock(height int64) *types.Block { // LoadBlockByHash returns the block with the given hash. // If no block is found for that hash, it returns nil. // Panics if it fails to parse height associated with the given hash. -func (bs *BlockStore) LoadBlockByHash(hash []byte) *types.Block { +func (bs *BlockStore) LoadBlockByHash(hash []byte) *block.Block { bz, err := bs.db.Get(blockHashKey(hash)) if err != nil { panic(err) @@ -184,7 +186,7 @@ func (bs *BlockStore) LoadBlockByHash(hash []byte) *types.Block { // LoadBlockPart returns the Part at the given index // from the block at the given height. // If no part is found for the given height and index, it returns nil. -func (bs *BlockStore) LoadBlockPart(height int64, index int) *types.Part { +func (bs *BlockStore) LoadBlockPart(height int64, index int) *meta.Part { var pbpart = new(tmproto.Part) bz, err := bs.db.Get(blockPartKey(height, index)) @@ -199,7 +201,7 @@ func (bs *BlockStore) LoadBlockPart(height int64, index int) *types.Part { if err != nil { panic(fmt.Errorf("unmarshal to tmproto.Part failed: %w", err)) } - part, err := types.PartFromProto(pbpart) + part, err := meta.PartFromProto(pbpart) if err != nil { panic(fmt.Sprintf("Error reading block part: %v", err)) } @@ -209,7 +211,7 @@ func (bs *BlockStore) LoadBlockPart(height int64, index int) *types.Part { // LoadBlockMeta returns the BlockMeta for the given height. // If no block is found for the given height, it returns nil. -func (bs *BlockStore) LoadBlockMeta(height int64) *types.BlockMeta { +func (bs *BlockStore) LoadBlockMeta(height int64) *block.BlockMeta { var pbbm = new(tmproto.BlockMeta) bz, err := bs.db.Get(blockMetaKey(height)) @@ -226,7 +228,7 @@ func (bs *BlockStore) LoadBlockMeta(height int64) *types.BlockMeta { panic(fmt.Errorf("unmarshal to tmproto.BlockMeta: %w", err)) } - blockMeta, err := types.BlockMetaFromProto(pbbm) + blockMeta, err := block.BlockMetaFromProto(pbbm) if err != nil { panic(fmt.Errorf("error from proto blockMeta: %w", err)) } @@ -238,7 +240,7 @@ func (bs *BlockStore) LoadBlockMeta(height int64) *types.BlockMeta { // This commit consists of the +2/3 and other Precommit-votes for block at `height`, // and it comes from the block.LastCommit for `height+1`. // If no commit is found for the given height, it returns nil. -func (bs *BlockStore) LoadBlockCommit(height int64) *types.Commit { +func (bs *BlockStore) LoadBlockCommit(height int64) *meta.Commit { var pbc = new(tmproto.Commit) bz, err := bs.db.Get(blockCommitKey(height)) if err != nil { @@ -251,7 +253,7 @@ func (bs *BlockStore) LoadBlockCommit(height int64) *types.Commit { if err != nil { panic(fmt.Errorf("error reading block commit: %w", err)) } - commit, err := types.CommitFromProto(pbc) + commit, err := meta.CommitFromProto(pbc) if err != nil { panic(fmt.Sprintf("Error reading block commit: %v", err)) } @@ -262,7 +264,7 @@ func (bs *BlockStore) LoadBlockCommit(height int64) *types.Commit { // cannonicalized. This is useful when we've seen a commit, but there // has not yet been a new block at `height + 1` that includes this // commit in its block.LastCommit. -func (bs *BlockStore) LoadSeenCommit() *types.Commit { +func (bs *BlockStore) LoadSeenCommit() *meta.Commit { var pbc = new(tmproto.Commit) bz, err := bs.db.Get(seenCommitKey()) if err != nil { @@ -276,7 +278,7 @@ func (bs *BlockStore) LoadSeenCommit() *types.Commit { panic(fmt.Sprintf("error reading block seen commit: %v", err)) } - commit, err := types.CommitFromProto(pbc) + commit, err := meta.CommitFromProto(pbc) if err != nil { panic(fmt.Errorf("error from proto commit: %w", err)) } @@ -302,7 +304,7 @@ func (bs *BlockStore) PruneBlocks(height int64) (uint64, error) { return fmt.Errorf("unmarshal to tmproto.BlockMeta: %w", err) } - blockMeta, err := types.BlockMetaFromProto(pbbm) + blockMeta, err := block.BlockMetaFromProto(pbbm) if err != nil { return fmt.Errorf("error from proto blockMeta: %w", err) } @@ -426,15 +428,15 @@ func (bs *BlockStore) batchDelete( // If all the nodes restart after committing a block, // we need this to reload the precommits to catch-up nodes to the // most recent height. Otherwise they'd stall at H-1. -func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) { - if block == nil { +func (bs *BlockStore) SaveBlock(blk *block.Block, blockParts *meta.PartSet, seenCommit *meta.Commit) { + if blk == nil { panic("BlockStore can only save a non-nil block") } batch := bs.db.NewBatch() - height := block.Height - hash := block.Hash() + height := blk.Height + hash := blk.Hash() if g, w := height, bs.Height()+1; bs.Base() > 0 && g != w { panic(fmt.Sprintf("BlockStore can only save contiguous blocks. Wanted %v, got %v", w, g)) @@ -452,7 +454,7 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s bs.saveBlockPart(height, i, part, batch) } - blockMeta := types.NewBlockMeta(block, blockParts) + blockMeta := block.NewBlockMeta(blk, blockParts) pbm := blockMeta.ToProto() if pbm == nil { panic("nil blockmeta") @@ -467,7 +469,7 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s panic(err) } - pbc := block.LastCommit.ToProto() + pbc := blk.LastCommit.ToProto() blockCommitBytes := mustEncode(pbc) if err := batch.Set(blockCommitKey(height-1), blockCommitBytes); err != nil { panic(err) @@ -489,7 +491,7 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s } } -func (bs *BlockStore) saveBlockPart(height int64, index int, part *types.Part, batch dbm.Batch) { +func (bs *BlockStore) saveBlockPart(height int64, index int, part *meta.Part, batch dbm.Batch) { pbp, err := part.ToProto() if err != nil { panic(fmt.Errorf("unable to make part into proto: %w", err)) @@ -501,7 +503,7 @@ func (bs *BlockStore) saveBlockPart(height int64, index int, part *types.Part, b } // SaveSeenCommit saves a seen commit, used by e.g. the state sync reactor when bootstrapping node. -func (bs *BlockStore) SaveSeenCommit(height int64, seenCommit *types.Commit) error { +func (bs *BlockStore) SaveSeenCommit(height int64, seenCommit *meta.Commit) error { pbc := seenCommit.ToProto() seenCommitBytes, err := proto.Marshal(pbc) if err != nil { @@ -510,7 +512,7 @@ func (bs *BlockStore) SaveSeenCommit(height int64, seenCommit *types.Commit) err return bs.db.Set(seenCommitKey(), seenCommitBytes) } -func (bs *BlockStore) SaveSignedHeader(sh *types.SignedHeader, blockID types.BlockID) error { +func (bs *BlockStore) SaveSignedHeader(sh *light.SignedHeader, blockID meta.BlockID) error { // first check that the block store doesn't already have the block bz, err := bs.db.Get(blockMetaKey(sh.Height)) if err != nil { @@ -524,7 +526,7 @@ func (bs *BlockStore) SaveSignedHeader(sh *types.SignedHeader, blockID types.Blo // doesn't have complete parity with block meta's thus block size and num // txs are filled with negative numbers. We should aim to find a solution to // this. - blockMeta := &types.BlockMeta{ + blockMeta := &block.BlockMeta{ BlockID: blockID, BlockSize: -1, Header: *sh.Header,