Fix event handler leaks, reduce allocations, and refactor Network layer (#10753)

* Fix event handler leaks, reduce allocations, and refactor Network layer

- Fix event handler leaks in ProtocolsManager, PeerManager, RlpxHost,
  DiscoveryManager, DiscoveryApp, CompositeDiscoveryApp,
  NodeLifecycleManager, and NettyDiscoveryHandler by tracking
  subscriptions and unsubscribing on disposal
- Implement IDisposable on ProtocolsManager with
  SessionSubscription/ProtocolHandlerSubscription wrappers for
  deterministic event handler cleanup
- Cache event handler and Func<> delegates in fields to avoid repeated
  allocations on subscribe/unsubscribe
- Replace Action<T> delegates in MessageQueue/MessageDictionary with
  IMessageSender<T> interface to eliminate per-construction allocations
- Rewrite PacketSender from async to synchronous with ContinueWith,
  removing async state machine overhead on the hot send path
- Add custom DisconnectEventHandlers collection enabling safe
  unsubscribe during dispatch
- Extract channel initializer classes in RlpxHost, replace
  closure-capturing lambdas with cached delegates and named methods
- Convert NodeTable sort lambda to struct comparer
- Improve shutdown ordering in PeerManager
- Add RLP deserialization collection-size limits to Snap and Eth message
  serializers
- Apply [MethodImpl(NoInlining)] to logging local functions with
  interpolated strings
- Apply [DoesNotReturn, StackTraceHidden] to throw helpers
- Replace ArgumentNullException constructor with ThrowIfNull

* Add tests for event handler cleanup, disposal, shutdown, and deserialization limits

- Add tests for event handler cleanup on session disposal
- Add tests for ProtocolsManager.Dispose
- Add tests for PeerPool shutdown and SessionMonitor stop
- Add tests for SnapCapabilitySwitcher lifecycle
- Add tests for CompositeNodeSource event forwarding
- Add tests for ZeroFrameDecoder oversized frame rejection
- Add tests for deserialization limit validation in Snap/Eth serializers
- Add tests for MessageDictionary and MessageQueue with IMessageSender
- Deduplicate test setup across Eth protocol handler tests
  (EthProtocolTestHelper)
- Refactor NettyDiscoveryHandlerTests, HelloMessageSerializerTests, and
  GetTrieNodesMessageSerializerTests

* Fix RetryCache use-after-return race, CTS leak, and shutdown ordering

- Replace TryGetValue/TryAdd with GetOrAdd in RetryCache.Announced to
  eliminate TOCTOU race where timer thread could remove and pool-return
  a ConcurrentHashSet while another thread still holds a reference
- Add using to CancellationTokenSource in RlpxHost.ConnectAsync to
  prevent timer resource leak on repeated outbound connect attempts
- Move subscription detachment in RlpxHost.Shutdown to after channel
  close and event loop shutdown so Disconnected handlers fire during
  shutdown and sessions are properly disposed

* Fix P2PMessageKey label caching on record struct

Record struct copies lose mutable field state, so the `_labels ??=`
pattern silently re-computed labels on every access. Replace with a
static ConcurrentDictionary keyed by the struct value — the key space is bounded by (protocol versions × SpaceSize) which is a few dozen entries total.

* Await PeerManager loop task in StopAsync and fix StartNew anti-pattern

Task.Factory.StartNew(LongRunning) with an async delegate returns
Task<Task> — the outer task completes at the first await, making
LongRunning meaningless and ContinueWith premature. Replace with
async Task + Task.Yield().

StopAsync previously returned Task.CompletedTask while the loop was
still running. Now it awaits the loop task for clean shutdown.

Add regression tests: Stop_does_not_hang, Disconnect_triggers_refill,
No_slot_available_before_deadline_does_not_deadlock_refill.

* Fix double-stop of DiscoveryApp during shutdown

DiscoveryApp and DiscoveryV5App implement IStoppableService via
IDiscoveryApp. When registered as container-owned singletons,
ServiceStopperMiddleware calls StopAsync on them directly AND
CompositeDiscoveryApp also calls StopAsync on them — producing
the duplicate "Discovery shutdown complete" log message.

Mark both as ExternallyOwned so only CompositeDiscoveryApp
manages their lifecycle.

fx

Propagate cancellation through discovery node location

LocateNodesAsync did not pass its CancellationToken into SendFindNode
or WasMessageReceived. During shutdown, the discovery loop had to wait
for all in-flight FindNode requests to time out (up to 8 rounds × N
nodes × 500ms each ≈ 70 seconds) before the task completed.

Thread the token through LocateNodesAsync → SendFindNodes →
SendFindNode → WasMessageReceived, and add ThrowIfCancellationRequested
at the top of each discovery round. WasMessageReceived now creates a
linked CTS so the Task.Delay is cancelled immediately on shutdown.

* fix: address Claude reviewer feedback on network PR

- RetryCache: document that first announcer is not added to retry bag
  (intentional — it receives RequestRequired and requests directly)
- RlpxHost.Shutdown: dispose orphaned sessions via DetachAndDispose
  (sessions whose Disconnected event never fired were leaked)
- DisconnectEventHandlers: document unordered invocation due to
  swap-and-null removal pattern
- PacketSender: document thread-safety of lazy-init (Netty guarantees
  single-threaded channel event delivery)
- ZeroFrameDecoder: document 12 MiB frame size cap rationale
- RlpxHost: remove commented-out logging setup code
- PeerManager: replace var with explicit KeyValuePair type,
  add locking comment on _isStopping field

* fix: CI failures from review feedback changes

- DetachAndDispose: mark session as disconnected before disposing to
  avoid InvalidOperationException on sessions still in Initialized state
- PeerManagerTests: relax assertion from exact count (1) to > 0 since
  ConnectTimeoutMs=0 allows multiple connections to be queued before
  the slot count updates

* fix: dispose CancellationTokenSource in RlpxHost.Shutdown

The delayCancellation CTS in the shutdown path was not disposed.
Add using declaration to match the ConnectAsync path.

* fix: address PR review feedback (round 2)

- Move IsIPv4Multicast to NodeFilter as a shared helper, delegate
  from DiscoveryV5App
- Rename LogTrace to TraceNodeFilteredByRateLimit for clarity
- Replace magic number 4097/1025 in snap serializer tests with
  SnapMessageLimits constants + 1

* fix: use ArrayPoolListRef instead of allocating Task[] in WhenAllDiscoveryApps

Author: benaadams

src/Nethermind/Nethermind.Core.Test/Modules/TestEnvironmentModule.cs

@@ -45,6 +45,7 @@ protected override void Load(ContainerBuilder builder)
             .AddKeyedSingleton<IFullDb>(DbNames.DiscoveryNodes, (_) => new MemDb())
             .AddKeyedSingleton<IFullDb>(DbNames.DiscoveryV5Nodes, (_) => new MemDb())
             .AddSingleton<IChannelFactory, INetworkConfig>(networkConfig => new LocalChannelFactory(networkGroup ?? nameof(TestEnvironmentModule), networkConfig))
+            .AddSingleton(NodeFilter.AcceptAll) // Disable inbound rate limiting for in-memory channels
 
             .AddSingleton<PseudoNethermindRunner>()
             .AddSingleton<TestBlockchainUtil>()

src/Nethermind/Nethermind.Core/Collections/ArrayListCore.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Buffers;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Runtime.CompilerServices;
@@ -138,6 +139,13 @@ public static void Sort(T[] array, int count, Comparison<T> comparison)
         if (count > 1) array.AsSpan(0, count).Sort(comparison);
     }
 
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    public static void Sort<TComparer>(T[] array, int count, TComparer comparer)
+        where TComparer : IComparer<T>
+    {
+        if (count > 1) array.AsSpan(0, count).Sort(comparer);
+    }
+
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
     public static void Reverse(T[] array, int count)
     {

src/Nethermind/Nethermind.Core/Collections/ArrayPoolList.cs

@@ -145,6 +145,13 @@ public void Sort(Comparison<T> comparison)
         ArrayPoolListCore<T>.Sort(_array, _count, comparison);
     }
 
