Fixing replica checkpointing divergence with explicit CommitAOF calls (#1305)

* repro for failure

* move test to appropriate file; implement RoleRead lock for recovery, fix replica commits (still have other paths to lock down)

* cleanup

* extend role prevention changes to all ops on StoreApi; refactor a bit so we can express this with a using; save reusable instance of the locking object, acquire in one place

* derp; restore checking the actual role -_-

* failing test for if ReplicaSyncTask dies - killing it the same we've seen in practice, which is the replica throwing causing the connection from the primary to be torn down

* restore ignored parameter

* this works, but I don't love the polling

* Revert "this works, but I don't love the polling"

This reverts commit 7a05d61.

* attempt to resume replication in response to a GOSSIP and the fact that no active replication task exists

* throttle re-establishingment attempts; put behind a setting

* include new setting in .conf

* introduce notion of an upgradeable read lock for role reading, and hold it while resuming replication; prevents a TOCTOU issue with resuming replication from a replica

* downgrade logic was jacked, correct

* big ol' test for our particular setup (1:1 primary replica, matched settings, out of band commits); example of Replica losing all data if intervention for promotion not timely; allow Replica to come up AS a Replica even if its Primary is down

* wait for replica takeover to complete before proceeding in test

* add test for Replica comes back up as Replica, even if Primary unreachable

* fault on Primary side too, knock out that todo

* formatting

* on Replicas, wait until we can connect to a Primary to toss any loaded data; punch a config in to allow Replicas to load checkpoint & aof from disk; clean up some new tests for reliability

* update defaults.conf with new setting

* restore cancellation for test

* add tests for new configs

* address feedback; wrap this giant list of parameters in a record struct

* address feedback; do not block GOSSIP message while resyncing, move that work onto a background task

* formatting

* address feedback; default values in GarnetServerOptions

* kick up KeraLua so Windows Lua bits have CFG

* per call discussion; bump version

---------

Co-authored-by: Vasileios Zois <[email protected]>

Author: kevin-montrose

Directory.Packages.props

@@ -9,7 +9,7 @@
     <PackageVersion Include="BenchmarkDotNet.Diagnostics.Windows" Version="0.13.12" />
     <PackageVersion Include="CommandLineParser" Version="2.9.1" />
     <PackageVersion Include="JsonPath.Net" Version="1.1.6" />
-    <PackageVersion Include="KeraLua" Version="1.4.4" />
+    <PackageVersion Include="KeraLua" Version="1.4.6" />
     <PackageVersion Include="Microsoft.Identity.Client" Version="4.73.1" />
     <PackageVersion Include="NUnit" Version="4.1.0" />
     <PackageVersion Include="NUnit3TestAdapter" Version="4.6.0" />

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- VersionPrefix property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.79</VersionPrefix>
+		<VersionPrefix>1.0.80</VersionPrefix>
 	</PropertyGroup>
 </Project>

benchmark/BDN.benchmark/Embedded/GarnetServerEmbedded.cs

@@ -18,6 +18,28 @@ public GarnetServerEmbedded() : base(new IPEndPoint(IPAddress.Loopback, 0), 1 <<
         {
         }
 
+        /// <inheritdoc/>
+        public override IEnumerable<IMessageConsumer> ActiveConsumers()
+        {
+            foreach (var kvp in activeHandlers)
+            {
+                var consumer = kvp.Key.Session;
+                if (consumer != null)
+                    yield return consumer;
+            }
+        }
+
+        /// <inheritdoc/>
+        public override IEnumerable<IClusterSession> ActiveClusterSessions()
+        {
+            foreach (var kvp in activeHandlers)
+            {
+                var consumer = kvp.Key.Session;
+                if (consumer != null)
+                    yield return ((RespServerSession)consumer).clusterSession;
+            }
+        }
+
         public EmbeddedNetworkHandler CreateNetworkHandler(SslClientAuthenticationOptions tlsOptions = null, string remoteEndpointName = null)
         {
             var networkSender = new EmbeddedNetworkSender();

libs/cluster/Server/ClusterManagerWorkerState.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Text;
 using System.Threading;
 using Garnet.common;
@@ -141,11 +142,17 @@ public ReadOnlySpan<byte> TryReset(bool soft, int expirySeconds = 60)
         /// Try to make this node a replica of node with nodeid
         /// </summary>
         /// <param name="nodeid"></param>
-        /// <param name="force">Check if node is clean (i.e. is PRIMARY without any assigned nodes)</param>
+        /// <param name="force">If false, checks if node is clean (i.e. is PRIMARY without any assigned nodes) before making changes.</param>
+        /// <param name="upgradeLock">If true, allows for a <see cref="RecoveryStatus.ReadRole"/> read lock to be upgraded to <see cref="RecoveryStatus.ClusterReplicate"/>.</param>
         /// <param name="errorMessage">The ASCII encoded error response if the method returned <see langword="false"/>; otherwise <see langword="default"/></param>
         /// <param name="logger"></param>
-        public bool TryAddReplica(string nodeid, bool force, out ReadOnlySpan<byte> errorMessage, ILogger logger = null)
+        public bool TryAddReplica(string nodeid, bool force, bool upgradeLock, out ReadOnlySpan<byte> errorMessage, ILogger logger = null)
         {
+            Debug.Assert(
+                !upgradeLock || clusterProvider.replicationManager.currentRecoveryStatus == RecoveryStatus.ReadRole,
+                "Lock upgrades are only allowed if caller holds a ReadRole lock"
+            );
+
             errorMessage = default;
             while (true)
             {
@@ -188,7 +195,7 @@ public bool TryAddReplica(string nodeid, bool force, out ReadOnlySpan<byte> erro
 
                 // Transition to recovering state
                 // Only one caller will succeed in becoming a replica for the provided node-id
-                if (!clusterProvider.replicationManager.BeginRecovery(RecoveryStatus.ClusterReplicate))
+                if (!clusterProvider.replicationManager.BeginRecovery(RecoveryStatus.ClusterReplicate, upgradeLock))
                 {
                     logger?.LogError($"{nameof(TryAddReplica)}: {{logMessage}}", Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
                     errorMessage = CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK;
@@ -200,7 +207,14 @@ public bool TryAddReplica(string nodeid, bool force, out ReadOnlySpan<byte> erro
                     break;
 
                 // If we reach here then we failed to update config so we need to suspend recovery and retry to update the config
-                clusterProvider.replicationManager.EndRecovery(RecoveryStatus.NoRecovery);
+                if (upgradeLock)
+                {
+                    clusterProvider.replicationManager.EndRecovery(RecoveryStatus.ReadRole, downgradeLock: true);
+                }
+                else
+                {
+                    clusterProvider.replicationManager.EndRecovery(RecoveryStatus.NoRecovery, downgradeLock: false);
+                }
             }
             FlushConfig();
             return true;

libs/cluster/Server/ClusterProvider.cs

@@ -84,6 +84,14 @@ public void Recover()
             replicationManager.Recover();
         }
 
+        /// <inheritdoc />
+        public bool PreventRoleChange()
+        => replicationManager.BeginRecovery(RecoveryStatus.ReadRole, upgradeLock: false);
+
+        /// <inheritdoc />
+        public void AllowRoleChange()
+        => replicationManager.EndRecovery(RecoveryStatus.NoRecovery, downgradeLock: false);
+
         /// <inheritdoc />
         public void Start()
         {

libs/cluster/Server/Failover/ReplicaFailoverSession.cs

@@ -155,7 +155,7 @@ private bool TakeOverAsPrimary()
             try
             {
                 // Make replica syncing unavailable by setting recovery flag
-                if (!clusterProvider.replicationManager.BeginRecovery(RecoveryStatus.ClusterFailover))
+                if (!clusterProvider.replicationManager.BeginRecovery(RecoveryStatus.ClusterFailover, upgradeLock: false))
                 {
                     logger?.LogWarning($"{nameof(TakeOverAsPrimary)}: {{logMessage}}", Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
                     return false;
@@ -181,7 +181,7 @@ private bool TakeOverAsPrimary()
             finally
             {
                 // Disable recovering as now this node has become a primary or failed in its attempt earlier
-                if (acquiredLock) clusterProvider.replicationManager.EndRecovery(RecoveryStatus.NoRecovery);
+                if (acquiredLock) clusterProvider.replicationManager.EndRecovery(RecoveryStatus.NoRecovery, downgradeLock: false);
             }
 
             return true;

libs/cluster/Server/Replication/PrimaryOps/AofSyncTaskInfo.cs

@@ -5,6 +5,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Garnet.client;
+using Garnet.common;
 using Microsoft.Extensions.Logging;
 using Tsavorite.core;
 
@@ -69,6 +70,8 @@ public unsafe void Consume(byte* payloadPtr, int payloadLength, long currentAddr
         {
             try
             {
+                ExceptionInjectionHelper.TriggerException(ExceptionInjectionType.Aof_Sync_Task_Consume);
+
                 // logger?.LogInformation("Sending {payloadLength} bytes to {remoteNodeId} at address {currentAddress}-{nextAddress}", payloadLength, remoteNodeId, currentAddress, nextAddress);
 
                 // This is called under epoch protection, so we have to wait for appending to complete

libs/cluster/Server/Replication/RecoveryStatus.cs

@@ -32,5 +32,9 @@ public enum RecoveryStatus : byte
         /// Replica has recovered the checkpoint after signal from primary
         /// </summary>
         CheckpointRecoveredAtReplica,
+        /// <summary>
+        /// Need to ensure a node does not change its role during a commit or checkpoint
+        /// </summary>
+        ReadRole,
     }
 }

libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs

@@ -6,6 +6,7 @@
 using System.Text;
 using System.Threading.Tasks;
 using Garnet.client;
+using Garnet.cluster.Server.Replication;
 using Microsoft.Extensions.Logging;
 
 namespace Garnet.cluster
@@ -15,36 +16,30 @@ internal sealed partial class ReplicationManager : IDisposable
         /// <summary>
         /// Try to replicate using diskless sync
         /// </summary>
-        /// <param name="session"></param>
-        /// <param name="nodeId"></param>
-        /// <param name="background"></param>
-        /// <param name="force"></param>
-        /// <param name="tryAddReplica"></param>
-        /// <param name="errorMessage"></param>
-        /// <returns></returns>
+        /// <param name="session">ClusterSession for this connection.</param>
+        /// <param name="options">Options for the sync.</param>
+        /// <param name="errorMessage">The ASCII encoded error message if the method returned <see langword="false"/>; otherwise <see langword="default"/></param>
+        /// <returns>A boolean indicating whether replication initiation was successful.</returns>
         public bool TryReplicateDisklessSync(
             ClusterSession session,
-            string nodeId,
-            bool background,
-            bool force,
-            bool tryAddReplica,
+            ReplicateSyncOptions options,
             out ReadOnlySpan<byte> errorMessage)
         {
             errorMessage = default;
 
             try
             {
-                logger?.LogTrace("CLUSTER REPLICATE {nodeid}", nodeId);
-                if (!clusterProvider.clusterManager.TryAddReplica(nodeId, force: force, out errorMessage, logger: logger))
+                logger?.LogTrace("CLUSTER REPLICATE {nodeid}", options.NodeId);
+                if (options.TryAddReplica && !clusterProvider.clusterManager.TryAddReplica(options.NodeId, options.Force, options.UpgradeLock, out errorMessage, logger: logger))
                     return false;
 
                 // Wait for threads to agree configuration change of this node
                 session.UnsafeBumpAndWaitForEpochTransition();
-                if (background)
-                    _ = Task.Run(() => TryBeginReplicaSync());
+                if (options.Background)
+                    _ = Task.Run(() => TryBeginReplicaSync(options.UpgradeLock));
                 else
                 {
-                    var result = TryBeginReplicaSync().Result;
+                    var result = TryBeginReplicaSync(options.UpgradeLock).Result;
                     if (result != null)
                     {
                         errorMessage = Encoding.ASCII.GetBytes(result);
@@ -58,7 +53,7 @@ public bool TryReplicateDisklessSync(
             }
             return true;
 
-            async Task<string> TryBeginReplicaSync()
+            async Task<string> TryBeginReplicaSync(bool downgradeLock)
             {
                 var disklessSync = clusterProvider.serverOptions.ReplicaDisklessSync;
                 var disableObjects = clusterProvider.serverOptions.DisableObjects;
@@ -129,12 +124,24 @@ async Task<string> TryBeginReplicaSync()
                 catch (Exception ex)
                 {
                     logger?.LogError(ex, $"{nameof(TryBeginReplicaSync)}");
-                    clusterProvider.clusterManager.TryResetReplica();
+
+                    if (options.AllowReplicaResetOnFailure)
+                    {
+                        clusterProvider.clusterManager.TryResetReplica();
+                    }
+
                     return ex.Message;
                 }
                 finally
                 {
-                    EndRecovery(RecoveryStatus.NoRecovery);
+                    if (downgradeLock)
+                    {
+                        EndRecovery(RecoveryStatus.ReadRole, downgradeLock: true);
+                    }
+                    else
+                    {
+                        EndRecovery(RecoveryStatus.NoRecovery, downgradeLock: false);
+                    }
                     gcs?.Dispose();
                     recvCheckpointHandler?.Dispose();
                 }
@@ -188,7 +195,7 @@ public long ReplicaRecoverDiskless(SyncMetadata primarySyncMetadata, out ReadOnl
             finally
             {
                 // Done with recovery at this point
-                EndRecovery(RecoveryStatus.CheckpointRecoveredAtReplica);
+                EndRecovery(RecoveryStatus.CheckpointRecoveredAtReplica, downgradeLock: false);
             }
         }
     }