fix: fix SnapSync regression from network layer refactor (#10753)

Fix four issues introduced in #10753 that caused SnapSync to regress
from 25% in 5 minutes to 1% in 15 minutes on Hoodi:

1. Session.Handshake: only disconnect on NodeId mismatch for static/bootnode
   peers (operator-verified identities). Discovered peers accept the new
   identity as before — stale discovery data is common and benign. Also
   fix HandshakeComplete firing after InitiateDisconnect on doomed sessions.

2. PeerManager: stop EnsureAvailableActivePeerSlotAsync from consuming
   SemaphoreSlim signals meant for the main peer update loop. Use
   Task.Delay for polling instead of WaitAsync on the shared semaphore.

3. PeerManager: reduce _sessionLock scope in OnDisconnected and
   OnHandshakeComplete to only guard session bookkeeping. Peer processing
   runs outside the lock to avoid serializing all session events.

4. PeerManager: restore dedicated thread for peer update loop via
   Task.Factory.StartNew(LongRunning) with .Unwrap() — fixes the
   original Task<Task> bug while keeping the loop off the thread pool.

Author: kamilchodola

src/Nethermind/Nethermind.Network.Test/P2P/SessionTests.cs

@@ -427,10 +427,37 @@ public void On_incoming_sessions_can_fill_remote_id_on_handshake()
     }
 
     [Test]
-    public void Outgoing_handshake_with_unexpected_identity_is_rejected()
+    public void Outgoing_handshake_with_unexpected_identity_updates_node_id_for_discovered_peer()
     {
-        DisconnectReason? disconnectReason = null;
         Session session = new(30312, new Node(TestItem.PublicKeyA, "127.0.0.1", 8545), _channel, NullDisconnectsAnalyzer.Instance, LimboLogs.Instance);
+
+        session.Handshake(TestItem.PublicKeyB);
+
+        Assert.That(session.ObsoleteRemoteNodeId, Is.EqualTo(TestItem.PublicKeyA));
+        Assert.That(session.RemoteNodeId, Is.EqualTo(TestItem.PublicKeyB));
+    }
+
+    [Test]
+    public void Outgoing_handshake_with_unexpected_identity_is_rejected_for_static_peer()
+    {
+        DisconnectReason? disconnectReason = null;
+        Node node = new(TestItem.PublicKeyA, "127.0.0.1", 8545) { IsStatic = true };
+        Session session = new(30312, node, _channel, NullDisconnectsAnalyzer.Instance, LimboLogs.Instance);
+        session.Disconnecting += (_, e) => disconnectReason = e.DisconnectReason;
+
+        session.Handshake(TestItem.PublicKeyB);
+
+        Assert.That(session.RemoteNodeId, Is.EqualTo(TestItem.PublicKeyA));
+        session.Init(5, _channelHandlerContext, _packetSender);
+        Assert.That(disconnectReason, Is.EqualTo(DisconnectReason.UnexpectedIdentity));
+    }
+
+    [Test]
+    public void Outgoing_handshake_with_unexpected_identity_is_rejected_for_bootnode()
+    {
+        DisconnectReason? disconnectReason = null;
+        Node node = new(TestItem.PublicKeyA, "127.0.0.1", 8545) { IsBootnode = true };
+        Session session = new(30312, node, _channel, NullDisconnectsAnalyzer.Instance, LimboLogs.Instance);
         session.Disconnecting += (_, e) => disconnectReason = e.DisconnectReason;
 
         session.Handshake(TestItem.PublicKeyB);

src/Nethermind/Nethermind.Network/P2P/Session.cs

@@ -336,18 +336,19 @@ public void Handshake(PublicKey? handshakeRemoteNodeId)
             }
             else if (handshakeRemoteNodeId is not null && RemoteNodeId != handshakeRemoteNodeId)
             {
-                if (Direction == ConnectionDirection.Out)
+                // Static/bootnode peers have operator-verified identities — a mismatch is suspicious.
+                if (Direction == ConnectionDirection.Out && (Node?.IsStatic == true || Node?.IsBootnode == true))
                 {
                     if (_logger.IsDebug) DebugUnexpectedNodeId(handshakeRemoteNodeId);
                     InitiateDisconnect(DisconnectReason.UnexpectedIdentity, $"expected {RemoteNodeId}, received {handshakeRemoteNodeId}");
+                    return;
                 }
-                else
-                {
-                    if (_logger.IsTrace) TraceDifferentNodeId(handshakeRemoteNodeId);
-                    ObsoleteRemoteNodeId = RemoteNodeId;
-                    RemoteNodeId = handshakeRemoteNodeId;
-                    Node = new Node(RemoteNodeId, RemoteHost, RemotePort);
-                }
+
+                // Discovered peers frequently have stale identity data — accept the new identity.
+                if (_logger.IsTrace) TraceDifferentNodeId(handshakeRemoteNodeId);
+                ObsoleteRemoteNodeId = RemoteNodeId;
+                RemoteNodeId = handshakeRemoteNodeId;
+                Node = new Node(RemoteNodeId, RemoteHost, RemotePort);
             }
 
             Metrics.Handshakes++;
