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/5] 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/5] 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/5] 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(); From f77280b3951f881b0ba0943d2536926ae038a5eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 21:01:21 +0200 Subject: [PATCH 4/5] fix(storage): discard pages from a WAL another process has truncated (#197) Multi-process mode lost committed data once another process had checkpointed and truncated the shared WAL: - Pages an engine had replayed (or read through the SHM page index) stayed in its _walIndex. Its reads returned them, and its next checkpoint wrote them over the newer images on disk (#197). That needed no further transaction: CheckpointAsync never caught up first. - Truncation was detected only as "SHM end < bookmark". Once the new WAL grew past the old bookmark, the engine replayed it from that offset, skipped its first records, believed itself caught up and truncated them away. - RollbackTransactionAsync appended an advisory abort record outside the commit path, without the SHM writer lock and at the engine's own, possibly stale, write position. After another process's truncation that left a gap of zeros in the shared WAL, which ended other processes' replay early, so they skipped later commits and truncated them. Changes: - The SHM header gets a WAL generation (offset 48, previously reserved; files without it read 0), incremented as the last step of every truncation, under the writer lock. - The catch-up compares it with the generation the engine's state belongs to. On a change (or an end before the bookmark) it drops everything it holds from the old WAL, which the truncating process had checkpointed, and replays the new WAL from its start. Without the writer lock (BeginTransaction), a truncation during the replay discards its result. - A page read through the SHM index is cached only if no truncation or discard happened meanwhile (generation and a local epoch, under a small lock). - In multi-process mode CheckpointAsync catches up and writes its pages under _commitLock and the SHM writer lock, so it holds the newest image of every page it writes and no other process can commit or truncate meanwhile. The flush runs after the locks are released. Backup, which already holds _commitLock, uses the same path without taking it again. - Backup truncates through the same path as a checkpoint (SHM writer lock, no truncation over records other processes have not replayed, new generation) instead of truncating the WAL under _commitLock alone. - In multi-process mode the abort record is no longer written. Replay and recovery ignore it; only transactions with a commit record are applied. - README notes that BeginTransaction may wait up to ReadTimeoutMs and that a checkpoint makes commits wait for its page writes; the SHM layout in MULTI_PROCESS_WAL.md lists the new field. Tests (MultiProcessWalTruncationTests): the #197 sequence, the same without a second transaction, a WAL regrown past the old bookmark, an own commit after another process's truncation, a rollback after a truncation, and 25 seeded random interleavings of commits, empty and rolled-back transactions, reads and checkpoints on two engines, plus a truncation by a backup. All seven fail on b721fee. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 + roadmap/v5/MULTI_PROCESS_WAL.md | 4 +- src/BLite.Core/Storage/StorageEngine.Pages.cs | 58 +--- .../Storage/StorageEngine.Recovery.cs | 117 +++++--- .../Storage/StorageEngine.Transactions.cs | 94 +++++- src/BLite.Core/Storage/StorageEngine.cs | 8 + .../Transactions/WalSharedMemory.cs | 27 +- .../Transactions/WalSharedMemoryLayout.cs | 4 +- .../MultiProcessWalTruncationTests.cs | 271 ++++++++++++++++++ 9 files changed, 492 insertions(+), 93 deletions(-) create mode 100644 tests/BLite.Tests/MultiProcessWalTruncationTests.cs diff --git a/README.md b/README.md index 5969d25..58f0df7 100644 --- a/README.md +++ b/README.md @@ -360,6 +360,8 @@ var config = new PageFileConfig { AllowMultiProcessAccess = true }; using var db = new AppDb("shared.db", config); // open from multiple processes ``` +Each transaction starts by replaying what other processes have committed since. If a commit in the same process is running, it waits up to `LockTimeout.ReadTimeoutMs` for it; after that it starts anyway and may not see those other commits until the next transaction. A checkpoint writes its pages under the cross-process writer lock, so commits in every process wait for those writes (not for the flush that follows). + ### �📡 Change Data Capture (CDC) Real-time event streaming for database changes with transactional consistency. diff --git a/roadmap/v5/MULTI_PROCESS_WAL.md b/roadmap/v5/MULTI_PROCESS_WAL.md index 31022bf..2054408 100644 --- a/roadmap/v5/MULTI_PROCESS_WAL.md +++ b/roadmap/v5/MULTI_PROCESS_WAL.md @@ -135,7 +135,9 @@ Offset Size Field 24 8 WalEndOffset [last byte written to WAL, Interlocked] 32 8 CheckpointedOffset [last byte safely checkpointed, Interlocked] 40 4 WriterOwnerPid [0 = unlocked] -44 20 Reserved [pad to 64 bytes] +44 4 Reserved +48 8 WalGeneration [incremented on every WAL truncation, Interlocked] +56 8 Reserved [pad to 64 bytes] ───────────────────────────────────────────────────────── 64 MaxReaders × 16 Reader Slot Array Per slot: [long ProcessId][long MaxReadOffset] diff --git a/src/BLite.Core/Storage/StorageEngine.Pages.cs b/src/BLite.Core/Storage/StorageEngine.Pages.cs index 298a8db..1c68b43 100644 --- a/src/BLite.Core/Storage/StorageEngine.Pages.cs +++ b/src/BLite.Core/Storage/StorageEngine.Pages.cs @@ -39,23 +39,11 @@ public void ReadPage(uint pageId, ulong? transactionId, Span destination) // When another process committed a page that our local _walIndex doesn't have, // consult the SHM hash table to find the WAL byte offset, then read the page // data directly from the WAL file. - if (_shm != null && _wal is Transactions.WriteAheadLog walFile) + if (ReadFromSharedWal(pageId) is { } pageData) { - long walOffset = _shm.LookupPageOffset(pageId); - if (walOffset > 0) - { - var pageData = walFile.ReadPageAt(walOffset, pageId); - if (pageData != null) - { - // Populate local _walIndex and _walOffsets to amortise future reads - // and ensure CheckpointAsync respects the reader boundary (Phase 6). - _walIndex[pageId] = pageData; - _walOffsets?[pageId] = walOffset; - var length = Math.Min(pageData.Length, destination.Length); - pageData.AsSpan(0, length).CopyTo(destination); - return; - } - } + var length = Math.Min(pageData.Length, destination.Length); + pageData.AsSpan(0, length).CopyTo(destination); + return; } // 3. Read committed baseline from PageFile @@ -95,21 +83,11 @@ public void ReadPageHeader(uint pageId, ulong? transactionId, Span destina } // Cross-process WAL lookup (Phase 4): check SHM index on miss. - if (_shm != null && _wal is Transactions.WriteAheadLog walFileH) + if (ReadFromSharedWal(pageId) is { } pageData) { - long walOffset = _shm.LookupPageOffset(pageId); - if (walOffset > 0) - { - var pageData = walFileH.ReadPageAt(walOffset, pageId); - if (pageData != null) - { - _walIndex[pageId] = pageData; - _walOffsets?[pageId] = walOffset; - var length = Math.Min(pageData.Length, destination.Length); - pageData.AsSpan(0, length).CopyTo(destination); - return; - } - } + var length = Math.Min(pageData.Length, destination.Length); + pageData.AsSpan(0, length).CopyTo(destination); + return; } GetPageFile(pageId, out var physId).ReadPageHeader(physId, destination); @@ -200,25 +178,15 @@ public ValueTask ReadPageAsync(uint pageId, ulong? transactionId, Memory d } // 2b. Cross-process WAL index lookup (Phase 4) — synchronous read via RandomAccess. - if (_shm != null && _wal is Transactions.WriteAheadLog walFileAsync) + if (ReadFromSharedWal(pageId) is { } pageData) { - long walOffset = _shm.LookupPageOffset(pageId); - if (walOffset > 0) - { - var pageData = walFileAsync.ReadPageAt(walOffset, pageId); - if (pageData != null) - { - _walIndex[pageId] = pageData; - _walOffsets?[pageId] = walOffset; - var length = Math.Min(pageData.Length, destination.Length); - pageData.AsMemory(0, length).CopyTo(destination); + var length = Math.Min(pageData.Length, destination.Length); + pageData.AsMemory(0, length).CopyTo(destination); #if NET5_0_OR_GREATER - return ValueTask.CompletedTask; + return ValueTask.CompletedTask; #else - return default; + return default; #endif - } - } } // 3. PageFile — true async OS read diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index 98366a3..7172393 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -36,7 +36,14 @@ public void FlushPageFile() /// Async opportunistic checkpoint — NEVER blocks commits. /// See for the strategy. /// - public async Task CheckpointAsync(CancellationToken ct = default) + /// + /// In multi-process mode the page writes run under the commit lock and the cross-process + /// writer lock, so commits in this and other processes wait for them (not for the flush). + /// + public Task CheckpointAsync(CancellationToken ct = default) => CheckpointCoreAsync(commitLockHeld: false, ct); + + /// The caller holds _commitLock (backup). + private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct) { if (_walIndex.IsEmpty) { @@ -44,7 +51,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).ConfigureAwait(false); } + try { await TryTruncateWalCoreAsync(commitLockHeld, ct).ConfigureAwait(false); } finally { Interlocked.Exchange(ref _checkpointRunning, 0); } } return; @@ -53,40 +60,66 @@ public async Task CheckpointAsync(CancellationToken ct = default) var sw = _metrics != null ? Metrics.ValueStopwatch.StartNew() : default; try { - // Phase 6: Compute the safe upper boundary for this checkpoint run. - // In multi-process mode we must not flush WAL entries whose records are - // still within the read window of an active cross-process reader. The SHM - // reader-slot array gives us the minimum offset any live reader registered; - // we only checkpoint entries whose WAL offset is at or below that boundary. - long safeOffset = long.MaxValue; - if (_shm != null) - safeOffset = Math.Min(_wal.GetCurrentSize(), _shm.GetMinReaderOffset()); - - var snapshot = _walIndex.ToArray(); - if (snapshot.Length == 0) return; + List> toCheckpoint; - // Build the set of entries that are safe to checkpoint. - var toCheckpoint = new List>(snapshot.Length); - foreach (var kvp in snapshot) + // Multi-process: catch up with the shared WAL and write the pages under _commitLock and + // the SHM writer lock. Otherwise this engine could write an image older than one another + // process already checkpointed (from a record it has not replayed yet), or one from a WAL + // that process has checkpointed and truncated since (#197). Same order as the commit path. + bool writeCommitHeld = false, writeShmHeld = false; + try { - if (_shm != null && _walOffsets != null) + if (_shm != null && _wal is Transactions.WriteAheadLog sharedWal) { - // Only include entries whose WAL record is within the safe window. - if (_walOffsets.TryGetValue(kvp.Key, out long entryOffset) && entryOffset > safeOffset) - continue; // reader still needs this WAL record — skip for now + if (!commitLockHeld) + { + writeCommitHeld = await _commitLock.WaitAsync(_config.LockTimeout.WriteTimeoutMs, ct).ConfigureAwait(false); + if (!writeCommitHeld) return; // a later checkpoint retries + } + writeShmHeld = _shm.TryAcquireWriterLock(_config.LockTimeout.WriteTimeoutMs); + if (!writeShmHeld || !TryCatchUpWithSharedWal(sharedWal)) return; } - toCheckpoint.Add(kvp); - } - if (toCheckpoint.Count == 0) return; + // Phase 6: Compute the safe upper boundary for this checkpoint run. + // In multi-process mode we must not flush WAL entries whose records are + // still within the read window of an active cross-process reader. The SHM + // reader-slot array gives us the minimum offset any live reader registered; + // we only checkpoint entries whose WAL offset is at or below that boundary. + long safeOffset = long.MaxValue; + if (_shm != null) + safeOffset = Math.Min(_wal.GetCurrentSize(), _shm.GetMinReaderOffset()); + + var snapshot = _walIndex.ToArray(); + if (snapshot.Length == 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); + // Build the set of entries that are safe to checkpoint. + toCheckpoint = new List>(snapshot.Length); + foreach (var kvp in snapshot) + { + if (_shm != null && _walOffsets != null) + { + // Only include entries whose WAL record is within the safe window. + if (_walOffsets.TryGetValue(kvp.Key, out long entryOffset) && entryOffset > safeOffset) + continue; // reader still needs this WAL record — skip for now + } + toCheckpoint.Add(kvp); + } - foreach (var kvp in toCheckpoint) - GetPageFile(kvp.Key, out var physId).WritePage(physId, kvp.Value); + 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); + } + finally + { + if (writeShmHeld) { try { _shm!.ReleaseWriterLock(); } catch { /* best-effort */ } } + if (writeCommitHeld) _commitLock.Release(); + } // Phase 6: Parallel flush in multi-file mode. var flushTasks = new List(); @@ -125,7 +158,7 @@ public async Task CheckpointAsync(CancellationToken ct = default) // Truncate WAL only when _walIndex is fully drained. if (_walIndex.IsEmpty) { - await TryTruncateWalAsync(ct).ConfigureAwait(false); + await TryTruncateWalCoreAsync(commitLockHeld, ct).ConfigureAwait(false); } else if (_shm != null && !_walIndex.IsEmpty) { @@ -176,7 +209,10 @@ 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) + private Task TryTruncateWalAsync(CancellationToken ct) => TryTruncateWalCoreAsync(commitLockHeld: false, ct); + + /// The caller holds _commitLock (backup). + private async Task TryTruncateWalCoreAsync(bool commitLockHeld, 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: @@ -184,7 +220,7 @@ private async Task TryTruncateWalAsync(CancellationToken ct) // 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; + if (!_walIndex.IsEmpty || (!commitLockHeld && !_commitLock.Wait(0))) return; bool shmHeld = false; try @@ -219,6 +255,10 @@ private async Task TryTruncateWalAsync(CancellationToken ct) _shm.WriteWalEndOffset(0); _shm.WriteCheckpointedOffset(0); Volatile.Write(ref _lastKnownWalEndOffset, 0); + + // Last, so a process that sees the new generation also sees the reset end. It + // then discards what it replayed from the old WAL, which is now all on disk. + DiscardSharedWalState(_shm.IncrementWalGeneration()); } OnWalTruncated(); } @@ -226,7 +266,7 @@ private async Task TryTruncateWalAsync(CancellationToken ct) finally { if (shmHeld) { try { _shm!.ReleaseWriterLock(); } catch { /* best-effort */ } } - _commitLock.Release(); + if (!commitLockHeld) _commitLock.Release(); } } @@ -265,12 +305,11 @@ internal async Task BackupDetailedAsync(string destinationDb List? operations = null; try { - await CheckpointAsync(ct); - if (_walIndex.IsEmpty) - { - await _wal.TruncateAsync(ct).ConfigureAwait(false); - OnWalTruncated(); - } + await CheckpointCoreAsync(commitLockHeld: true, ct).ConfigureAwait(false); + + // Through the same path as a checkpoint: in multi-process mode it takes the SHM writer + // lock, keeps records other processes have not replayed, and starts a new WAL generation. + await TryTruncateWalCoreAsync(commitLockHeld: true, ct).ConfigureAwait(false); operations = BuildBackupPlan(destinationDbPath, includeIndexes); diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 790d99e..2228fd1 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -17,7 +17,8 @@ public Transaction BeginTransaction(IsolationLevel isolationLevel = IsolationLev // 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) + && (_shm.ReadWalEndOffset() != Volatile.Read(ref _lastKnownWalEndOffset) + || _shm.ReadWalGeneration() != Volatile.Read(ref _knownWalGeneration)) && _commitLock.Wait(_config.LockTimeout.ReadTimeoutMs)) { try { TryCatchUpWithSharedWal(walForReplay); } @@ -67,16 +68,28 @@ public Transaction BeginTransaction(IsolationLevel isolationLevel = IsolationLev /// 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. /// + /// + /// If another process has truncated the WAL since (a new WAL generation), everything this + /// engine holds from the old WAL is discarded first and the new WAL is replayed from its start. + /// Without the SHM writer lock, a truncation during the replay makes it return false. + /// /// private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) { - long shmEnd = _shm!.ReadWalEndOffset(); + long generation = _shm!.ReadWalGeneration(); + 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) + // Another process truncated the WAL: a new generation, or an end before the bookmark (as + // a process that does not count generations leaves it). It had checkpointed every record + // first, so the page files hold each page this engine replayed or read from the old WAL + // in that version or a newer one. Keeping them would serve stale reads and let a + // checkpoint write them back over the newer data (#197). Replay the new WAL from its start. + if (generation != Volatile.Read(ref _knownWalGeneration) || shmEnd < lastKnown) + { + DiscardSharedWalState(generation); lastKnown = 0; + } if (shmEnd > lastKnown) { @@ -91,6 +104,11 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) return false; } + // Without the SHM writer lock (BeginTransaction), the WAL can be truncated and + // rewritten while it is read; the bytes read then belong to neither generation. + if (_shm.ReadWalGeneration() != generation) + return false; + foreach (var (pid, data, walOffset) in newPages) { _walIndex[pid] = data; @@ -104,6 +122,64 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) return true; } + /// + /// Drops every page this engine holds from an earlier generation of the shared WAL and + /// records as the current one. Each of them is on disk in that + /// version or a newer one: the process that truncated the WAL had checkpointed it. Its own + /// commits are no exception, since a commit replays the WAL (and so notices a truncation) + /// before appending. + /// + private void DiscardSharedWalState(long generation) + { + lock (_replayLock) + { + _walIndex.Clear(); + _walOffsets?.Clear(); + _replayEpoch++; + Volatile.Write(ref _knownWalGeneration, generation); + Volatile.Write(ref _lastKnownWalEndOffset, 0); + } + } + + /// + /// Reads the latest committed image of straight from the shared WAL + /// through the SHM page index (Phase 4), or returns null if the WAL has none or was + /// truncated while it was read. The image is cached in _walIndex only if this engine's + /// view of the WAL is still the one it was read from: a page cached after a discard would be a + /// stale image that a checkpoint could write back. + /// + private byte[]? ReadFromSharedWal(uint pageId) + { + if (_shm == null || _wal is not Transactions.WriteAheadLog walFile) + return null; + + long epoch = Interlocked.Read(ref _replayEpoch); + long generation = _shm.ReadWalGeneration(); + long walOffset = _shm.LookupPageOffset(pageId); + if (walOffset <= 0) + return null; + + var pageData = walFile.ReadPageAt(walOffset, pageId); + if (pageData == null) + return null; + + lock (_replayLock) + { + // Truncated meanwhile: the page files hold every record of the old WAL. + if (_shm.ReadWalGeneration() != generation) + return null; + + // Populate local _walIndex and _walOffsets to amortise future reads + // and ensure CheckpointAsync respects the reader boundary (Phase 6). + if (_replayEpoch == epoch && _knownWalGeneration == generation) + { + _walIndex[pageId] = pageData; + _walOffsets?[pageId] = walOffset; + } + } + return pageData; + } + /// /// Forgets a transaction once it has committed or rolled back through the ulong /// overloads, which callers use directly without going through the ones. @@ -418,7 +494,13 @@ public async Task RollbackTransactionAsync(ulong transactionId) _walCache.TryRemove(transactionId, out _); try { - await _wal.WriteAbortRecordAsync(transactionId); + // The abort record is advisory: replay and recovery only apply transactions with a + // commit record. In multi-process mode it would be appended outside the commit path, + // without the SHM writer lock and at this engine's own write position, which other + // processes' appends or a truncation may have made stale. Written there, it overwrites + // their records or leaves a gap of zeros that ends every other process's replay early. + if (_shm == null) + await _wal.WriteAbortRecordAsync(transactionId); } finally { diff --git a/src/BLite.Core/Storage/StorageEngine.cs b/src/BLite.Core/Storage/StorageEngine.cs index b432f2b..1bfb011 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -47,6 +47,13 @@ public sealed partial class StorageEngine : IDisposable // Compared against _shm.ReadWalEndOffset() on BeginTransaction to detect // cross-process commits and trigger incremental WAL replay (Phase 7). private long _lastKnownWalEndOffset; + + // Multi-process: the generation of the shared WAL (WalSharedMemory.ReadWalGeneration) that the + // pages this engine replayed or read from it belong to, and how often it has discarded them. + // Both change only under _replayLock, which also guards caching a page read from the WAL. + private long _knownWalGeneration; + private long _replayEpoch; + private readonly object _replayLock = new(); // Collection-per-file: collectionName → IPageStorage dedicated // Null if CollectionDataDirectory not configured (embedded mode, single file) @@ -204,6 +211,7 @@ public StorageEngine(string databasePath, PageFileConfig config) var shmPath = walPath + "-shm"; _shm = Transactions.WalSharedMemory.Open(shmPath, config.PageSize); _walOffsets = new ConcurrentDictionary(); + _knownWalGeneration = _shm.ReadWalGeneration(); // If a previous writer crashed without releasing the writer lock, the // recorded PID will not be alive — clear it so this process (or another) diff --git a/src/BLite.Core/Transactions/WalSharedMemory.cs b/src/BLite.Core/Transactions/WalSharedMemory.cs index 377db0a..2df8662 100644 --- a/src/BLite.Core/Transactions/WalSharedMemory.cs +++ b/src/BLite.Core/Transactions/WalSharedMemory.cs @@ -38,7 +38,9 @@ namespace BLite.Core.Transactions; /// 24 8 WalEndOffset (Interlocked) /// 32 8 CheckpointedOffset (Interlocked) /// 40 4 WriterOwnerPid (0 = unlocked; set under writer lock) -/// 44 20 Reserved (padded to 64-byte header) +/// 44 4 Reserved +/// 48 8 WalGeneration (Interlocked; incremented on every WAL truncation) +/// 56 8 Reserved (padded to 64-byte header) /// 64 N×16 Reader-slot array (N = MaxReaders; per slot: long PID + long MaxReadOffset) /// /// @@ -236,6 +238,7 @@ private void InitializeHeader() _accessor.Write(WalSharedMemoryLayout.OffsetWalEndOffset, 0L); _accessor.Write(WalSharedMemoryLayout.OffsetCheckpointedOffset, 0L); _accessor.Write(WalSharedMemoryLayout.OffsetWriterOwnerPid, 0); + _accessor.Write(WalSharedMemoryLayout.OffsetWalGeneration, 0L); _accessor.Flush(); } @@ -316,6 +319,28 @@ public void WriteCheckpointedOffset(long offset) Volatile.Write(ref RefLong(WalSharedMemoryLayout.OffsetCheckpointedOffset), offset); } + /// + /// Reads the WAL generation: how many times the shared WAL has been truncated. Records a + /// process replayed from an earlier generation are stale, because the process that truncated + /// had checkpointed every record first. SHM files written before this field existed read 0. + /// + public long ReadWalGeneration() + { + ThrowIfDisposed(); + return Volatile.Read(ref RefLong(WalSharedMemoryLayout.OffsetWalGeneration)); + } + + /// + /// Starts a new WAL generation and returns it. Called under the writer lock once the WAL has + /// been truncated and WalEndOffset reset, so a process that sees the new generation + /// also sees the reset offset. + /// + public long IncrementWalGeneration() + { + ThrowIfDisposed(); + return Interlocked.Increment(ref RefLong(WalSharedMemoryLayout.OffsetWalGeneration)); + } + /// /// Returns the PID currently recorded as the writer owner, or 0 if no writer is registered. /// diff --git a/src/BLite.Core/Transactions/WalSharedMemoryLayout.cs b/src/BLite.Core/Transactions/WalSharedMemoryLayout.cs index f2ad5d9..2d19169 100644 --- a/src/BLite.Core/Transactions/WalSharedMemoryLayout.cs +++ b/src/BLite.Core/Transactions/WalSharedMemoryLayout.cs @@ -26,7 +26,9 @@ internal static class WalSharedMemoryLayout public const int OffsetWalEndOffset = 24; public const int OffsetCheckpointedOffset = 32; public const int OffsetWriterOwnerPid = 40; - // bytes 44..63 reserved for future use (padding to 64-byte cache line) + // bytes 44..47 reserved + public const int OffsetWalGeneration = 48; + // bytes 56..63 reserved for future use (padding to 64-byte cache line) /// Total header size in bytes (also the offset of the first reader slot). public const int HeaderSize = 64; diff --git a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs new file mode 100644 index 0000000..6403348 --- /dev/null +++ b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs @@ -0,0 +1,271 @@ +using BLite.Core.Storage; + +namespace BLite.Tests; + +/// +/// Multi-process mode: once another process has checkpointed and truncated the shared WAL, the +/// pages this engine replayed from it are older than the page file and must never be read or +/// checkpointed again (#197). Two engines on the same files stand in for two processes. +/// +public class MultiProcessWalTruncationTests : IDisposable +{ + private readonly string _dir; + private readonly string _dbPath; + private static readonly PageFileConfig Config = PageFileConfig.Default with { AllowMultiProcessAccess = true }; + + public MultiProcessWalTruncationTests() + { + _dir = Path.Combine(Path.GetTempPath(), $"blite_mp_truncation_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_dir); + _dbPath = Path.Combine(_dir, "shared.db"); + } + + public void Dispose() + { + try { Directory.Delete(_dir, recursive: true); } catch { /* best-effort */ } + } + + private long WalLength => new FileInfo(Path.ChangeExtension(_dbPath, ".wal")).Length; + + [Fact] + public async Task ReplayedPage_AfterAnotherProcessTruncated_IsNotWrittenBackOnCheckpoint() + { + // The sequence from #197. + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var page = a.AllocatePage(); + + await WriteAndCommitAsync(a, page, 0x11); + using (b.BeginTransaction()) + AssertPage(b, page, 0x11); // B replays 0x11 + + await WriteAndCommitAsync(a, page, 0x22); + await a.CheckpointAsync(); + Assert.Equal(0, WalLength); // A truncated the WAL + + using (b.BeginTransaction()) + AssertPage(b, page, 0x22); + await b.CheckpointAsync(); + + AssertPage(b, page, 0x22); + AssertFreshEngineReads(page, 0x22); + } + + [Fact] + public async Task ReplayedPage_AfterAnotherProcessTruncated_IsNotWrittenBack_EvenWithoutANewTransaction() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var page = a.AllocatePage(); + + await WriteAndCommitAsync(a, page, 0x11); + using (b.BeginTransaction()) + AssertPage(b, page, 0x11); + + await WriteAndCommitAsync(a, page, 0x22); + await a.CheckpointAsync(); + Assert.Equal(0, WalLength); + + // B checkpoints straight away, without a transaction that would catch up first. + await b.CheckpointAsync(); + + AssertFreshEngineReads(page, 0x22); + AssertPage(b, page, 0x22); + } + + [Fact] + public async Task WalRegrownPastTheReplayPosition_IsReplayedFromItsStart() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var first = a.AllocatePage(); + var second = a.AllocatePage(); + var third = a.AllocatePage(); + + await WriteAndCommitAsync(a, first, 0x11); + using (b.BeginTransaction()) + AssertPage(b, first, 0x11); // B's replay position: the end of one transaction + long bPosition = WalLength; + + await a.CheckpointAsync(); + Assert.Equal(0, WalLength); + + // A appends more than B had replayed, so the new WAL's end passes B's old position. + await WriteAndCommitAsync(a, second, 0x22); + await WriteAndCommitAsync(a, third, 0x33); + Assert.True(WalLength > bPosition); + + // B catches up without reading the pages (a read would fetch them from the WAL on its + // own). Replaying from B's old position would skip the new WAL's first record; B would + // then believe it holds everything, checkpoint, and truncate that record away. + using (b.BeginTransaction()) { } + await b.CheckpointAsync(); + AssertFreshEngineReads(first, 0x11); + AssertFreshEngineReads(second, 0x22); + AssertFreshEngineReads(third, 0x33); + } + + [Fact] + public async Task OwnCommit_AfterAnotherProcessTruncated_IsKept() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var shared = a.AllocatePage(); + var own = a.AllocatePage(); + + await WriteAndCommitAsync(a, shared, 0x11); + using (b.BeginTransaction()) + AssertPage(b, shared, 0x11); + + await WriteAndCommitAsync(a, shared, 0x22); + await a.CheckpointAsync(); + Assert.Equal(0, WalLength); + + // B commits without beginning in between; its commit lands in the new WAL. + using (var tx = b.BeginTransaction()) + { + b.WritePage(own, tx.TransactionId, Filled(b.PageSize, 0xBB)); + await tx.CommitAsync(); + } + AssertPage(b, own, 0xBB); + AssertPage(b, shared, 0x22); + + await b.CheckpointAsync(); + AssertFreshEngineReads(shared, 0x22); + AssertFreshEngineReads(own, 0xBB); + } + + [Fact] + public async Task TruncationByABackup_IsSeenByTheOtherProcess() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var page = a.AllocatePage(); + + await WriteAndCommitAsync(a, page, 0x11); + using (b.BeginTransaction()) + AssertPage(b, page, 0x11); + await WriteAndCommitAsync(a, page, 0x22); + + // A backup checkpoints and truncates the WAL, like a checkpoint does. + await a.BackupAsync(Path.Combine(_dir, "backup.db")); + Assert.Equal(0, WalLength); + + await b.CheckpointAsync(); + AssertFreshEngineReads(page, 0x22); + AssertPage(b, page, 0x22); + } + + [Fact] + public async Task RollbackAfterAnotherProcessTruncated_LeavesNoGapThatHidesLaterCommits() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var p1 = a.AllocatePage(); + var p2 = a.AllocatePage(); + var p3 = a.AllocatePage(); + + // A writes two transactions, so its own write position is past one transaction's length. + await WriteAndCommitAsync(a, p1, 0x11); + await WriteAndCommitAsync(a, p1, 0x12); + using (b.BeginTransaction()) { } + await b.CheckpointAsync(); + Assert.Equal(0, WalLength); + await WriteAndCommitAsync(b, p2, 0x22); + + // A rolls back. An abort record written at A's stale position would leave a gap of zeros + // after B's record, and B's replay would stop there and miss A's next commit. + using (a.BeginTransaction()) { } + await WriteAndCommitAsync(a, p3, 0x33); + + await b.CheckpointAsync(); + AssertFreshEngineReads(p1, 0x12); + AssertFreshEngineReads(p2, 0x22); + AssertFreshEngineReads(p3, 0x33); + } + + [Fact] + public async Task RandomInterleavings_KeepEveryCommit() + { + // Seeded, so a failure reproduces. Commits, empty and rolled-back transactions, reads and + // checkpoints (each of which may truncate the WAL) from either engine. + for (int seed = 0; seed < 25; seed++) + { + var dbPath = Path.Combine(_dir, $"stress_{seed}.db"); + var rnd = new Random(seed); + var expected = new Dictionary(); + using (var a = new StorageEngine(dbPath, Config)) + using (var b = new StorageEngine(dbPath, Config)) + { + var engines = new[] { a, b }; + var pages = Enumerable.Range(0, 8).Select(_ => a.AllocatePage()).ToArray(); + for (int step = 0; step < 300; step++) + { + var engine = engines[rnd.Next(2)]; + var page = pages[rnd.Next(pages.Length)]; + switch (rnd.Next(6)) + { + case 0: + case 1: + var fill = (byte)rnd.Next(1, 255); + await WriteAndCommitAsync(engine, page, fill); + expected[page] = fill; + break; + case 2: + using (engine.BeginTransaction()) { } + break; + case 3: + using (var tx = engine.BeginTransaction()) + engine.WritePage(page, tx.TransactionId, Filled(engine.PageSize, 0xEE)); // rolled back + break; + case 4: + await engine.CheckpointAsync(); + break; + case 5: + using (engine.BeginTransaction()) + if (expected.TryGetValue(page, out var current)) + AssertPage(engine, page, current); + break; + } + } + await a.CheckpointAsync(); + await b.CheckpointAsync(); + await a.CheckpointAsync(); + } + + using var c = new StorageEngine(dbPath, Config); + using (c.BeginTransaction()) + foreach (var (page, fill) in expected) + AssertPage(c, page, fill); + } + } + + private void AssertFreshEngineReads(uint page, byte fill) + { + using var c = new StorageEngine(_dbPath, Config); + using (c.BeginTransaction()) + AssertPage(c, page, fill); + } + + 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)[32], read[32]); + Assert.Equal(Filled(storage.PageSize, fill), read); + } + + 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 829b471267ec53788b70af8423262f6bb00f7419 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Tue, 13 Oct 2026 15:47:25 +0200 Subject: [PATCH 5/5] fix(storage): address review of the multi-process truncation fix (#201) - Direct commit (CommitTransactionAsync(ulong)) and two-phase commit (both PrepareTransactionAsync overloads, MarkTransactionCommittedAsync) throw NotSupportedException in multi-process mode. They append without the SHM writer lock, without catching up and without publishing the WAL end (#188), so another process could truncate past their records and this engine would then discard them. Nothing in the library calls them; Transaction.CommitAsync uses group commit. - A checkpoint that cannot take its locks or catch up with the shared WAL publishes the Checkpoint metric with Success = false; the new MetricsSnapshot.CheckpointsFailedTotal counts these, and they no longer count towards CheckpointsTotal. A failed catch-up is traced as a warning (the first, then every power of two in a row). - WalSharedMemory.Open no longer re-initialises a well-formed SHM with another page size or reader count while it shows signs of use (WAL end offset, live writer or reader PID); it throws instead. Otherwise the re-initialised header keeps NextTransactionId and starts a new WAL generation instead of resetting it to 0. - Backup no longer calls TryTruncateWalCoreAsync a second time: the checkpoint truncates, and whatever stays in the WAL is copied with it. - README and MULTI_PROCESS_WAL.md: every process must run a version later than 5.2.3; the gated commit paths; the configuration-mismatch rule. Tests: the review's direct-commit sequence, the 2PC paths in multi-process and single-process mode, a checkpoint blocked by the writer lock, and SHM re-opening with another page size while in use and while idle. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 + roadmap/v5/MULTI_PROCESS_WAL.md | 6 + src/BLite.Core/Metrics/MetricsDispatcher.cs | 7 + src/BLite.Core/Metrics/MetricsSnapshot.cs | 7 + .../Storage/StorageEngine.Recovery.cs | 24 ++- .../Storage/StorageEngine.Transactions.cs | 58 ++++++- src/BLite.Core/Storage/StorageEngine.cs | 3 + .../Transactions/WalSharedMemory.cs | 54 +++++- src/BLite.Core/Transactions/WriteAheadLog.cs | 3 + .../MultiProcessWalTruncationTests.cs | 160 ++++++++++++++++++ 10 files changed, 309 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 90f22af..1a868fd 100644 --- a/README.md +++ b/README.md @@ -362,6 +362,8 @@ using var db = new AppDb("shared.db", config); // open from multiple processes Each transaction starts by replaying what other processes have committed since. If a commit in the same process is running, it waits up to `LockTimeout.ReadTimeoutMs` for it; after that it starts anyway and may not see those other commits until the next transaction. A checkpoint writes its pages under the cross-process writer lock, so commits in every process wait for those writes (not for the flush that follows). +Every process that opens the database must run a BLite version later than 5.2.3. Older versions don't track the WAL generation in the `.wal-shm`, so in a mixed deployment a process can still read, and checkpoint back over newer data, pages from a WAL another process has already checkpointed and truncated ([#197](https://github.com/EntglDb/BLite/issues/197)). Commit through `Transaction.CommitAsync` (group commit): the direct commit path `CommitTransactionAsync(ulong)` and two-phase commit (`PrepareTransactionAsync`, `MarkTransactionCommittedAsync`) throw `NotSupportedException` in multi-process mode ([#188](https://github.com/EntglDb/BLite/issues/188)). + ### �📡 Change Data Capture (CDC) Real-time event streaming for database changes with transactional consistency. diff --git a/roadmap/v5/MULTI_PROCESS_WAL.md b/roadmap/v5/MULTI_PROCESS_WAL.md index 2054408..cb220e9 100644 --- a/roadmap/v5/MULTI_PROCESS_WAL.md +++ b/roadmap/v5/MULTI_PROCESS_WAL.md @@ -148,6 +148,12 @@ Offset Size Field The WAL index is a double-buffered open-addressing hash table mapping `uint pageId → long walByteOffset`. The double-buffer allows the writer to rebuild one half while readers continue using the other, then atomically flip a pointer in the SHM header. +**Version compatibility.** `WalGeneration` was added in a reserved field without changing `Version`, so older processes can open the same `.wal-shm`, but they neither increment it when they truncate the WAL nor discard what they replayed when another process does. Every process that opens the database must therefore run a BLite version later than 5.2.3; in a mixed deployment the older processes keep the bug described in [#197](https://github.com/EntglDb/BLite/issues/197) (stale pages read and checkpointed back after another process truncated the WAL). + +**Configuration mismatch.** A well-formed `.wal-shm` whose page size or reader-slot count differs from the opening process's configuration is re-initialised only if nothing shows it in use (a WAL end offset of 0, no live writer or reader PID); otherwise `Open` throws `InvalidOperationException`. On re-initialisation `WalGeneration` is incremented and `NextTransactionId` kept, so neither ever goes back. + +**Commit paths.** Only group commit appends under the cross-process writer lock, after catching up with the shared WAL, and publishes the new end offset. The direct commit path `CommitTransactionAsync(ulong)` and two-phase commit (`PrepareTransactionAsync`, `MarkTransactionCommittedAsync`) do none of that and throw `NotSupportedException` in multi-process mode ([#188](https://github.com/EntglDb/BLite/issues/188)). + --- ## Platform Matrix diff --git a/src/BLite.Core/Metrics/MetricsDispatcher.cs b/src/BLite.Core/Metrics/MetricsDispatcher.cs index ead7a78..00afef3 100644 --- a/src/BLite.Core/Metrics/MetricsDispatcher.cs +++ b/src/BLite.Core/Metrics/MetricsDispatcher.cs @@ -37,6 +37,7 @@ internal sealed class MetricsDispatcher : IDisposable // ── Checkpoint counters ───────────────────────────────────────────────── private long _checkpoints; private long _checkpointLatencySum; // microseconds + private long _checkpointsFailed; // ── Global collection counters ────────────────────────────────────────── private long _inserts; @@ -185,6 +186,7 @@ public MetricsSnapshot GetSnapshot() GroupCommitAvgBatchSize = gcBatches > 0 ? (double)gcSizeSum / gcBatches : 0, CheckpointsTotal = cpCount, AvgCheckpointLatencyUs = cpCount > 0 ? (double)cpLatSum / cpCount : 0, + CheckpointsFailedTotal = Interlocked.Read(ref _checkpointsFailed), InsertsTotal = inserts, UpdatesTotal = updates, DeletesTotal = deletes, @@ -247,6 +249,11 @@ private void Aggregate(MetricEvent evt) break; case MetricEventType.Checkpoint: + if (!evt.Success) + { + Interlocked.Increment(ref _checkpointsFailed); + break; + } Interlocked.Increment(ref _checkpoints); Interlocked.Add(ref _checkpointLatencySum, micros); break; diff --git a/src/BLite.Core/Metrics/MetricsSnapshot.cs b/src/BLite.Core/Metrics/MetricsSnapshot.cs index 27c1760..0477a04 100644 --- a/src/BLite.Core/Metrics/MetricsSnapshot.cs +++ b/src/BLite.Core/Metrics/MetricsSnapshot.cs @@ -37,6 +37,13 @@ public sealed class MetricsSnapshot /// Average checkpoint duration in microseconds. 0 if no checkpoints yet. public double AvgCheckpointLatencyUs { get; init; } + /// + /// Total number of checkpoints that wrote nothing because they failed: a lock timeout, an + /// error, or (multi-process mode) a shared WAL that could not be replayed first. Not included + /// in or . + /// + public long CheckpointsFailedTotal { get; init; } + // ── Collection operations ─────────────────────────────────────────────── /// Total document inserts across all collections. diff --git a/src/BLite.Core/Storage/StorageEngine.Recovery.cs b/src/BLite.Core/Storage/StorageEngine.Recovery.cs index 7172393..e7c848b 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -58,6 +58,11 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct } if (Interlocked.CompareExchange(ref _checkpointRunning, 1, 0) != 0) return; var sw = _metrics != null ? Metrics.ValueStopwatch.StartNew() : default; + + // Reported in the Checkpoint metric. A checkpoint that could not take its locks or catch + // up with the shared WAL writes nothing, not even this engine's own pages; a later one + // retries. A failed catch-up is also reported as a warning (ReportFailedCatchUp). + bool success = true; try { List> toCheckpoint; @@ -74,10 +79,10 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct if (!commitLockHeld) { writeCommitHeld = await _commitLock.WaitAsync(_config.LockTimeout.WriteTimeoutMs, ct).ConfigureAwait(false); - if (!writeCommitHeld) return; // a later checkpoint retries + if (!writeCommitHeld) { success = false; return; } // a later checkpoint retries } writeShmHeld = _shm.TryAcquireWriterLock(_config.LockTimeout.WriteTimeoutMs); - if (!writeShmHeld || !TryCatchUpWithSharedWal(sharedWal)) return; + if (!writeShmHeld || !TryCatchUpWithSharedWal(sharedWal)) { success = false; return; } } // Phase 6: Compute the safe upper boundary for this checkpoint run. @@ -191,6 +196,11 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct } } } + catch + { + success = false; + throw; + } finally { Interlocked.Exchange(ref _checkpointRunning, 0); @@ -200,7 +210,7 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct Timestamp = sw.StartTimestamp, Type = Metrics.MetricEventType.Checkpoint, ElapsedMicros = sw.GetElapsedMicros(), - Success = true, + Success = success, }); } } @@ -305,12 +315,12 @@ internal async Task BackupDetailedAsync(string destinationDb List? operations = null; try { + // Truncates the WAL through the checkpoint path once every page is written: in + // multi-process mode under the SHM writer lock, keeping records other processes have + // not replayed, and starting a new WAL generation. Whatever stays in the WAL is + // copied with it below. await CheckpointCoreAsync(commitLockHeld: true, ct).ConfigureAwait(false); - // Through the same path as a checkpoint: in multi-process mode it takes the SHM writer - // lock, keeps records other processes have not replayed, and starts a new WAL generation. - await TryTruncateWalCoreAsync(commitLockHeld: true, ct).ConfigureAwait(false); - operations = BuildBackupPlan(destinationDbPath, includeIndexes); foreach (var operation in operations) diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index 2228fd1..b334377 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -98,9 +98,10 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) { newPages = wal.ReadCommittedPagesSince(lastKnown, shmEnd); } - catch + catch (Exception ex) { // Stale reads until the next catch-up retries, never a skipped record. + ReportFailedCatchUp(wal, lastKnown, shmEnd, ex.Message); return false; } @@ -119,9 +120,28 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) } Volatile.Write(ref _lastKnownWalEndOffset, shmEnd); + _failedCatchUps = 0; return true; } + /// + /// Counts the catch-ups in a row that could not replay the shared WAL, and reports the first + /// one and then every power of two as a warning (): a WAL that stays + /// unreadable is visible without flooding the log. Until a catch-up succeeds, this engine's + /// reads may miss other processes' commits and its checkpoints write nothing, so it never + /// truncates the WAL. Called under _commitLock. + /// + private void ReportFailedCatchUp(Transactions.WriteAheadLog wal, long fromOffset, long toOffset, string reason) + { + int failures = ++_failedCatchUps; + if ((failures & (failures - 1)) != 0) + return; + Trace.TraceWarning( + $"BLite: could not replay the shared WAL '{wal.FilePath}' from offset {fromOffset} to {toOffset} " + + $"({failures} attempt(s) in a row): {reason}. Until it can, reads may miss other processes' commits " + + "and checkpoints write nothing, so the WAL is not truncated."); + } + /// /// Drops every page this engine holds from an earlier generation of the shared WAL and /// records as the current one. Each of them is on disk in that @@ -220,8 +240,10 @@ public async Task RollbackTransactionAsync(Transaction transaction) /// Transaction ID /// All writes to record in WAL /// True if preparation succeeded + /// Multi-process mode (see ). public async Task PrepareTransactionAsync(ulong transactionId) { + ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try { await _wal.WriteBeginRecordAsync(transactionId); @@ -241,8 +263,10 @@ public async Task PrepareTransactionAsync(ulong transactionId) } } + /// Multi-process mode (see ). public async Task PrepareTransactionAsync(ulong transactionId, CancellationToken ct = default) { + ThrowIfMultiProcessAccess(nameof(PrepareTransactionAsync)); try { await _wal.WriteBeginRecordAsync(transactionId, ct); @@ -274,11 +298,16 @@ public async Task PrepareTransactionAsync(ulong transactionId, Cancellatio /// /// Transaction to commit /// All writes performed in this transaction (unused, kept for compatibility) + /// + /// Multi-process mode (see ); commit through + /// or CommitTransactionAsync(transactionId, cancellationToken). + /// public async Task CommitTransactionAsync(ulong transactionId) { - // NOTE: No admission gate here — the caller (CommitTransactionAsync(Transaction, ct)) - // already acquired the gate. This method is also called from the group commit writer - // which should not be gated. + ThrowIfMultiProcessAccess(nameof(CommitTransactionAsync) + "(ulong)"); + + // NOTE: No admission gate here. Nothing in the library calls this direct path; + // Transaction.CommitAsync goes through group commit (the CancellationToken overload). bool needsCheckpoint = false; bool completed = false; @@ -448,8 +477,10 @@ public async Task CommitTransactionAsync(ulong transactionId, CancellationToken /// Used for 2PC: after PrepareAsync() writes to WAL, this finalizes the commit. /// /// Transaction to mark committed + /// Multi-process mode (see ). public async Task MarkTransactionCommittedAsync(ulong transactionId) { + ThrowIfMultiProcessAccess(nameof(MarkTransactionCommittedAsync)); bool needsCheckpoint = false; _commitLock.Wait(); @@ -485,6 +516,25 @@ public async Task MarkTransactionCommittedAsync(ulong transactionId) } } + /// + /// Rejects the direct commit path (CommitTransactionAsync(ulong)) and two-phase commit + /// (, ) in + /// multi-process mode. They append to the shared WAL at this engine's own write position, + /// without the cross-process writer lock, without catching up with the shared WAL first and + /// without publishing the new WAL end (#188). Their records can overwrite another process's, + /// and stay invisible to other processes, which then checkpoint and truncate the WAL without + /// them. This engine then discards them along with everything else it holds from the old WAL, + /// so the commit is lost. Group commit () does all of + /// that under both locks. + /// + private void ThrowIfMultiProcessAccess(string operation) + { + if (_shm != null) + throw new NotSupportedException( + $"{operation} is not supported with PageFileConfig.AllowMultiProcessAccess. " + + "Commit through Transaction.CommitAsync or CommitTransactionAsync(transactionId, cancellationToken)."); + } + /// /// Rolls back a transaction: discards all uncommitted changes. /// diff --git a/src/BLite.Core/Storage/StorageEngine.cs b/src/BLite.Core/Storage/StorageEngine.cs index 1bfb011..bac33f8 100644 --- a/src/BLite.Core/Storage/StorageEngine.cs +++ b/src/BLite.Core/Storage/StorageEngine.cs @@ -54,6 +54,9 @@ public sealed partial class StorageEngine : IDisposable private long _knownWalGeneration; private long _replayEpoch; private readonly object _replayLock = new(); + + // Multi-process: catch-ups in a row that could not replay the shared WAL (under _commitLock). + private int _failedCatchUps; // Collection-per-file: collectionName → IPageStorage dedicated // Null if CollectionDataDirectory not configured (embedded mode, single file) diff --git a/src/BLite.Core/Transactions/WalSharedMemory.cs b/src/BLite.Core/Transactions/WalSharedMemory.cs index 2df8662..2ea7e7c 100644 --- a/src/BLite.Core/Transactions/WalSharedMemory.cs +++ b/src/BLite.Core/Transactions/WalSharedMemory.cs @@ -208,11 +208,33 @@ private void OpenOrInitialize() if (!valid) { + long generation = 0; + long nextTransactionId = 0; + if (magic == WalSharedMemoryLayout.Magic && version == WalSharedMemoryLayout.CurrentVersion) + { + // A well-formed SHM for another page size or reader count: a configuration + // mismatch, not corruption. Re-initialising it under processes that use it + // would reset the WAL end offset and the WAL generation they rely on, so they + // would take it for a truncation and drop the commits they have not + // checkpointed. Refuse while there is any sign of such a process. + if (IsInUse(fileMaxReaders)) + throw new InvalidOperationException( + $"The shared WAL memory file '{_shmPath}' is in use with page size {filePageSize} " + + $"and {fileMaxReaders} reader slots; this process was configured with page size " + + $"{_pageSize} and {_maxReaders} reader slots. Every process must use the same configuration."); + + // Otherwise keep the counters monotonic: a new generation, so an idle process + // still attached sees a change (it holds nothing from the WAL, whose end is 0), + // and never the generation of an earlier WAL again; and no reused transaction ids. + generation = Volatile.Read(ref RefLong(WalSharedMemoryLayout.OffsetWalGeneration)) + 1; + nextTransactionId = Volatile.Read(ref RefLong(WalSharedMemoryLayout.OffsetNextTransactionId)); + } + // Treat as corruption: zero the header and re-initialize. Reader slots // (which carry only PID + offset) are also reset — any live readers will // re-register on their next BeginTransaction. ZeroFile(); - InitializeHeader(); + InitializeHeader(generation, nextTransactionId); } } } @@ -228,20 +250,44 @@ private void ZeroFile() _accessor!.Flush(); } - private void InitializeHeader() + private void InitializeHeader(long walGeneration = 0, long nextTransactionId = 0) { _accessor!.Write(WalSharedMemoryLayout.OffsetMagic, WalSharedMemoryLayout.Magic); _accessor.Write(WalSharedMemoryLayout.OffsetVersion, WalSharedMemoryLayout.CurrentVersion); _accessor.Write(WalSharedMemoryLayout.OffsetPageSize, _pageSize); _accessor.Write(WalSharedMemoryLayout.OffsetMaxReaders, _maxReaders); - _accessor.Write(WalSharedMemoryLayout.OffsetNextTransactionId, 0L); + _accessor.Write(WalSharedMemoryLayout.OffsetNextTransactionId, nextTransactionId); _accessor.Write(WalSharedMemoryLayout.OffsetWalEndOffset, 0L); _accessor.Write(WalSharedMemoryLayout.OffsetCheckpointedOffset, 0L); _accessor.Write(WalSharedMemoryLayout.OffsetWriterOwnerPid, 0); - _accessor.Write(WalSharedMemoryLayout.OffsetWalGeneration, 0L); + _accessor.Write(WalSharedMemoryLayout.OffsetWalGeneration, walGeneration); _accessor.Flush(); } + /// + /// Whether a well-formed SHM written with reader slots shows + /// signs of use: WAL records not yet truncated away, a live writer, or a live reader. A + /// process that is attached but idle, with an empty WAL, leaves no trace and is not detected. + /// + private bool IsInUse(int fileMaxReaders) + { + if (Volatile.Read(ref RefLong(WalSharedMemoryLayout.OffsetWalEndOffset)) != 0) + return true; + if (IsProcessAlive(Volatile.Read(ref RefInt(WalSharedMemoryLayout.OffsetWriterOwnerPid)))) + return true; + + // Only the slots that lie inside this process's view (the file's layout may be larger). + int slots = Math.Min(Math.Min(fileMaxReaders, MaxAllowedReaders), + (_shmFileSize - WalSharedMemoryLayout.HeaderSize) / WalSharedMemoryLayout.ReaderSlotSize); + for (int i = 0; i < slots; i++) + { + long pid = Volatile.Read(ref RefLong(WalSharedMemoryLayout.HeaderSize + i * WalSharedMemoryLayout.ReaderSlotSize)); + if (pid > 0 && pid <= int.MaxValue && IsProcessAlive((int)pid)) + return true; + } + return false; + } + // ── Atomic counters ────────────────────────────────────────────────────── /// diff --git a/src/BLite.Core/Transactions/WriteAheadLog.cs b/src/BLite.Core/Transactions/WriteAheadLog.cs index c211d03..c8c563b 100644 --- a/src/BLite.Core/Transactions/WriteAheadLog.cs +++ b/src/BLite.Core/Transactions/WriteAheadLog.cs @@ -42,6 +42,9 @@ public sealed class WriteAheadLog : IWriteAheadLog { private readonly string _walPath; private FileStream? _walStream; + + /// Path of the WAL file (diagnostics). + internal string FilePath => _walPath; private readonly SemaphoreSlim _lock = new(1, 1); private bool _disposed; private readonly int _writeTimeoutMs; diff --git a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs index 6403348..6dd7d92 100644 --- a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs +++ b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs @@ -1,4 +1,5 @@ using BLite.Core.Storage; +using BLite.Core.Transactions; namespace BLite.Tests; @@ -27,6 +28,8 @@ public void Dispose() private long WalLength => new FileInfo(Path.ChangeExtension(_dbPath, ".wal")).Length; + private string ShmPath => Path.ChangeExtension(_dbPath, ".wal") + "-shm"; + [Fact] public async Task ReplayedPage_AfterAnotherProcessTruncated_IsNotWrittenBackOnCheckpoint() { @@ -240,6 +243,163 @@ public async Task RandomInterleavings_KeepEveryCommit() } } + [Fact] + public async Task DirectCommitPath_IsRejected_SoAnotherProcessesTruncationCannotLoseIt() + { + // The sequence from the review of #201. The direct path neither catches up with the + // shared WAL nor publishes its end: A would commit, checkpoint and truncate without + // seeing B's record, and B would then discard it along with the rest of the old WAL. + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var p = a.AllocatePage(); + var q = a.AllocatePage(); + + await WriteAndCommitAsync(a, p, 0x11); + using (b.BeginTransaction()) + AssertPage(b, p, 0x11); + + long walBefore = WalLength; + using (var tx = b.BeginTransaction()) + { + b.WritePage(p, tx.TransactionId, Filled(b.PageSize, 0x44)); + await Assert.ThrowsAsync(() => b.CommitTransactionAsync(tx.TransactionId)); + Assert.Equal(walBefore, WalLength); // nothing appended outside group commit + + // The transaction is untouched and commits through group commit. + await tx.CommitAsync(); + } + + await WriteAndCommitAsync(a, q, 0x55); + await a.CheckpointAsync(); + Assert.Equal(0, WalLength); + + using (b.BeginTransaction()) + AssertPage(b, p, 0x44); + AssertFreshEngineReads(p, 0x44); + AssertFreshEngineReads(q, 0x55); + } + + [Fact] + public async Task TwoPhaseCommitPaths_AreRejected() + { + using var engine = new StorageEngine(_dbPath, Config); + var page = engine.AllocatePage(); + long walBefore = WalLength; + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(page, tx.TransactionId, Filled(engine.PageSize, 0x66)); + await Assert.ThrowsAsync(() => engine.PrepareTransactionAsync(tx.TransactionId)); + await Assert.ThrowsAsync(() => engine.PrepareTransactionAsync(tx.TransactionId, CancellationToken.None)); + await Assert.ThrowsAsync(() => engine.MarkTransactionCommittedAsync(tx.TransactionId)); + await Assert.ThrowsAsync(() => tx.PrepareAsync()); + } + + Assert.Equal(walBefore, WalLength); + } + + [Fact] + public async Task DirectAndTwoPhaseCommitPaths_StillWork_InSingleProcessMode() + { + using var engine = new StorageEngine(_dbPath, PageFileConfig.Default); + var direct = engine.AllocatePage(); + var prepared = engine.AllocatePage(); + + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(direct, tx.TransactionId, Filled(engine.PageSize, 0x77)); + await engine.CommitTransactionAsync(tx.TransactionId); + } + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(prepared, tx.TransactionId, Filled(engine.PageSize, 0x78)); + Assert.True(await engine.PrepareTransactionAsync(tx.TransactionId)); + await engine.MarkTransactionCommittedAsync(tx.TransactionId); + } + + AssertPage(engine, direct, 0x77); + AssertPage(engine, prepared, 0x78); + } + + [Fact] + public async Task CheckpointThatCannotTakeTheWriterLock_IsReportedAsFailed() + { + using var engine = new StorageEngine(_dbPath, Config); + var metrics = engine.EnsureMetrics(); + var page = engine.AllocatePage(); + await WriteAndCommitAsync(engine, page, 0x11); + + // Another holder of the cross-process writer lock (an instance in the same process + // excludes it like another process would). + using (var other = WalSharedMemory.Open(ShmPath, Config.PageSize)) + { + Assert.True(other.TryAcquireWriterLock(1_000)); + await engine.CheckpointAsync(); + other.ReleaseWriterLock(); + } + Assert.NotEqual(0, WalLength); // nothing was written or truncated + + await WaitUntilAsync(() => metrics.GetSnapshot().CheckpointsFailedTotal == 1); + Assert.Equal(0, metrics.GetSnapshot().CheckpointsTotal); + + await engine.CheckpointAsync(); + Assert.Equal(0, WalLength); + await WaitUntilAsync(() => metrics.GetSnapshot().CheckpointsTotal == 1); + Assert.Equal(1, metrics.GetSnapshot().CheckpointsFailedTotal); + } + + [Fact] + public async Task ShmOpenedWithAnotherPageSize_WhileInUse_ThrowsAndKeepsTheHeader() + { + using var a = new StorageEngine(_dbPath, Config); + var page = a.AllocatePage(); + await WriteAndCommitAsync(a, page, 0x11); + await a.CheckpointAsync(); // generation 1 + await WriteAndCommitAsync(a, page, 0x22); // the WAL has records again + + Assert.Throws(() => WalSharedMemory.Open(ShmPath, Config.PageSize * 2)); + + using (var shm = WalSharedMemory.Open(ShmPath, Config.PageSize)) + { + Assert.Equal(1, shm.ReadWalGeneration()); + Assert.Equal(WalLength, shm.ReadWalEndOffset()); + } + using (a.BeginTransaction()) + AssertPage(a, page, 0x22); + AssertFreshEngineReads(page, 0x22); + } + + [Fact] + public async Task ShmOpenedWithAnotherPageSize_WhenIdle_IsReinitialisedWithoutGoingBack() + { + // E.g. a .wal-shm left behind by a database recreated with another page size. + ulong nextTransactionId; + using (var a = new StorageEngine(_dbPath, Config)) + { + var page = a.AllocatePage(); + await WriteAndCommitAsync(a, page, 0x11); + await a.CheckpointAsync(); // generation 1, WAL end 0 + using var shm = WalSharedMemory.Open(ShmPath, Config.PageSize); + Assert.Equal(1, shm.ReadWalGeneration()); + nextTransactionId = shm.ReadNextTransactionId(); + } + + using var reopened = WalSharedMemory.Open(ShmPath, Config.PageSize * 2); + // A generation never seen before: a process still attached to the old one sees a change. + Assert.Equal(2, reopened.ReadWalGeneration()); + Assert.Equal(nextTransactionId, reopened.ReadNextTransactionId()); + } + + private static async Task WaitUntilAsync(Func condition) + { + var deadline = DateTime.UtcNow.AddSeconds(10); + while (!condition()) + { + Assert.True(DateTime.UtcNow < deadline, "Condition not met within 10 s."); + await Task.Delay(20); + } + } + private void AssertFreshEngineReads(uint page, byte fill) { using var c = new StorageEngine(_dbPath, Config);