MultiLog replication: drift-barrier throttling, inline tail-witness pulses, and AOF shipping backpressure

benchmark/Resp.benchmark/OfflineBench/AOFBench/AofBench.cs

@@ -28,7 +28,7 @@ public static GarnetServerOptions GetServerOptions(Options options)
                 CommitFrequencyMs = options.CommitFrequencyMs,
                 AofPhysicalSublogCount = options.AofPhysicalSublogCount,
                 AofReplayTaskCount = options.AofReplayTaskCount,
-                ReplicationOffsetMaxLag = 0,
+                AofReplayMaxLagBytes = 0,
                 CheckpointDir = OperatingSystem.IsLinux() ? "/tmp" : null,
             };
             return serverOptions;

benchmark/Resp.benchmark/OfflineBench/AOFBench/AofGen.cs

@@ -234,7 +234,7 @@ public AofGen(Options options)
                 CommitFrequencyMs = -1,
                 FastAofTruncate = true,
                 EnableCluster = true,
-                ReplicationOffsetMaxLag = 0,
+                AofReplayMaxLagBytes = 0,
                 AofPhysicalSublogCount = options.AofPhysicalSublogCount
             };
             aofServerOptions.GetAofSettings(0, out var logSettings);

benchmark/Resp.benchmark/OfflineBench/GarnetServerInstance.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft Corporation.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
 using System.Net;
@@ -27,7 +27,7 @@ public static GarnetServerOptions GetServerOptions(Options options)
                 CommitFrequencyMs = options.CommitFrequencyMs,
                 AofPhysicalSublogCount = options.AofPhysicalSublogCount,
                 AofReplayTaskCount = options.AofReplayTaskCount,
-                ReplicationOffsetMaxLag = 0,
+                AofReplayMaxLagBytes = 0,
                 CheckpointDir = OperatingSystem.IsLinux() ? "/tmp" : null,
             };
             return serverOptions;

libs/client/ClientSession/GarnetClientSessionReplicationExtensions.cs

@@ -548,19 +548,16 @@ public void SetClusterSyncHeader(string sourceNodeId)
         }
 
         /// <summary>
-        /// Issue CLUSTER ADVANCE_TIME
+        /// Issue an in-band CLUSTER ADVANCE_TIME pulse for one physical sublog. No response is
+        /// expected, so the pulse remains ordered with APPENDLOG traffic on this connection.
         /// </summary>
+        /// <param name="physicalSublogIdx"></param>
         /// <param name="sequenceNumber"></param>
-        /// <param name="aofAddress"></param>
-        /// <returns></returns>
         /// <seealso cref="M:Garnet.cluster.ClusterSession.NetworkClusterAdvanceTime"/>
