Merge with 2.5.0 release

Author: sakno

CHANGELOG.md

@@ -1,6 +1,15 @@
 Release Notes
 ====
 
+# 05-04-2024
+<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.5.0">DotNext.Net.Cluster 5.5.0</a>
+* Introduced `IRaftCluster.WaitForLeadershipAsync` method that waits for the local node to be elected as a leader of the cluster
+* Fixed [233](https://github.com/dotnet/dotNext/issues/233)
+* Fixed correctness of appending no-op entry by a leader used as a write barrier
+
+<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.5.0">DotNext.AspNetCore.Cluster 5.5.0</a>
+* Updated dependencies
+
 # 04-20-2024
 <a href="https://www.nuget.org/packages/dotnext.io/5.4.0">DotNext.IO 5.4.0</a>
 * Added `FileWriter.WrittenBuffer` property

README.md

@@ -44,16 +44,14 @@ 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: 04-20-2024
+Release Date: 05-04-2024
 
-<a href="https://www.nuget.org/packages/dotnext.io/5.4.0">DotNext.IO 5.4.0</a>
-* Added `FileWriter.WrittenBuffer` property
+<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.5.0">DotNext.Net.Cluster 5.5.0</a>
+* Introduced `IRaftCluster.WaitForLeadershipAsync` method that waits for the local node to be elected as a leader of the cluster
+* Fixed [233](https://github.com/dotnet/dotNext/issues/233)
+* Fixed correctness of appending no-op entry by a leader used as a write barrier
 
-<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.4.0">DotNext.Net.Cluster 5.4.0</a>
-* Changed binary file format for WAL for more efficient I/O. A new format is incompatible with all previous versions. To enable legacy format, set `PersistentState.Options.UseLegacyBinaryFormat` property to **true**
-* Introduced a new experimental binary format for WAL based on sparse files. Can be enabled with `PersistentState.Options.MaxLogEntrySize` property
-
-<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.4.0">DotNext.AspNetCore.Cluster 5.4.0</a>
+<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.5.0">DotNext.AspNetCore.Cluster 5.5.0</a>
 * Updated dependencies
 
 Changelog for previous versions located [here](./CHANGELOG.md).

src/Directory.Packages.props

@@ -19,8 +19,8 @@
   </ItemGroup>
   <ItemGroup>
     <!--Microsoft packages-->
-    <PackageVersion Include="Microsoft.AspNetCore.Connections.Abstractions" Version="8.0.2" />
-    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.0" />
+    <PackageVersion Include="Microsoft.AspNetCore.Connections.Abstractions" Version="8.0.4" />
+    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Http" Version="8.0.0" />
@@ -33,8 +33,8 @@
     <!--Misc packages-->
     <PackageVersion Include="BenchmarkDotNet" Version="0.13.12" />
     <PackageVersion Include="FastMember.Signed" Version="1.5.0" />
-    <PackageVersion Include="xunit" Version="2.7.0" />
-    <PackageVersion Include="xunit.runner.visualstudio" Version="2.5.7"/>
-    <PackageVersion Include="coverlet.collector" Version="6.0.0"/>
+    <PackageVersion Include="xunit" Version="2.8.0" />
+    <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.0"/>
+    <PackageVersion Include="coverlet.collector" Version="6.0.2"/>
   </ItemGroup>
 </Project>

src/DotNext.Tests/Net/Cluster/Consensus/Raft/Http/RaftHttpClusterTests.cs

@@ -803,7 +803,7 @@ public static async Task ConsensusToken()
         True(GetLocalClusterView(host2).ConsensusToken.IsCancellationRequested);
         await host2.StartAsync();
 
-        await GetLocalClusterView(host1).WaitForLeaderAsync(DefaultTimeout);
+        await GetLocalClusterView(host1).WaitForLeadershipAsync(DefaultTimeout);
         Equal(GetLocalClusterView(host1).LeadershipToken, GetLocalClusterView(host1).ConsensusToken);
         True(await GetLocalClusterView(host1).AddMemberAsync(GetLocalClusterView(host2).LocalMemberAddress));
 

src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj

@@ -8,7 +8,7 @@
     <ImplicitUsings>true</ImplicitUsings>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.4.0</VersionPrefix>
+    <VersionPrefix>5.5.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>

src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj

@@ -8,7 +8,7 @@
     <Nullable>enable</Nullable>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.4.0</VersionPrefix>
+    <VersionPrefix>5.5.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/CustomTransport/GenericClient.cs

@@ -81,6 +81,7 @@ private protected override async ValueTask<IConnectionContext> ConnectAsync(Canc
     {
         // connection has separated timeout
         using var connectDurationTracker = CancellationTokenSource.CreateLinkedTokenSource(token);
+        connectDurationTracker.CancelAfter(ConnectTimeout);
         return new GenericConnectionContext(await factory.ConnectAsync(EndPoint, connectDurationTracker.Token).ConfigureAwait(false), defaultAllocator);
     }
 }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/CustomTransport/GenericServer.cs

@@ -74,12 +74,12 @@ private async void HandleConnection(ConnectionContext connection, int transmissi
             // reset by client
             logger.ConnectionWasResetByClient(clientAddress);
         }
-        catch (OperationCanceledException) when (tokenSource is not null && tokenSource.CancellationOrigin == connection.ConnectionClosed)
+        catch (OperationCanceledException) when (tokenSource?.CancellationOrigin == connection.ConnectionClosed)
         {
             // closed by client
             logger.ConnectionWasResetByClient(clientAddress);
         }
-        catch (OperationCanceledException) when (tokenSource is not null && tokenSource.CancellationOrigin == lifecycleToken)
+        catch (OperationCanceledException) when (tokenSource?.CancellationOrigin == lifecycleToken)
         {
             // server stopped, suppress exception
         }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/IRaftCluster.cs

@@ -53,7 +53,7 @@ public interface IRaftCluster : IReplicationCluster<IRaftLogEntry>, IPeerMesh<IR
     /// has communication with the leader.
     /// </summary>
     /// <remarks>
-    /// The token moves to canceled state if the current node upgrades to the candidate state or looses connection with the leader.
+    /// The token moves to canceled state if the current node upgrades to the candidate state or loses connection with the leader.
     /// </remarks>
     CancellationToken ConsensusToken { get; }
 
@@ -72,4 +72,16 @@ public interface IRaftCluster : IReplicationCluster<IRaftLogEntry>, IPeerMesh<IR
     /// <returns>The task representing asynchronous result.</returns>
     /// <exception cref="OperationCanceledException">The operation has been canceled.</exception>
     ValueTask ApplyReadBarrierAsync(CancellationToken token = default);
+
+    /// <summary>
+    /// Waits until the local node is elected as the leader.
+    /// </summary>
+    /// <param name="timeout">The time to wait; or <see cref="Timeout.InfiniteTimeSpan"/>.</param>
+    /// <param name="token">The token that can be used to cancel the operation.</param>
+    /// <returns>The leadership token.</returns>
+    /// <exception cref="TimeoutException">The operation is timed out.</exception>
+    /// <exception cref="OperationCanceledException">The operation has been canceled.</exception>
+    /// <exception cref="ObjectDisposedException">The local node is disposed.</exception>
+    /// <seealso cref="LeadershipToken"/>
+    ValueTask<CancellationToken> WaitForLeadershipAsync(TimeSpan timeout, CancellationToken token = default);
 }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Replication.cs

@@ -111,7 +111,7 @@ internal ValueTask<Result<bool>> ReplicateAsync(IAuditTrail<IRaftLogEntry> audit
             var startIndex = replicationIndex + 1L;
             Debug.Assert(startIndex == Member.State.NextIndex);
 
-            logger.ReplicationStarted(Member.EndPoint, startIndex);
+            logger.ReplicationStarted(Member.EndPoint, startIndex, currentIndex);
             return auditTrail.ReadAsync(this, startIndex, currentIndex, token);
         }
 
@@ -168,7 +168,7 @@ private ConfiguredTaskAwaitable<Result<HeartbeatResult>>.ConfiguredTaskAwaiter R
         {
             Debug.Assert(snapshot.IsSnapshot);
 
-            logger.InstallingSnapshot(replicationIndex = snapshotIndex);
+            logger.InstallingSnapshot(Member.EndPoint, replicationIndex = snapshotIndex);
             var result = Member.InstallSnapshotAsync(term, snapshot, snapshotIndex, token).ConfigureAwait(false).GetAwaiter();
             fingerprint = Member.State.ConfigurationFingerprint; // keep local version unchanged
             return result;

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.cs

@@ -163,30 +163,33 @@ private async Task DoHeartbeats(TimeSpan period, IAuditTrail<IRaftLogEntry> audi
                                     goto case MemberResponse.Canceled;
                                 case MemberResponse.Canceled:
                                     return;
-                            }
-
-                            if (++quorum == majority)
-                            {
-                                RenewLease(startTime.Elapsed);
-                                UpdateLeaderStickiness();
-                                await configurationStorage.ApplyAsync(Token).ConfigureAwait(false);
-                            }
-
-                            if (result.Value && ++commitQuorum == majority)
-                            {
-                                // majority of nodes accept entries with at least one entry from the current term
-                                var count = await auditTrail.CommitAsync(currentIndex, Token).ConfigureAwait(false); // commit all entries starting from the first uncommitted index to the end
-                                Logger.CommitSuccessful(currentIndex, count);
+                                case MemberResponse.Successful when ++quorum == majority:
+                                    RenewLease(startTime.Elapsed);
+                                    UpdateLeaderStickiness();
+                                    goto default;
+                                default:
+                                    commitQuorum += Unsafe.BitCast<bool, byte>(result.Value);
+                                    continue;
                             }
                         }
                     }
 
