Fix to delta log for recovery (#856)

* Fix to delta log for recovery
* If we are in fast-commit, we may not write every metadata to disk. However, when we are deleting files on disk, we have to write metadata for the new start location on disk so we know where to scan forward from.

Author: badrishc

cs/src/core/FasterLog/FasterLog.cs

@@ -2072,8 +2072,13 @@ private void UpdateCommittedState(FasterLogRecoveryInfo recoveryInfo)
         private void WriteCommitMetadata(FasterLogRecoveryInfo recoveryInfo)
         {
             // TODO: can change to write this in separate thread for fast commit
+
+            // If we are in fast-commit, we may not write every metadata to disk. However, when we are deleting files
+            // on disk, we have to write metadata for the new start location on disk so we know where to scan forward from.
+            bool forceWriteMetadata = fastCommitMode && (allocator.BeginAddress < recoveryInfo.BeginAddress);
             logCommitManager.Commit(recoveryInfo.BeginAddress, recoveryInfo.UntilAddress,
-                recoveryInfo.ToByteArray(), recoveryInfo.CommitNum);
+                recoveryInfo.ToByteArray(), recoveryInfo.CommitNum, forceWriteMetadata);
+
             // If not fast committing, set committed state as we commit metadata explicitly only after metadata commit
             if (!fastCommitMode)
                 UpdateCommittedState(recoveryInfo);

cs/src/core/FasterLog/ILogCommitManager.cs

@@ -19,7 +19,8 @@ public interface ILogCommitManager : IDisposable
         /// <param name="untilAddress">Address committed until (for information only, not necessary to persist)</param>
         /// <param name="commitMetadata">Commit metadata - should be persisted</param>
         /// <param name="commitNum">commit num</param>
-        void Commit(long beginAddress, long untilAddress, byte[] commitMetadata, long commitNum);
+        /// <param name="forceWriteMetadata">force writing of metadata in case of fast commit</param>
+        void Commit(long beginAddress, long untilAddress, byte[] commitMetadata, long commitNum, bool forceWriteMetadata);
 
         /// <summary>
         /// Return commit metadata

cs/src/core/Index/CheckpointManagement/DeviceLogCommitCheckpointManager.cs

@@ -109,9 +109,9 @@ public DeviceLogCommitCheckpointManager(INamedDeviceFactory deviceFactory, strin
         #region ILogCommitManager
 
         /// <inheritdoc />
-        public unsafe void Commit(long beginAddress, long untilAddress, byte[] commitMetadata, long commitNum)
+        public unsafe void Commit(long beginAddress, long untilAddress, byte[] commitMetadata, long commitNum, bool forceWriteMetadata)
         {
-            if (fastCommitThrottleFreq > 0 && (commitCount++ % fastCommitThrottleFreq != 0)) return;
+            if (!forceWriteMetadata && fastCommitThrottleFreq > 0 && (commitCount++ % fastCommitThrottleFreq != 0)) return;
 
             using var device = deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(commitNum));
 

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

@@ -228,8 +228,15 @@ public unsafe bool GetNext(out long physicalAddress, out int entryLength, out De
 
                 if (entryLength == 0)
                 {
+                    if (_currentOffset == 0)
+                    {
+                        // We found a hole at beginning of page, this must imply end of delta log
+                        return false;
+                    }
+
+                    // Hole at end of page, skip to next page
                     currentAddress = (1 + (currentAddress >> LogPageSizeBits)) << LogPageSizeBits;
-                    if (Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
+                    if (!Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
                         return false;
                     else
                         continue;
@@ -239,7 +246,7 @@ public unsafe bool GetNext(out long physicalAddress, out int entryLength, out De
                 if (entryLength < 0 || (_currentOffset + recordSize > PageSize))
                 {
                     currentAddress = (1 + (currentAddress >> LogPageSizeBits)) << LogPageSizeBits;
-                    if (Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
+                    if (!Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
                         return false;
                     else
                         continue;
@@ -249,7 +256,7 @@ public unsafe bool GetNext(out long physicalAddress, out int entryLength, out De
                 if (!VerifyBlockChecksum((byte*)physicalAddress, entryLength))
                 {
                     currentAddress = (1 + (currentAddress >> LogPageSizeBits)) << LogPageSizeBits;
-                    if (Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
+                    if (!Utility.MonotonicUpdate(ref nextAddress, currentAddress, out _))
                         return false;
                     else
                         continue;