+    public void Sort<TComparer>(TComparer comparer)
+        where TComparer : IComparer<T>
+    {
+        GuardDispose();
+        ArrayPoolListCore<T>.Sort(_array, _count, comparer);
+    }
+
     public int Capacity => _capacity;
 
     bool IList.IsFixedSize => false;

src/Nethermind/Nethermind.Core/Collections/ArrayPoolListRef.cs

@@ -67,6 +67,7 @@ public void AddRange(params IEnumerable<T> items)
     public void ReduceCount(int newCount) => ArrayPoolListCore<T>.ReduceCount(SafeArrayPool<T>.Shared, ref _array, ref _capacity, ref _count, newCount);
     public void Truncate(int newLength) => ArrayPoolListCore<T>.Truncate(newLength, _array, ref _count);
     public readonly void Sort(Comparison<T> comparison) => ArrayPoolListCore<T>.Sort(_array, _count, comparison);
+    public readonly void Sort<TComparer>(TComparer comparer) where TComparer : IComparer<T> => ArrayPoolListCore<T>.Sort(_array, _count, comparer);
     public readonly void Reverse() => ArrayPoolListCore<T>.Reverse(_array, _count);
     public readonly ref T GetRef(int index) => ref ArrayPoolListCore<T>.GetRef(_array, index, _count);
     public readonly Span<T> AsSpan() => _array.AsSpan(0, _count);