-                    if (commitQuorum < majority)
+                    if (commitQuorum >= majority)
+                    {
+                        // majority of nodes accept entries with at least one entry from the current term
+                        var count = await auditTrail.CommitAsync(currentIndex, Token).ConfigureAwait(false); // commit all entries starting from the first uncommitted index to the end
+                        Logger.CommitSuccessful(currentIndex, count);
+                    }
+                    else
                     {
                         Logger.CommitFailed(quorum, commitIndex);
                     }
 
-                    if (quorum < majority)
+                    if (quorum >= majority)
+                    {
+                        await configurationStorage.ApplyAsync(Token).ConfigureAwait(false);
+                    }
+                    else
                     {
                         MoveToFollowerState(randomizeTimeout: false);
                         return;

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LogMessages.cs

@@ -110,10 +110,10 @@ internal static partial class LogMessages
     [LoggerMessage(
         EventIdOffset + 12,
         LogLevel.Debug,
-        "Replication of member {Member} started at log index {EntryIndex}",
+        "Replication of member {Member} started from {StartIndex} log index to {EndIndex}",
         EventName = $"{EventIdPrefix}.{nameof(ReplicationStarted)}"
     )]
-    public static partial void ReplicationStarted(this ILogger logger, EndPoint member, long entryIndex);
+    public static partial void ReplicationStarted(this ILogger logger, EndPoint member, long startIndex, long endIndex);
 
     [LoggerMessage(
         EventIdOffset + 13,
@@ -158,10 +158,10 @@ internal static partial class LogMessages
     [LoggerMessage(
         EventIdOffset + 18,
         LogLevel.Information,
-        "Installing snapshot at index {EntryIndex}",
+        "Installing snapshot at index {EntryIndex} for member {Member}",
         EventName = $"{EventIdPrefix}.{nameof(InstallingSnapshot)}"
     )]
-    public static partial void InstallingSnapshot(this ILogger logger, long entryIndex);
+    public static partial void InstallingSnapshot(this ILogger logger, EndPoint member, long entryIndex);
 
     public const string SnapshotInstallationFailed = "Critical error detected while installing snapshot of audit trail";
 

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.cs

@@ -35,6 +35,7 @@ public abstract partial class RaftCluster<TMember> : Disposable, IUnresponsiveCl
 
     private volatile RaftState<TMember> state;
     private volatile TaskCompletionSource<TMember> electionEvent;
+    private volatile TaskCompletionSource<CancellationToken> leadershipEvent;
     private InvocationList<Action<RaftCluster<TMember>, TMember?>> leaderChangedHandlers;
     private InvocationList<Action<RaftCluster<TMember>, TMember>> replicationHandlers;
     private volatile int electionTimeout;
@@ -75,6 +76,7 @@ protected RaftCluster(IClusterMemberConfiguration config, in TagList measurement
         readinessProbe = new(TaskCreationOptions.RunContinuationsAsynchronously);
         aggressiveStickiness = config.AggressiveLeaderStickiness;
         electionEvent = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        leadershipEvent = new(TaskCreationOptions.RunContinuationsAsynchronously);
         state = new StandbyState<TMember>(this, TimeSpan.FromMilliseconds(electionTimeout));
         EndPointComparer = config.EndPointComparer;
         this.measurementTags = measurementTags;
@@ -256,22 +258,18 @@ private set
         }
     }
 
-    /// <summary>
-    /// Waits for the leader election asynchronously.
-    /// </summary>
-    /// <param name="timeout">The time to wait; or <see cref="System.Threading.Timeout.InfiniteTimeSpan"/>.</param>
-    /// <param name="token">The token that can be used to cancel the operation.</param>
-    /// <returns>The elected leader.</returns>
-    /// <exception cref="TimeoutException">The operation is timed out.</exception>
-    /// <exception cref="OperationCanceledException">The operation has been canceled.</exception>
-    /// <exception cref="ObjectDisposedException">The local node is disposed.</exception>
+    /// <inheritdoc cref="ICluster.WaitForLeaderAsync(TimeSpan, CancellationToken)"/>
     public Task<TMember> WaitForLeaderAsync(TimeSpan timeout, CancellationToken token = default)
         => electionEvent.Task.WaitAsync(timeout, token);
 
     /// <inheritdoc />
     ValueTask<IClusterMember> ICluster.WaitForLeaderAsync(TimeSpan timeout, CancellationToken token)
         => new(WaitForLeaderAsync(timeout, token).Convert<TMember, IClusterMember>());
 
+    /// <inheritdoc cref="IRaftCluster.WaitForLeadershipAsync(TimeSpan, CancellationToken)"/>
+    public ValueTask<CancellationToken> WaitForLeadershipAsync(TimeSpan timeout, CancellationToken token = default)
+        => new(leadershipEvent.Task.WaitAsync(timeout, token));
+
     private ValueTask UnfreezeAsync()
     {
         ValueTask result;
@@ -441,7 +439,14 @@ private async Task CancelPendingRequestsAsync()
     }
 
     private ValueTask UpdateStateAsync(RaftState<TMember> newState)
-        => Interlocked.Exchange(ref state, newState).DisposeAsync();
+    {
+        if (leadershipEvent is { Task.IsCompletedSuccessfully: true } leadershipEventCopy)
+        {
+            Interlocked.CompareExchange(ref leadershipEvent, new(TaskCreationOptions.RunContinuationsAsynchronously), leadershipEventCopy);
+        }
+
+        return Interlocked.Exchange(ref state, newState).DisposeAsync();
+    }
 
     /// <summary>
     /// Stops serving local member.
@@ -1118,11 +1123,13 @@ async void IRaftStateMachine<TMember>.MoveToLeaderState(IRaftStateMachine.IWeakC
                 };
 
                 await UpdateStateAsync(newState).ConfigureAwait(false);
+                await auditTrail.AppendNoOpEntry(LifecycleToken).ConfigureAwait(false);
 
+                // ensure that the leader is visible to the consumers after no-op entry is added to the log (which acts as a write barrier)
                 Leader = newLeader;
-                await auditTrail.AppendNoOpEntry(LifecycleToken).ConfigureAwait(false);
-                newState.StartLeading(HeartbeatTimeout, auditTrail, ConfigurationStorage);
+                leadershipEvent.TrySetResult(newState.Token);
 
+                newState.StartLeading(HeartbeatTimeout, auditTrail, ConfigurationStorage);
                 Logger.TransitionToLeaderStateCompleted(currentTerm);
             }
         }
@@ -1293,6 +1300,7 @@ protected override void Dispose(bool disposing)
             memberAddedHandlers = memberRemovedHandlers = default;
             leaderChangedHandlers = default;
             TrySetDisposedException(electionEvent);
+            TrySetDisposedException(leadershipEvent);
         }
 
         base.Dispose(disposing);

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/Tcp/TcpClient.cs

@@ -121,7 +121,7 @@ private protected override async ValueTask<IConnectionContext> ConnectAsync(Canc
         try
         {
             connectDurationTracker.CancelAfter(ConnectTimeout);
-            await socket.ConnectAsync(EndPoint, token).ConfigureAwait(false);
+            await socket.ConnectAsync(EndPoint, connectDurationTracker.Token).ConfigureAwait(false);
         }
         catch
         {