Rework SimpleSetupCluster in cluster tests, and longer timeouts for select tests (#1762)

* rework SimpleSetupCluster so it's 1) async 2) checks for success throughout; async's a few more tests; adds some null checks

* bump timeouts on longer running cluster vector set tests; plumb [CancelAfter] so ClusterTestContext knows of it

* formatting

* address feedback

* this CancelAfter was not being enforced, as is too aggressive - remove it

* remove some debugging changes that slipped in

Author: kevin-montrose

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -251,7 +251,7 @@ or FailoverStatus.FAILOVER_IN_PROGRESS
                     return;
                 }
 
-                logger.LogInformation("Beginning resync to {primaryId} after replication session failed", primaryId);
+                logger?.LogInformation("Beginning resync to {primaryId} after replication session failed", primaryId);
 
                 // At this point we need to hold the lock until this upcoming task completes
                 suppressUnlock = true;
@@ -270,16 +270,16 @@ or FailoverStatus.FAILOVER_IN_PROGRESS
 
                             if (success)
                             {
-                                logger.LogInformation("Resync to {primaryId} successfully started", primaryId);
+                                logger?.LogInformation("Resync to {primaryId} successfully started", primaryId);
                             }
                             else
                             {
-                                logger.LogWarning("Failed to resync to {primaryId} after replication session failed: {errorMessage}", primaryId, Encoding.UTF8.GetString(errorMessage.Span));
+                                logger?.LogWarning("Failed to resync to {primaryId} after replication session failed: {errorMessage}", primaryId, Encoding.UTF8.GetString(errorMessage.Span));
                             }
                         }
                         catch (Exception ex)
                         {
-                            logger.LogError(ex, "Error encountered on replication recovery background task");
+                            logger?.LogError(ex, "Error encountered on replication recovery background task");
                         }
                         finally
                         {

libs/cluster/Session/RespClusterBasicCommands.cs

@@ -374,7 +374,9 @@ private bool NetworkClusterGossip(out bool invalidParameters)
 
                 Debug.Assert(withMeetSpan.EqualsUpperCaseSpanIgnoringCase(CmdStrings.WITHMEET));
                 if (withMeetSpan.EqualsUpperCaseSpanIgnoringCase(CmdStrings.WITHMEET))
+                {
                     gossipWithMeet = true;
+                }
             }
 
             var gossipMessage = parseState.GetArgSliceByRef(currTokenIdx).SpanByte.ToByteArray();
@@ -406,7 +408,9 @@ private bool NetworkClusterGossip(out bool invalidParameters)
                     RemoteNodeId = other.LocalNodeId;
                 }
                 else
+                {
                     logger?.LogWarning("Received gossip from unknown node: {node-id}", other.LocalNodeId);
+                }
             }
 
             // Respond if configuration has changed or gossipWithMeet option is specified

test/Garnet.test.cluster/ClusterConfigTests.cs

@@ -63,7 +63,7 @@ public void ClusterConfigInitializesUnassignedWorkerTest()
         }
 
         [Test, Order(2)]
