fix: Replication checkpoint recovery fix (#227)

* add replication recovery state fix

* add metrics to update primary recovery time

* fix whitespace formatting error

* Address comments - change DateTime to timestamp and formatting fixes

libs/cluster/Server/ClusterManager.cs

@@ -286,6 +286,7 @@ public void TryResetReplica()
                 if (Interlocked.CompareExchange(ref currentConfig, newConfig, current) == current)
                     break;
             }
+            replicationManager.Reset();
             FlushConfig();
         }
 

libs/cluster/Server/ClusterProvider.cs

@@ -194,7 +194,7 @@ public MetricsItem[] GetReplicationInfo()
             int replicaCount = clusterEnabled ? replicaInfo.Count : 0;
             var role = clusterEnabled ? config.GetLocalNodeRole() : NodeRole.PRIMARY;
             int commonInfoCount = 11;
-            int replicaInfoCount = 9;
+            int replicaInfoCount = 10;
             int replicationInfoCount = commonInfoCount + replicaCount;
             replicationInfoCount += role == NodeRole.REPLICA ? replicaInfoCount : 0;
 
@@ -228,6 +228,7 @@ public MetricsItem[] GetReplicationInfo()
                     replicationInfo[commonInfoCount + 6] = new("slave_priority", "100");
                     replicationInfo[commonInfoCount + 7] = new("slave_read_only", "1");
                     replicationInfo[commonInfoCount + 8] = new("replica_announced", "1");
+                    replicationInfo[commonInfoCount + 9] = new("master_sync_last_io_seconds_ago", replicationManager.LastPrimarySyncSeconds.ToString());
                 }
                 else
                 {

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

@@ -49,6 +49,7 @@ public void CloseDevice()
         /// <param name="segmentId"></param>
         public void ProcessFileSegments(int segmentId, Guid token, CheckpointFileType type, long startAddress, Span<byte> data)
         {
+            clusterProvider.replicationManager.UpdateLastPrimarySyncTime();
             if (writeIntoCkptDevice == null)
             {
                 Debug.Assert(writeIntoCkptDevice == null);

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

@@ -167,6 +167,7 @@ private async Task<string> InitiateReplicaSync()
         /// <exception cref="Exception">Throws invalid type checkpoint metadata.</exception>
         public void ProcessCheckpointMetadata(Guid fileToken, CheckpointFileType fileType, byte[] checkpointMetadata)
         {
+            UpdateLastPrimarySyncTime();
             ReplicationLogCheckpointManager ckptManager = fileType switch
             {
                 CheckpointFileType.STORE_SNAPSHOT or
@@ -290,6 +291,7 @@ public long BeginReplicaRecover(
             long beginAddress,
             long recoveredReplicationOffset)
         {
+            UpdateLastPrimarySyncTime();
             storeWrapper.RecoverCheckpoint(recoverMainStoreFromToken, recoverObjectStoreFromToken,
                 remoteCheckpoint.storeIndexToken, remoteCheckpoint.storeHlogToken, remoteCheckpoint.objectStoreIndexToken, remoteCheckpoint.objectStoreHlogToken);
 

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -24,9 +24,16 @@ internal sealed partial class ReplicationManager : IDisposable
 
         readonly ILogger logger;
         bool _disposed;
+
+        private long primary_sync_last_time;
+
+        internal long LastPrimarySyncSeconds => recovering ? (DateTime.UtcNow.Ticks - primary_sync_last_time) / TimeSpan.TicksPerSecond : 0;
+
+        internal void UpdateLastPrimarySyncTime() => this.primary_sync_last_time = DateTime.UtcNow.Ticks;
+
+
         public bool recovering;
         private long replicationOffset;
-
         public long ReplicationOffset
         {
             get
@@ -126,6 +133,11 @@ void CheckpointVersionShift(bool isMainStore, long oldVersion, long newVersion)
             storeWrapper.EnqueueCommit(isMainStore, newVersion);
         }
 
+        public void Reset()
+        {
+            recovering = false;
+        }
+
         public void Dispose()
         {
             _disposed = true;