@@ -358,7 +359,7 @@ public void Handshake(PublicKey? handshakeRemoteNodeId)
             void TraceHandshakeCalled() => _logger.Trace($"{nameof(Handshake)} called on {this}");
 
             [MethodImpl(MethodImplOptions.NoInlining)]
-            void DebugUnexpectedNodeId(PublicKey remoteNodeId) => _logger.Debug($"Unexpected remote node id in handshake: expected {RemoteNodeId}, received {remoteNodeId}");
+            void DebugUnexpectedNodeId(PublicKey remoteNodeId) => _logger.Debug($"Unexpected remote node id in handshake with static/bootnode peer: expected {RemoteNodeId}, received {remoteNodeId}");
 
             [MethodImpl(MethodImplOptions.NoInlining)]
             void TraceDifferentNodeId(PublicKey remoteNodeId) => _logger.Trace($"Different NodeId received in handshake: old: {RemoteNodeId}, new: {remoteNodeId}");

src/Nethermind/Nethermind.Network/PeerManager.cs

@@ -191,22 +191,23 @@ public void Start()
             SignalPeerUpdateNeeded();
         }
 
-        private async Task RunPeerUpdateLoopAsync()
+        private Task RunPeerUpdateLoopAsync()
         {
-            await Task.Yield();
-
-            try
-            {
-                await RunPeerUpdateLoop();
-            }
-            catch (Exception e) when (e is not OperationCanceledException)
+            return Task.Factory.StartNew(async () =>
             {
-                if (_logger.IsError) _logger.Error("Peer update loop encountered an exception.", e);
-            }
-            catch (OperationCanceledException)
-            {
-                if (_logger.IsDebug) _logger.Debug("Peer update loop stopped.");
-            }
+                try
+                {
+                    await RunPeerUpdateLoop();
+                }
+                catch (Exception e) when (e is not OperationCanceledException)
+                {
+                    if (_logger.IsError) _logger.Error("Peer update loop encountered an exception.", e);
+                }
+                catch (OperationCanceledException)
+                {
+                    if (_logger.IsDebug) _logger.Debug("Peer update loop stopped.");
+                }
+            }, _cancellationTokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap();
         }
 
         public async Task StopAsync()
@@ -507,8 +508,9 @@ private async Task<bool> EnsureAvailableActivePeerSlotAsync()
                                       TimeSpan.FromMilliseconds(_networkConfig.ConnectTimeoutMs);
             while (DateTimeOffset.UtcNow < deadline && (AvailableActivePeersCount - _pending) <= 0)
             {
-                // Wait for a signal or poll every 100ms.
-                await _peerUpdateRequested.WaitAsync(TimeSpan.FromMilliseconds(100), _cancellationTokenSource.Token);
+                // Poll every 100ms. Do not consume _peerUpdateRequested here — consuming the
+                // semaphore signal would starve the main peer update loop (WaitForPeerUpdateRequestAsync).
+                await Task.Delay(TimeSpan.FromMilliseconds(100), _cancellationTokenSource.Token);
             }
 
             return AvailableActivePeersCount - _pending > 0;
@@ -1154,42 +1156,42 @@ private void OnDisconnected(object sender, DisconnectEventArgs e)
                 {
                     return;
                 }
+            }
 
-                if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.Closing);
+            if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.Closing);
 
-                if (session.State != SessionState.Disconnected)
-                {
-                    ThrowInvalidOnDisconnectedState(session);
-                }
+            if (session.State != SessionState.Disconnected)
+            {
+                ThrowInvalidOnDisconnectedState(session);
+            }
 
-                if (_logger.IsTrace) TracePeerDisconnected();
+            if (_logger.IsTrace) TracePeerDisconnected();
 
-                if (session.RemoteNodeId is null)
-                {
-                    // this happens when we have a disconnect on incoming connection before handshake
-                    if (_logger.IsTrace) TraceDisconnectWithoutRemoteNodeId();
-                    return;
-                }
+            if (session.RemoteNodeId is null)
+            {
+                // this happens when we have a disconnect on incoming connection before handshake
+                if (_logger.IsTrace) TraceDisconnectWithoutRemoteNodeId();
+                return;
+            }
 