src/Nethermind/Nethermind.Crypto/PrivateKeyGenerator.cs

@@ -1,54 +1,56 @@
-// SPDX-FileCopyrightText: 2022 Demerzel Solutions Limited
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
 // SPDX-License-Identifier: LGPL-3.0-only
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 
-namespace Nethermind.Crypto
+namespace Nethermind.Crypto;
+
+public sealed class PrivateKeyGenerator : IPrivateKeyGenerator, IDisposable
 {
-    public class PrivateKeyGenerator : IPrivateKeyGenerator, IDisposable
-    {
-        private readonly ICryptoRandom _cryptoRandom;
-        private readonly bool _disposeRandom = false;
+    private readonly ICryptoRandom _cryptoRandom;
+    private readonly bool _disposeRandom = false;
 
-        public PrivateKeyGenerator()
-        {
-            _cryptoRandom = new CryptoRandom();
-            _disposeRandom = true;
-        }
+    public PrivateKeyGenerator()
+    {
+        _cryptoRandom = new CryptoRandom();
+        _disposeRandom = true;
+    }
 
-        public PrivateKeyGenerator(ICryptoRandom cryptoRandom)
-        {
-            _cryptoRandom = cryptoRandom;
-        }
+    public PrivateKeyGenerator(ICryptoRandom cryptoRandom)
+    {
+        _cryptoRandom = cryptoRandom;
+    }
 
-        public PrivateKey Generate()
-        {
-            return Generate(1).First();
-        }
-        public IEnumerable<PrivateKey> Generate(int number)
+    public PrivateKey Generate()
+    {
+        do
         {
-            do
+            byte[] bytes = _cryptoRandom.GenerateRandomBytes(32);
+            if (SecP256k1.VerifyPrivateKey(bytes))
             {
-                var bytes = _cryptoRandom.GenerateRandomBytes(32);
-                if (SecP256k1.VerifyPrivateKey(bytes))
-                {
-                    yield return new PrivateKey(bytes);
-                    if (--number == 0)
-                    {
-                        yield break;
-                    }
-                }
-            } while (true);
-        }
+                return new PrivateKey(bytes);
+            }
+        } while (true);
+    }
 
