feat: clean up p2p & implement missing peering functionality

gno.land/cmd/gnoland/secrets_init.go

@@ -201,9 +201,5 @@ func generateLastSignValidatorState() *privval.FilePVLastSignState {
 
 // generateNodeKey generates the p2p node key
 func generateNodeKey() *p2p.NodeKey {
-	privKey := ed25519.GenPrivKey()
-
-	return &p2p.NodeKey{
-		PrivKey: privKey,
-	}
+	return p2p.GenerateNodeKey()
 }

go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/rogpeppe/go-internal v1.12.0
 	github.com/rs/cors v1.11.1
 	github.com/rs/xid v1.6.0
+	github.com/sig-0/insertion-queue v0.0.0-20241004125609-6b3ca841346b
 	github.com/stretchr/testify v1.9.0
 	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
 	go.etcd.io/bbolt v1.3.11

tm2/pkg/bft/blockchain/pool.go

@@ -12,7 +12,7 @@
 	"github.com/gnolang/gno/tm2/pkg/bft/types"
 	"github.com/gnolang/gno/tm2/pkg/flow"
 	"github.com/gnolang/gno/tm2/pkg/log"
-	"github.com/gnolang/gno/tm2/pkg/p2p"
+	types2 "github.com/gnolang/gno/tm2/pkg/p2p"
 	"github.com/gnolang/gno/tm2/pkg/service"
 )
 
@@ -69,7 +69,7 @@
 	requesters map[int64]*bpRequester
 	height     int64 // the lowest key in requesters.
 	// peers
-	peers         map[p2p.ID]*bpPeer
+	peers         map[types2.ID]*bpPeer
 	maxPeerHeight int64 // the biggest reported height
 
 	// atomic
