wait for vector set replay to quiesce when waiting for replica sync to catch up

Author: kevin-montrose

test/Garnet.test.cluster/ClusterTestUtils.cs

@@ -9,12 +9,14 @@
 using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
+using System.Runtime.CompilerServices;
 using System.Security.Cryptography.X509Certificates;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Garnet.client;
 using Garnet.common;
+using Garnet.server;
 using Garnet.server.TLS;
 using GarnetClusterManagement;
 using Microsoft.Extensions.Logging;
@@ -2983,8 +2985,22 @@ public void WaitForReplicaAofSync(int primaryIndex, int secondaryIndex, ILogger
                 BackOff(cancellationToken: context.cts.Token, msg: $"[{endpoints[primaryIndex]}]: {primaryMainStoreVersion},{primaryReplicationOffset} != [{endpoints[secondaryIndex]}]: {replicaMainStoreVersion},{secondaryReplicationOffset1}");
             }
             logger?.LogInformation("[{primaryEndpoint}]{primaryReplicationOffset} ?? [{endpoints[secondaryEndpoint}]{secondaryReplicationOffset1}", endpoints[primaryIndex], primaryReplicationOffset, endpoints[secondaryIndex], secondaryReplicationOffset1);
+
+            WaitForVectorSetReplayQuiescence(secondaryIndex, logger);
+        }
+
+        public void WaitForVectorSetReplayQuiescence(int index, ILogger logger = null)
+        {
+            var server = this.context.nodes[index];
+            var storeWrapper = GetStoreWrapper(server);
+            var vectorManager = storeWrapper.DefaultDatabase.VectorManager;
+
+            vectorManager?.WaitForVectorOperationsToComplete();
         }
 
+        [UnsafeAccessor(UnsafeAccessorKind.Field, Name = "storeWrapper")]
+        private static extern ref StoreWrapper GetStoreWrapper(GarnetServer server);
+
         public void WaitForConnectedReplicaCount(int primaryIndex, long minCount, ILogger logger = null)
         {
             while (true)