Release 5.22.0

Author: sakno

CHANGELOG.md

@@ -1,7 +1,7 @@
 Release Notes
 ====
 
-# 05-25-2025
+# 05-26-2025
 <a href="https://www.nuget.org/packages/dotnext/5.22.0">DotNext 5.22.0</a>
 * Added `!` operator overloading for the result and optional types: [261](https://github.com/dotnet/dotNext/pull/261)
 

README.md

@@ -44,7 +44,7 @@ All these things are implemented in 100% managed code on top of existing .NET AP
 * [NuGet Packages](https://www.nuget.org/profiles/rvsakno)
 
 # What's new
-Release Date: 05-25-2025
+Release Date: 05-26-2025
 
 <a href="https://www.nuget.org/packages/dotnext/5.22.0">DotNext 5.22.0</a>
 * Added `!` operator overloading for the result and optional types: [261](https://github.com/dotnet/dotNext/pull/261)

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/StateMachine/WriteAheadLog.Cleaner.cs

@@ -8,20 +8,18 @@ namespace DotNext.Net.Cluster.Consensus.Raft.StateMachine;
 partial class WriteAheadLog
 {
     [AsyncMethodBuilder(typeof(SpawningAsyncTaskMethodBuilder))]
-    private async Task CleanUpAsync(CancellationToken token)
+    private async Task CleanUpAsync(long upToIndex, CancellationToken token)
     {
-        var nextIndex = stateMachine.Snapshot?.Index ?? 0L;
-
         // After the barrier, we know that there is no competing reader that reads the old snapshot version
         lockManager.SetCallerInformation("Remove Pages");
         await lockManager.AcquireReadBarrierAsync(token).ConfigureAwait(false);
         try
         {
             // The barrier can suspend this async flow. However, the OS flushes the pages in the background
-            RemoveSquashedPages(nextIndex);
+            RemoveSquashedPages(upToIndex);
 
             // ensure that garbage reclamation is not running concurrently with the snapshot installation process
-            await stateMachine.ReclaimGarbageAsync(nextIndex, token).ConfigureAwait(false);
+            await stateMachine.ReclaimGarbageAsync(upToIndex, token).ConfigureAwait(false);
         }
         catch (Exception e) when (e is not OperationCanceledException canceledEx || canceledEx.CancellationToken != token)
         {
@@ -45,6 +43,7 @@ private void RemoveSquashedPages(long toIndex)
         toPage = metadataPages.GetEndPageIndex(toIndex);
         removedBytes += metadataPages.Delete(toPage) * (long)MetadataPageManager.PageSize;
 
-        BytesDeletedMeter.Record(removedBytes, measurementTags);
+        if (removedBytes > 0L)
+            BytesDeletedMeter.Record(removedBytes, measurementTags);
     }
 }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/StateMachine/WriteAheadLog.Flusher.cs

@@ -28,15 +28,21 @@ partial class WriteAheadLog
     [AsyncMethodBuilder(typeof(SpawningAsyncTaskMethodBuilder))]
     private async Task FlushAsync(long previousIndex, TimeSpan timeout, CancellationToken token)
     {
-        var cleanupTask = Task.CompletedTask;
+        // Weak ref tracks the task, but allows GC to collect associated state machine
+        // as soon as possible. While the task is running, it cannot be collected, because it's referenced
+        // by the async state machine.
+        var cleanupTask = GCHandle.Alloc(Task.CompletedTask, GCHandleType.Weak);
         try
         {
-            for (long newIndex; !token.IsCancellationRequested && backgroundTaskFailure is null; previousIndex = newIndex)
+            for (long newIndex, oldSnapshot = SnapshotIndex, newSnapshot;
+                 !token.IsCancellationRequested && backgroundTaskFailure is null;
+                 previousIndex = long.Max(oldSnapshot = newSnapshot, newIndex) + 1L)
             {
-                manualFlushQueue.SwitchValve();
+                newSnapshot = SnapshotIndex;
                 newIndex = LastCommittedEntryIndex;
+                manualFlushQueue.SwitchValve();
 
-                if (newIndex > previousIndex)
+                if (newIndex >= previousIndex)
                 {
                     // Ensure that the flusher is not running with the snapshot installation process concurrently
                     lockManager.SetCallerInformation("Flush Pages");
@@ -55,18 +61,22 @@ private async Task FlushAsync(long previousIndex, TimeSpan timeout, Cancellation
                     {
                         lockManager.ReleaseReadLock();
                     }
-                    
-                    if (cleanupTask.IsCompleted)
-                        cleanupTask = CleanUpAsync(token);
                 }
 
+                if (cleanupTask.Target is null or Task { IsCompletedSuccessfully: true } && oldSnapshot < newSnapshot)
+                    cleanupTask.Target = CleanUpAsync(newSnapshot, token);
+
                 manualFlushQueue.Drain();
                 await flushTrigger.WaitAsync(timeout, token).ConfigureAwait(false);
             }
         }
-        catch (OperationCanceledException e) when (e.CancellationToken == token)
+        catch (OperationCanceledException e) when (e.CancellationToken == token && cleanupTask.Target is Task target)
+        {
+            await target.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+        }
+        finally
         {
-            await cleanupTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+            cleanupTask.Free();
         }
     }
 

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/StateMachine/WriteAheadLog.cs

@@ -92,7 +92,7 @@ public WriteAheadLog(Options configuration, IStateMachine stateMachine)
                 flushOnCommit = false;
             }
 
-            flusherTask = FlushAsync(commitIndex, interval, lifetimeTokenSource.Token);
+            flusherTask = FlushAsync(commitIndex + 1L, interval, lifetimeTokenSource.Token);
         }
 
         // applier
@@ -110,6 +110,8 @@ static void CreateIfNeeded(DirectoryInfo directory)
         }
     }
 
+    private long SnapshotIndex => stateMachine.Snapshot?.Index ?? 0L;
+
     /// <summary>
     /// Initializes the log asynchronously.
     /// </summary>