-        public Task<string> ExecuteClusterAdvanceTime(long sequenceNumber, Span<byte> aofAddress)
+        public void ExecuteClusterAdvanceTime(int physicalSublogIdx, long sequenceNumber)
         {
-            var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
-            tcsQueue.Enqueue(tcs);
             var curr = offset;
-            var argCount = 2;
-            var arraySize = 2 + argCount;
+            const int arraySize = 4;
 
             while (!RespWriteUtils.TryWriteArrayLength(arraySize, ref curr, end))
             {
@@ -586,24 +583,20 @@ public Task<string> ExecuteClusterAdvanceTime(long sequenceNumber, Span<byte> ao
             offset = curr;
 
             //3
-            while (!RespWriteUtils.TryWriteArrayItem(sequenceNumber, ref curr, end))
+            while (!RespWriteUtils.TryWriteArrayItem(physicalSublogIdx, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
 
             //4
-            while (!RespWriteUtils.TryWriteBulkString(aofAddress, ref curr, end))
+            while (!RespWriteUtils.TryWriteArrayItem(sequenceNumber, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
-
-            Flush();
-            Interlocked.Increment(ref numCommands);
-            return tcs.Task;
         }
 
         /// <summary>

libs/cluster/Server/ClusterProvider.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft Corporation.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
 using System;
@@ -239,7 +239,9 @@ public MetricsItem[] GetReplicationInfo()
                 {
                     var (address, port) = config.GetLocalNodePrimaryAddress();
                     var primaryLinkStatus = clusterManager.GetPrimaryLinkStatus(config);
-                    var replicationOffsetLag = storeWrapper.appendOnlyFile.Log.TailAddress.AggregateDiff(replicationManager.ReplicationOffset);
+                    var replicationOffsetAccLag = storeWrapper.appendOnlyFile.Log.TailAddress.AggregateDiff(replicationManager.ReplicationOffset);
+                    var replicationOffsetVectorLag = storeWrapper.appendOnlyFile.Log.TailAddress.Diff(replicationManager.ReplicationOffset);
+                    var rcm = storeWrapper.appendOnlyFile.readConsistencyManager;
                     replicationInfo.Add(new("master_host", address));
                     replicationInfo.Add(new("master_port", port.ToString()));
                     replicationInfo.Add(primaryLinkStatus[0]);
@@ -250,8 +252,11 @@ public MetricsItem[] GetReplicationInfo()
                     replicationInfo.Add(new("slave_read_only", "1"));
                     replicationInfo.Add(new("replica_announced", "1"));
                     replicationInfo.Add(new("master_sync_last_io_seconds_ago", replicationManager.LastPrimarySyncSeconds.ToString()));
-                    replicationInfo.Add(new("replication_offset_lag", replicationOffsetLag.ToString()));
-                    replicationInfo.Add(new("replication_offset_max_lag", storeWrapper.runtimeConfig.GetInt(ServerConfigType.REPLICATION_OFFSET_MAX_LAG).ToString()));
+                    replicationInfo.Add(new("replication_offset_vector_lag", replicationOffsetVectorLag.ToString()));
+                    replicationInfo.Add(new("replication_offset_acc_lag", replicationOffsetAccLag.ToString()));
+                    replicationInfo.Add(new("aof_replay_max_lag_bytes", storeWrapper.serverOptions.AofReplayMaxLagBytes.ToString()));
+                    replicationInfo.Add(new("physical_sublog_max_sequence_vector", rcm == null ? "-1" : rcm.GetPhysicalSublogMaxSequenceVector()));
+                    replicationInfo.Add(new("physical_sublog_max_drift_sequence_vector", rcm == null ? "-1" : rcm.GetPhysicalSublogMaxDriftSequenceVector()));
                 }
                 else
                 {

libs/cluster/Server/Failover/ReplicaFailoverSession.cs

@@ -152,7 +152,6 @@ private async Task<bool> TakeOverAsPrimaryAsync()
                 if (clusterProvider.serverOptions.AofPhysicalSublogCount > 1)
                 {
                     clusterProvider.storeWrapper.appendOnlyFile.ResetSequenceNumberGenerator();
-                    await clusterProvider.storeWrapper.TaskManager.CancelAsync(TaskType.AdvanceTimeReplicaTask).ConfigureAwait(false);
                 }
 
                 // Initialize checkpoint history
@@ -161,10 +160,6 @@ private async Task<bool> TakeOverAsPrimaryAsync()
 
                 _ = clusterProvider.BumpAndWaitForEpochTransitionAsync().ConfigureAwait(false);
 
-                // Stop advance time task when reconfiguring node to be replica
-                if (clusterProvider.storeWrapper.serverOptions.AofPhysicalSublogCount > 1)
-                    await clusterProvider.storeWrapper.TaskManager.CancelAsync(TaskType.AdvanceTimeReplicaTask).ConfigureAwait(false);
-
                 // Resume all background maintenance that were possibly shutdown when this node became a replica
                 clusterProvider.storeWrapper.StartPrimaryTasks();
             }

libs/cluster/Server/Replication/PrimaryOps/AofOperations/AofSyncDriver.cs

@@ -69,23 +69,28 @@ public AofAddress PreviousAddress
         }
 
         /// <summary>
-        /// Return previous address for a specific sublog without copying the full AofAddress struct
+        /// Return previous address for a specific sublog without copying the full AofAddress struct.
+        /// This is the address actually shipped (Consume'd) to the replica, used to bound AOF
+        /// truncation conservatively.
         /// </summary>
         /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
         /// <returns>The previous address of the specified sublog's sync task.</returns>
         public long GetPreviousAddress(int physicalSublogIdx) => aofSyncTasks[physicalSublogIdx].PreviousAddress;
 
         /// <summary>
-        /// Return start address for a specific sublog without copying the full AofAddress struct
+        /// Return the shipped watermark address for a specific sublog (the monotonic high-water this
+        /// replica's sync task has shipped/scanned past), used to drive the backpressure gate.
         /// </summary>
         /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
-        /// <returns>The start address of the specified sublog's sync task.</returns>
-        public long GetStartAddress(int physicalSublogIdx) => aofSyncTasks[physicalSublogIdx].StartAddress;
+        /// <returns>The shipped watermark address of the specified sublog's sync task.</returns>
+        public long GetShippedWatermarkAddress(int physicalSublogIdx) => aofSyncTasks[physicalSublogIdx].ShippedWatermarkAddress;
 
         /// <summary>
-        /// Replica endpoint
+        /// Return start address for a specific sublog without copying the full AofAddress struct
         /// </summary>
-        readonly IPEndPoint endPoint;
+        /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
+        /// <returns>The start address of the specified sublog's sync task.</returns>
+        public long GetStartAddress(int physicalSublogIdx) => aofSyncTasks[physicalSublogIdx].StartAddress;
 
         public AofSyncDriver(
             ClusterProvider clusterProvider,
@@ -100,13 +105,12 @@ public AofSyncDriver(
             this.aofSyncDriverStore = aofSyncDriverStore;
             this.localNodeId = localNodeId;
             this.remoteNodeId = remoteNodeId;
-            this.endPoint = endPoint;
             cts = new();
             this.logger = logger;
 
             aofSyncTasks = new AofSyncTask[clusterProvider.serverOptions.AofPhysicalSublogCount];
             for (var physicalSublogIdx = 0; physicalSublogIdx < aofSyncTasks.Length; physicalSublogIdx++)
-                aofSyncTasks[physicalSublogIdx] = new AofSyncTask(clusterProvider, physicalSublogIdx, endPoint, startAddress[physicalSublogIdx], localNodeId, remoteNodeId, cts, logger);
+                aofSyncTasks[physicalSublogIdx] = new AofSyncTask(clusterProvider, aofSyncDriverStore, physicalSublogIdx, endPoint, startAddress[physicalSublogIdx], localNodeId, remoteNodeId, cts, logger);
         }
 
         /// <summary>
@@ -152,10 +156,9 @@ public async Task RunAsync()
                 }
                 else
                 {
-                    var tasks = new Task[aofSyncTasks.Length + 1];
-                    tasks[0] = AdvancePhysicalSublogTimeAsync();
+                    var tasks = new Task[aofSyncTasks.Length];
                     for (var i = 0; i < aofSyncTasks.Length; i++)
-                        tasks[i + 1] = aofSyncTasks[i].RunAofSyncTaskAsync(this);
+                        tasks[i] = aofSyncTasks[i].RunAofSyncTaskAsync(this);
 
                     _ = await Task.WhenAny(tasks).ConfigureAwait(false);
                 }
@@ -174,60 +177,6 @@ public async Task RunAsync()
             }
         }
 
-        /// <summary>
-        /// Advance physical sublog time background task.
-        /// </summary>
-        /// <returns></returns>
-        /// <exception cref="GarnetException"></exception>
-        /// <seealso cref="T:Garnet.cluster.ReplicationManager.AdvanceTime"/>
-        async Task AdvancePhysicalSublogTimeAsync()
-        {
-            var enteredMonitor = false;
-            var client = new GarnetClientSession(
-                        endPoint,
-                        clusterProvider.replicationManager.GetAofSyncNetworkBufferSettings,
-                        clusterProvider.replicationManager.GetNetworkPool,
-                        tlsOptions: this.clusterProvider.serverOptions.TlsOptions?.TlsClientOptions,
-                        authUsername: this.clusterProvider.ClusterUsername,
-                        authPassword: this.clusterProvider.ClusterPassword,
-                        logger: logger);
-
-            try
-            {
-                enteredMonitor = activeWorkerMonitor.TryEnter();
-                if (!enteredMonitor)
-                    throw new GarnetException($"Failed to acquire read lock at {nameof(AdvancePhysicalSublogTimeAsync)}");
-
-                // Connect to replica
-                await client.ConnectAsync((int)clusterProvider.serverOptions.ReplicaSyncTimeout.TotalMilliseconds, cts.Token).ConfigureAwait(false);
-
-                var appendOnlyFile = clusterProvider.storeWrapper.appendOnlyFile;
-                var previousTailAddress = AofAddress.Create(appendOnlyFile.Log.Size, 0);
-
-                while (!cts.IsCancellationRequested)
-                {
-                    await Task.Delay(clusterProvider.storeWrapper.runtimeConfig.GetInt(ServerConfigType.AOF_TAIL_WITNESS_FREQ), cts.Token).ConfigureAwait(false);
-                    var currentTailAddress = appendOnlyFile.Log.TailAddress;
-                    var newWrites = previousTailAddress.AnyLesser(currentTailAddress);
-
-                    if (newWrites)
-                    {
-                        var sequenceNumber = appendOnlyFile.GetLargerThanMaximumSequenceNumber();
-                        _ = await client.ExecuteClusterAdvanceTime(sequenceNumber, currentTailAddress.Span).
-                            WaitAsync(clusterProvider.serverOptions.ReplicaSyncTimeout, cts.Token).
-                            ConfigureAwait(false);
-                        previousTailAddress.MonotonicUpdate(ref currentTailAddress);
-                    }
-                }
-            }
-            finally
-            {
-                if (enteredMonitor)
-                    _ = activeWorkerMonitor.Exit();
-                client?.Dispose();
-            }
-        }
-
         #region DisklesSyncInterface
         public async Task ConnectClientsAsync()
         {

libs/cluster/Server/Replication/PrimaryOps/AofOperations/AofSyncDriverStore.cs

@@ -206,6 +206,80 @@ public List<RoleInfo> GetReplicaInfo(in AofAddress PrimaryReplicationOffset)
             return replicaInfo;
         }
 
+        /// <summary>
+        /// Publish one physical sublog's minimum shipped address across all attached replicas into
+        /// the AOF backpressure gate (long.MaxValue when none is attached, which releases it). A
+        /// sync task calls this for its own sublog as it ships, so only the watermark of the sublog
+        /// that made progress is recomputed and written. Appenders self-check their own tail against
+        /// the watermark, so this is a freshness hint, never a correctness action: a stale value
+        /// makes appenders stall sooner, not later. Must not be called while holding the store lock.
+        /// </summary>
+        /// <param name="physicalSublogIdx"></param>
+        internal void PublishShippedAddress(int physicalSublogIdx)
+        {
+            var appendOnlyFile = clusterProvider.storeWrapper.appendOnlyFile;
+            var backpressure = appendOnlyFile?.backpressure;
+            if (backpressure == null)
+                return;
+
+            var minShipped = long.MaxValue;
+            _lock.ReadLock();
+            try
+            {
+                if (!_disposed && numDrivers > 0)
+                {
+                    for (var i = 0; i < numDrivers; i++)
+                        minShipped = Math.Min(minShipped, syncDrivers[i].GetShippedWatermarkAddress(physicalSublogIdx));
+                }
+            }
+            finally
+            {
+                _lock.ReadUnlock();
+            }
+
+            backpressure.PublishShippedAddress(physicalSublogIdx, minShipped);
+        }
+
+        /// <summary>
+        /// Publish every physical sublog's minimum shipped address across all attached replicas into
+        /// the AOF backpressure gate (long.MaxValue per sublog when none is attached, or during
+        /// dispose, which releases the gate). The store calls this whenever the driver set changes,
+        /// since attaching or detaching a replica can change every sublog's minimum at once. Must
+        /// not be called while holding the store lock.
+        /// </summary>
+        internal void PublishShippedAddresses()
+        {
+            var appendOnlyFile = clusterProvider.storeWrapper.appendOnlyFile;
+            var backpressure = appendOnlyFile?.backpressure;
+            if (backpressure == null)
+                return;
+
+            var sublogCount = clusterProvider.serverOptions.AofPhysicalSublogCount;
+            _lock.ReadLock();
+            try
+            {
+                if (_disposed || numDrivers == 0)
+                {
+                    for (var physicalSublogIdx = 0; physicalSublogIdx < sublogCount; physicalSublogIdx++)
+                        backpressure.PublishShippedAddress(physicalSublogIdx, long.MaxValue);
+                }
+                else
+                {
+                    for (var physicalSublogIdx = 0; physicalSublogIdx < sublogCount; physicalSublogIdx++)
+                    {
+                        var minShipped = long.MaxValue;
+                        for (var i = 0; i < numDrivers; i++)
+                            minShipped = Math.Min(minShipped, syncDrivers[i].GetShippedWatermarkAddress(physicalSublogIdx));
+                        backpressure.PublishShippedAddress(physicalSublogIdx, minShipped);
+                    }
+                }
+            }
+            finally
+            {
+                _lock.ReadUnlock();
+            }
+        }
+
         /// <summary>
         /// Dispose the AofSyncDriverStore
         /// </summary>
@@ -234,6 +308,10 @@ public void Dispose()
             }
             numDrivers = 0;
             Array.Clear(syncDrivers);
+
+            // With no drivers attached, PublishShippedAddresses writes a max watermark per sublog,
+            // making every appender's computed lag non-positive so none stalls on the gate.
+            PublishShippedAddresses();
         }
 
         /// <summary>
@@ -333,6 +411,9 @@ public bool TryAddReplicationDriver(string remoteNodeId, ref AofAddress startAdd
                 }
             }
 
+            if (success)
+                PublishShippedAddresses();
+
             return success;
         }
 
@@ -442,6 +523,9 @@ public bool TryAddReplicationDrivers(ReplicaSyncSession[] replicaSyncSessions, r
                 }
             }
 
+            if (success)
+                PublishShippedAddresses();
+
             return true;
         }
 
@@ -491,6 +575,10 @@ public bool TryRemove(AofSyncDriver aofSyncDriver)
             {
                 _lock.WriteUnlock();
             }
+
+            if (success)
+                PublishShippedAddresses();
+
             return success;
         }
 
@@ -562,6 +650,10 @@ public void Reset()
             {
                 _lock.WriteUnlock();
             }
+
+            // No drivers remain, so PublishShippedAddresses writes a max watermark per sublog and
+            // no appender stays stalled on the gate.
+            PublishShippedAddresses();
         }
 
         [Conditional("DEBUG")]