Replication Misc Fixes (#1467)

* reset iterator at failover

* change order of dispose to improve unit test reliability

Author: vazois

libs/cluster/Server/Failover/ReplicaFailoverSession.cs

@@ -138,6 +138,9 @@ private bool TakeOverAsPrimary()
                 // Update replicationIds and replicationOffset2
                 clusterProvider.replicationManager.TryUpdateForFailover();
 
+                // Reset replay iterators
+                clusterProvider.replicationManager.ResetReplayIterator();
+
                 // Initialize checkpoint history
                 if (!clusterProvider.replicationManager.InitializeCheckpointStore())
                     logger?.LogWarning("Failed acquiring latest memory checkpoint metadata at {method}", nameof(TakeOverAsPrimary));

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

@@ -33,8 +33,17 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
         {
             // logger?.LogInformation("Processing {recordLength} bytes; previousAddress {previousAddress}, currentAddress {currentAddress}, nextAddress {nextAddress}, current AOF tail {tail}", recordLength, previousAddress, currentAddress, nextAddress, storeWrapper.appendOnlyFile.TailAddress);
             var currentConfig = clusterProvider.clusterManager.CurrentConfig;
+            var syncReplay = clusterProvider.serverOptions.ReplicationOffsetMaxLag == 0;
+
+            // Need to ensure that this replay task is allowed to complete before the replicaReplayGroup is disposed
+            // NOTE: this should not be expensive because every replay task has its own lock copy
+            // Cache invalidation happens only on dispose which is rare operation
+            var failReplay = syncReplay && !activeReplay.TryReadLock();
             try
             {
+                if (failReplay)
+                    throw new GarnetException($"Failed to acquire activeReplay lock!", LogLevel.Warning, clientResponse: false);
+
                 if (clusterProvider.replicationManager.CannotStreamAOF)
                 {
                     logger?.LogError("Replica is recovering cannot sync AOF");
@@ -120,9 +129,13 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
             catch (Exception ex)
             {
                 logger?.LogWarning(ex, "An exception occurred at ReplicationManager.ProcessPrimaryStream");
-                ResetReplayIterator();
                 throw new GarnetException(ex.Message, ex, LogLevel.Warning, clientResponse: false);
             }
+            finally
+            {
+                if (syncReplay && !failReplay)
+                    activeReplay.ReadUnlock();
+            }
         }
     }
 }

libs/server/StoreWrapper.cs

@@ -910,13 +910,13 @@ public void Dispose()
             disposed = true;
 
             itemBroker?.Dispose();
+            clusterProvider?.Dispose();
             monitor?.Dispose();
             luaTimeoutManager?.Dispose();
             ctsCommit?.Cancel();
             databaseManager.Dispose();
 
             ctsCommit?.Dispose();
-            clusterProvider?.Dispose();
         }
     }
 }