From 39a2960eba8b48f48ca46c705c2fb64dc96a6eb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 12:55:01 +0200 Subject: [PATCH 1/3] fix(storage): don't reuse freed pages that still have a WAL image (#168) ReadPage prefers the committed WAL image over the page file, and recovery replays it over the page file. FreePage put a page straight back on the free list, so when it was reused and initialized with WritePageImmediate (e.g. a new B-Tree root), reads kept returning the old page until the next checkpoint. Dropping and recreating an index made range scans follow the old tree's leaf chain and return entries several times. A freed page that is in the WAL index, was checkpointed since the last WAL truncation, or is written by an open transaction is now queued and returned to the free list right after the next truncation. Every other page is freed immediately, as before. Queued pages are skipped by GetCollectionPageIds, discarded when their collection file is dropped, and released by a checkpoint on Dispose so a clean close does not leak them. Co-Authored-By: Claude Opus 5.5 --- src/BLite.Core/Indexing/BTreeIndex.cs | 5 +- .../Storage/StorageEngine.Memory.cs | 112 ++++++++++++-- .../Storage/StorageEngine.Recovery.cs | 120 +++++++++------ src/BLite.Core/Storage/StorageEngine.cs | 7 + tests/BLite.Tests/FreePageReuseTests.cs | 139 ++++++++++++++++++ 5 files changed, 324 insertions(+), 59 deletions(-) create mode 100644 tests/BLite.Tests/FreePageReuseTests.cs diff --git a/src/BLite.Core/Indexing/BTreeIndex.cs b/src/BLite.Core/Indexing/BTreeIndex.cs index e657c24..dedf5aa 100644 --- a/src/BLite.Core/Indexing/BTreeIndex.cs +++ b/src/BLite.Core/Indexing/BTreeIndex.cs @@ -1658,9 +1658,8 @@ private void MergeWithSibling(uint nodeId, uint parentId, int childIndex, ListFrees a page, routing to the correct sub-file via the encoded page ID. + // Pages checkpointed out of _walIndex whose records are still in the WAL file, i.e. since + // the last truncation. See FreePage. + private readonly ConcurrentDictionary _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 _pendingFreePages = new(); + + /// + /// Frees a page, routing to the correct sub-file via the encoded page ID. + /// + /// A page that still has a committed image in the WAL is not reusable right away: + /// 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 + /// 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. + /// + /// 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); + } + + /// + /// Whether the WAL holds, or an open transaction is about to commit, an image of the page. + /// + private bool HasWalImage(uint pageId) + { + if (_walIndex.ContainsKey(pageId) || _checkpointedWalPages.ContainsKey(pageId)) + return true; + + foreach (var txnPages in _walCache.Values) + if (txnPages.ContainsKey(pageId)) + return true; + + return false; + } + + /// + /// Returns every page queued by to its file's free list. Call only + /// right after the WAL was truncated, before any new commit can append to it. + /// + private void OnWalTruncated() + { + _checkpointedWalPages.Clear(); + + HashSet pages; + lock (_pendingFreeLock) + { + if (_pendingFreePages.Count == 0) return; + pages = _pendingFreePages; + _pendingFreePages = new HashSet(); + } + + foreach (var pageId in pages) + GetPageFile(pageId, out var physId).FreePage(physId); + } + + /// Snapshot of the queued frees, or null when there are none. + private HashSet? SnapshotPendingFreePages() + { + lock (_pendingFreeLock) + return _pendingFreePages.Count == 0 ? null : new HashSet(_pendingFreePages); + } + + /// + /// 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. + /// + 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 +248,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 +459,9 @@ internal IReadOnlyCollection 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 +477,15 @@ internal IReadOnlyCollection FreeCollectionPagesCore(string collectionName /// public IEnumerable 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 +496,11 @@ public IEnumerable 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; + } } // ----------------------------------------------------------------------- diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index aa8358d..5c75551 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -38,7 +38,16 @@ public void FlushPageFile() /// public async Task CheckpointAsync(CancellationToken ct = default) { - if (_walIndex.IsEmpty) return; + if (_walIndex.IsEmpty) + { + // Nothing to flush, but pages freed since the last truncation wait for the next one. + if (HasPendingFreePages && Interlocked.CompareExchange(ref _checkpointRunning, 1, 0) == 0) + { + try { await TryTruncateWalAsync(ct).ConfigureAwait(false); } + finally { Interlocked.Exchange(ref _checkpointRunning, 0); } + } + return; + } if (Interlocked.CompareExchange(ref _checkpointRunning, 1, 0) != 0) return; var sw = _metrics != null ? Metrics.ValueStopwatch.StartNew() : default; try @@ -70,6 +79,11 @@ public async Task CheckpointAsync(CancellationToken ct = default) if (toCheckpoint.Count == 0) return; + // The WAL file keeps these records until it is truncated; FreePage must still treat + // the pages as having a WAL image once they leave _walIndex below. + foreach (var kvp in toCheckpoint) + _checkpointedWalPages.TryAdd(kvp.Key, 0); + foreach (var kvp in toCheckpoint) GetPageFile(kvp.Key, out var physId).WritePage(physId, kvp.Value); @@ -108,53 +122,9 @@ public async Task CheckpointAsync(CancellationToken ct = default) _shm.WriteCheckpointedOffset(maxCheckpointedOffset); // Truncate WAL only when _walIndex is fully drained. - // _commitLock is still needed here to prevent truncating while the - // group-commit writer is appending new WAL records. - // - // In multi-process mode the SHM cross-process writer lock must also be held - // for the duration of the truncation: another process could be appending - // WAL records concurrently (its commit path acquires only its own - // _commitLock, not ours), and truncating the shared WAL file underneath it - // would corrupt the on-disk format. The lock order matches the commit path - // (in-process _commitLock → cross-process SHM writer lock) to avoid any - // possible cross-lock deadlock. - if (_walIndex.IsEmpty && _commitLock.Wait(0)) + if (_walIndex.IsEmpty) { - bool shmHeld = false; - try - { - if (_shm != null) - { - shmHeld = _shm.TryAcquireWriterLock(_config.LockTimeout.WriteTimeoutMs); - if (!shmHeld) - { - // Skip this truncation cycle rather than throw — a future - // checkpoint will retry. Concurrent commits from another - // process are a normal, recoverable condition. - return; - } - } - // Double-check: new commits may have promoted entries - // between our IsEmpty check and acquiring the lock. - if (_walIndex.IsEmpty) - { - await _wal.TruncateAsync(ct); - // Reset the SHM WAL index and end offset so other processes - // know the WAL has been cleared (Phase 4 / Phase 6). - if (_shm != null) - { - _shm.RebuildIndex(System.Array.Empty<(uint, long)>()); - _shm.WriteWalEndOffset(0); - _shm.WriteCheckpointedOffset(0); - Volatile.Write(ref _lastKnownWalEndOffset, 0); - } - } - } - finally - { - if (shmHeld) { try { _shm!.ReleaseWriterLock(); } catch { /* best-effort */ } } - _commitLock.Release(); - } + await TryTruncateWalAsync(ct); } else if (_shm != null && !_walIndex.IsEmpty) { @@ -201,6 +171,58 @@ public async Task CheckpointAsync(CancellationToken ct = default) } } + /// + /// Truncates the WAL if every committed record has been checkpointed, then releases the + /// pages queued by . Skipped when a commit holds the lock. + /// + private async Task TryTruncateWalAsync(CancellationToken ct) + { + // _commitLock prevents truncating while the group-commit writer is appending new WAL + // records. In multi-process mode the SHM cross-process writer lock must also be held: + // another process could be appending WAL records concurrently (its commit path acquires + // only its own _commitLock, not ours), and truncating the shared WAL file underneath it + // would corrupt the on-disk format. The lock order matches the commit path (in-process + // _commitLock → cross-process SHM writer lock) to avoid any possible cross-lock deadlock. + if (!_walIndex.IsEmpty || !_commitLock.Wait(0)) return; + + bool shmHeld = false; + try + { + if (_shm != null) + { + shmHeld = _shm.TryAcquireWriterLock(_config.LockTimeout.WriteTimeoutMs); + if (!shmHeld) + { + // Skip this truncation cycle rather than throw — a future + // checkpoint will retry. Concurrent commits from another + // process are a normal, recoverable condition. + return; + } + } + // Double-check: new commits may have promoted entries + // between our IsEmpty check and acquiring the lock. + if (_walIndex.IsEmpty) + { + await _wal.TruncateAsync(ct); + // Reset the SHM WAL index and end offset so other processes + // know the WAL has been cleared (Phase 4 / Phase 6). + if (_shm != null) + { + _shm.RebuildIndex(System.Array.Empty<(uint, long)>()); + _shm.WriteWalEndOffset(0); + _shm.WriteCheckpointedOffset(0); + Volatile.Write(ref _lastKnownWalEndOffset, 0); + } + OnWalTruncated(); + } + } + finally + { + if (shmHeld) { try { _shm!.ReleaseWriterLock(); } catch { /* best-effort */ } } + _commitLock.Release(); + } + } + /// /// Creates a consistent backup of this database to . /// @@ -238,7 +260,10 @@ internal async Task BackupDetailedAsync(string destinationDb { await CheckpointAsync(ct); if (_walIndex.IsEmpty) + { await _wal.TruncateAsync(ct).ConfigureAwait(false); + OnWalTruncated(); + } operations = BuildBackupPlan(destinationDbPath, includeIndexes); @@ -349,6 +374,7 @@ public async Task RecoverAsync(CancellationToken ct = default) // 5. Truncate WAL (all changes now in PageFile) await _wal.TruncateAsync(ct); + OnWalTruncated(); } finally { diff --git a/src/BLite.Core/Storage/StorageEngine.cs b/src/BLite.Core/Storage/StorageEngine.cs index b245765..b432f2b 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -355,6 +355,13 @@ public void Dispose() _activeTransactions.Clear(); } + // 2b. Pages freed since the last WAL truncation only reach the free list after the next + // one (see FreePage); checkpoint now so they are not leaked when the engine closes. + if (HasPendingFreePages) + { + try { Task.Run(() => CheckpointAsync()).GetAwaiter().GetResult(); } catch { /* best-effort: the pages stay leaked */ } + } + // 3. Close WAL and PageFile. try { _wal?.Dispose(); } catch { /* best-effort */ } try { _pageFile?.Dispose(); } catch { /* best-effort */ } diff --git a/tests/BLite.Tests/FreePageReuseTests.cs b/tests/BLite.Tests/FreePageReuseTests.cs new file mode 100644 index 0000000..69eb248 --- /dev/null +++ b/tests/BLite.Tests/FreePageReuseTests.cs @@ -0,0 +1,139 @@ +using BLite.Core; +using BLite.Core.Storage; +using BLite.Shared; + +namespace BLite.Tests; + +/// +/// A freed page must not be handed out again while the WAL still holds a committed image of +/// it: reads prefer the WAL, so the reused page would read back its old contents (#168). +/// +public class FreePageReuseTests : IDisposable +{ + private readonly string _dbPath; + + public FreePageReuseTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), $"freepage_{Guid.NewGuid():N}.db"); + } + + public void Dispose() + { + TryDelete(_dbPath); + TryDelete(Path.ChangeExtension(_dbPath, ".wal")); + var collDir = _dbPath + "_coll"; + if (Directory.Exists(collDir)) + try { Directory.Delete(collDir, recursive: true); } catch { /* best-effort */ } + } + + private static void TryDelete(string path) + { + try { if (File.Exists(path)) File.Delete(path); } catch { /* best-effort */ } + } + + [Fact] + public async Task DropAndRecreateIndex_RangeScan_ReturnsEachDocumentOnce() + { + using var db = new TestDbContext(_dbPath); + for (int i = 0; i < 200; i++) + await db.PeopleV2.InsertAsync(new PersonV2 { Name = $"p{i}", Age = i % 50 }); + + for (int pass = 0; pass < 4; pass++) + { + await db.PeopleV2.DropIndexAsync("idx_Age"); + await db.PeopleV2.CreateIndexAsync(p => p.Age, "idx_Age"); + + Assert.Equal(200, (await db.PeopleV2.QueryIndexAsync("idx_Age", null, null).ToListAsync()).Count); + } + + Assert.Equal(200, db.PeopleV2.AsQueryable().Where(p => p.Age >= 0).ToList().Count); + Assert.Equal(4, db.PeopleV2.AsQueryable().Where(p => p.Age == 49).ToList().Count); + } + + [Fact] + public async Task FreedPage_IsReusedOnlyAfterWalTruncation_AndReadsItsNewContents() + { + using var storage = new StorageEngine(_dbPath, PageFileConfig.Default); + var page = await AllocateAndCommitAsync(storage, fill: 0xAA); + + storage.FreePage(page); + Assert.NotEqual(page, storage.AllocatePage()); + + await storage.CheckpointAsync(); + Assert.Equal(page, storage.AllocatePage()); + + var fresh = Filled(storage.PageSize, 0xBB); + storage.WritePageImmediate(page, fresh); + var read = new byte[storage.PageSize]; + storage.ReadPage(page, null, read); + Assert.Equal(fresh, read); + } + + [Fact] + public async Task FreedPage_WithEmptyWalIndex_IsReleasedByCheckpoint() + { + using var storage = new StorageEngine(_dbPath, PageFileConfig.Default); + var page = await AllocateAndCommitAsync(storage, fill: 0xAA); + await storage.CheckpointAsync(); + + storage.FreePage(page); + await storage.CheckpointAsync(); + + Assert.Equal(page, storage.AllocatePage()); + } + + [Fact] + public async Task FreedPage_StillQueuedAtDispose_IsNotLeaked() + { + uint page; + using (var storage = new StorageEngine(_dbPath, PageFileConfig.Default)) + { + page = await AllocateAndCommitAsync(storage, fill: 0xAA); + storage.FreePage(page); + } + + using (var storage = new StorageEngine(_dbPath, PageFileConfig.Default)) + Assert.Equal(page, storage.AllocatePage()); + } + + [Fact] + public async Task DroppedCollectionFile_DiscardsItsQueuedFrees() + { + var config = PageFileConfig.Default with { CollectionDataDirectory = _dbPath + "_coll" }; + using var storage = new StorageEngine(_dbPath, config); + + var oldPage = storage.AllocateCollectionPage("items"); + await WriteAndCommitAsync(storage, oldPage, fill: 0xAA); + storage.FreePage(oldPage); + storage.DropCollectionFile("items"); + + // The recreated file reuses the slot, so its first page gets the same encoded ID. + var livePage = storage.AllocateCollectionPage("items"); + await WriteAndCommitAsync(storage, livePage, fill: 0xCC); + await storage.CheckpointAsync(); + + // Releasing the stale free would have put the live page on the new file's free list. + Assert.NotEqual(livePage, storage.AllocateCollectionPage("items")); + } + + private static async Task AllocateAndCommitAsync(StorageEngine storage, byte fill) + { + var page = storage.AllocatePage(); + await WriteAndCommitAsync(storage, page, fill); + return page; + } + + private static async Task WriteAndCommitAsync(StorageEngine storage, uint page, byte fill) + { + using var txn = storage.BeginTransaction(); + storage.WritePage(page, txn.TransactionId, Filled(storage.PageSize, fill)); + await txn.CommitAsync(); + } + + private static byte[] Filled(int size, byte value) + { + var buffer = new byte[size]; + buffer.AsSpan(32).Fill(value); // keep a zeroed page header + return buffer; + } +} From 7ae1bbd0795bcf16db19f4026cd1cfb6fcdef4cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 15:28:47 +0200 Subject: [PATCH 2/3] fix(storage): address review of deferred page frees (#168) - OnWalTruncated keeps a queued page that an open transaction has written: its image reaches the WAL only at that commit, so the page is released after the truncation that follows. Before, a B-Tree merge followed by a checkpoint before the commit released the page early (same failure as #168). - Multi-process: TryTruncateWalAsync skips the truncation while the shared WAL holds records this process has not replayed, so an empty local _walIndex can no longer discard another process's committed records. - Commit promotion writes _walIndex before removing the transaction from _walCache (group commit and both direct paths), so a page is never in neither while FreePage looks for its WAL image. - OnWalTruncated releases page by page and keeps a page whose release throws queued for the next truncation, instead of losing the rest of the batch. Tests for the open-transaction sequence (storage level and through a real B-Tree merge) and for the multi-process truncation; all three fail without these changes. Co-Authored-By: Claude Opus 5.5 --- .../Storage/StorageEngine.GroupCommit.cs | 4 +- .../Storage/StorageEngine.Memory.cs | 51 +++++++-- .../Storage/StorageEngine.Recovery.cs | 9 +- .../Storage/StorageEngine.Transactions.cs | 11 +- tests/BLite.Tests/FreePageReuseTests.cs | 105 ++++++++++++++++++ 5 files changed, 163 insertions(+), 17 deletions(-) diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index 20559b7..9245d37 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -193,9 +193,11 @@ private async Task ProcessBatchAsync(List 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 _); } } diff --git a/src/BLite.Core/Storage/StorageEngine.Memory.cs b/src/BLite.Core/Storage/StorageEngine.Memory.cs index cc11592..ca1f740 100644 --- a/src/BLite.Core/Storage/StorageEngine.Memory.cs +++ b/src/BLite.Core/Storage/StorageEngine.Memory.cs @@ -108,19 +108,17 @@ public void FreePage(uint pageId) /// private bool HasWalImage(uint pageId) { - if (_walIndex.ContainsKey(pageId) || _checkpointedWalPages.ContainsKey(pageId)) - return true; - - foreach (var txnPages in _walCache.Values) - if (txnPages.ContainsKey(pageId)) - return true; - - return false; + return _walIndex.ContainsKey(pageId) + || _checkpointedWalPages.ContainsKey(pageId) + || IsInOpenTransaction(pageId); } /// - /// Returns every page queued by to its file's free list. Call only - /// right after the WAL was truncated, before any new commit can append to it. + /// Returns the pages queued by to their file's free list. Call only + /// right after the WAL was truncated, under _commitLock, 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. /// private void OnWalTruncated() { @@ -134,8 +132,39 @@ private void OnWalTruncated() _pendingFreePages = new HashSet(); } + var keep = new List(); foreach (var pageId in pages) - GetPageFile(pageId, out var physId).FreePage(physId); + { + 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; } /// Snapshot of the queued frees, or null when there are none. diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index 5c75551..d214ae6 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -41,9 +41,10 @@ public async Task CheckpointAsync(CancellationToken ct = default) if (_walIndex.IsEmpty) { // Nothing to flush, but pages freed since the last truncation wait for the next one. + // No Checkpoint metric is published for this path: nothing is written to the page files. if (HasPendingFreePages && Interlocked.CompareExchange(ref _checkpointRunning, 1, 0) == 0) { - try { await TryTruncateWalAsync(ct).ConfigureAwait(false); } + try { await TryTruncateWalAsync(ct); } finally { Interlocked.Exchange(ref _checkpointRunning, 0); } } return; @@ -198,6 +199,12 @@ private async Task TryTruncateWalAsync(CancellationToken ct) // process are a normal, recoverable condition. return; } + + // Another process may have committed records this process has not replayed + // into its _walIndex yet; an empty local index says nothing about them, and + // truncating would discard them. + if (_shm.ReadWalEndOffset() > Volatile.Read(ref _lastKnownWalEndOffset)) + return; } // Double-check: new commits may have promoted entries // between our IsEmpty check and acquiring the lock. diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index ef40b21..9765a00 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -219,12 +219,13 @@ public async Task CommitTransactionAsync(ulong transactionId) await _wal.WriteCommitRecordAsync(transactionId); await _wal.FlushAsync(); // Durability: ensure WAL is on disk - // 3. Move pages from cache to WAL index (for reads) - _walCache.TryRemove(transactionId, out _); + // 3. Move pages from cache to WAL index (for reads). Index first, then drop the + // cache: a page must never be in neither (FreePage relies on finding its WAL image). foreach (var kvp in pages) { _walIndex[kvp.Key] = kvp.Value; } + _walCache.TryRemove(transactionId, out _); // Check if checkpoint is needed, but defer it until after releasing the lock needsCheckpoint = _wal.GetCurrentSize() > MaxWalSize; @@ -368,13 +369,15 @@ public async Task MarkTransactionCommittedAsync(ulong transactionId) await _wal.WriteCommitRecordAsync(transactionId); await _wal.FlushAsync(); - // Move from cache to WAL index - if (_walCache.TryRemove(transactionId, out var pages)) + // Move from cache to WAL index. Index first, then drop the cache: a page must + // never be in neither (FreePage relies on finding its WAL image). + if (_walCache.TryGetValue(transactionId, out var pages)) { foreach (var kvp in pages) { _walIndex[kvp.Key] = kvp.Value; } + _walCache.TryRemove(transactionId, out _); } // Check if checkpoint is needed, but defer it until after releasing the lock diff --git a/tests/BLite.Tests/FreePageReuseTests.cs b/tests/BLite.Tests/FreePageReuseTests.cs index 69eb248..c941d2f 100644 --- a/tests/BLite.Tests/FreePageReuseTests.cs +++ b/tests/BLite.Tests/FreePageReuseTests.cs @@ -1,4 +1,6 @@ +using BLite.Bson; using BLite.Core; +using BLite.Core.Indexing; using BLite.Core.Storage; using BLite.Shared; @@ -116,6 +118,109 @@ public async Task DroppedCollectionFile_DiscardsItsQueuedFrees() Assert.NotEqual(livePage, storage.AllocateCollectionPage("items")); } + [Fact] + public async Task FreedPage_WrittenByAnOpenTransaction_IsNotReleasedBeforeThatCommitIsCheckpointed() + { + using var storage = new StorageEngine(_dbPath, PageFileConfig.Default); + var page = storage.AllocatePage(); + + using (var txn = storage.BeginTransaction()) + { + storage.WritePage(page, txn.TransactionId, Filled(storage.PageSize, 0xAA)); + storage.FreePage(page); + + // A truncation while the transaction is open must not release the page: its image + // only reaches the WAL when the transaction commits. + await storage.CheckpointAsync(); + Assert.NotEqual(page, storage.AllocatePage()); + + await txn.CommitAsync(); + } + Assert.NotEqual(page, storage.AllocatePage()); + + // The checkpoint after the commit flushes that image and truncates; now the page is free. + await storage.CheckpointAsync(); + Assert.Equal(page, storage.AllocatePage()); + + var fresh = Filled(storage.PageSize, 0xBB); + storage.WritePageImmediate(page, fresh); + var read = new byte[storage.PageSize]; + storage.ReadPage(page, null, read); + Assert.Equal(fresh, read); + } + + [Fact] + public async Task BTreeMergesInAnOpenTransaction_SurviveACheckpointBeforeTheCommit() + { + using var storage = new StorageEngine(_dbPath, PageFileConfig.Default); + var tree = new BTreeIndex(storage, IndexOptions.CreateUnique("_id")); + var keys = Enumerable.Range(0, 400).Select(_ => new IndexKey(ObjectId.NewObjectId())).OrderBy(k => k).ToArray(); + using (var txn = storage.BeginTransaction()) + { + for (int i = 0; i < keys.Length; i++) + tree.Insert(keys[i], new DocumentLocation((uint)(i + 1), 0), txn.TransactionId); + await txn.CommitAsync(); + } + + // Deleting most keys merges leaves, which frees their pages inside the transaction. + using (var txn = storage.BeginTransaction()) + { + for (int i = 0; i < 350; i++) + tree.Delete(keys[i], new DocumentLocation((uint)(i + 1), 0), txn.TransactionId); + await storage.CheckpointAsync(); + await txn.CommitAsync(); + } + + // A second tree allocates pages; none of them may still belong to the first one. + var other = new BTreeIndex(storage, IndexOptions.CreateUnique("_id")); + var otherKeys = Enumerable.Range(0, 400).Select(_ => new IndexKey(ObjectId.NewObjectId())).ToArray(); + using (var txn = storage.BeginTransaction()) + { + for (int i = 0; i < otherKeys.Length; i++) + other.Insert(otherKeys[i], new DocumentLocation((uint)(i + 1000), 0), txn.TransactionId); + await txn.CommitAsync(); + } + await storage.CheckpointAsync(); + + Assert.Equal(keys.Skip(350), tree.Range(IndexKey.MinKey, IndexKey.MaxKey).Select(e => e.Key)); + Assert.Equal(400, other.Range(IndexKey.MinKey, IndexKey.MaxKey).Count()); + } + + [Fact] + public async Task MultiProcess_Truncation_KeepsRecordsAnotherProcessHasNotReplayed() + { + var dir = _dbPath + "_mp"; + Directory.CreateDirectory(dir); + var dbPath = Path.Combine(dir, "shared.db"); + var walPath = Path.ChangeExtension(dbPath, ".wal"); + var cfg = PageFileConfig.Default with { AllowMultiProcessAccess = true }; + try + { + using var engineA = new StorageEngine(dbPath, cfg); + using var engineB = new StorageEngine(dbPath, cfg); + + // A queues a free (every free is queued in multi-process mode) without replaying B. + engineA.FreePage(engineA.AllocatePage()); + + using (var txB = engineB.BeginTransaction()) + { + var page = engineB.AllocatePage(); + engineB.WritePage(page, txB.TransactionId, Filled(engineB.PageSize, 0xCC)); + await txB.CommitAsync(); + } + long walAfterB = new FileInfo(walPath).Length; + Assert.True(walAfterB > 0); + + // A's WAL index is empty, but B's committed records must stay in the shared WAL. + await engineA.CheckpointAsync(); + Assert.Equal(walAfterB, new FileInfo(walPath).Length); + } + finally + { + try { Directory.Delete(dir, recursive: true); } catch { /* best-effort */ } + } + } + private static async Task AllocateAndCommitAsync(StorageEngine storage, byte fill) { var page = storage.AllocatePage(); From b721feef0e719bad85ab006d4081684258eef0cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 18:56:59 +0200 Subject: [PATCH 3/3] fix(storage): replay other processes' WAL records before a commit (#168) Address review of #169: in multi-process mode, group commit advanced _lastKnownWalEndOffset past records another process had appended since this engine's last replay, so the truncation check passed and a checkpoint could drop that process's committed records. - Group commit catches up with the shared WAL under _commitLock and the SHM writer lock before appending: it replays [bookmark, SHM end) and only advances the bookmark past its batch if that replay succeeded. - Group commit appends at the current end of the WAL file instead of at this engine's stale stream position, which overwrote records other processes had appended (WriteAheadLog.SeekToEnd). - BeginTransaction replays through the same helper, under _commitLock, and the newest image of a page in the replayed range now replaces one this engine committed itself; it used to be skipped, leaving a stale image in _walIndex that a checkpoint wrote back before truncating. - Restore ConfigureAwait(false) on both TryTruncateWalAsync call sites. - Tests: begin B, commit A, commit B, checkpoint B; and a page committed by B then by A, replayed and checkpointed by B. Co-Authored-By: Claude Opus 5.5 --- .../Storage/StorageEngine.GroupCommit.cs | 20 +++- .../Storage/StorageEngine.Recovery.cs | 4 +- .../Storage/StorageEngine.Transactions.cs | 103 ++++++++++-------- src/BLite.Core/Transactions/WriteAheadLog.cs | 21 ++++ tests/BLite.Tests/FreePageReuseTests.cs | 81 ++++++++++++++ 5 files changed, 179 insertions(+), 50 deletions(-) diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index 9245d37..b1622db 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/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 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) @@ -223,8 +235,10 @@ private async Task ProcessBatchAsync(List 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 diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index d214ae6..98366a3 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -44,7 +44,7 @@ public async Task CheckpointAsync(CancellationToken ct = default) // No Checkpoint metric is published for this path: nothing is written to the page files. if (HasPendingFreePages && Interlocked.CompareExchange(ref _checkpointRunning, 1, 0) == 0) { - try { await TryTruncateWalAsync(ct); } + try { await TryTruncateWalAsync(ct).ConfigureAwait(false); } finally { Interlocked.Exchange(ref _checkpointRunning, 0); } } return; @@ -125,7 +125,7 @@ public async Task CheckpointAsync(CancellationToken ct = default) // Truncate WAL only when _walIndex is fully drained. if (_walIndex.IsEmpty) { - await TryTruncateWalAsync(ct); + await TryTruncateWalAsync(ct).ConfigureAwait(false); } else if (_shm != null && !_walIndex.IsEmpty) { diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 9765a00..790d99e 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -13,52 +13,15 @@ public Transaction BeginTransaction(IsolationLevel isolationLevel = IsolationLev // Compare the SHM WAL end offset against the last offset this engine has // replayed. If another process has committed since then, replay the new // WAL records into _walIndex so reads in this transaction see fresh data. - if (_shm != null && _wal is Transactions.WriteAheadLog walForReplay) + // The replay runs under _commitLock (see TryCatchUpWithSharedWal); if a local commit + // holds it for longer than the read timeout, the transaction starts without the + // replay: its reads may be stale, and the next transaction or commit catches up. + if (_shm != null && _wal is Transactions.WriteAheadLog walForReplay + && _shm.ReadWalEndOffset() != Volatile.Read(ref _lastKnownWalEndOffset) + && _commitLock.Wait(_config.LockTimeout.ReadTimeoutMs)) { - long shmEnd = _shm.ReadWalEndOffset(); - long lastKnown = Volatile.Read(ref _lastKnownWalEndOffset); - - // Detect WAL truncation by another process: shmEnd < lastKnown means the WAL - // was reset to 0 by a checkpoint and new records are already being appended. - // Reset our bookmark so the next iteration replays from the new base. - if (shmEnd < lastKnown) - { - Interlocked.CompareExchange(ref _lastKnownWalEndOffset, 0, lastKnown); - lastKnown = Volatile.Read(ref _lastKnownWalEndOffset); - } - - if (shmEnd > lastKnown) - { - // Attempt to claim the replay range; only the first thread wins — - // others will see the updated _lastKnownWalEndOffset and skip. - if (Interlocked.CompareExchange(ref _lastKnownWalEndOffset, shmEnd, lastKnown) == lastKnown) - { - try - { - var newPages = walForReplay.ReadCommittedPagesSince(lastKnown, shmEnd); - foreach (var (pid, data, walOffset) in newPages) - { - // Only populate _walIndex if this engine hasn't already committed - // a newer local version of this page. `_walOffsets` tracks pages - // committed by THIS engine; if the page is present there, our - // local _walIndex already has the fresher version. - if (_walOffsets == null || !_walOffsets.ContainsKey(pid)) - { - _walIndex[pid] = data; - // Record the WAL offset so CheckpointAsync can apply the - // GetMinReaderOffset() safe boundary for this page (Phase 6). - _walOffsets?[pid] = walOffset; - } - } - } - catch - { - // Best-effort: a failed replay only means stale reads, - // never data corruption. Reset so the next transaction retries. - Interlocked.CompareExchange(ref _lastKnownWalEndOffset, lastKnown, shmEnd); - } - } - } + try { TryCatchUpWithSharedWal(walForReplay); } + finally { _commitLock.Release(); } } // In multi-process mode, allocate the transaction id from the shared SHM counter @@ -91,6 +54,56 @@ public Transaction BeginTransaction(IsolationLevel isolationLevel = IsolationLev return transaction; } + /// + /// Replays into _walIndex the records other processes have committed to the shared + /// WAL since this engine last caught up with it, [_lastKnownWalEndOffset, SHM WAL end), + /// and advances _lastKnownWalEndOffset to that end. Returns false, leaving the + /// bookmark where it was, if the records could not be read. + /// + /// The caller must hold _commitLock, so no local commit or truncation interleaves. + /// The bookmark then never runs ahead of what this engine holds, which + /// relies on: every record before it is in + /// _walIndex or already checkpointed. The replayed range runs to the end of the WAL, + /// so the latest committed image of each page in it is the newest there is and replaces + /// whatever this engine holds for that page, including an image it committed itself. + /// + /// + private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) + { + long shmEnd = _shm!.ReadWalEndOffset(); + long lastKnown = Volatile.Read(ref _lastKnownWalEndOffset); + + // shmEnd < lastKnown: another process truncated the WAL after checkpointing it, and + // may have appended to it since. Replay the new WAL from its start. + if (shmEnd < lastKnown) + lastKnown = 0; + + if (shmEnd > lastKnown) + { + List<(uint pageId, byte[] data, long walOffset)> newPages; + try + { + newPages = wal.ReadCommittedPagesSince(lastKnown, shmEnd); + } + catch + { + // Stale reads until the next catch-up retries, never a skipped record. + return false; + } + + foreach (var (pid, data, walOffset) in newPages) + { + _walIndex[pid] = data; + // Record the WAL offset so CheckpointAsync can apply the + // GetMinReaderOffset() safe boundary for this page (Phase 6). + _walOffsets?[pid] = walOffset; + } + } + + Volatile.Write(ref _lastKnownWalEndOffset, shmEnd); + return true; + } + /// /// Forgets a transaction once it has committed or rolled back through the ulong /// overloads, which callers use directly without going through the ones. diff --git a/src/BLite.Core/Transactions/WriteAheadLog.cs b/src/BLite.Core/Transactions/WriteAheadLog.cs index 8857515..c211d03 100644 --- a/src/BLite.Core/Transactions/WriteAheadLog.cs +++ b/src/BLite.Core/Transactions/WriteAheadLog.cs @@ -470,6 +470,27 @@ public long GetCurrentSize() /// internal long GetCurrentPositionNoLock() => _walStream?.Position ?? 0; + /// + /// Moves the write position to the current end of the WAL file. In multi-process mode + /// another process may have appended to the file since this instance last wrote to it, so + /// the position left by this instance's own writes is stale; call this under the + /// cross-process writer lock before appending. + /// + internal void SeekToEnd() + { + if (!_lock.Wait(_writeTimeoutMs)) + throw new TimeoutException("Timed out acquiring WAL write lock."); + try + { + if (_walStream != null) + _walStream.Position = _walStream.Length; + } + finally + { + _lock.Release(); + } + } + /// /// Writes a WAL data record and returns the byte offset at which the record /// starts in the WAL file. The offset is captured atomically inside the WAL diff --git a/tests/BLite.Tests/FreePageReuseTests.cs b/tests/BLite.Tests/FreePageReuseTests.cs index c941d2f..a621a1e 100644 --- a/tests/BLite.Tests/FreePageReuseTests.cs +++ b/tests/BLite.Tests/FreePageReuseTests.cs @@ -221,6 +221,87 @@ public async Task MultiProcess_Truncation_KeepsRecordsAnotherProcessHasNotReplay } } + [Fact] + public async Task MultiProcess_CommitAfterAnotherProcessCommitted_CheckpointKeepsThatCommit() + { + var dir = _dbPath + "_mp"; + Directory.CreateDirectory(dir); + var dbPath = Path.Combine(dir, "shared.db"); + var cfg = PageFileConfig.Default with { AllowMultiProcessAccess = true }; + try + { + using var engineA = new StorageEngine(dbPath, cfg); + using var engineB = new StorageEngine(dbPath, cfg); + var pageA = engineA.AllocatePage(); + var pageB = engineA.AllocatePage(); + + // B begins before A commits, so its begin-time replay does not see A's record. + using (var txB = engineB.BeginTransaction()) + { + await WriteAndCommitAsync(engineA, pageA, fill: 0xAA); + + engineB.WritePage(pageB, txB.TransactionId, Filled(engineB.PageSize, 0xBB)); + await txB.CommitAsync(); + } + + // B's commit must not have skipped (or overwritten) A's record: after B checkpoints, + // A's page is still there for B and for a process that opens the database now. + await engineB.CheckpointAsync(); + + AssertPage(engineB, pageA, 0xAA); + AssertPage(engineB, pageB, 0xBB); + using var engineC = new StorageEngine(dbPath, cfg); + using (engineC.BeginTransaction()) + { + AssertPage(engineC, pageA, 0xAA); + AssertPage(engineC, pageB, 0xBB); + } + } + finally + { + try { Directory.Delete(dir, recursive: true); } catch { /* best-effort */ } + } + } + + [Fact] + public async Task MultiProcess_PageCommittedByBothProcesses_CheckpointKeepsTheNewerImage() + { + var dir = _dbPath + "_mp"; + Directory.CreateDirectory(dir); + var dbPath = Path.Combine(dir, "shared.db"); + var cfg = PageFileConfig.Default with { AllowMultiProcessAccess = true }; + try + { + using var engineA = new StorageEngine(dbPath, cfg); + using var engineB = new StorageEngine(dbPath, cfg); + var page = engineA.AllocatePage(); + + await WriteAndCommitAsync(engineB, page, fill: 0xBB); + await WriteAndCommitAsync(engineA, page, fill: 0xAA); + + // B replays A's newer image of the page it has committed itself, then checkpoints. + using (engineB.BeginTransaction()) + AssertPage(engineB, page, 0xAA); + await engineB.CheckpointAsync(); + + AssertPage(engineB, page, 0xAA); + using var engineC = new StorageEngine(dbPath, cfg); + using (engineC.BeginTransaction()) + AssertPage(engineC, page, 0xAA); + } + finally + { + try { Directory.Delete(dir, recursive: true); } catch { /* best-effort */ } + } + } + + private static void AssertPage(StorageEngine storage, uint page, byte fill) + { + var read = new byte[storage.PageSize]; + storage.ReadPage(page, null, read); + Assert.Equal(Filled(storage.PageSize, fill), read); + } + private static async Task AllocateAndCommitAsync(StorageEngine storage, byte fill) { var page = storage.AllocatePage();