feat: updated dial scheduler to track pending inbound connections

Author: the-shadow-coder

p2p/dial.go

@@ -76,6 +76,7 @@ var (
 	errSelf             = errors.New("is self")
 	errAlreadyDialing   = errors.New("already dialing")
 	errAlreadyConnected = errors.New("already connected")
+	errPendingInbound   = errors.New("peer has pending inbound connection")
 	errRecentlyDialed   = errors.New("recently dialed")
 	errNetRestrict      = errors.New("not contained in netrestrict list")
 	errNoPort           = errors.New("node does not provide TCP port")
@@ -104,12 +105,15 @@ type dialScheduler struct {
 	remStaticCh   chan *enode.Node
 	addPeerCh     chan *conn
 	remPeerCh     chan *conn
+	addPendingCh  chan enode.ID
+	remPendingCh  chan enode.ID
 
 	// Everything below here belongs to loop and
 	// should only be accessed by code on the loop goroutine.
-	dialing   map[enode.ID]*dialTask // active tasks
-	peers     map[enode.ID]struct{}  // all connected peers
-	dialPeers int                    // current number of dialed peers
+	dialing        map[enode.ID]*dialTask // active tasks
+	peers          map[enode.ID]struct{}  // all connected peers
+	pendingInbound map[enode.ID]struct{}  // in-progress inbound connections
+	dialPeers      int                    // current number of dialed peers
 
 	// The static map tracks all static dial tasks. The subset of usable static dial tasks
 	// (i.e. those passing checkDial) is kept in staticPool. The scheduler prefers
@@ -163,19 +167,22 @@ func (cfg dialConfig) withDefaults() dialConfig {
 func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupFunc) *dialScheduler {
 	cfg := config.withDefaults()
 	d := &dialScheduler{
-		dialConfig:    cfg,
-		historyTimer:  mclock.NewAlarm(cfg.clock),
-		setupFunc:     setupFunc,
-		dnsLookupFunc: net.DefaultResolver.LookupNetIP,
-		dialing:       make(map[enode.ID]*dialTask),
-		static:        make(map[enode.ID]*dialTask),
-		peers:         make(map[enode.ID]struct{}),
-		doneCh:        make(chan *dialTask),
-		nodesIn:       make(chan *enode.Node),
-		addStaticCh:   make(chan *enode.Node),
-		remStaticCh:   make(chan *enode.Node),
-		addPeerCh:     make(chan *conn),
-		remPeerCh:     make(chan *conn),
+		dialConfig:     cfg,
+		historyTimer:   mclock.NewAlarm(cfg.clock),
+		setupFunc:      setupFunc,
+		dnsLookupFunc:  net.DefaultResolver.LookupNetIP,
+		dialing:        make(map[enode.ID]*dialTask),
+		static:         make(map[enode.ID]*dialTask),
+		peers:          make(map[enode.ID]struct{}),
+		pendingInbound: make(map[enode.ID]struct{}),
+		doneCh:         make(chan *dialTask),
+		nodesIn:        make(chan *enode.Node),
+		addStaticCh:    make(chan *enode.Node),
+		remStaticCh:    make(chan *enode.Node),
+		addPeerCh:      make(chan *conn),
+		remPeerCh:      make(chan *conn),
+		addPendingCh:   make(chan enode.ID),
+		remPendingCh:   make(chan enode.ID),
 	}
 	d.lastStatsLog = d.clock.Now()
 	d.ctx, d.cancel = context.WithCancel(context.Background())
@@ -223,6 +230,22 @@ func (d *dialScheduler) peerRemoved(c *conn) {
 	}
 }
 
+// inboundPending notifies the scheduler about a pending inbound connection.
+func (d *dialScheduler) inboundPending(id enode.ID) {
+	select {
+	case d.addPendingCh <- id:
+	case <-d.ctx.Done():
+	}
+}
+
+// inboundCompleted notifies the scheduler that an inbound connection completed or failed.
+func (d *dialScheduler) inboundCompleted(id enode.ID) {
+	select {
+	case d.remPendingCh <- id:
+	case <-d.ctx.Done():
+	}
+}
+
 // loop is the main loop of the dialer.
 func (d *dialScheduler) loop(it enode.Iterator) {
 	var (
@@ -276,6 +299,15 @@ loop:
 			delete(d.peers, c.node.ID())
 			d.updateStaticPool(c.node.ID())
 
+		case id := <-d.addPendingCh:
+			d.pendingInbound[id] = struct{}{}
+			d.log.Trace("Marked node as pending inbound", "id", id)
+
+		case id := <-d.remPendingCh:
+			delete(d.pendingInbound, id)
+			d.updateStaticPool(id)
+			d.log.Trace("Unmarked node as pending inbound", "id", id)
+
 		case node := <-d.addStaticCh:
 			id := node.ID()
 			_, exists := d.static[id]
@@ -390,6 +422,9 @@ func (d *dialScheduler) checkDial(n *enode.Node) error {
 	if _, ok := d.peers[n.ID()]; ok {
 		return errAlreadyConnected
 	}
+	if _, ok := d.pendingInbound[n.ID()]; ok {
+		return errPendingInbound
+	}
 	if d.netRestrict != nil && !d.netRestrict.ContainsAddr(n.IPAddr()) {
 		return errNetRestrict
 	}

p2p/server.go

@@ -708,12 +708,14 @@ running:
 				}
 
 				pendingInbound[nodeID] = time.Now()
+				srv.dialsched.inboundPending(nodeID)
 				srv.log.Trace("Tracking pending inbound connection", "id", nodeID, "pending_count", len(pendingInbound))
 			}
 
 			err := srv.postHandshakeChecks(peers, inboundCount, c)
 			if err != nil && c.flags&inboundConn != 0 && c.flags&trustedConn == 0 {
 				delete(pendingInbound, nodeID)
+				srv.dialsched.inboundCompleted(nodeID)
 				srv.log.Trace("Removed failed pending inbound connection", "id", nodeID, "err", err)
 			}
 			c.cont <- err
@@ -732,6 +734,7 @@ running:
 					if startTime, exists := pendingInbound[nodeID]; exists {
 						duration := time.Since(startTime)
 						delete(pendingInbound, nodeID)
+						srv.dialsched.inboundCompleted(nodeID)
 						srv.log.Trace("Promoted pending inbound to peer", "id", nodeID,
 							"handshake_duration", duration, "pending_count", len(pendingInbound))
 					}
@@ -755,6 +758,7 @@ running:
 				// Failed to add peer. Clean up pending tracking if it was inbound.
 				if c.flags&inboundConn != 0 {
 					delete(pendingInbound, nodeID)
+					srv.dialsched.inboundCompleted(nodeID)
 					srv.log.Trace("Removed failed pending inbound at add peer stage",
 						"id", nodeID, "err", err)
 				}
@@ -771,6 +775,7 @@ running:
 			// Remove from pending tracking if present (defensive cleanup).
 			if _, exists := pendingInbound[nodeID]; exists {
 				delete(pendingInbound, nodeID)
+				srv.dialsched.inboundCompleted(nodeID)
 				srv.log.Trace("Cleaned up pending entry on peer deletion", "id", nodeID)
 			}