Fix/error on shutdown (#7873)

Co-authored-by: Lukasz Rozmej <[email protected]>

src/Nethermind/Nethermind.Synchronization.Test/FastSync/SnapProtocolTests/StateSyncDispatcherTests.cs

@@ -58,6 +58,7 @@ public void Setup()
     public async Task TearDown()
     {
         await _pool.DisposeAsync();
+        await _dispatcher.DisposeAsync();
     }
 
     [Test]

src/Nethermind/Nethermind.Synchronization.Test/ParallelSync/SyncDispatcherTests.cs

@@ -270,6 +270,38 @@ public async Task Simple_test_sync()
         }
     }
 
+    [Retry(tryCount: 5)]
+    [Test]
+    public async Task When_ConcurrentHandleResponseIsRunning_Then_BlockDispose()
+    {
+        TestSyncFeed syncFeed = new(isMultiFeed: true);
+        syncFeed.LockResponse();
+        TestDownloader downloader = new TestDownloader();
+        SyncDispatcher<TestBatch> dispatcher = new(
+            new TestSyncConfig(),
+            syncFeed,
+            downloader,
+            new TestSyncPeerPool(),
+            new StaticPeerAllocationStrategyFactory<TestBatch>(FirstFree.Instance),
+            LimboLogs.Instance);
+        Task executorTask = dispatcher.Start(CancellationToken.None);
+
+        // Load some requests
+        syncFeed.Activate();
+        await Task.Delay(100);
+        syncFeed.Finish();
+
+        // Dispose
+        Task disposeTask = Task.Run(() => dispatcher.DisposeAsync());
+        await Task.Delay(100);
+
+        disposeTask.IsCompletedSuccessfully.Should().BeFalse();
+
+        syncFeed.UnlockResponse();
+        await disposeTask;
+        await executorTask;
+    }
+
     [Retry(tryCount: 5)]
     [TestCase(false, 1, 1, 8)]
     [TestCase(true, 1, 1, 24)]
@@ -294,6 +326,7 @@ public async Task Test_release_before_processing_complete(bool isMultiSync, int
 
         Task _ = dispatcher.Start(CancellationToken.None);
         syncFeed.Activate();
+
         await Task.Delay(100);
 
         Assert.That(() => syncFeed.HighestRequested, Is.EqualTo(expectedHighestRequest).After(4000, 100));

src/Nethermind/Nethermind.Synchronization/ParallelSync/SyncDispatcher.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using Nethermind.Blockchain.Synchronization;
@@ -12,10 +13,11 @@
 
 namespace Nethermind.Synchronization.ParallelSync
 {
-    public class SyncDispatcher<T>
+    public class SyncDispatcher<T> : IAsyncDisposable
     {
         private readonly Lock _feedStateManipulation = new();
         private SyncFeedState _currentFeedState = SyncFeedState.Dormant;
+        private static readonly TimeSpan ActiveTaskDisposeTimeout = TimeSpan.FromSeconds(10);
 
         private IPeerAllocationStrategyFactory<T> PeerAllocationStrategyFactory { get; }
 
@@ -24,10 +26,13 @@ public class SyncDispatcher<T>
         private ISyncDownloader<T> Downloader { get; }
         private ISyncPeerPool SyncPeerPool { get; }
 
+        private readonly CountdownEvent _activeTasks = new CountdownEvent(1);
         private readonly SemaphoreSlim _concurrentProcessingSemaphore;
         private readonly TimeSpan _emptyRequestDelay;
         private readonly int _allocateTimeoutMs;
 
+        private bool _disposed = false;
+
         public SyncDispatcher(
             ISyncConfig syncConfig,
             ISyncFeed<T>? syncFeed,
@@ -109,7 +114,29 @@ public async Task Start(CancellationToken cancellationToken)
                             if (Logger.IsTrace) Logger.Trace($"SyncDispatcher request: {request}, AllocatedPeer {allocation.Current}");
 
                             // Use Task.Run to make sure it queues it instead of running part of it synchronously.
-                            Task task = Task.Run(() => DoDispatch(cancellationToken, allocatedPeer, request, allocation), cancellationToken);
+                            _activeTasks.AddCount();
+
+                            Task task;
+                            try
+                            {
+                                task = Task.Run(
+                                    () =>
+                                    {
+                                        try
+                                        {
+                                            return DoDispatch(cancellationToken, allocatedPeer, request, allocation);
+                                        }
+                                        finally
+                                        {
+                                            _activeTasks.Signal();
+                                        }
+                                    });
+                            }
+                            catch
+                            {
+                                _activeTasks.Signal();
+                                throw;
+                            }
 
                             if (!Feed.IsMultiFeed)
                             {
@@ -272,5 +299,22 @@ private void UpdateState(SyncFeedState state)
                 }
             }
         }
+
+        public ValueTask DisposeAsync()
+        {
+            if (_disposed)
+            {
+                return ValueTask.CompletedTask;
+            }
+            _disposed = true;
+
+            _activeTasks.Signal();
+            if (!_activeTasks.Wait(ActiveTaskDisposeTimeout))
+            {
+                if (Logger.IsWarn) Logger.Warn($"Timeout on waiting for active tasks for feed {Feed.GetType().Name} {_activeTasks.CurrentCount}");
+            }
+            _concurrentProcessingSemaphore.Dispose();
+            return ValueTask.CompletedTask;
+        }
     }
 }

