From bf1cb89bb7c34d75e9aade09f7304fd376ce084b Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Tue, 14 Jun 2022 16:55:10 -0400 Subject: [PATCH 1/2] Revert "p2p: self-add node should not error (tendermint#8753)" (#8757) --- internal/p2p/peermanager.go | 12 +----------- internal/p2p/peermanager_test.go | 14 ++++++-------- node/node.go | 2 +- node/seed.go | 2 +- node/setup.go | 2 -- 5 files changed, 9 insertions(+), 23 deletions(-) diff --git a/internal/p2p/peermanager.go b/internal/p2p/peermanager.go index 65741f63f..7391de4ea 100644 --- a/internal/p2p/peermanager.go +++ b/internal/p2p/peermanager.go @@ -15,7 +15,6 @@ import ( dbm "github.com/tendermint/tm-db" tmsync "github.com/tendermint/tendermint/internal/libs/sync" - "github.com/tendermint/tendermint/libs/log" p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p" "github.com/tendermint/tendermint/types" ) @@ -146,8 +145,6 @@ type PeerManagerOptions struct { // persistentPeers provides fast PersistentPeers lookups. It is built // by optimize(). persistentPeers map[types.NodeID]bool - - Logger log.Logger } // Validate validates the options. @@ -267,7 +264,6 @@ type PeerManager struct { rand *rand.Rand dialWaker *tmsync.Waker // wakes up DialNext() on relevant peer changes evictWaker *tmsync.Waker // wakes up EvictNext() on relevant peer changes - logger log.Logger mtx sync.Mutex store *peerStore @@ -302,7 +298,6 @@ func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptio rand: rand.New(rand.NewSource(time.Now().UnixNano())), // nolint:gosec dialWaker: tmsync.NewWaker(), evictWaker: tmsync.NewWaker(), - logger: log.NewNopLogger(), store: store, dialing: map[types.NodeID]bool{}, @@ -313,11 +308,6 @@ func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptio evicting: map[types.NodeID]bool{}, subscriptions: map[*PeerUpdates]*PeerUpdates{}, } - - if options.Logger != nil { - peerManager.logger = options.Logger - } - if err = peerManager.configurePeers(); err != nil { return nil, err } @@ -400,7 +390,7 @@ func (m *PeerManager) Add(address NodeAddress) (bool, error) { return false, err } if address.NodeID == m.selfID { - return false, nil + return false, fmt.Errorf("can't add self (%v) to peer store", m.selfID) } m.mtx.Lock() diff --git a/internal/p2p/peermanager_test.go b/internal/p2p/peermanager_test.go index bb79fe771..47e8462a4 100644 --- a/internal/p2p/peermanager_test.go +++ b/internal/p2p/peermanager_test.go @@ -265,9 +265,8 @@ func TestPeerManager_Add(t *testing.T) { require.Error(t, err) // Adding self should error - ok, err := peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID}) - require.False(t, ok) - require.NoError(t, err) + _, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID}) + require.Error(t, err) } func TestPeerManager_DialNext(t *testing.T) { @@ -842,14 +841,13 @@ func TestPeerManager_Dialed_Connected(t *testing.T) { require.Error(t, peerManager.Dialed(b)) } -func TestPeerManager_Adding_Self(t *testing.T) { +func TestPeerManager_Dialed_Self(t *testing.T) { peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{}) require.NoError(t, err) - // Ingesting self should not error. - ok, err := peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID}) - require.False(t, ok) - require.NoError(t, err) + // Dialing self should error. + _, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID}) + require.Error(t, err) } func TestPeerManager_Dialed_MaxConnected(t *testing.T) { diff --git a/node/node.go b/node/node.go index f2c4cd6a8..1bda1f0f7 100644 --- a/node/node.go +++ b/node/node.go @@ -203,7 +203,7 @@ func makeNode( } } - peerManager, peerCloser, err := createPeerManager(logger, cfg, dbProvider, nodeKey.ID) + peerManager, peerCloser, err := createPeerManager(cfg, dbProvider, nodeKey.ID) closers = append(closers, peerCloser) if err != nil { return nil, combineCloseError( diff --git a/node/seed.go b/node/seed.go index 3ba4d86d4..a0b71e411 100644 --- a/node/seed.go +++ b/node/seed.go @@ -67,7 +67,7 @@ func makeSeedNode( // Setup Transport and Switch. p2pMetrics := p2p.PrometheusMetrics(cfg.Instrumentation.Namespace, "chain_id", genDoc.ChainID) - peerManager, closer, err := createPeerManager(logger, cfg, dbProvider, nodeKey.ID) + peerManager, closer, err := createPeerManager(cfg, dbProvider, nodeKey.ID) if err != nil { return nil, combineCloseError( fmt.Errorf("failed to create peer manager: %w", err), diff --git a/node/setup.go b/node/setup.go index 3240ba0e7..51a048249 100644 --- a/node/setup.go +++ b/node/setup.go @@ -199,7 +199,6 @@ func createEvidenceReactor( } func createPeerManager( - logger log.Logger, cfg *config.Config, dbProvider config.DBProvider, nodeID types.NodeID, @@ -227,7 +226,6 @@ func createPeerManager( maxUpgradeConns := uint16(4) options := p2p.PeerManagerOptions{ - Logger: logger.With("module", "peermanager"), SelfAddress: selfAddr, MaxConnected: maxConns, MaxConnectedUpgrade: maxUpgradeConns, From 979a6a1b13331ae90efb1c03b9dc69bc227d1909 Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Tue, 14 Jun 2022 19:12:53 -0400 Subject: [PATCH 2/2] p2p: accept should not abort on first error (#8759) --- internal/p2p/router.go | 17 ++++--- internal/p2p/router_test.go | 96 +++++++++++++------------------------ 2 files changed, 43 insertions(+), 70 deletions(-) diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 8b77541de..511fa0fb9 100644 --- a/internal/p2p/router.go +++ b/internal/p2p/router.go @@ -470,14 +470,17 @@ func (r *Router) dialSleep(ctx context.Context) { func (r *Router) acceptPeers(ctx context.Context, transport Transport) { for { conn, err := transport.Accept(ctx) - switch err { - case nil: - case io.EOF: - r.logger.Debug("stopping accept routine", "transport", transport) + switch { + case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + r.logger.Debug("stopping accept routine", "transport", transport, "err", "context canceled") return - default: + case errors.Is(err, io.EOF): + r.logger.Debug("stopping accept routine", "transport", transport, "err", "EOF") + return + case err != nil: + // in this case we got an error from the net.Listener. r.logger.Error("failed to accept connection", "transport", transport, "err", err) - return + continue } incomingIP := conn.RemoteEndpoint().IP @@ -489,7 +492,7 @@ func (r *Router) acceptPeers(ctx context.Context, transport Transport) { "close_err", closeErr, ) - return + continue } // Spawn a goroutine for the handshake, to avoid head-of-line blocking. diff --git a/internal/p2p/router_test.go b/internal/p2p/router_test.go index 663e6b81c..e0910fb21 100644 --- a/internal/p2p/router_test.go +++ b/internal/p2p/router_test.go @@ -442,78 +442,48 @@ func TestRouter_AcceptPeers(t *testing.T) { } } -func TestRouter_AcceptPeers_Error(t *testing.T) { - t.Cleanup(leaktest.Check(t)) +func TestRouter_AcceptPeers_Errors(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + for _, err := range []error{io.EOF, context.Canceled, context.DeadlineExceeded} { + t.Run(err.Error(), func(t *testing.T) { + t.Cleanup(leaktest.Check(t)) - // Set up a mock transport that returns an error, which should prevent - // the router from calling Accept again. - mockTransport := &mocks.Transport{} - mockTransport.On("String").Maybe().Return("mock") - mockTransport.On("Accept", mock.Anything).Once().Return(nil, errors.New("boom")) - mockTransport.On("Close").Return(nil) - mockTransport.On("Listen", mock.Anything).Return(nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() - // Set up and start the router. - peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{}) - require.NoError(t, err) + // Set up a mock transport that returns io.EOF once, which should prevent + // the router from calling Accept again. + mockTransport := &mocks.Transport{} + mockTransport.On("String").Maybe().Return("mock") + mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF) + mockTransport.On("Close").Return(nil) + mockTransport.On("Listen", mock.Anything).Return(nil) - router, err := p2p.NewRouter( - log.NewNopLogger(), - p2p.NopMetrics(), - selfKey, - peerManager, - func() *types.NodeInfo { return &selfInfo }, - mockTransport, - nil, - p2p.RouterOptions{}, - ) - require.NoError(t, err) + // Set up and start the router. + peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{}) + require.NoError(t, err) - require.NoError(t, router.Start(ctx)) - time.Sleep(time.Second) - router.Stop() + router, err := p2p.NewRouter( + log.NewNopLogger(), + p2p.NopMetrics(), + selfKey, + peerManager, + func() *types.NodeInfo { return &selfInfo }, + mockTransport, + nil, + p2p.RouterOptions{}, + ) + require.NoError(t, err) - mockTransport.AssertExpectations(t) -} + require.NoError(t, router.Start(ctx)) + time.Sleep(time.Second) + router.Stop() -func TestRouter_AcceptPeers_ErrorEOF(t *testing.T) { - t.Cleanup(leaktest.Check(t)) + mockTransport.AssertExpectations(t) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + }) - // Set up a mock transport that returns io.EOF once, which should prevent - // the router from calling Accept again. - mockTransport := &mocks.Transport{} - mockTransport.On("String").Maybe().Return("mock") - mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF) - mockTransport.On("Close").Return(nil) - mockTransport.On("Listen", mock.Anything).Return(nil) - - // Set up and start the router. - peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{}) - require.NoError(t, err) - - router, err := p2p.NewRouter( - log.NewNopLogger(), - p2p.NopMetrics(), - selfKey, - peerManager, - func() *types.NodeInfo { return &selfInfo }, - mockTransport, - nil, - p2p.RouterOptions{}, - ) - require.NoError(t, err) - - require.NoError(t, router.Start(ctx)) - time.Sleep(time.Second) - router.Stop() - - mockTransport.AssertExpectations(t) + } } func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {