fix(storage): don't reuse freed pages that still have a WAL image (#168)

src/BLite.Core/Indexing/BTreeIndex.cs

@@ -1658,9 +1658,8 @@ private void MergeWithSibling(uint nodeId, uint parentId, int childIndex, List<I
         WriteInternalNode(parentId, p0, parentEntries, transactionId);
 
         // Free the empty Right Node
-        _storage.FreePage(rightNodeId); // Need to verify this works safely with Txn logic?
-                                         // Actually, FreePage is immediate in current impl. Might need TransactionalFreePage.
-                                         // Or just leave it allocated but unused for now.
+        _storage.FreePage(rightNodeId); // Not transactional: the page is released even if this
+                                         // transaction rolls back. Might need TransactionalFreePage.
 
         // Recursive Underflow Check on Parent
         int minInternal = MaxEntriesPerNode / 2;

src/BLite.Core/Storage/StorageEngine.GroupCommit.cs

@@ -18,7 +18,8 @@ public sealed partial class StorageEngine
     //   • The WAL flush cost (~5–30 µs) is amortised across N concurrent commits.
     //   • _commitLock is held only by the writer, for the duration of the batch
     //     write + flush — not by every individual caller for the full round trip.
-    //   • BeginTransaction no longer competes with an in-flight flush.
+    //   • BeginTransaction no longer competes with an in-flight flush (except in
+    //     multi-process mode, when it has another process's commits to replay).
     //   • The sync CommitTransactionAsync(ulong) path and checkpoint operations still
     //     use _commitLock normally and are fully compatible.
     // -------------------------------------------------------------------------
@@ -158,6 +159,17 @@ private async Task ProcessBatchAsync(List<PendingCommit> batch)
             var wal = (_shm != null) ? (_wal as Transactions.WriteAheadLog) : null;
             var pageOffsets = wal != null ? new List<(uint pageId, long walOffset)>() : null;
 
+            // Another process may have appended to the shared WAL since this engine last wrote
+            // to it or replayed it. Replay those records first, so that advancing the bookmark
+            // past this batch below never skips them, and append after them instead of at this
+            // engine's own, stale, write position (which would overwrite them).
+            bool caughtUp = false;
+            if (wal != null)
+            {
+                caughtUp = TryCatchUpWithSharedWal(wal);
+                wal.SeekToEnd();
+            }
+
             foreach (var commit in batch)
             {
                 if (commit.Pages == null || commit.Pages.IsEmpty)
@@ -193,9 +205,11 @@ private async Task ProcessBatchAsync(List<PendingCommit> batch)
             {
                 if (commit.Pages != null)
                 {
-                    _walCache.TryRemove(commit.TransactionId, out _);
+                    // Index first, then drop the cache: a page must never be in neither
+                    // (FreePage relies on finding its WAL image in one of them).
                     foreach (var kvp in commit.Pages)
                         _walIndex[kvp.Key] = kvp.Value;
+                    _walCache.TryRemove(commit.TransactionId, out _);
                 }
             }
 
@@ -221,8 +235,10 @@ private async Task ProcessBatchAsync(List<PendingCommit> batch)
             // this engine does not redundantly replay WAL records that we just committed
             // (they are already in _walIndex from the lines above). This is safe because
             // we hold _commitLock + SHM writer lock: no concurrent writer can insert
-            // records between our last write and the position we record here.
-            if (_shm != null)
+            // records between our last write and the position we record here, and the
+            // catch-up above replayed everything before this batch. If that replay failed,
+            // the bookmark stays put and the next catch-up replays this batch as well.
+            if (caughtUp)
                 Volatile.Write(ref _lastKnownWalEndOffset, newWalEnd);
 
             // Check if checkpoint is needed, but defer until after releasing the lock

src/BLite.Core/Storage/StorageEngine.Memory.cs

@@ -68,10 +68,126 @@ public uint AllocateCollectionPage(string collectionName)
         return CollectionPageMarker | ((uint)(slot & 0x3F) << CollSlotShift) | (localId & CollLocalMask);
     }
 
-    /// <summary>Frees a page, routing to the correct sub-file via the encoded page ID.</summary>
+    // Pages checkpointed out of _walIndex whose records are still in the WAL file, i.e. since
+    // the last truncation. See FreePage.
+    private readonly ConcurrentDictionary<uint, byte> _checkpointedWalPages = new();
+
+    // Freed pages that had a WAL image, held off the free list until the next truncation.
+    private readonly object _pendingFreeLock = new();
+    private HashSet<uint> _pendingFreePages = new();
+
+    /// <summary>
+    /// Frees a page, routing to the correct sub-file via the encoded page ID.
+    /// <para>
+    /// A page that still has a committed image in the WAL is not reusable right away:
+    /// <see cref="ReadPage"/> prefers that image over the page file and recovery replays it over
+    /// the page file, so a page handed out again would read back, or be overwritten with, its
+    /// old contents. Such a page is queued instead and returned to its file's free list right
+    /// after the next WAL truncation (checkpoint, backup or recovery); until then
+    /// <see cref="GetCollectionPageIds"/> skips it. A queued page is leaked, never reused
+    /// unsafely, if the process dies first. In multi-process mode every page is queued, since
+    /// another process may have written it.
+    /// </para>
+    /// </summary>
     public void FreePage(uint pageId)
     {
-        GetPageFile(pageId, out var physId).FreePage(physId);
+        if (pageId == 0) throw new InvalidOperationException("Cannot free header page 0");
+
+        if (_shm == null && !HasWalImage(pageId))
+        {
+            GetPageFile(pageId, out var physId).FreePage(physId);
+            return;
+        }
+
+        lock (_pendingFreeLock)
+            _pendingFreePages.Add(pageId);
+    }
+
+    /// <summary>
+    /// Whether the WAL holds, or an open transaction is about to commit, an image of the page.
+    /// </summary>
+    private bool HasWalImage(uint pageId)
+    {
+        return _walIndex.ContainsKey(pageId)
+            || _checkpointedWalPages.ContainsKey(pageId)
+            || IsInOpenTransaction(pageId);
+    }
+
+    /// <summary>
+    /// Returns the pages queued by <see cref="FreePage"/> to their file's free list. Call only
+    /// right after the WAL was truncated, under <c>_commitLock</c>, so no commit can append to it.
+    /// A page that an open transaction has written stays queued: its image reaches the WAL only
+    /// when that transaction commits, so it is released after the truncation that follows. A
+    /// page whose release fails stays queued as well, and is retried at the next truncation.
+    /// </summary>
+    private void OnWalTruncated()
+    {
+        _checkpointedWalPages.Clear();
+
+        HashSet<uint> pages;
+        lock (_pendingFreeLock)
+        {
+            if (_pendingFreePages.Count == 0) return;
+            pages = _pendingFreePages;
+            _pendingFreePages = new HashSet<uint>();
+        }
+
+        var keep = new List<uint>();
+        foreach (var pageId in pages)
+        {
+            if (IsInOpenTransaction(pageId))
+            {
+                keep.Add(pageId);
+                continue;
+            }
+
+            try
+            {
+                GetPageFile(pageId, out var physId).FreePage(physId);
+            }
+            catch
+            {
+                keep.Add(pageId);
+            }
+        }
+
+        if (keep.Count > 0)
+        {
+            lock (_pendingFreeLock)
+                foreach (var pageId in keep)
+                    _pendingFreePages.Add(pageId);
+        }
+    }
+
+    private bool IsInOpenTransaction(uint pageId)
+    {
+        foreach (var txnPages in _walCache.Values)
+            if (txnPages.ContainsKey(pageId))
+                return true;
+        return false;
+    }
+
+    /// <summary>Snapshot of the queued frees, or <c>null</c> when there are none.</summary>
+    private HashSet<uint>? SnapshotPendingFreePages()
+    {
+        lock (_pendingFreeLock)
+            return _pendingFreePages.Count == 0 ? null : new HashSet<uint>(_pendingFreePages);
+    }
+
+    /// <summary>
+    /// Drops queued frees of pages in the given collection file, which is being deleted: the
+    /// slot keeps its number, so releasing them later would free pages of a recreated file.
+    /// </summary>
+    private void DiscardPendingFreePages(int collectionSlot)
+    {
+        var slotBits = CollectionPageMarker | ((uint)(collectionSlot & 0x3F) << CollSlotShift);
+        lock (_pendingFreeLock)
+            _pendingFreePages.RemoveWhere(p => (p & (SubFileTypeMask | CollSlotMask)) == slotBits);
+    }
+
+    private bool HasPendingFreePages
+    {
+        get { lock (_pendingFreeLock) return _pendingFreePages.Count > 0; }
     }
 
     // -----------------------------------------------------------------------
@@ -161,6 +277,9 @@ public void DropCollectionFile(string collectionName)
     {
         if (_collectionFiles == null) return;
 
+        if (_collectionNameToSlot != null && _collectionNameToSlot.TryGetValue(collectionName, out var slot))
+            DiscardPendingFreePages(slot);
+
         if (_collectionFiles.TryRemove(collectionName, out var lazy))
         {
             if (lazy.IsValueCreated) lazy.Value.Dispose();
@@ -369,13 +488,9 @@ internal IReadOnlyCollection<uint> FreeCollectionPagesCore(string collectionName
 
         beforeFree?.Invoke(toFree);
 
-        // Free all collected pages and remove stale WAL-index entries so that
-        // a re-allocated page is not served stale data from the WAL index.
+        // FreePage defers reuse until the WAL holds no stale image of these pages.
         foreach (var pageId in toFree)
-        {
             FreePage(pageId);
-            _walIndex.TryRemove(pageId, out _);
-        }
 
         return toFree;
     }
@@ -391,11 +506,15 @@ internal IReadOnlyCollection<uint> FreeCollectionPagesCore(string collectionName
     /// </summary>
     public IEnumerable<uint> GetCollectionPageIds(string collectionName)
     {
+        // Freed pages waiting for a WAL truncation still carry their old contents on disk.
+        var pending = SnapshotPendingFreePages();
+
         if (_collectionFiles == null)
         {
             var mainCount = _pageFile.NextPageId;
             for (uint i = 0; i < mainCount; i++)
-                yield return i;
+                if (pending == null || !pending.Contains(i))
+                    yield return i;
             yield break;
         }
 
@@ -406,7 +525,11 @@ public IEnumerable<uint> GetCollectionPageIds(string collectionName)
         var localCount = file.NextPageId;
         var encBase = CollectionPageMarker | ((uint)(slot & 0x3F) << CollSlotShift);
         for (uint localId = 0; localId < localCount; localId++)
-            yield return encBase | (localId & CollLocalMask);
+        {
+            var pageId = encBase | (localId & CollLocalMask);
+            if (pending == null || !pending.Contains(pageId))
+                yield return pageId;
+        }
     }
 
     // -----------------------------------------------------------------------