-        [Category("CLUSTER-CONFIG"), CancelAfter(1000)]
+        [Category("CLUSTER-CONFIG")]
         public void ClusterForgetAfterNodeRestartTest()
         {
             int nbInstances = 4;

test/Garnet.test.cluster/ClusterMigrateTests.cs

@@ -1404,7 +1404,7 @@ public void ClusterSimpleMigrateWithAuth(List<int> migrateRange)
 
         [Test, Order(15)]
         [Category("CLUSTER")]
-        public void ClusterAllowWritesDuringMigrateTest()
+        public async Task ClusterAllowWritesDuringMigrateTestAsync()
         {
             context.logger.LogDebug("0. ClusterSimpleMigrateTestWithReadWrite started");
             var Shards = defaultShards;
@@ -1428,7 +1428,7 @@ public void ClusterAllowWritesDuringMigrateTest()
 
             // Get slot mapping
             var slot = context.clusterTestUtils.ClusterKeySlot(srcNode.EndPoint.ToIPEndPoint(), Encoding.ASCII.GetString(keyExists));
-            var tgtNode = context.clusterTestUtils.GetAnyOtherNode(srcNode.EndPoint.ToIPEndPoint(), logger: context.logger);
+            var tgtNode = await context.clusterTestUtils.GetAnyOtherNodeAsync(srcNode.EndPoint.ToIPEndPoint(), logger: context.logger).ConfigureAwait(false);
 
             // Set slot to IMPORTING state
             var resp = context.clusterTestUtils.SetSlot(tgtNode.EndPoint.ToIPEndPoint(), slot, "IMPORTING", srcNode.NodeId);

test/Garnet.test.cluster/ClusterNegativeTests.cs

@@ -628,7 +628,7 @@ public void ClusterReplicateFails()
         }
         [Test, Order(14), CancelAfter(testTimeout)]
         [Category("REPLICATION")]
-        public void ClusterFailoverSucceedsDuringEnsureReplication(CancellationToken cancellationToken)
+        public async Task ClusterFailoverSucceedsDuringEnsureReplicationAsync(CancellationToken cancellationToken)
         {
             // Verify that EnsureReplication does not block an in-flight failover.
             // EnsureReplication polls for dropped replication sessions and attempts auto-resync.
@@ -653,7 +653,7 @@ public void ClusterFailoverSucceedsDuringEnsureReplication(CancellationToken can
             var resp = context.clusterTestUtils.ClusterReplicate(replicaNodeIndex: replicaIndex, primaryNodeIndex: primaryIndex, logger: context.logger);
             ClassicAssert.AreEqual("OK", resp);
             context.clusterTestUtils.WaitForReplicaRecovery(replicaIndex, context.logger);
-            context.clusterTestUtils.WaitForConnectedReplicaCount(primaryIndex, 1, context.logger);
+            await context.clusterTestUtils.WaitForConnectedReplicaCountAsync(primaryIndex, 1, context.logger).ConfigureAwait(false);
 
             // Populate primary and wait for sync
             context.kvPairs = [];
@@ -686,7 +686,7 @@ public void ClusterFailoverSucceedsDuringEnsureReplication(CancellationToken can
 
         [Test, Order(15), CancelAfter(testTimeout)]
         [Category("REPLICATION")]
-        public void ClusterEnsureReplicationWorksAfterFailover(CancellationToken cancellationToken)
+        public async Task ClusterEnsureReplicationWorksAfterFailoverAsync(CancellationToken cancellationToken)
         {
             // Verify that EnsureReplication still functions after a failover completes.
             // The failover guard should only suppress auto-resync during active failover states,
@@ -710,7 +710,7 @@ public void ClusterEnsureReplicationWorksAfterFailover(CancellationToken cancell
             var resp = context.clusterTestUtils.ClusterReplicate(replicaNodeIndex: replicaIndex, primaryNodeIndex: primaryIndex, logger: context.logger);
             ClassicAssert.AreEqual("OK", resp);
             context.clusterTestUtils.WaitForReplicaRecovery(replicaIndex, context.logger);
-            context.clusterTestUtils.WaitForConnectedReplicaCount(primaryIndex, 1, context.logger);
+            await context.clusterTestUtils.WaitForConnectedReplicaCountAsync(primaryIndex, 1, context.logger).ConfigureAwait(false);
 
             // Populate primary data and sync
             context.kvPairs = [];
@@ -746,7 +746,7 @@ public void ClusterEnsureReplicationWorksAfterFailover(CancellationToken cancell
 #if DEBUG
         [Test, Order(16), CancelAfter(testTimeout)]
         [Category("REPLICATION")]
-        public void ClusterFailoverResetsPrimaryOnTakeOverFailure(CancellationToken cancellationToken)
+        public async Task ClusterFailoverResetsPrimaryOnTakeOverFailureAsync(CancellationToken cancellationToken)
         {
             // Verify that when TakeOverAsPrimary fails after PauseWritesAndWaitForSync
             // has already sent failstopwrites to the primary, the primary is reset back
@@ -772,7 +772,7 @@ public void ClusterFailoverResetsPrimaryOnTakeOverFailure(CancellationToken canc
             var resp = context.clusterTestUtils.ClusterReplicate(replicaNodeIndex: replicaIndex, primaryNodeIndex: primaryIndex, logger: context.logger);
             ClassicAssert.AreEqual("OK", resp);
             context.clusterTestUtils.WaitForReplicaRecovery(replicaIndex, context.logger);
-            context.clusterTestUtils.WaitForConnectedReplicaCount(primaryIndex, 1, context.logger);
+            await context.clusterTestUtils.WaitForConnectedReplicaCountAsync(primaryIndex, 1, context.logger).ConfigureAwait(false);
 
             // Populate data and sync
             context.kvPairs = [];

test/Garnet.test.cluster/ClusterTestContext.cs

@@ -46,6 +46,12 @@ public class ClusterTestContext
 
         public void Setup(Dictionary<string, LogLevel> monitorTests, int testTimeoutSeconds = 60)
         {
+            // Pull timeout off [CancelAfter] if its specified, otherwise use default
+            if (TestContext.CurrentContext.Test.Properties.ContainsKey("Timeout"))
+            {
+                testTimeoutSeconds = ((int)TestContext.CurrentContext.Test.Properties["Timeout"].First()) / 1_000;
+            }
+
             cts = new CancellationTokenSource(TimeSpan.FromSeconds(testTimeoutSeconds));
 
             TestFolder = TestUtils.UnitTestWorkingDir() + "\\";
@@ -779,7 +785,7 @@ public void ClusterFailoverSpinWait(int replicaNodeIndex, ILogger logger)
             }
         }
 
-        public void AttachAndWaitForSync(int primary_count, int replica_count, bool disableObjects)
+        public async Task AttachAndWaitForSyncAsync(int primary_count, int replica_count, bool disableObjects)
         {
             var primaryId = clusterTestUtils.GetNodeIdFromNode(0, logger);
 
@@ -800,7 +806,7 @@ public void AttachAndWaitForSync(int primary_count, int replica_count, bool disa
                 clusterTestUtils.WaitForReplicaAofSync(0, i, logger);
             }
 
-            clusterTestUtils.WaitForConnectedReplicaCount(0, replica_count, logger: logger);
+            await clusterTestUtils.WaitForConnectedReplicaCountAsync(0, replica_count, logger: logger).ConfigureAwait(false);
 
             // Validate data on replicas
             for (var i = primary_count; i < replica_count; i++)