fix a potential concurrency issue when handling new heads

aboutcircles · Sep 25, 2024 · 73fb628 · 73fb628
1 parent 5271423
commit 73fb628
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 16 deletions.
diff --git a/Circles.Index/Circles.Index.csproj b/Circles.Index/Circles.Index.csproj
@@ -21,6 +21,7 @@
 
 
     <ItemGroup>
+      <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="8.0.8" />
       <PackageReference Include="Nethermind.Numerics.Int256" Version="1.2.0" />
       <PackageReference Include="Nethermind.ReferenceAssemblies" Version="1.28.0" />
     </ItemGroup>

diff --git a/Circles.Index/Plugin.cs b/Circles.Index/Plugin.cs
@@ -26,7 +26,7 @@ public class Plugin : INethermindPlugin
 
     private StateMachine? _indexerMachine;
     private Context? _indexerContext;
-    private Task? _currentMachineExecution;
+    private int _isProcessing;
     private int _newItemsArrived;
     private long _latestHeadToIndex = -1;
 
@@ -88,8 +88,6 @@ public async Task Init(INethermindApi nethermindApi)
 
         await _indexerMachine.TransitionTo(StateMachine.State.Initial);
 
-        _currentMachineExecution = Task.CompletedTask;
-
         nethermindApi.BlockTree!.NewHeadBlock += (_, args) =>
         {
             var fullSyncInfo = nethermindApi.EthSyncingInfo?.GetFullInfo();
@@ -191,28 +189,46 @@ private static void InitCaches(InterfaceLogger logger, IDatabase database)
 
     private void HandleNewHead()
     {
-        if (_currentMachineExecution is { IsCompleted: false })
+        // Start the processing task if not already running
+        if (Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0)
         {
-            // If there is an ongoing execution, we don't need to start a new one
-            return;
+            Task.Run(ProcessBlocksAsync, _cancellationTokenSource.Token);
         }
-
-        _currentMachineExecution = Task.Run(ProcessBlocksAsync, _cancellationTokenSource.Token);
     }
 
     private async Task ProcessBlocksAsync()
     {
-        // This loop is to ensure that we process all the new heads that arrive while we are processing the current head
-        do
+        try
         {
-            long toIndex = Interlocked.Exchange(ref _latestHeadToIndex, -1);
-            if (toIndex == -1)
+            do
             {
-                continue;
-            }
+                _cancellationTokenSource.Token.ThrowIfCancellationRequested();
+
+                long toIndex = Interlocked.Exchange(ref _latestHeadToIndex, -1);
+                if (toIndex == -1)
+                {
+                    continue;
+                }
+
+                await _indexerMachine!.HandleEvent(new StateMachine.NewHead(toIndex));
 
-            await _indexerMachine!.HandleEvent(new StateMachine.NewHead(toIndex));
-        } while (Interlocked.CompareExchange(ref _newItemsArrived, 0, 1) == 1);
+                // As long as new items arrive, keep processing
+            } while (Interlocked.CompareExchange(ref _newItemsArrived, 0, 1) == 1);
+        }
+        catch (OperationCanceledException)
+        {
+            _indexerContext!.Logger.Info("Processing was canceled.");
+        }
+        catch (Exception e)
+        {
+            _indexerContext!.Logger.Error("Error processing blocks", e);
+            throw;
+        }
+        finally
+        {
+            // Mark processing as complete
+            Interlocked.Exchange(ref _isProcessing, 0);
+        }
     }
 
     public Task InitNetworkProtocol()