-        public void Dispose()
+    public IEnumerable<PrivateKey> Generate(int number)
+    {
+        do
         {
-            if (_disposeRandom)
+            yield return Generate();
+            if (--number == 0)
             {
-                _cryptoRandom.Dispose();
+                yield break;
             }
+        } while (true);
+    }
+
+    public void Dispose()
+    {
+        if (_disposeRandom)
+        {
+            _cryptoRandom.Dispose();
         }
     }
 }

src/Nethermind/Nethermind.Init/Modules/DiscoveryModule.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 using Autofac;
+using Autofac.Features.AttributeFilters;
 using Nethermind.Api;
 using Nethermind.Core;
 using Nethermind.Crypto;
@@ -105,7 +106,6 @@ protected override void Load(ContainerBuilder builder)
 
                 .AddNetworkStorage(DbNames.DiscoveryNodes, "discoveryNodes")
                 .AddNetworkStorage(DbNames.DiscoveryV5Nodes, "discoveryV5Nodes")
-                .AddSingleton<DiscoveryV5App>()
 
                 .AddSingleton<INodeDistanceCalculator, NodeDistanceCalculator>()
                 .AddSingleton<INodeTable, NodeTable>()
@@ -114,9 +114,14 @@ protected override void Load(ContainerBuilder builder)
                 .AddSingleton<IDiscoveryManager, DiscoveryManager>()
                 .AddSingleton<INodesLocator, NodesLocator>()
                 .AddSingleton<DiscoveryPersistenceManager>()
-                .AddSingleton<DiscoveryApp>()
 
                 ;
+
+            // DiscoveryApp and DiscoveryV5App implement IStoppableService via IDiscoveryApp,
+            // but their lifecycle is owned by CompositeDiscoveryApp. Mark ExternallyOwned so
+            // ServiceStopperMiddleware does not double-stop them.
+            builder.RegisterType<DiscoveryV5App>().AsSelf().WithAttributeFiltering().SingleInstance().ExternallyOwned();
+            builder.RegisterType<DiscoveryApp>().AsSelf().WithAttributeFiltering().SingleInstance().ExternallyOwned();
         }
 
 

src/Nethermind/Nethermind.Init/Steps/InitializeNetwork.cs

@@ -140,6 +140,7 @@ await InitPeer().ContinueWith(initPeerTask =>
         {
             SnapCapabilitySwitcher snapCapabilitySwitcher =
                 new(_api.ProtocolsManager, _api.SyncModeSelector, _api.LogManager);
+            _api.DisposeStack.Push(snapCapabilitySwitcher);
             snapCapabilitySwitcher.EnableSnapCapabilityUntilSynced();
         }
 

src/Nethermind/Nethermind.Network.Discovery.Test/DiscoveryManagerTests.cs

@@ -6,16 +6,20 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Net;
+using System.Threading;
 using System.Threading.Tasks;
 using FluentAssertions;
 using Nethermind.Core;
 using Nethermind.Core.Crypto;
 using Nethermind.Core.Test;
 using Nethermind.Core.Test.Builders;
+using Nethermind.Core.Test.Modules;
+using Nethermind.Config;
 using Nethermind.Core.Timers;
 using Nethermind.Crypto;
 using Nethermind.Db;
 using Nethermind.Logging;
+using Nethermind.Network;
 using Nethermind.Network.Config;
 using Nethermind.Network.Discovery.Lifecycle;
 using Nethermind.Network.Discovery.Messages;
@@ -94,6 +98,20 @@ public async Task OnPingMessageTest()
             await _msgSender.Received().SendMsg(Arg.Is<PingMsg>(static m => m.FarAddress!.Address.ToString() == Host && m.FarAddress.Port == Port));
         }
 
