wip

vazois · vazois · commit e8169d562267 · 2025-03-18T16:15:49.000-07:00
diff --git a/libs/cluster/Server/Replication/PrimaryOps/DisklessReplication/ReplicaSyncSession.cs b/libs/cluster/Server/Replication/PrimaryOps/DisklessReplication/ReplicaSyncSession.cs
@@ -264,6 +264,9 @@ public async Task BeginAofSync()
 
                 logger?.LogSyncMetadata(LogLevel.Trace, "BeginAofSync", replicaSyncMetadata, recoverSyncMetadata);
 
+                // Check what happens if we fail after recovery and start AOF stream
+                ExceptionScenarioHelper.TriggerException(ExceptionScenario.FAIL_RIGHT_BEFORE_AOF_STREAM_STARTS);
+
                 // We have already added the iterator for the covered address above but replica might request an address
                 // that is ahead of the covered address so we should start streaming from that address in order not to
                 // introduce duplicate insertions.
@@ -275,7 +278,7 @@ public async Task BeginAofSync()
             catch (Exception ex)
             {
                 logger?.LogError(ex, "{method}", $"{nameof(ReplicaSyncSession.BeginAofSync)}");
-                SetStatus(SyncStatus.FAILED, "Failed to begin AOF sync");
+                SetStatus(SyncStatus.FAILED, ex.Message);
                 AofSyncTask?.Dispose();
             }
         }
diff --git a/libs/cluster/Server/Replication/PrimaryOps/ReplicaSyncSession.cs b/libs/cluster/Server/Replication/PrimaryOps/ReplicaSyncSession.cs
@@ -308,7 +308,7 @@ public async Task<bool> SendCheckpoint()
             {
                 logger?.LogError(ex, "acquiredEntry: {cEntryDump}", localEntry.GetCheckpointEntryDump());
                 if (aofSyncTaskInfo != null) _ = clusterProvider.replicationManager.TryRemoveReplicationTask(aofSyncTaskInfo);
-                errorMsg = "ERR " + ex.Message;// this is error sent to remote client
+                errorMsg = ex.Message;// this is error sent to remote client
                 return false;
             }
             finally
diff --git a/libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs b/libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs
@@ -38,15 +38,26 @@ public bool TryReplicateDisklessSync(
 
                 // Wait for threads to agree configuration change of this node
                 session.UnsafeBumpAndWaitForEpochTransition();
-                _ = Task.Run(() => TryBeginReplicaSync());
+
+                if (background)
+                    _ = Task.Run(() => TryBeginReplicaSync());
+                else
+                {
+                    var resp = TryBeginReplicaSync().GetAwaiter().GetResult();
+                    if (resp != null)
+                    {
+                        errorMessage = Encoding.ASCII.GetBytes(resp);
+                        return false;
+                    }
+                }
             }
             catch (Exception ex)
             {
                 logger?.LogError(ex, $"{nameof(TryReplicateDisklessSync)}");
                 replicateLock.WriteUnlock();
             }
 
-            async Task TryBeginReplicaSync()
+            async Task<string> TryBeginReplicaSync()
             {
                 var disklessSync = clusterProvider.serverOptions.ReplicaDisklessSync;
                 var disableObjects = clusterProvider.serverOptions.DisableObjects;
@@ -84,7 +95,7 @@ async Task TryBeginReplicaSync()
                     {
                         var errorMsg = Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_NOT_ASSIGNED_PRIMARY_ERROR);
                         logger?.LogError("{msg}", errorMsg);
-                        return;
+                        return errorMsg;
                     }
 
                     gcs = new(
@@ -118,6 +129,7 @@ async Task TryBeginReplicaSync()
                 {
                     logger?.LogError(ex, $"{nameof(TryBeginReplicaSync)}");
                     clusterProvider.clusterManager.TryResetReplica();
+                    return ex.Message;
                 }
                 finally
                 {
@@ -126,6 +138,7 @@ async Task TryBeginReplicaSync()
                     gcs?.Dispose();
                     recvCheckpointHandler?.Dispose();
                 }
+                return null;
             }
 
             return true;