src/Nethermind/Nethermind.Synchronization/SyncFeedComponent.cs

@@ -12,10 +12,10 @@ namespace Nethermind.Synchronization;
 /// <summary>
 /// Container to make it simpler to get a bunch of components within a same named scoped.
 /// </summary>
-public class SyncFeedComponent<TBatch>(Lazy<ISyncFeed<TBatch>> feed, Lazy<SyncDispatcher<TBatch>> dispatcher, Lazy<ISyncDownloader<TBatch>> blockDownloader, ILifetimeScope lifetimeScope) : IDisposable, IAsyncDisposable
+public class SyncFeedComponent<TBatch>(ISyncFeed<TBatch> feed, SyncDispatcher<TBatch> dispatcher, Lazy<ISyncDownloader<TBatch>> blockDownloader, ILifetimeScope lifetimeScope) : IDisposable, IAsyncDisposable
 {
-    public ISyncFeed<TBatch> Feed => feed.Value;
-    public SyncDispatcher<TBatch> Dispatcher => dispatcher.Value;
+    public ISyncFeed<TBatch> Feed => feed;
+    public SyncDispatcher<TBatch> Dispatcher => dispatcher;
     public BlockDownloader BlockDownloader => (BlockDownloader)blockDownloader.Value;
 
     public void Dispose()

src/Nethermind/Nethermind.Synchronization/Synchronizer.cs

@@ -240,8 +240,9 @@ public async ValueTask DisposeAsync()
         {
             _syncCancellation?.Cancel();
 
-            await Task.WhenAny(
-                Task.Delay(FeedsTerminationTimeout),
+            Task timeout = Task.Delay(FeedsTerminationTimeout);
+            Task completedFirst = await Task.WhenAny(
+                timeout,
                 Task.WhenAll(
                     fullSyncComponent.Feed.FeedTask,
                     fastSyncComponent.Feed.FeedTask,
@@ -251,6 +252,11 @@ await Task.WhenAny(
                     oldBodiesComponent.Feed.FeedTask,
                     oldReceiptsComponent.Feed.FeedTask));
 
+            if (completedFirst == timeout)
+            {
+                if (_logger.IsWarn) _logger.Warn("Sync feeds dispose timeout");
+            }
+
             CancellationTokenExtensions.CancelDisposeAndClear(ref _syncCancellation);
         }