fix failover release lock and add recovery status tracking

Author: vazois

libs/cluster/Server/ClusterManagerWorkerState.cs

@@ -188,7 +188,7 @@ public bool TryAddReplica(string nodeid, bool force, out ReadOnlySpan<byte> erro
 
                 // Transition to recovering state
                 // Only one caller will succeed in becoming a replica for the provided node-id
-                if (!clusterProvider.replicationManager.StartRecovery())
+                if (!clusterProvider.replicationManager.StartRecovery(RecoveryStatus.ClusterReplicate))
                 {
                     logger?.LogError($"{nameof(TryAddReplica)}: {{logMessage}}", Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
                     errorMessage = CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK;

libs/cluster/Server/ClusterProvider.cs

@@ -246,6 +246,7 @@ public MetricsItem[] GetReplicationInfo()
                     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.serverOptions.ReplicationOffsetMaxLag.ToString()));
+                    replicationInfo.Add(new("recoverStatus", replicationManager.recoverStatus.ToString()));
                 }
                 else
                 {

libs/cluster/Server/Failover/ReplicaFailoverSession.cs

@@ -150,13 +150,15 @@ private bool TakeOverAsPrimary()
         {
             // Take over as primary and inform old primary
             status = FailoverStatus.TAKING_OVER_AS_PRIMARY;
+            var acquiredLock = true;
 
             try
             {
                 // Make replica syncing unavailable by setting recovery flag
-                if (!clusterProvider.replicationManager.StartRecovery())
+                if (!clusterProvider.replicationManager.StartRecovery(RecoveryStatus.ClusterFailover))
                 {
                     logger?.LogWarning($"{nameof(TakeOverAsPrimary)}: {{logMessage}}", Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
+                    acquiredLock = false;
                     return false;
                 }
                 _ = clusterProvider.BumpAndWaitForEpochTransition();
@@ -178,7 +180,7 @@ private bool TakeOverAsPrimary()
             finally
             {
                 // Disable recovering as now this node has become a primary or failed in its attempt earlier
-                clusterProvider.replicationManager.SuspendRecovery();
+                if (acquiredLock) clusterProvider.replicationManager.SuspendRecovery();
             }
 
             return true;

libs/cluster/Server/Replication/RecoveryStatus.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+namespace Garnet.cluster
+{
+    /// <summary>
+    /// Recovery status
+    /// </summary>
+    public enum RecoveryStatus : byte
+    {
+        /// <summary>
+        /// No recovery
+        /// </summary>
+        NoRecovery,
+        /// <summary>
+        /// Recovery at initilization
+        /// </summary>
+        InitializeRecover,
+        /// <summary>
+        /// Recovery at cluster replicate
+        /// </summary>
+        ClusterReplicate,
+        /// <summary>
+        /// Recovery at cluster failover
+        /// </summary>
+        ClusterFailover,
+        /// <summary>
+        /// Recovery at replicaof no one
+        /// </summary>
+        ReplicaOfNoOne
+    }
+}

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -64,6 +64,11 @@ public long ReplicationOffset2
         public string PrimaryReplId => currentReplicationConfig.primary_replid;
         public string PrimaryReplId2 => currentReplicationConfig.primary_replid2;
 
+        /// <summary>
+        /// Recovery status
+        /// </summary>
+        public RecoveryStatus recoverStatus;
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public ReplicationLogCheckpointManager GetCkptManager(StoreType storeType)
         {
@@ -112,7 +117,7 @@ public ReplicationManager(ClusterProvider clusterProvider, ILogger logger = null
                 clusterProvider.GetReplicationLogCheckpointManager(StoreType.Object).checkpointVersionShift = CheckpointVersionShift;
 
             // If this node starts as replica, it cannot serve requests until it is connected to primary
-            if (clusterProvider.clusterManager.CurrentConfig.LocalNodeRole == NodeRole.REPLICA && clusterProvider.serverOptions.Recover && !StartRecovery())
+            if (clusterProvider.clusterManager.CurrentConfig.LocalNodeRole == NodeRole.REPLICA && clusterProvider.serverOptions.Recover && !StartRecovery(RecoveryStatus.InitializeRecover))
                 throw new Exception(Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
 
             checkpointStore = new CheckpointStore(storeWrapper, clusterProvider, true, logger);
@@ -165,22 +170,24 @@ void CheckpointVersionShift(bool isMainStore, long oldVersion, long newVersion)
         /// <summary>
         /// Acquire recovery and checkpoint locks to prevent checkpoints and parallel recovery tasks
         /// </summary>
-        public bool StartRecovery()
+        public bool StartRecovery(RecoveryStatus recoverStatus)
         {
             if (!clusterProvider.storeWrapper.TryPauseCheckpoints())
             {
-                logger?.LogError("Error could not acquire checkpoint lock");
+                logger?.LogError("Error could not acquire checkpoint lock [{recoverStatus}]", recoverStatus);
                 return false;
             }
 
             if (!recoverLock.TryWriteLock())
             {
-                logger?.LogError("Error could not acquire recover lock");
+                logger?.LogError("Error could not acquire recover lock [{recoverStatus}]", recoverStatus);
                 // If failed to acquire recoverLock re-enable checkpoint taking
                 clusterProvider.storeWrapper.ResumeCheckpoints();
                 return false;
             }
 
+            this.recoverStatus = recoverStatus;
+            logger?.LogTrace("Success recover lock [{recoverStatus}]", recoverStatus);
             return true;
         }
 
@@ -189,6 +196,8 @@ public bool StartRecovery()
         /// </summary>
         public void SuspendRecovery()
         {
+            logger?.LogTrace("Release recover lock [{recoverStatus}]", recoverStatus);
+            recoverStatus = RecoveryStatus.NoRecovery;
             recoverLock.WriteUnlock();
             clusterProvider.storeWrapper.ResumeCheckpoints();
         }

libs/cluster/Session/ReplicaOfCommand.cs

@@ -24,13 +24,13 @@ private bool TryREPLICAOF(out bool invalidParameters)
             var addressSpan = parseState.GetArgSliceByRef(0).ReadOnlySpan;
             var portSpan = parseState.GetArgSliceByRef(1).ReadOnlySpan;
 
-            //Turn of replication and make replica into a primary but do not delete data
+            // Turn of replication and make replica into a primary but do not delete data
             if (addressSpan.EqualsUpperCaseSpanIgnoringCase("NO"u8) &&
                 portSpan.EqualsUpperCaseSpanIgnoringCase("ONE"u8))
             {
                 try
                 {
-                    if (!clusterProvider.replicationManager.StartRecovery())
+                    if (!clusterProvider.replicationManager.StartRecovery(RecoveryStatus.ReplicaOfNoOne))
                     {
                         logger?.LogError($"{nameof(TryREPLICAOF)}: {{logMessage}}", Encoding.ASCII.GetString(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK));
                         while (!RespWriteUtils.TryWriteError(CmdStrings.RESP_ERR_GENERIC_CANNOT_ACQUIRE_RECOVERY_LOCK, ref dcurr, dend))