diff --git a/libs/cluster/Session/RespClusterReplicationCommands.cs b/libs/cluster/Session/RespClusterReplicationCommands.cs
@@ -59,7 +59,7 @@ private bool NetworkClusterReplicate(out bool invalidParameters)
                 return true;
             }
 
-            var background = false;
+            var background = true;
             var nodeId = parseState.GetString(0);
 
             if (parseState.Count > 1)
diff --git a/test/Garnet.test.cluster/ClusterTestContext.cs b/test/Garnet.test.cluster/ClusterTestContext.cs
@@ -65,7 +65,7 @@ public void TearDown()
             waiter?.Dispose();
             clusterTestUtils?.Dispose();
             loggerFactory?.Dispose();
-            var timeoutSeconds = 5;
+            var timeoutSeconds = 60;
             if (!Task.Run(() => DisposeCluster()).Wait(TimeSpan.FromSeconds(timeoutSeconds)))
                 logger?.LogError("Timed out waiting for DisposeCluster");
             if (!Task.Run(() => TestUtils.DeleteDirectory(TestFolder, true)).Wait(TimeSpan.FromSeconds(timeoutSeconds)))
diff --git a/test/Garnet.test.cluster/ClusterTestUtils.cs b/test/Garnet.test.cluster/ClusterTestUtils.cs
@@ -1879,7 +1879,7 @@ public List<SlotItem> ClusterSlots(IPEndPoint endPoint, ILogger logger = null)
             }
         }
 