+        [Test]
+        public void OnPingMessage_WithUnexpectedDestination_IsIgnored()
+        {
+            IPEndPoint address = new(IPAddress.Parse(Host), Port);
+            PingMsg ping = new(_publicKey, GetExpirationTime(), address, new IPEndPoint(IPAddress.Parse("203.0.113.9"), _nodeTable.MasterNode!.Port), new byte[32])
+            {
+                FarAddress = address
+            };
+
+            _discoveryManager.OnIncomingMsg(ping);
+
+            _msgSender.DidNotReceive().SendMsg(Arg.Any<DiscoveryMsg>());
+        }
+
         [Test, Ignore("Add bonding"), Retry(3)]
         public void OnPongMessageTest()
         {
@@ -152,6 +170,34 @@ public void MemoryTest()
             }
         }
 
+        [Test]
+        public void OnPingMessage_FromFilteredIp_IsIgnored()
+        {
+            // Setup with filtering enabled (default) — the filter has a 5-minute timeout per IP/subnet.
+            // First ping from this IP creates a lifecycle manager and is processed.
+            IPEndPoint address = new(IPAddress.Parse("203.0.113.50"), Port);
+            PingMsg ping1 = new(TestItem.PublicKeyC, GetExpirationTime(), address, _nodeTable.MasterNode!.Address, new byte[32])
+            {
+                FarAddress = address
+            };
+            _discoveryManager.OnIncomingMsg(ping1);
+
+            // Second ping from the same IP but different node ID should be filtered
+            // because the IP is already in the NodeFilter cache.
+            PingMsg ping2 = new(TestItem.PublicKeyD, GetExpirationTime(), address, _nodeTable.MasterNode!.Address, new byte[32])
+            {
+                FarAddress = address
+            };
+            _discoveryManager.OnIncomingMsg(ping2);
+
+            // Only one lifecycle manager should have been created (for the first ping)
+            INodeLifecycleManager? manager1 = _discoveryManager.GetNodeLifecycleManager(new Node(TestItem.PublicKeyC, "203.0.113.50", Port));
+            INodeLifecycleManager? manager2 = _discoveryManager.GetNodeLifecycleManager(new Node(TestItem.PublicKeyD, "203.0.113.50", Port));
+
+            Assert.That(manager1, Is.Not.Null, "First node from the IP should be accepted");
+            Assert.That(manager2, Is.Null, "Second node from the same IP should be filtered");
+        }
+
         private static long GetExpirationTime() => Timestamper.Default.UnixTime.SecondsLong + 20;
 
         [Test, Ignore("Add bonding"), Retry(3)]
@@ -190,6 +236,33 @@ private void ReceiveSomePong()
             _discoveryManager.OnIncomingMsg(pongMsg);
         }
 
+        [Test]
+        public async Task SendMessage_DropOldest_WhenQueueIsFull()
+        {
+            // Use a very slow rate (1/sec) so messages pile up in the queue
+            SetupDiscoveryManager(new DiscoveryConfig()
+            {
+                MaxOutgoingMessagePerSecond = 1
+            });
+
+            // Fire 600 fire-and-forget messages (exceeds bounded channel capacity of 512).
+            // With DropOldest semantics, the oldest messages are evicted when full,
+            // keeping the channel bounded and preventing unbounded task/memory growth.
+            for (int i = 0; i < 600; i++)
+            {
+                FindNodeMsg msg = new(_publicKey, i, []);
+                _discoveryManager.SendMessage(msg);
+            }
+
+            // Allow the consumer to process a few messages
+            await Task.Delay(50);
+
+            // With rate=1/sec, only ~1 message should have been actually sent in 50ms.
+            // The key property: the channel is bounded at 512, so ~88 oldest messages were dropped.
+            int sent = _msgSender.ReceivedCalls().Count(c => c.GetMethodInfo().Name == nameof(IMsgSender.SendMsg));
+            Assert.That(sent, Is.LessThan(520), "Bounded channel should prevent unbounded send accumulation");
+        }
+
         [Test]
         [Repeat(10)]
         public async Task RateLimitOutgoingMessage()
