Store delta log tail address (#861)

* Store delta log tail address

* fix + comment

Author: badrishc

cs/src/core/Index/Common/Contexts.cs

@@ -396,7 +396,9 @@ public struct HybridLogRecoveryInfo
 
 
         /// <summary>
-        /// Tail address of delta file
+        /// Tail address of delta file: -1 indicates this is not a delta checkpoint metadata
+        /// At recovery, this value denotes the delta tail address excluding the metadata record for the checkpoint
+        /// because we create the metadata before writing to the delta file.
         /// </summary>
         public long deltaTailAddress;
 
@@ -415,7 +417,7 @@ public void Initialize(Guid token, long _version)
             startLogicalAddress = 0;
             finalLogicalAddress = 0;
             snapshotFinalLogicalAddress = 0;
-            deltaTailAddress = 0;
+            deltaTailAddress = -1; // indicates this is not a delta checkpoint metadata
             headAddress = 0;
 
             checkpointTokens = new();
@@ -564,6 +566,11 @@ internal void Recover(Guid token, ICheckpointManager checkpointManager, out byte
                 throw new FasterException("Invalid log commit metadata for ID " + token.ToString());
             using StreamReader s = new(new MemoryStream(metadata));
             Initialize(s);
+            if (scanDelta && deltaLog != null && deltaTailAddress >= 0)
+            {
+                // Adjust delta tail address to include the metadata record
+                deltaTailAddress = deltaLog.NextAddress;
+            }
             var cookie = s.ReadToEnd();
             commitCookie =  cookie.Length == 0 ? null : Convert.FromBase64String(cookie);
         }

cs/src/core/Index/Recovery/ICheckpointManager.cs

@@ -86,7 +86,7 @@ public interface ICheckpointManager :  IDisposable
         /// </summary>
         /// <param name="logToken">Token</param>
         /// <param name="deltaLog">Delta log</param>
-        /// <param name="scanDelta"> whether or not to scan through the delta log to acquire latest entry</param>
+        /// <param name="scanDelta"> whether or not to scan through the delta log to acquire latest entry. make sure the delta log points to the tail address immediately following the returned metadata.</param>
         /// <param name="recoverTo"> version upper bound to scan for in the delta log. Function will return the largest version metadata no greater than the given version.</param>
         /// <returns>Metadata, or null if invalid</returns>
         byte[] GetLogCheckpointMetadata(Guid logToken, DeltaLog deltaLog, bool scanDelta = false, long recoverTo = -1);

cs/src/core/Index/Recovery/Recovery.cs

@@ -150,6 +150,10 @@ public struct LogFileInfo
         /// Hybrid log file end address
         /// </summary>
         public long hybridLogFileEndAddress;
+        /// <summary>
+        /// Delta log tail address
+        /// </summary>
+        public long deltaLogTailAddress;
     }
 
     public partial class FasterKV<Key, Value> : FasterBase, IFasterKV<Key, Value>
@@ -188,19 +192,21 @@ public long GetLatestCheckpointVersion()
         /// Get size of snapshot files for token
         /// </summary>
         /// <param name="token"></param>
+        /// <param name="version"></param>
         /// <returns></returns>
-        public LogFileInfo GetLogFileSize(Guid token)
+        public LogFileInfo GetLogFileSize(Guid token, long version = -1)
         {
             using var current = new HybridLogCheckpointInfo();
             // We find the latest checkpoint metadata for the given token, including scanning the delta log for the latest metadata
             current.Recover(token, checkpointManager, hlog.LogPageSizeBits,
-                out var _, true);
+                out var _, true, version);
             long snapshotDeviceOffset = hlog.GetPage(current.info.snapshotStartFlushedLogicalAddress) << hlog.LogPageSizeBits;
             return new LogFileInfo
             {
                 snapshotFileEndAddress = current.info.snapshotFinalLogicalAddress - snapshotDeviceOffset,
                 hybridLogFileStartAddress = hlog.GetPage(current.info.beginAddress) << hlog.LogPageSizeBits,
-                hybridLogFileEndAddress = current.info.flushedLogicalAddress
+                hybridLogFileEndAddress = current.info.flushedLogicalAddress,
+                deltaLogTailAddress = current.info.deltaTailAddress,
             };
         }
 

cs/src/core/Index/Synchronization/HybridLogCheckpointTask.cs

@@ -346,6 +346,7 @@ public override void GlobalBeforeEnteringState<Key, Value>(SystemState next, Fas
                     break;
                 case Phase.PERSISTENCE_CALLBACK:
                     CollectMetadata(next, faster);
+                    faster._hybridLogCheckpoint.info.deltaTailAddress = faster._hybridLogCheckpoint.deltaLog.TailAddress;
                     faster.WriteHybridLogIncrementalMetaInfo(faster._hybridLogCheckpoint.deltaLog);
                     faster._hybridLogCheckpoint.info.deltaTailAddress = faster._hybridLogCheckpoint.deltaLog.TailAddress;
                     faster._lastSnapshotCheckpoint = faster._hybridLogCheckpoint.Transfer();