-        public string ClusterReplicate(int replicaNodeIndex, int primaryNodeIndex, bool failEx = true, ILogger logger = null)
+        public string ClusterReplicate(int replicaNodeIndex, int primaryNodeIndex, bool async = false, bool failEx = true, ILogger logger = null)
         {
             var primaryId = ClusterMyId(primaryNodeIndex, logger: logger);
             return ClusterReplicate(replicaNodeIndex, primaryId, failEx: failEx, logger: logger);
@@ -1893,7 +1893,7 @@ public string ClusterReplicate(IPEndPoint endPoint, string primaryNodeId, bool a
             try
             {
                 var server = redis.GetServer(endPoint);
-                List<object> args = async ? ["replicate", primaryNodeId, "async"] : ["replicate", primaryNodeId];
+                List<object> args = async ? ["replicate", primaryNodeId, "async"] : ["replicate", primaryNodeId, "sync"];
                 var result = (string)server.Execute("cluster", args);
                 ClassicAssert.AreEqual("OK", result);
                 return result;
diff --git a/test/Garnet.test.cluster/ReplicationTests/ClusterReplicationBaseTests.cs b/test/Garnet.test.cluster/ReplicationTests/ClusterReplicationBaseTests.cs
@@ -3,7 +3,6 @@
 
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.IO;
 using System.Linq;
 using System.Net;
@@ -1174,16 +1173,16 @@ public void ClusterReplicationCheckpointAlignmentTest([Values] bool performRMW)
             }
         }
 
-        [Test, Order(23)]
-        [Conditional("DEBUG")]
-        public void ClusterExceptionAtReplicaAofSyncStart()
+        //[Test, Order(23)]
+        //[Conditional("DEBUG")]
+        public void ClusterExceptionAtReplicaAofSyncStart([Values] bool enableDisklessSync)
         {
             ExceptionScenarioHelper.EnableException(ExceptionScenario.FAIL_RIGHT_BEFORE_AOF_STREAM_STARTS);
 
             var primaryIndex = 0;
             var replicaIndex = 1;
             var nodes_count = 2;
-            context.CreateInstances(nodes_count, disableObjects: false, enableAOF: true, useTLS: useTLS);
+            context.CreateInstances(nodes_count, disableObjects: false, enableAOF: true, useTLS: useTLS, enableDisklessSync: enableDisklessSync, timeout: timeout);
             context.CreateConnection(useTLS: useTLS);
 
             _ = context.clusterTestUtils.AddDelSlotsRange(primaryIndex, [(0, 16383)], addslot: true, logger: context.logger);
@@ -1195,13 +1194,13 @@ public void ClusterExceptionAtReplicaAofSyncStart()
             context.clusterTestUtils.WaitUntilNodeIsKnown(primaryIndex, replicaIndex, logger: context.logger);
 
             var resp = context.clusterTestUtils.ClusterReplicate(replicaNodeIndex: replicaIndex, primaryNodeIndex: primaryIndex, failEx: false, logger: context.logger);
-            ClassicAssert.AreEqual("ERR Debug scenario triggered FAIL_AFTER_RECOVERY_AND_BEFORE_ATTACH", resp);
+            ClassicAssert.AreEqual("Debug scenario triggered FAIL_RIGHT_BEFORE_AOF_STREAM_STARTS", resp);
 
             var role = context.clusterTestUtils.RoleCommand(replicaIndex, logger: context.logger);
             ClassicAssert.AreEqual("master", role.Value);
 
             resp = context.clusterTestUtils.ClusterReplicate(replicaNodeIndex: replicaIndex, primaryNodeIndex: primaryIndex, failEx: false, logger: context.logger);
-            ClassicAssert.AreEqual("ERR Debug scenario triggered FAIL_AFTER_RECOVERY_AND_BEFORE_ATTACH", resp);
+            ClassicAssert.AreEqual("Debug scenario triggered FAIL_RIGHT_BEFORE_AOF_STREAM_STARTS", resp);
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -264,6 +264,9 @@ public async Task BeginAofSync()`
`264`	`264`
`265`	`265`	`logger?.LogSyncMetadata(LogLevel.Trace, "BeginAofSync", replicaSyncMetadata, recoverSyncMetadata);`
`266`	`266`
	`267`	`+ // Check what happens if we fail after recovery and start AOF stream`
	`268`	`+ ExceptionScenarioHelper.TriggerException(ExceptionScenario.FAIL_RIGHT_BEFORE_AOF_STREAM_STARTS);`
	`269`	`+`
`267`	`270`	`// We have already added the iterator for the covered address above but replica might request an address`
`268`	`271`	`// that is ahead of the covered address so we should start streaming from that address in order not to`
`269`	`272`	`// introduce duplicate insertions.`
`@@ -275,7 +278,7 @@ public async Task BeginAofSync()`
`275`	`278`	`catch (Exception ex)`
`276`	`279`	`{`
`277`	`280`	`logger?.LogError(ex, "{method}", $"{nameof(ReplicaSyncSession.BeginAofSync)}");`
`278`		`- SetStatus(SyncStatus.FAILED, "Failed to begin AOF sync");`
	`281`	`+ SetStatus(SyncStatus.FAILED, ex.Message);`
`279`	`282`	`AofSyncTask?.Dispose();`
`280`	`283`	`}`
`281`	`284`	`}`
Original file line number	Diff line number	Diff line change
`@@ -308,7 +308,7 @@ public async Task<bool> SendCheckpoint()`
`308`	`308`	`{`
`309`	`309`	`logger?.LogError(ex, "acquiredEntry: {cEntryDump}", localEntry.GetCheckpointEntryDump());`
`310`	`310`	`if (aofSyncTaskInfo != null) _ = clusterProvider.replicationManager.TryRemoveReplicationTask(aofSyncTaskInfo);`
`311`		`- errorMsg = "ERR " + ex.Message;// this is error sent to remote client`
	`311`	`+ errorMsg = ex.Message;// this is error sent to remote client`
`312`	`312`	`return false;`
`313`	`313`	`}`
`314`	`314`	`finally`
Original file line number	Diff line number	Diff line change
`@@ -38,15 +38,26 @@ public bool TryReplicateDisklessSync(`
`38`	`38`
`39`	`39`	`// Wait for threads to agree configuration change of this node`
`40`	`40`	`session.UnsafeBumpAndWaitForEpochTransition();`
`41`		`- _ = Task.Run(() => TryBeginReplicaSync());`
	`41`	`+`
	`42`	`+ if (background)`
	`43`	`+ _ = Task.Run(() => TryBeginReplicaSync());`
	`44`	`+ else`
	`45`	`+ {`
	`46`	`+ var resp = TryBeginReplicaSync().GetAwaiter().GetResult();`
	`47`	`+ if (resp != null)`
	`48`	`+ {`
	`49`	`+ errorMessage = Encoding.ASCII.GetBytes(resp);`
	`50`	`+ return false;`
	`51`	`+ }`
	`52`	`+ }`
`42`	`53`	`}`
`43`	`54`	`catch (Exception ex)`
`44`	`55`	`{`
`45`	`56`	`logger?.LogError(ex, $"{nameof(TryReplicateDisklessSync)}");`
`46`	`57`	`replicateLock.WriteUnlock();`
`47`	`58`	`}`
`48`	`59`
`49`		`- async Task TryBeginReplicaSync()`
	`60`	`+ async Task<string> TryBeginReplicaSync()`
`50`	`61`	`{`
`51`	`62`	`var disklessSync = clusterProvider.serverOptions.ReplicaDisklessSync;`
`52`	`63`	`var disableObjects = clusterProvider.serverOptions.DisableObjects;`
`@@ -84,7 +95,7 @@ async Task TryBeginReplicaSync()`
`84`	`95`	`{`
`85`	`96`	`var errorMsg = Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_NOT_ASSIGNED_PRIMARY_ERROR);`
`86`	`97`	`logger?.LogError("{msg}", errorMsg);`
`87`		`- return;`
	`98`	`+ return errorMsg;`
`88`	`99`	`}`
`89`	`100`
`90`	`101`	`gcs = new(`
`@@ -118,6 +129,7 @@ async Task TryBeginReplicaSync()`
`118`	`129`	`{`
`119`	`130`	`logger?.LogError(ex, $"{nameof(TryBeginReplicaSync)}");`
`120`	`131`	`clusterProvider.clusterManager.TryResetReplica();`
	`132`	`+ return ex.Message;`
`121`	`133`	`}`
`122`	`134`	`finally`
`123`	`135`	`{`
`@@ -126,6 +138,7 @@ async Task TryBeginReplicaSync()`
`126`	`138`	`gcs?.Dispose();`
`127`	`139`	`recvCheckpointHandler?.Dispose();`
`128`	`140`	`}`
	`141`	`+ return null;`
`129`	`142`	`}`
`130`	`143`
`131`	`144`	`return true;`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ private bool NetworkClusterReplicate(out bool invalidParameters)`
`59`	`59`	`return true;`
`60`	`60`	`}`
`61`	`61`
`62`		`- var background = false;`
	`62`	`+ var background = true;`
`63`	`63`	`var nodeId = parseState.GetString(0);`
`64`	`64`
`65`	`65`	`if (parseState.Count > 1)`
Original file line number	Diff line number	Diff line change
`@@ -1879,7 +1879,7 @@ public List<SlotItem> ClusterSlots(IPEndPoint endPoint, ILogger logger = null)`
`1879`	`1879`	`}`
`1880`	`1880`	`}`
`1881`	`1881`
`1882`		`- public string ClusterReplicate(int replicaNodeIndex, int primaryNodeIndex, bool failEx = true, ILogger logger = null)`
	`1882`	`+ public string ClusterReplicate(int replicaNodeIndex, int primaryNodeIndex, bool async = false, bool failEx = true, ILogger logger = null)`
`1883`	`1883`	`{`
`1884`	`1884`	`var primaryId = ClusterMyId(primaryNodeIndex, logger: logger);`
`1885`	`1885`	`return ClusterReplicate(replicaNodeIndex, primaryId, failEx: failEx, logger: logger);`
`@@ -1893,7 +1893,7 @@ public string ClusterReplicate(IPEndPoint endPoint, string primaryNodeId, bool a`
`1893`	`1893`	`try`
`1894`	`1894`	`{`
`1895`	`1895`	`var server = redis.GetServer(endPoint);`
`1896`		`- List<object> args = async ? ["replicate", primaryNodeId, "async"] : ["replicate", primaryNodeId];`
	`1896`	`+ List<object> args = async ? ["replicate", primaryNodeId, "async"] : ["replicate", primaryNodeId, "sync"];`
`1897`	`1897`	`var result = (string)server.Execute("cluster", args);`
`1898`	`1898`	`ClassicAssert.AreEqual("OK", result);`
`1899`	`1899`	`return result;`