@@ -83,7 +83,7 @@
 // requests and errors will be sent to requestsCh and errorsCh accordingly.
 func NewBlockPool(start int64, requestsCh chan<- BlockRequest, errorsCh chan<- peerError) *BlockPool {
 	bp := &BlockPool{
-		peers: make(map[p2p.ID]*bpPeer),
+		peers: make(map[types2.ID]*bpPeer),
 
 		requesters: make(map[int64]*bpRequester),
 		height:     start,
@@ -226,13 +226,13 @@
 // RedoRequest invalidates the block at pool.height,
 // Remove the peer and redo request from others.
 // Returns the ID of the removed peer.
-func (pool *BlockPool) RedoRequest(height int64) p2p.ID {
+func (pool *BlockPool) RedoRequest(height int64) types2.ID {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
 	request := pool.requesters[height]
 	peerID := request.getPeerID()
-	if peerID != p2p.ID("") {
+	if peerID != types2.ID("") {
 		// RemovePeer will redo all requesters associated with this peer.
 		pool.removePeer(peerID)
 	}
@@ -241,7 +241,7 @@
 
 // AddBlock validates that the block comes from the peer it was expected from and calls the requester to store it.
 // TODO: ensure that blocks come in order for each peer.
-func (pool *BlockPool) AddBlock(peerID p2p.ID, block *types.Block, blockSize int) {
+func (pool *BlockPool) AddBlock(peerID types2.ID, block *types.Block, blockSize int) {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
@@ -278,7 +278,7 @@
 }
 
 // SetPeerHeight sets the peer's alleged blockchain height.
-func (pool *BlockPool) SetPeerHeight(peerID p2p.ID, height int64) {
+func (pool *BlockPool) SetPeerHeight(peerID types2.ID, height int64) {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
@@ -298,14 +298,14 @@
 
 // RemovePeer removes the peer with peerID from the pool. If there's no peer
 // with peerID, function is a no-op.
-func (pool *BlockPool) RemovePeer(peerID p2p.ID) {
+func (pool *BlockPool) RemovePeer(peerID types2.ID) {
 	pool.mtx.Lock()
 	defer pool.mtx.Unlock()
 
 	pool.removePeer(peerID)
 }
 
-func (pool *BlockPool) removePeer(peerID p2p.ID) {
+func (pool *BlockPool) removePeer(peerID types2.ID) {
 	for _, requester := range pool.requesters {
 		if requester.getPeerID() == peerID {
 			requester.redo(peerID)
@@ -386,14 +386,14 @@
 	return int64(len(pool.requesters))
 }
 
-func (pool *BlockPool) sendRequest(height int64, peerID p2p.ID) {
+func (pool *BlockPool) sendRequest(height int64, peerID types2.ID) {
 	if !pool.IsRunning() {
 		return
 	}
 	pool.requestsCh <- BlockRequest{height, peerID}
 }
 
-func (pool *BlockPool) sendError(err error, peerID p2p.ID) {
+func (pool *BlockPool) sendError(err error, peerID types2.ID) {
 	if !pool.IsRunning() {
 		return
 	}
@@ -424,7 +424,7 @@
 
 type bpPeer struct {
 	pool        *BlockPool
-	id          p2p.ID
+	id          types2.ID
 	recvMonitor *flow.Monitor
 
 	height     int64
@@ -435,7 +435,7 @@
 	logger *slog.Logger
 }
 
-func newBPPeer(pool *BlockPool, peerID p2p.ID, height int64) *bpPeer {
+func newBPPeer(pool *BlockPool, peerID types2.ID, height int64) *bpPeer {
 	peer := &bpPeer{
 		pool:       pool,
 		id:         peerID,
@@ -499,10 +499,10 @@
 	pool       *BlockPool
 	height     int64
 	gotBlockCh chan struct{}
-	redoCh     chan p2p.ID // redo may send multitime, add peerId to identify repeat
+	redoCh     chan types2.ID // redo may send multitime, add peerId to identify repeat
 
 	mtx    sync.Mutex
-	peerID p2p.ID
+	peerID types2.ID
 	block  *types.Block
 }
 
@@ -511,7 +511,7 @@
 		pool:       pool,
 		height:     height,
 		gotBlockCh: make(chan struct{}, 1),
-		redoCh:     make(chan p2p.ID, 1),
+		redoCh:     make(chan types2.ID, 1),
 
 		peerID: "",
 		block:  nil,
@@ -526,7 +526,7 @@
 }
 
 // Returns true if the peer matches and block doesn't already exist.
-func (bpr *bpRequester) setBlock(block *types.Block, peerID p2p.ID) bool {
+func (bpr *bpRequester) setBlock(block *types.Block, peerID types2.ID) bool {
 	bpr.mtx.Lock()
 	if bpr.block != nil || bpr.peerID != peerID {
 		bpr.mtx.Unlock()
@@ -548,7 +548,7 @@
 	return bpr.block
 }
 
-func (bpr *bpRequester) getPeerID() p2p.ID {
+func (bpr *bpRequester) getPeerID() types2.ID {
 	bpr.mtx.Lock()
 	defer bpr.mtx.Unlock()
 	return bpr.peerID
@@ -570,7 +570,7 @@
 // Tells bpRequester to pick another peer and try again.
 // NOTE: Nonblocking, and does nothing if another redo
 // was already requested.
-func (bpr *bpRequester) redo(peerID p2p.ID) {
+func (bpr *bpRequester) redo(peerID types2.ID) {
 	select {
 	case bpr.redoCh <- peerID:
 	default:
@@ -631,5 +631,5 @@
 // delivering the block
 type BlockRequest struct {
 	Height int64
-	PeerID p2p.ID
+	PeerID types2.ID
 }

tm2/pkg/bft/blockchain/pool_test.go

@@ -5,12 +5,12 @@ import (
 	"testing"
 	"time"
 
+	types2 "github.com/gnolang/gno/tm2/pkg/p2p"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/gnolang/gno/tm2/pkg/bft/types"
 	"github.com/gnolang/gno/tm2/pkg/log"
-	"github.com/gnolang/gno/tm2/pkg/p2p"
 	"github.com/gnolang/gno/tm2/pkg/random"
 )
 
@@ -19,7 +19,7 @@ func init() {
 }
 
 type testPeer struct {
-	id        p2p.ID
+	id        types2.ID
 	height    int64
 	inputChan chan inputData // make sure each peer's data is sequential
 }
@@ -47,7 +47,7 @@ func (p testPeer) simulateInput(input inputData) {
 	// input.t.Logf("Added block from peer %v (height: %v)", input.request.PeerID, input.request.Height)
 }
 
-type testPeers map[p2p.ID]testPeer
+type testPeers map[types2.ID]testPeer
 
 func (ps testPeers) start() {
 	for _, v := range ps {
@@ -64,7 +64,7 @@ func (ps testPeers) stop() {
 func makePeers(numPeers int, minHeight, maxHeight int64) testPeers {
 	peers := make(testPeers, numPeers)
 	for i := 0; i < numPeers; i++ {
-		peerID := p2p.ID(random.RandStr(12))
+		peerID := types2.ID(random.RandStr(12))
 		height := minHeight + random.RandInt63n(maxHeight-minHeight)
 		peers[peerID] = testPeer{peerID, height, make(chan inputData, 10)}
 	}
@@ -172,7 +172,7 @@ func TestBlockPoolTimeout(t *testing.T) {
 
 	// Pull from channels
 	counter := 0
-	timedOut := map[p2p.ID]struct{}{}
+	timedOut := map[types2.ID]struct{}{}
 	for {
 		select {
 		case err := <-errorsCh:
@@ -195,7 +195,7 @@ func TestBlockPoolRemovePeer(t *testing.T) {
 
 	peers := make(testPeers, 10)
 	for i := 0; i < 10; i++ {
-		peerID := p2p.ID(fmt.Sprintf("%d", i+1))
+		peerID := types2.ID(fmt.Sprintf("%d", i+1))
 		height := int64(i + 1)
 		peers[peerID] = testPeer{peerID, height, make(chan inputData)}
 	}
@@ -215,10 +215,10 @@ func TestBlockPoolRemovePeer(t *testing.T) {
 	assert.EqualValues(t, 10, pool.MaxPeerHeight())
 
 	// remove not-existing peer
-	assert.NotPanics(t, func() { pool.RemovePeer(p2p.ID("Superman")) })
+	assert.NotPanics(t, func() { pool.RemovePeer(types2.ID("Superman")) })
 
 	// remove peer with biggest height
-	pool.RemovePeer(p2p.ID("10"))
+	pool.RemovePeer(types2.ID("10"))
 	assert.EqualValues(t, 9, pool.MaxPeerHeight())
 
 	// remove all peers

tm2/pkg/bft/blockchain/reactor.go

@@ -37,11 +37,7 @@ const (
 		bcBlockResponseMessageFieldKeySize
 )
 
-type consensusReactor interface {
-	// for when we switch from blockchain reactor and fast sync to
-	// the consensus machine
-	SwitchToConsensus(sm.State, int)
-}
+type SwitchToConsensusFn func(sm.State, int)
 
 type peerError struct {
 	err    error
@@ -66,11 +62,17 @@ type BlockchainReactor struct {
 
 	requestsCh <-chan BlockRequest
 	errorsCh   <-chan peerError
+
+	switchToConsensusFn SwitchToConsensusFn
 }
 
 // NewBlockchainReactor returns new reactor instance.
-func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore,
+func NewBlockchainReactor(
+	state sm.State,
+	blockExec *sm.BlockExecutor,
+	store *store.BlockStore,
 	fastSync bool,
+	switchToConsensusFn SwitchToConsensusFn,
 ) *BlockchainReactor {
 	if state.LastBlockHeight != store.Height() {
 		panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight,
@@ -89,13 +91,14 @@ func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *st
 	)
 
 	bcR := &BlockchainReactor{
-		initialState: state,
-		blockExec:    blockExec,
-		store:        store,
-		pool:         pool,
-		fastSync:     fastSync,
-		requestsCh:   requestsCh,
-		errorsCh:     errorsCh,
+		initialState:        state,
+		blockExec:           blockExec,
+		store:               store,
+		pool:                pool,
+		fastSync:            fastSync,
+		requestsCh:          requestsCh,
+		errorsCh:            errorsCh,
+		switchToConsensusFn: switchToConsensusFn,
 	}
 	bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
 	return bcR
@@ -257,16 +260,13 @@ FOR_LOOP:
 		select {
 		case <-switchToConsensusTicker.C:
 			height, numPending, lenRequesters := bcR.pool.GetStatus()
-			outbound, inbound, _ := bcR.Switch.NumPeers()
-			bcR.Logger.Debug("Consensus ticker", "numPending", numPending, "total", lenRequesters,
-				"outbound", outbound, "inbound", inbound)
+
+			bcR.Logger.Debug("Consensus ticker", "numPending", numPending, "total", lenRequesters)
 			if bcR.pool.IsCaughtUp() {
 				bcR.Logger.Info("Time to switch to consensus reactor!", "height", height)
 				bcR.pool.Stop()
-				conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
-				if ok {
-					conR.SwitchToConsensus(state, blocksSynced)
-				}
+
+				bcR.switchToConsensusFn(state, blocksSynced)
 				// else {
 				// should only happen during testing
 				// }

tm2/pkg/bft/blockchain/reactor_test.go

@@ -110,7 +110,7 @@ func newBlockchainReactor(logger *slog.Logger, genDoc *types.GenesisDoc, privVal
 		blockStore.SaveBlock(thisBlock, thisParts, lastCommit)
 	}
 
-	bcReactor := NewBlockchainReactor(state.Copy(), blockExec, blockStore, fastSync)
+	bcReactor := NewBlockchainReactor(state.Copy(), blockExec, blockStore, fastSync, nil)
 	bcReactor.SetLogger(logger.With("module", "blockchain"))
 
 	return BlockchainReactorPair{bcReactor, proxyApp}