diff --git a/node/defaults.go b/node/defaults.go index 34b31f63ab2..4210c7a2247 100644 --- a/node/defaults.go +++ b/node/defaults.go @@ -46,7 +46,7 @@ var DefaultConfig = Config{ ListenAddr: ":30303", ListenAddr65: ":30304", MaxPeers: 100, - MaxPendingPeers: 50, + MaxPendingPeers: 1000, NAT: nat.Any(), }, } diff --git a/p2p/server.go b/p2p/server.go index 595052fc06b..b7b152b8547 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -786,13 +786,12 @@ running: // Ensure that the trusted flag is set before checking against MaxPeers. c.flags |= trustedConn } - // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them. - c.cont <- srv.postHandshakeChecks(peers, inboundCount, c) + c.cont <- nil case c := <-srv.checkpointAddPeer: // At this point the connection is past the protocol handshake. // Its capabilities are known and the remote identity is verified. - err := srv.addPeerChecks(peers, inboundCount, c) + err := srv.postHandshakeChecks(peers, inboundCount, c) if err == nil { // The handshakes are done and it passed all checks. p := srv.launchPeer(c, c.pubkey) @@ -850,21 +849,13 @@ func (srv *Server) postHandshakeChecks(peers map[enode.ID]*Peer, inboundCount in return DiscAlreadyConnected case c.node.ID() == srv.localnode.ID(): return DiscSelf + case (len(srv.Protocols) > 0) && (countMatchingProtocols(srv.Protocols, c.caps) == 0): + return DiscUselessPeer default: return nil } } -func (srv *Server) addPeerChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error { - // Drop connections with no matching protocols. - if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 { - return DiscUselessPeer - } - // Repeat the post-handshake checks because the - // peer set might have changed since those checks were performed. - return srv.postHandshakeChecks(peers, inboundCount, c) -} - // listenLoop runs in its own goroutine and accepts // inbound connections. func (srv *Server) listenLoop(ctx context.Context) { diff --git a/p2p/server_test.go b/p2p/server_test.go index ad4a92f5c24..40ef5c2c631 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -274,22 +274,30 @@ func TestServerAtCap(t *testing.T) { } // Inject a few connections to fill up the peer set. - for i := 0; i < 10; i++ { + for i := 0; i < srv.Config.MaxPeers; i++ { c := newconn(randomID()) if err := srv.checkpoint(c, srv.checkpointAddPeer); err != nil { t.Fatalf("could not add conn %d: %v", i, err) } } + // Try inserting a non-trusted connection. anotherID := randomID() c := newconn(anotherID) - if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != DiscTooManyPeers { + if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != nil { + t.Error("unexpected error @ checkpointPostHandshake:", err) + } + if err := srv.checkpoint(c, srv.checkpointAddPeer); err != DiscTooManyPeers { t.Error("wrong error for insert:", err) } + // Try inserting a trusted connection. c = newconn(trustedID) if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != nil { - t.Error("unexpected error for trusted conn @posthandshake:", err) + t.Error("unexpected error @ checkpointPostHandshake:", err) + } + if err := srv.checkpoint(c, srv.checkpointAddPeer); err != nil { + t.Error("unexpected error for trusted conn:", err) } if !c.is(trustedConn) { t.Error("Server did not set trusted flag") @@ -298,7 +306,10 @@ func TestServerAtCap(t *testing.T) { // Remove from trusted set and try again srv.RemoveTrustedPeer(newNode(trustedID, "")) c = newconn(trustedID) - if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != DiscTooManyPeers { + if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != nil { + t.Error("unexpected error @ checkpointPostHandshake:", err) + } + if err := srv.checkpoint(c, srv.checkpointAddPeer); err != DiscTooManyPeers { t.Error("wrong error for insert:", err) } @@ -306,7 +317,10 @@ func TestServerAtCap(t *testing.T) { srv.AddTrustedPeer(newNode(anotherID, "")) c = newconn(anotherID) if err := srv.checkpoint(c, srv.checkpointPostHandshake); err != nil { - t.Error("unexpected error for trusted conn @posthandshake:", err) + t.Error("unexpected error @ checkpointPostHandshake:", err) + } + if err := srv.checkpoint(c, srv.checkpointAddPeer); err != nil { + t.Error("unexpected error for trusted conn:", err) } if !c.is(trustedConn) { t.Error("Server did not set trusted flag") @@ -322,8 +336,7 @@ func TestServerPeerLimits(t *testing.T) { pubkey: &clientkey.PublicKey, phs: protoHandshake{ Pubkey: crypto.MarshalPubkey(&clientkey.PublicKey), - // Force "DiscUselessPeer" due to unmatching caps - // Caps: []Cap{discard.cap()}, + Caps: []Cap{discard.cap()}, }, } @@ -348,35 +361,31 @@ func TestServerPeerLimits(t *testing.T) { flags := dynDialedConn dialDest := clientnode conn, _ := net.Pipe() - srv.SetupConn(conn, flags, dialDest) - if tp.closeErr != DiscTooManyPeers { - t.Errorf("unexpected close error: %q", tp.closeErr) + err := srv.SetupConn(conn, flags, dialDest) + _ = conn.Close() + if !errors.Is(err, DiscTooManyPeers) { + t.Fatalf("expected DiscTooManyPeers, but got error: %q", err) } - conn.Close() srv.AddTrustedPeer(clientnode) // Check that server allows a trusted peer despite being full. conn, _ = net.Pipe() - srv.SetupConn(conn, flags, dialDest) - if tp.closeErr == DiscTooManyPeers { - t.Errorf("failed to bypass MaxPeers with trusted node: %q", tp.closeErr) + err = srv.SetupConn(conn, flags, dialDest) + _ = conn.Close() + if err != nil { + t.Fatalf("failed to bypass MaxPeers with trusted node: %q", err) } - if tp.closeErr != DiscUselessPeer { - t.Errorf("unexpected close error: %q", tp.closeErr) - } - conn.Close() - srv.RemoveTrustedPeer(clientnode) // Check that server is full again. conn, _ = net.Pipe() - srv.SetupConn(conn, flags, dialDest) - if tp.closeErr != DiscTooManyPeers { - t.Errorf("unexpected close error: %q", tp.closeErr) + err = srv.SetupConn(conn, flags, dialDest) + _ = conn.Close() + if !errors.Is(err, DiscTooManyPeers) { + t.Fatalf("expected DiscTooManyPeers, but got error: %q", err) } - conn.Close() } func TestServerSetupConn(t *testing.T) { @@ -423,7 +432,7 @@ func TestServerSetupConn(t *testing.T) { { tt: &setupTransport{pubkey: srvpub, phs: protoHandshake{Pubkey: crypto.MarshalPubkey(srvpub)}}, flags: inboundConn, - wantCalls: "doEncHandshake,close,", + wantCalls: "doEncHandshake,doProtoHandshake,close,", wantCloseErr: DiscSelf, }, { @@ -490,6 +499,7 @@ func (c *setupTransport) doProtoHandshake(our *protoHandshake) (*protoHandshake, } return &c.phs, nil } + func (c *setupTransport) close(err error) { c.calls += "close," c.closeErr = err @@ -497,10 +507,11 @@ func (c *setupTransport) close(err error) { // setupConn shouldn't write to/read from the connection. func (c *setupTransport) WriteMsg(Msg) error { - panic("WriteMsg called on setupTransport") + return errors.New("WriteMsg called on setupTransport") } + func (c *setupTransport) ReadMsg() (Msg, error) { - panic("ReadMsg called on setupTransport") + return Msg{}, errors.New("ReadMsg called on setupTransport") } func newkey() *ecdsa.PrivateKey {