Release 5.22.0

Author: sakno

azure-pipelines.yml

@@ -145,7 +145,7 @@ stages:
             artifact: config
           - task: AzureCLI@2
             inputs:
-              azureSubscription: '.NEXT Code Signing'
+              azureSubscription: .NEXT Code Signing
               scriptType: pscore
               scriptLocation: inlineScript
               inlineScript:

src/DotNext.Tests/Net/Cluster/Consensus/Raft/StateMachine/WriteAheadLogTests.cs

@@ -298,17 +298,13 @@ public static async Task IncrementalState()
 
         Memory<byte> buffer = new byte[sizeof(long)];
         const long count = 1000L;
-        using var cts = new CancellationTokenSource();
         for (var i = 0L; i < count; i++)
         {
             BinaryPrimitives.WriteInt64LittleEndian(buffer.Span, i);
-            
-            cts.CancelAfter(DefaultTimeout);
-            var index = await wal.AppendAsync(buffer, token: cts.Token);
-            await wal.CommitAsync(index, cts.Token);
-            await wal.WaitForApplyAsync(index, cts.Token);
 
-            True(cts.TryReset());
+            var index = await wal.AppendAsync(buffer);
+            await wal.CommitAsync(index);
+            await wal.WaitForApplyAsync(index);
         }
 
         Equal(count * (0L + count - 1L) / 2L, stateMachine.Value);

src/DotNext.Threading/Threading/Leases/LeaseConsumer.cs

@@ -184,18 +184,14 @@ public async Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TR
         ArgumentNullException.ThrowIfNull(worker);
 
         var exception = default(Exception?);
-        var leaseToken = Token;
-        var operationToken = token;
-        var cts = operationToken.LinkTo(leaseToken);
-
-        var task = Fork(worker, operationToken);
+        var task = Fork(worker, out var cts, out var leaseToken, token);
+        var timerToken = task.AsCancellationToken();
         try
         {
-            Task completedTask;
             do
             {
-                completedTask = await Task.WhenAny(Task.Delay(Expiration.Value / 2, operationToken), task).ConfigureAwait(false);
-            } while (!ReferenceEquals(completedTask, task) && await TryRenewAsync(token).ConfigureAwait(false));
+                await Task.Delay(Expiration.Value / 2, timerToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+            } while (!task.IsCompleted && await TryRenewAsync(token).ConfigureAwait(false));
         }
         catch (Exception e)
         {
@@ -222,6 +218,14 @@ public async Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TR
         {
             cts?.Dispose();
         }
+    }
+
+    private Task<TResult> Fork<TResult>(Func<CancellationToken, Task<TResult>> worker, out LinkedCancellationTokenSource? cts,
+        out CancellationToken leaseToken,
+        CancellationToken token)
+    {
+        cts = token.LinkTo(leaseToken = Token);
+        return Fork(worker, token);
 
         static Task<TResult> Fork(Func<CancellationToken, Task<TResult>> function, CancellationToken token)
             => Task.Run(() => function(token), token);

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

@@ -31,7 +31,7 @@ private async Task ApplyAsync(CancellationToken token)
             await lockManager.AcquireReadLockAsync(token).ConfigureAwait(false);
             try
             {
-                await ApplyAsync(appliedIndex + 1L, newIndex, token).ConfigureAwait(false);
+                await ApplyAsync(LastAppliedIndex + 1L, newIndex, token).ConfigureAwait(false);
             }
             catch (Exception e) when (e is not OperationCanceledException canceledEx || canceledEx.CancellationToken != token)
             {
@@ -54,9 +54,12 @@ private async Task ApplyAsync(CancellationToken token)
     public long LastAppliedIndex
     {
         get => Volatile.Read(in appliedIndex);
+        
         private set
         {
-            Volatile.Write(ref appliedIndex, value);
+            // Full memory barrier to ensure that the index cannot be changed later, somewhere
+            // within Signal due to inlining
+            Interlocked.Exchange(ref appliedIndex, value);
             appliedEvent.Signal(resumeAll: true);
         }
     }

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

@@ -169,7 +169,7 @@ private long Commit(long index)
         return index - oldCommitIndex;
     }
 
-    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    [MethodImpl(MethodImplOptions.NoInlining)]
     private void OnCommitted(long count)
     {
         if (flushOnCommit)

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

@@ -41,6 +41,9 @@ public WriteAheadLog(Options configuration, IStateMachine stateMachine)
         ArgumentNullException.ThrowIfNull(configuration);
         ArgumentNullException.ThrowIfNull(stateMachine);
 
+        if (nuint.Size < sizeof(ulong))
+            throw new PlatformNotSupportedException();
+
         lifetimeTokenSource = new();
         var rootPath = new DirectoryInfo(configuration.Location);
         CreateIfNeeded(rootPath);
@@ -252,7 +255,7 @@ public async ValueTask AppendAsync<TEntry>(TEntry entry, long startIndex, Cancel
                 if (startIndex <= LastCommittedEntryIndex)
                     throw new InvalidOperationException(ExceptionMessages.InvalidAppendIndex);
 
-                appliedIndex = await stateMachine.ApplyAsync(new LogEntry(entry, startIndex), token).ConfigureAwait(false);
+                LastAppliedIndex = await stateMachine.ApplyAsync(new LogEntry(entry, startIndex), token).ConfigureAwait(false);
                 var snapshotIndex = stateMachine.Snapshot?.Index ?? startIndex;
                 LastEntryIndex = long.Max(tailIndex, LastCommittedEntryIndex = snapshotIndex);
             }