@@ -209,5 +282,90 @@ public async Task RateLimitOutgoingMessage()
             await _discoveryManager.SendMessageAsync(msg);
             Stopwatch.GetElapsedTime(startTime).Should().BeGreaterThanOrEqualTo(TimeSpan.FromSeconds(0.9));
         }
+
+        [Test]
+        public async Task SendMessage_StartsOnlyOneQueueConsumer_WhenFirstUseIsConcurrent()
+        {
+            SetupDiscoveryManager(new DiscoveryConfig
+            {
+                MaxOutgoingMessagePerSecond = 1000
+            });
+
+            const int concurrency = 16;
+            Task[] sendTasks = new Task[concurrency];
+            ManualResetEventSlim start = new(false);
+
+            for (int i = 0; i < concurrency; i++)
+            {
+                int expiration = i;
+                sendTasks[i] = Task.Run(() =>
+                {
+                    start.Wait();
+                    _discoveryManager.SendMessage(new FindNodeMsg(_publicKey, expiration, []));
+                });
+            }
+
+            start.Set();
+            await Task.WhenAll(sendTasks);
+
+            ((DiscoveryManager)_discoveryManager).SendQueueConsumersCreated.Should().Be(1);
+        }
+    }
+
+    [Parallelizable(ParallelScope.Self)]
+    [TestFixture]
+    public class DiscoveryAppTests
+    {
+        private const string TestPrivateKeyHex = "0x3a1076bf45ab87712ad64ccb3b10217737f7faacbf2872e88fdd9a537d8fe266";
+
+        [Test]
+        public async Task StopAsync_ShouldIgnoreDiscoveredNodesAfterStop()
+        {
+            PrivateKey privateKey = new(TestPrivateKeyHex);
+            IProtectedPrivateKey nodeKey = new InsecureProtectedPrivateKey(privateKey);
+            INodesLocator nodesLocator = Substitute.For<INodesLocator>();
+            IDiscoveryManager discoveryManager = Substitute.For<IDiscoveryManager>();
+            INodeTable nodeTable = Substitute.For<INodeTable>();
+            IMessageSerializationService messageSerializationService = Substitute.For<IMessageSerializationService>();
+            ICryptoRandom cryptoRandom = Substitute.For<ICryptoRandom>();
+            INetworkStorage discoveryStorage = Substitute.For<INetworkStorage>();
+            IProcessExitSource processExitSource = Substitute.For<IProcessExitSource>();
+            processExitSource.Token.Returns(CancellationToken.None);
+
+            Node masterNode = new(privateKey.PublicKey, IPAddress.Loopback.ToString(), 30303);
+            nodeTable.MasterNode.Returns(masterNode);
+
+            DiscoveryConfig discoveryConfig = new();
+            NetworkConfig networkConfig = new();
+            ILogManager logManager = LimboLogs.Instance;
+            DiscoveryPersistenceManager persistenceManager = new(discoveryStorage, discoveryManager, discoveryConfig, logManager);
+
+            DiscoveryApp discoveryApp = new(
+                nodeKey,
+                nodesLocator,
+                discoveryManager,
+                nodeTable,
+                messageSerializationService,
+                cryptoRandom,
+                discoveryStorage,
+                persistenceManager,
+                processExitSource,
+                networkConfig,
+                discoveryConfig,
+                Timestamper.Default,
+                logManager);
+
+            int addedNodes = 0;
+            discoveryApp.NodeAdded += (_, _) => addedNodes++;
+
+            Node discoveredNode = new(TestItem.PublicKeyA, "192.168.10.5", 30303);
+            discoveryManager.NodeDiscovered += Raise.EventWith(new NodeEventArgs(discoveredNode));
+            Assert.That(addedNodes, Is.EqualTo(1));
+
+            await discoveryApp.StopAsync();
+
+            discoveryManager.NodeDiscovered += Raise.EventWith(new NodeEventArgs(discoveredNode));
+            Assert.That(addedNodes, Is.EqualTo(1));
+        }
     }
 }