-                Peer peer = _peerPool.GetOrAdd(session.Node);
-                if (session.Direction == ConnectionDirection.Out)
-                {
-                    peer.IsAwaitingConnection = false;
-                }
+            Peer peer = _peerPool.GetOrAdd(session.Node);
+            if (session.Direction == ConnectionDirection.Out)
+            {
+                peer.IsAwaitingConnection = false;
+            }
 
-                if (_peerPool.ActivePeers.TryGetValue(session.RemoteNodeId, out Peer activePeer))
+            if (_peerPool.ActivePeers.TryGetValue(session.RemoteNodeId, out Peer activePeer))
+            {
+                //we want to update reputation always
+                _stats.ReportDisconnect(session.Node, e.DisconnectType, e.DisconnectReason);
+                if (activePeer.InSession?.SessionId != session.SessionId && activePeer.OutSession?.SessionId != session.SessionId)
                 {
-                    //we want to update reputation always
-                    _stats.ReportDisconnect(session.Node, e.DisconnectType, e.DisconnectReason);
-                    if (activePeer.InSession?.SessionId != session.SessionId && activePeer.OutSession?.SessionId != session.SessionId)
-                    {
-                        if (_logger.IsTrace) TraceIgnoringDifferentSessionDisconnect(activePeer.Node.Id);
-                        return;
-                    }
-
-                    DeactivatePeerIfDisconnected(activePeer, "session disconnected");
-                    SignalPeerUpdateNeeded();
+                    if (_logger.IsTrace) TraceIgnoringDifferentSessionDisconnect(activePeer.Node.Id);
+                    return;
                 }
+
+                DeactivatePeerIfDisconnected(activePeer, "session disconnected");
+                SignalPeerUpdateNeeded();
             }
 
             [MethodImpl(MethodImplOptions.NoInlining)]
@@ -1228,38 +1230,38 @@ private void OnHandshakeComplete(object sender, EventArgs args)
                 {
                     return;
                 }
+            }
 
-                _stats.GetOrAdd(session.Node);
+            _stats.GetOrAdd(session.Node);
 
-                //In case of OUT connections and different RemoteNodeId we need to replace existing Active Peer with new peer
-                ManageNewRemoteNodeId(session);
+            //In case of OUT connections and different RemoteNodeId we need to replace existing Active Peer with new peer
+            ManageNewRemoteNodeId(session);
 
-                if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.HandshakeCompleted);
+            if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.HandshakeCompleted);
 
-                //This is the first moment we get confirmed publicKey of remote node in case of incoming connections
-                if (session.Direction == ConnectionDirection.In)
+            //This is the first moment we get confirmed publicKey of remote node in case of incoming connections
+            if (session.Direction == ConnectionDirection.In)
+            {
+                // For incoming connection, this is the entry point.
+                ProcessIncomingConnection(session);
+            }
+            else
+            {
+                if (!_peerPool.ActivePeers.TryGetValue(session.RemoteNodeId, out Peer peer))
                 {
-                    // For incoming connection, this is the entry point.
-                    ProcessIncomingConnection(session);
+                    //Can happen when peer sent Disconnect message before handshake is done, it takes us a while to disconnect
+                    if (_logger.IsTrace) TraceHandshakeWithoutActivePeer();
+                    return;
                 }
-                else
-                {
-                    if (!_peerPool.ActivePeers.TryGetValue(session.RemoteNodeId, out Peer peer))
-                    {
-                        //Can happen when peer sent Disconnect message before handshake is done, it takes us a while to disconnect
-                        if (_logger.IsTrace) TraceHandshakeWithoutActivePeer();
-                        return;
-                    }
 
-                    peer.Stats.AddNodeStatsHandshakeEvent(ConnectionDirection.Out);
-                }
+                peer.Stats.AddNodeStatsHandshakeEvent(ConnectionDirection.Out);
+            }
 
-                if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.HandshakeInitialized);
+            if (_logger.IsTrace) TraceSessionLifecycle(session, SessionLifecycleTraceEvent.HandshakeInitialized);
 
-                [MethodImpl(MethodImplOptions.NoInlining)]
-                void TraceHandshakeWithoutActivePeer()
-                    => _logger.Trace($"Initiated handshake (OUT) with a peer without adding it to the Active collection : {session}");
-            }
+            [MethodImpl(MethodImplOptions.NoInlining)]
+            void TraceHandshakeWithoutActivePeer()
+                => _logger.Trace($"Initiated handshake (OUT) with a peer without adding it to the Active collection : {session}");
         }
 
         private void ManageNewRemoteNodeId(ISession session)