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/7] 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 56e16518e147f267b8492e9aacded8aa5a190e3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 12:43:28 +0200 Subject: [PATCH 2/7] fix(vacuum): rebuild typed-collection indexes into fresh trees (#167) VacuumAsync(RebuildIndexes = true) re-inserted every document into the existing index without clearing it, so each rebuild duplicated every B-Tree entry and kept the stale spatial and vector entries from #164. Each index is now rebuilt into an empty copy that queries don't see. Only after the rebuild committed is it swapped in, its root persisted, and the old tree freed. A document that cannot be loaded or indexed fails the rebuild and leaves the existing index untouched, instead of being silently skipped. Co-Authored-By: Claude Opus 5.5 --- .../Collections/DocumentCollection.cs | 24 +++- .../Indexing/CollectionIndexManager.cs | 40 +++++- tests/BLite.Tests/VacuumTests.cs | 120 +++++++++++++++++- 3 files changed, 176 insertions(+), 8 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 19d77c2..34ebe1e 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -720,7 +720,7 @@ public async Task VacuumAsync(VacuumOptions? options = null, CancellationToken c if (options?.RebuildIndexes == true) { foreach (var index in _indexManager.GetAllIndexes()) - await RebuildIndexAsync(index, ct).ConfigureAwait(false); + await RebuildIndexFromScratchAsync(index.Name, ct).ConfigureAwait(false); } success = true; @@ -1594,6 +1594,28 @@ public async IAsyncEnumerable QueryIndexAsync( } } + /// + /// Populates an empty copy of index from the documents and swaps it in + /// only after the rebuild committed; the old tree is then freed. If any document cannot be + /// loaded or indexed, the rebuild fails and the existing index stays in place untouched, + /// instead of being replaced by an incomplete one. + /// + private async Task RebuildIndexFromScratchAsync(string name, CancellationToken ct) + { + var fresh = _indexManager.CreateUnregisteredCopy(name); + try + { + await RebuildIndexAsync(fresh, ct, skipUnreadableDocuments: false).ConfigureAwait(false); + } + catch + { + fresh.FreeAllPages(_storage); + fresh.Dispose(); + throw; + } + _indexManager.ReplaceIndex(name, fresh); + } + private async Task RebuildIndexAsync(CollectionSecondaryIndex index, CancellationToken ct = default, bool skipUnreadableDocuments = true) { var transaction = _storage.BeginTransaction(IsolationLevel.ReadCommitted); diff --git a/src/BLite.Core/Indexing/CollectionIndexManager.cs b/src/BLite.Core/Indexing/CollectionIndexManager.cs index d465365..bebaa8d 100644 --- a/src/BLite.Core/Indexing/CollectionIndexManager.cs +++ b/src/BLite.Core/Indexing/CollectionIndexManager.cs @@ -301,10 +301,41 @@ internal CollectionSecondaryIndex RecreateIndex(string name) if (!_indexes.TryGetValue(name, out var old)) throw new InvalidOperationException($"Index '{name}' does not exist"); - var fresh = new CollectionSecondaryIndex(old.Definition, _storage, _mapper) - { - KeyFormat = old.KeyFormat - }; + var fresh = CreateUnregisteredCopy(name); + fresh.KeyFormat = old.KeyFormat; + ReplaceIndex(name, fresh); + return fresh; + } + } + + /// + /// Creates an empty index with the same definition as without + /// registering it, so queries keep using the existing index while the copy is populated. + /// The copy owns its root page: hand it to or release it with + /// . + /// + internal CollectionSecondaryIndex CreateUnregisteredCopy(string name) + { + lock (_lock) + { + if (!_indexes.TryGetValue(name, out var old)) + throw new InvalidOperationException($"Index '{name}' does not exist"); + + return new CollectionSecondaryIndex(old.Definition, _storage, _mapper); + } + } + + /// + /// Registers under , persists its root page + /// and frees the pages of the index it replaces. + /// + internal void ReplaceIndex(string name, CollectionSecondaryIndex fresh) + { + lock (_lock) + { + if (!_indexes.TryGetValue(name, out var old)) + throw new InvalidOperationException($"Index '{name}' does not exist"); + _indexes[name] = fresh; _cachedIndexInfo = null; // invalidate cache @@ -313,7 +344,6 @@ internal CollectionSecondaryIndex RecreateIndex(string name) SaveMetadata(); old.FreeAllPages(_storage); old.Dispose(); - return fresh; } } diff --git a/tests/BLite.Tests/VacuumTests.cs b/tests/BLite.Tests/VacuumTests.cs index ed5303d..9debfdb 100644 --- a/tests/BLite.Tests/VacuumTests.cs +++ b/tests/BLite.Tests/VacuumTests.cs @@ -345,7 +345,7 @@ public async Task SecureErase_Immediate_DocumentCollection_FreeSpaceIsZero_After await db.Users.DeleteAsync(user.Id); await db.SaveChangesAsync(); - db.ForceCheckpoint(); + await db.CheckpointAsync(); AssertAllDataPageFreeSpaceIsZero(db.Storage, "users"); } @@ -399,7 +399,7 @@ public async Task Vacuum_DocumentCollection_FreeBytesAreZero_AfterVacuum() await db.SaveChangesAsync(); await db.Users.VacuumAsync(); - db.ForceCheckpoint(); + await db.CheckpointAsync(); AssertAllDataPageFreeSpaceIsZero(db.Storage, "users"); } @@ -586,6 +586,122 @@ public async Task Vacuum_Idempotent_DocumentsStillCorrectAfterTwoPasses() Assert.Equal(10, await col.CountAsync()); } + // ====================================================================== + // RebuildIndexes on DocumentCollection (#167) + // ====================================================================== + + /// + /// A rebuild must replace the B-Tree contents, not re-insert every document into the + /// existing tree: each document is found exactly once by equality and range queries. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_BTree_NoDuplicates() + { + using var db = new TestDbContext(_dbPath); + for (int i = 1; i <= 3; i++) + await db.PeopleV2.InsertAsync(new PersonV2 { Name = $"p{i}", Age = 30 + i }); + + await db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true }); + + Assert.Single(db.PeopleV2.AsQueryable().Where(p => p.Age == 32).ToList()); + Assert.Equal(3, db.PeopleV2.AsQueryable().Where(p => p.Age >= 31).ToList().Count); + } + + /// + /// Back-to-back rebuilds stay idempotent even though each one frees the previous tree while + /// the WAL still holds its pages (#168). Once the WAL is checkpointed, those pages are reused, + /// so further rebuilds do not grow the file. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_Repeated_NoDuplicatesNoGrowth() + { + 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 }); + + var options = new VacuumOptions { RebuildIndexes = true, TruncateFile = false }; + for (int pass = 0; pass < 5; pass++) + await db.PeopleV2.VacuumAsync(options); + + Assert.Equal(4, db.PeopleV2.AsQueryable().Where(p => p.Age == 7).ToList().Count); + Assert.Equal(200, db.PeopleV2.AsQueryable().Where(p => p.Age >= 0).ToList().Count); + + await db.CheckpointAsync(); + long sizeBefore = new FileInfo(_dbPath).Length; + for (int pass = 0; pass < 5; pass++) + { + await db.PeopleV2.VacuumAsync(options); + await db.CheckpointAsync(); + } + + Assert.Equal(200, db.PeopleV2.AsQueryable().Where(p => p.Age >= 0).ToList().Count); + Assert.Equal(sizeBefore, new FileInfo(_dbPath).Length); + } + + /// + /// The rebuilt tree's root page is persisted, so the index is still correct after reopening. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_SurvivesReopen() + { + using (var db = new TestDbContext(_dbPath)) + { + for (int i = 1; i <= 10; i++) + await db.PeopleV2.InsertAsync(new PersonV2 { Name = $"p{i}", Age = i }); + await db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true }); + } + + using (var db = new TestDbContext(_dbPath)) + { + Assert.Single(db.PeopleV2.AsQueryable().Where(p => p.Age == 5).ToList()); + Assert.Equal(10, db.PeopleV2.AsQueryable().Where(p => p.Age >= 1).ToList().Count); + + await db.PeopleV2.InsertAsync(new PersonV2 { Name = "late", Age = 5 }); + Assert.Equal(2, db.PeopleV2.AsQueryable().Where(p => p.Age == 5).ToList().Count); + } + } + + /// + /// A rebuilt spatial index holds only current positions, so a moved document is no longer + /// found where it used to be. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_Spatial_DropsStaleEntries() + { + using var db = new TestDbContext(_dbPath); + var e = new GeoEntity { Name = "mover", Location = (45.0, 9.0) }; + await db.GeoItems.InsertAsync(e); + e.Location = (10.0, 10.0); + await db.GeoItems.UpdateAsync(e); + + await db.GeoItems.VacuumAsync(new VacuumOptions { RebuildIndexes = true }); + + (double, double) oldMin = (44.0, 8.0), oldMax = (46.0, 10.0); + (double, double) newMin = (9.0, 9.0), newMax = (11.0, 11.0); + Assert.Empty(db.GeoItems.AsQueryable().Where(p => p.Location.Within(oldMin, oldMax)).ToList()); + Assert.Single(db.GeoItems.AsQueryable().Where(p => p.Location.Within(newMin, newMax)).ToList()); + } + + /// + /// A rebuilt vector index holds only current embeddings, so nearest-neighbour search sees + /// the updated vector. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_Vector_UsesCurrentEmbeddings() + { + using var db = new TestDbContext(_dbPath); + var a = new VectorEntity { Title = "A", Embedding = [0f, 0f, 0f] }; + await db.VectorItems.InsertAsync(a); + await db.VectorItems.InsertAsync(new VectorEntity { Title = "B", Embedding = [5f, 5f, 5f] }); + a.Embedding = [100f, 100f, 100f]; + await db.VectorItems.UpdateAsync(a); + + await db.VectorItems.VacuumAsync(new VacuumOptions { RebuildIndexes = true }); + + var top = await db.VectorItems.VectorSearchAsync("idx_vector", [0f, 0f, 0f], 1).ToListAsync(); + Assert.Equal("B", Assert.Single(top).Title); + } + // ====================================================================== // Helpers // ====================================================================== 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 3/7] 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 50ef14326f32d91d232d6bfa878a2c50c99e917a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 15:33:21 +0200 Subject: [PATCH 4/7] fix(vacuum): address review of index rebuild (#167) - A failed rebuild now frees the whole half-built copy: the rebuild runs its own transaction and collects the copy's pages through it before the rollback, so pages added by splits (allocated outside the transaction, linked only from nodes written in it) are found. Freeing never masks the original exception. - Spatial indexes are skipped by the rebuild: RTreeIndex cannot enumerate its pages, so replacing the tree would leak the old one on every vacuum. - CreateUnregisteredCopy and ReplaceIndex check for disposal. - VacuumOptions.RebuildIndexes documents that earlier index handles refer to the replaced tree, that a query running during the swap may read pages being freed, and that spatial indexes are not rebuilt. BTreeIndex and VectorSearchIndex gain an internal CollectAllPages overload that walks the tree as a transaction sees it. The spatial vacuum test now checks that the R-Tree is kept; a new test covers the failure path (one page freed before, the whole copy now). The previous commit message also switched two tests from ForceCheckpoint() to CheckpointAsync(). Co-Authored-By: Claude Opus 5.5 --- .../Collections/DocumentCollection.cs | 66 +++++++++++++------ src/BLite.Core/Indexing/BTreeIndex.cs | 18 +++-- .../Indexing/CollectionIndexManager.cs | 4 ++ .../Indexing/CollectionSecondaryIndex.cs | 20 ++++-- src/BLite.Core/Indexing/VectorSearchIndex.cs | 10 ++- src/BLite.Core/VacuumOptions.cs | 8 +++ tests/BLite.Tests/VacuumTests.cs | 61 +++++++++++++---- 7 files changed, 141 insertions(+), 46 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 34ebe1e..0ce3dcd 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -720,7 +720,12 @@ public async Task VacuumAsync(VacuumOptions? options = null, CancellationToken c if (options?.RebuildIndexes == true) { foreach (var index in _indexManager.GetAllIndexes()) + { + // RTreeIndex cannot enumerate its pages yet, so a rebuild would leak the + // whole old tree on every vacuum. Spatial indexes are kept as they are. + if (index.Definition.Type == IndexType.Spatial) continue; await RebuildIndexFromScratchAsync(index.Name, ct).ConfigureAwait(false); + } } success = true; @@ -1603,13 +1608,29 @@ public async IAsyncEnumerable QueryIndexAsync( private async Task RebuildIndexFromScratchAsync(string name, CancellationToken ct) { var fresh = _indexManager.CreateUnregisteredCopy(name); + var transaction = _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - await RebuildIndexAsync(fresh, ct, skipUnreadableDocuments: false).ConfigureAwait(false); + await FillIndexAsync(fresh, transaction, skipUnreadableDocuments: false, ct).ConfigureAwait(false); + await transaction.CommitAsync(ct); } catch { - fresh.FreeAllPages(_storage); + // Collect the copy's pages through the transaction before rolling it back: pages added + // by splits are allocated outside the transaction but linked only from nodes written + // in it, so after the rollback they could not be found any more. + IReadOnlyList pages = Array.Empty(); + try { pages = fresh.CollectAllPages(transaction.TransactionId); } + catch { /* best-effort: never mask the original failure */ } + + await transaction.RollbackAsync(); + + try + { + foreach (var pageId in pages) + _storage.FreePage(pageId); + } + catch { /* best-effort: never mask the original failure */ } fresh.Dispose(); throw; } @@ -1621,24 +1642,7 @@ private async Task RebuildIndexAsync(CollectionSecondaryIndex index, Can var transaction = _storage.BeginTransaction(IsolationLevel.ReadCommitted); try { - // Iterate all documents in the collection via primary index - foreach (var entry in _primaryIndex.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, transaction.TransactionId)) - { - ct.ThrowIfCancellationRequested(); - try - { - var document = await FindByLocationAsync(entry.Location, transaction.TransactionId, ct); - if (document != null) - { - index.Insert(document, entry.Location, transaction); - } - } - catch when (skipUnreadableDocuments) - { - // Skip documents that fail to load or index - // Production: should log errors - } - } + await FillIndexAsync(index, transaction, skipUnreadableDocuments, ct).ConfigureAwait(false); await transaction.CommitAsync(ct); } catch @@ -1648,6 +1652,28 @@ private async Task RebuildIndexAsync(CollectionSecondaryIndex index, Can } } + private async Task FillIndexAsync(CollectionSecondaryIndex index, ITransaction transaction, bool skipUnreadableDocuments, CancellationToken ct) + { + // Iterate all documents in the collection via primary index + foreach (var entry in _primaryIndex.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, transaction.TransactionId)) + { + ct.ThrowIfCancellationRequested(); + try + { + var document = await FindByLocationAsync(entry.Location, transaction.TransactionId, ct); + if (document != null) + { + index.Insert(document, entry.Location, transaction); + } + } + catch when (skipUnreadableDocuments) + { + // Skip documents that fail to load or index + // Production: should log errors + } + } + } + #endregion #region Data Page Management diff --git a/src/BLite.Core/Indexing/BTreeIndex.cs b/src/BLite.Core/Indexing/BTreeIndex.cs index dedf5aa..db20278 100644 --- a/src/BLite.Core/Indexing/BTreeIndex.cs +++ b/src/BLite.Core/Indexing/BTreeIndex.cs @@ -1967,30 +1967,36 @@ public async IAsyncEnumerable LikeAsync( /// Returns all physical page IDs owned by this B+Tree. /// Used when dropping an index to free storage space. /// - public IReadOnlyList CollectAllPages() + public IReadOnlyList CollectAllPages() => CollectAllPages(0); + + /// + /// Like , but walks the tree as + /// sees it, including nodes that transaction has written but not committed. + /// + internal IReadOnlyList CollectAllPages(ulong transactionId) { var result = new List(); var root = RootPageId; if (root != 0) - CollectPagesRecursive(root, result); + CollectPagesRecursive(root, result, transactionId); return result; } - private void CollectPagesRecursive(uint pageId, List result) + private void CollectPagesRecursive(uint pageId, List result, ulong transactionId) { if (pageId == 0) return; result.Add(pageId); var pageBuffer = System.Buffers.ArrayPool.Shared.Rent(_storage.PageSize); try { - ReadPage(pageId, 0, pageBuffer); + ReadPage(pageId, transactionId, pageBuffer); var header = BTreeNodeHeader.ReadFrom(pageBuffer.AsSpan(32)); if (!header.IsLeaf) { var (p0, entries) = ReadInternalEntries(pageBuffer, header.EntryCount); - CollectPagesRecursive(p0, result); + CollectPagesRecursive(p0, result, transactionId); foreach (var entry in entries) - CollectPagesRecursive(entry.PageId, result); + CollectPagesRecursive(entry.PageId, result, transactionId); } } finally diff --git a/src/BLite.Core/Indexing/CollectionIndexManager.cs b/src/BLite.Core/Indexing/CollectionIndexManager.cs index bebaa8d..53d00dc 100644 --- a/src/BLite.Core/Indexing/CollectionIndexManager.cs +++ b/src/BLite.Core/Indexing/CollectionIndexManager.cs @@ -318,6 +318,8 @@ internal CollectionSecondaryIndex CreateUnregisteredCopy(string name) { lock (_lock) { + if (_disposed) + throw new ObjectDisposedException(nameof(CollectionIndexManager)); if (!_indexes.TryGetValue(name, out var old)) throw new InvalidOperationException($"Index '{name}' does not exist"); @@ -333,6 +335,8 @@ internal void ReplaceIndex(string name, CollectionSecondaryIndex fresh) { lock (_lock) { + if (_disposed) + throw new ObjectDisposedException(nameof(CollectionIndexManager)); if (!_indexes.TryGetValue(name, out var old)) throw new InvalidOperationException($"Index '{name}' does not exist"); diff --git a/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs b/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs index 0afe4b9..d9de013 100644 --- a/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs +++ b/src/BLite.Core/Indexing/CollectionSecondaryIndex.cs @@ -638,12 +638,18 @@ public void Dispose() /// internal void FreeAllPages(StorageEngine storage) { - if (_btreeIndex != null) - foreach (var pageId in _btreeIndex.CollectAllPages()) - storage.FreePage(pageId); - else if (_vectorIndex != null) - foreach (var pageId in _vectorIndex.CollectAllPages()) - storage.FreePage(pageId); - // RTreeIndex page freeing: not yet implemented (no CollectAllPages on RTree). + foreach (var pageId in CollectAllPages(0)) + storage.FreePage(pageId); + } + + /// + /// The pages of this index as sees them (0: committed state). + /// Empty for a spatial index: RTreeIndex cannot enumerate its pages yet. + /// + internal IReadOnlyList CollectAllPages(ulong transactionId) + { + if (_btreeIndex != null) return _btreeIndex.CollectAllPages(transactionId); + if (_vectorIndex != null) return _vectorIndex.CollectAllPages(transactionId); + return Array.Empty(); } } diff --git a/src/BLite.Core/Indexing/VectorSearchIndex.cs b/src/BLite.Core/Indexing/VectorSearchIndex.cs index 87673b0..33ae152 100644 --- a/src/BLite.Core/Indexing/VectorSearchIndex.cs +++ b/src/BLite.Core/Indexing/VectorSearchIndex.cs @@ -566,7 +566,13 @@ private uint CreateNewPage(ITransaction? transaction) /// Returns all physical page IDs owned by this vector index. /// Used when dropping an index to free storage. /// - public IReadOnlyList CollectAllPages() + public IReadOnlyList CollectAllPages() => CollectAllPages(0); + + /// + /// Like , but follows the page chain as + /// sees it, including pages it has linked but not committed. + /// + internal IReadOnlyList CollectAllPages(ulong transactionId) { // Multi-page vector graphs store all pages starting from _rootPageId; // follow NextPageId chain to collect the full set. @@ -578,7 +584,7 @@ public IReadOnlyList CollectAllPages() while (current != 0) { pages.Add(current); - _storage.ReadPage(current, null, buffer); + _storage.ReadPage(current, transactionId == 0 ? null : transactionId, buffer); var header = PageHeader.ReadFrom(buffer); current = header.NextPageId; } diff --git a/src/BLite.Core/VacuumOptions.cs b/src/BLite.Core/VacuumOptions.cs index 8979c1d..57ee779 100644 --- a/src/BLite.Core/VacuumOptions.cs +++ b/src/BLite.Core/VacuumOptions.cs @@ -35,6 +35,14 @@ public sealed class VacuumOptions /// that accumulates over time as documents are inserted, updated, and deleted. /// Defaults to false because index rebuilds are expensive for large /// collections; set to true only when index fragmentation is a concern. + /// + /// On a typed collection each index is rebuilt into a new tree that replaces the old one once + /// the rebuild has committed; if a document cannot be loaded or indexed, the vacuum throws and + /// the existing index is kept. Index handles obtained earlier (for example from + /// GetIndexAsync or CreateIndexAsync) refer to the replaced tree afterwards and + /// must be fetched again. A query already running on the old tree while it is replaced may read + /// pages that are being freed. Spatial indexes are not rebuilt. + /// /// public bool RebuildIndexes { get; init; } = false; diff --git a/tests/BLite.Tests/VacuumTests.cs b/tests/BLite.Tests/VacuumTests.cs index 9debfdb..0bbf5c0 100644 --- a/tests/BLite.Tests/VacuumTests.cs +++ b/tests/BLite.Tests/VacuumTests.cs @@ -1,6 +1,7 @@ using BLite.Bson; using BLite.Core; using BLite.Core.Collections; +using BLite.Core.Indexing; using BLite.Core.Storage; using BLite.Shared; @@ -662,24 +663,62 @@ public async Task Vacuum_RebuildIndexes_DocumentCollection_SurvivesReopen() } /// - /// A rebuilt spatial index holds only current positions, so a moved document is no longer - /// found where it used to be. + /// Spatial indexes are not rebuilt (their pages cannot be enumerated and freed yet); the + /// existing R-Tree is kept and keeps answering queries. /// [Fact] - public async Task Vacuum_RebuildIndexes_DocumentCollection_Spatial_DropsStaleEntries() + public async Task Vacuum_RebuildIndexes_DocumentCollection_Spatial_IsKept() { using var db = new TestDbContext(_dbPath); - var e = new GeoEntity { Name = "mover", Location = (45.0, 9.0) }; - await db.GeoItems.InsertAsync(e); - e.Location = (10.0, 10.0); - await db.GeoItems.UpdateAsync(e); + await db.GeoItems.InsertAsync(new GeoEntity { Name = "milan", Location = (45.46, 9.19) }); + uint SpatialRoot() => db.Storage.GetAllCollectionsMetadata() + .SelectMany(m => m.Indexes).Single(i => i.Name == "idx_spatial").RootPageId; + var rootBefore = SpatialRoot(); await db.GeoItems.VacuumAsync(new VacuumOptions { RebuildIndexes = true }); - (double, double) oldMin = (44.0, 8.0), oldMax = (46.0, 10.0); - (double, double) newMin = (9.0, 9.0), newMax = (11.0, 11.0); - Assert.Empty(db.GeoItems.AsQueryable().Where(p => p.Location.Within(oldMin, oldMax)).ToList()); - Assert.Single(db.GeoItems.AsQueryable().Where(p => p.Location.Within(newMin, newMax)).ToList()); + Assert.Equal(rootBefore, SpatialRoot()); + (double, double) min = (45.0, 9.0), max = (46.0, 10.0); + Assert.Equal("milan", Assert.Single(db.GeoItems.AsQueryable().Where(p => p.Location.Within(min, max)).ToList()).Name); + } + + /// + /// A rebuild that fails part-way keeps the old index and returns every page of the half-built + /// copy, including the pages its splits added, to the free list. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIndexAndFreesCopy() + { + using var db = new TestDbContext(_dbPath); + var people = Enumerable.Range(0, 3000).Select(i => new PersonV2 { Name = $"p{i}", Age = i % 90 }).ToList(); + await db.PeopleV2.InsertBulkAsync(people); + await db.CheckpointAsync(); + + // Corrupt the last document in primary-key order, so the rebuild fails after most inserts. + var storage = db.Storage; + var last = people.Select(p => p.Id).Max(); + var primary = (BTreeIndex)typeof(DocumentCollection) + .GetField("_primaryIndex", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! + .GetValue(db.PeopleV2)!; + Assert.True(primary.TryFind(new IndexKey(last), out var location)); + var page = new byte[storage.PageSize]; + storage.ReadPage(location.PageId, null, page); + var slot = SlotEntry.ReadFrom(page.AsSpan(SlottedPageHeader.Size + location.SlotIndex * SlotEntry.Size)); + page.AsSpan(slot.Offset + 4, 12).Fill(0xFF); // invalid element types + storage.WritePageImmediate(location.PageId, page); + + await Assert.ThrowsAnyAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); + + // The old index still answers queries. + Assert.Equal(people.Count(p => p.Age == 10), (await db.PeopleV2.QueryIndexAsync("idx_Age", 10, 10).ToListAsync()).Count); + + // Pages handed out next come from the free list until it is empty; the half-built copy of + // a 3000-entry index spans dozens of pages, not just its root. + await db.CheckpointAsync(); + var highWater = (uint)storage.GetCollectionPageIds("peoplev2_collection").Count(); + int reused = 0; + while (storage.AllocatePage() < highWater) reused++; + Assert.True(reused > 20, $"only {reused} pages were returned to the free list"); } /// From 3dd35fb1d0686e2113508da325b8d53623c16b7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 15:53:19 +0200 Subject: [PATCH 5/7] test(vacuum): make the failed-rebuild leak test deterministic (#167) The test corrupted the document with the largest ObjectId, which is not always the last entry in primary-key byte order, so the rebuild sometimes failed early with a small copy. It also read the free-list high-water mark from GetCollectionPageIds (which leaves out queued frees) and could query the corrupted document's age. It now corrupts the last primary-index entry, reads NextPageId from the page file, queries a different age, and counts pages still queued for the next WAL truncation as freed. Co-Authored-By: Claude Opus 5.5 --- tests/BLite.Tests/VacuumTests.cs | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/tests/BLite.Tests/VacuumTests.cs b/tests/BLite.Tests/VacuumTests.cs index 0bbf5c0..60174cd 100644 --- a/tests/BLite.Tests/VacuumTests.cs +++ b/tests/BLite.Tests/VacuumTests.cs @@ -696,11 +696,13 @@ public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIn // Corrupt the last document in primary-key order, so the rebuild fails after most inserts. var storage = db.Storage; - var last = people.Select(p => p.Id).Max(); var primary = (BTreeIndex)typeof(DocumentCollection) .GetField("_primaryIndex", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! .GetValue(db.PeopleV2)!; - Assert.True(primary.TryFind(new IndexKey(last), out var location)); + var lastEntry = primary.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Backward).First(); + var location = lastEntry.Location; + var corruptedAge = (await db.PeopleV2.FindByIdAsync(lastEntry.Key.As()))!.Age; + var queryAge = (corruptedAge + 1) % 90; var page = new byte[storage.PageSize]; storage.ReadPage(location.PageId, null, page); var slot = SlotEntry.ReadFrom(page.AsSpan(SlottedPageHeader.Size + location.SlotIndex * SlotEntry.Size)); @@ -710,15 +712,19 @@ public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIn await Assert.ThrowsAnyAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); // The old index still answers queries. - Assert.Equal(people.Count(p => p.Age == 10), (await db.PeopleV2.QueryIndexAsync("idx_Age", 10, 10).ToListAsync()).Count); + Assert.Equal(people.Count(p => p.Age == queryAge), (await db.PeopleV2.QueryIndexAsync("idx_Age", queryAge, queryAge).ToListAsync()).Count); - // Pages handed out next come from the free list until it is empty; the half-built copy of - // a 3000-entry index spans dozens of pages, not just its root. + // The half-built copy of a 3000-entry index spans dozens of pages, not just its root. Each + // is either back on the free list (handed out next, below the high-water mark) or still + // queued for the next WAL truncation (a recycled page with a WAL image; a background + // checkpoint can make the explicit one below return without truncating). await db.CheckpointAsync(); - var highWater = (uint)storage.GetCollectionPageIds("peoplev2_collection").Count(); + var flags = System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance; + int queued = ((HashSet)typeof(StorageEngine).GetField("_pendingFreePages", flags)!.GetValue(storage)!).Count; + var highWater = ((IPageStorage)typeof(StorageEngine).GetField("_pageFile", flags)!.GetValue(storage)!).NextPageId; int reused = 0; while (storage.AllocatePage() < highWater) reused++; - Assert.True(reused > 20, $"only {reused} pages were returned to the free list"); + Assert.True(reused + queued > 20, $"only {reused} pages were returned to the free list and {queued} queued"); } /// From 6c4df5d6a29af28df0620b5957d79cd5139bc26e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Mon, 12 Oct 2026 17:06:11 +0200 Subject: [PATCH 6/7] fix(vacuum): keep the original error and free the copy when a rebuild cannot finish (#167) - RebuildIndexFromScratchAsync: a failing RollbackAsync no longer replaces the rebuild's exception or skips freeing and disposing the copy. - If ReplaceIndex fails after the copy committed, the copy's pages are freed unless the swap already went through. - ReplaceIndex restores the old index when the metadata cannot be saved, so memory and disk keep agreeing and the caller still owns the copy. - Tests for both paths (abort record write fails; metadata save times out). Co-Authored-By: Claude Opus 5.5 --- .../Collections/DocumentCollection.cs | 36 ++++- .../Indexing/CollectionIndexManager.cs | 16 ++- tests/BLite.Tests/VacuumTests.cs | 128 ++++++++++++++++-- 3 files changed, 160 insertions(+), 20 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 0ce3dcd..238994f 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -1623,18 +1623,40 @@ private async Task RebuildIndexFromScratchAsync(string name, CancellationToken c try { pages = fresh.CollectAllPages(transaction.TransactionId); } catch { /* best-effort: never mask the original failure */ } - await transaction.RollbackAsync(); + try { await transaction.RollbackAsync(); } + catch { /* best-effort: never mask the original failure */ } - try + FreeUnregisteredCopy(fresh, pages); + throw; + } + + try + { + _indexManager.ReplaceIndex(name, fresh); + } + catch + { + // The copy is committed but, unless the swap itself went through, nothing refers to it. + if (!ReferenceEquals(_indexManager.GetIndex(name), fresh)) { - foreach (var pageId in pages) - _storage.FreePage(pageId); + IReadOnlyList pages = Array.Empty(); + try { pages = fresh.CollectAllPages(0); } + catch { /* best-effort: never mask the original failure */ } + FreeUnregisteredCopy(fresh, pages); } - catch { /* best-effort: never mask the original failure */ } - fresh.Dispose(); throw; } - _indexManager.ReplaceIndex(name, fresh); + } + + private void FreeUnregisteredCopy(CollectionSecondaryIndex copy, IReadOnlyList pages) + { + try + { + foreach (var pageId in pages) + _storage.FreePage(pageId); + } + catch { /* best-effort: never mask the original failure */ } + copy.Dispose(); } private async Task RebuildIndexAsync(CollectionSecondaryIndex index, CancellationToken ct = default, bool skipUnreadableDocuments = true) diff --git a/src/BLite.Core/Indexing/CollectionIndexManager.cs b/src/BLite.Core/Indexing/CollectionIndexManager.cs index 53d00dc..1077ea4 100644 --- a/src/BLite.Core/Indexing/CollectionIndexManager.cs +++ b/src/BLite.Core/Indexing/CollectionIndexManager.cs @@ -329,7 +329,8 @@ internal CollectionSecondaryIndex CreateUnregisteredCopy(string name) /// /// Registers under , persists its root page - /// and frees the pages of the index it replaces. + /// and frees the pages of the index it replaces. If the metadata cannot be saved, the old + /// index stays registered and remains the caller's to release. /// internal void ReplaceIndex(string name, CollectionSecondaryIndex fresh) { @@ -345,7 +346,18 @@ internal void ReplaceIndex(string name, CollectionSecondaryIndex fresh) // Persist the new root before freeing the old tree: a crash in between leaks the old // pages, whereas the reverse order would let the next open free them a second time. - SaveMetadata(); + try + { + SaveMetadata(); + } + catch + { + // Keep the old index registered, matching the metadata on disk; the caller still + // owns the copy. + _indexes[name] = old; + UpdateMetadata(); + throw; + } old.FreeAllPages(_storage); old.Dispose(); } diff --git a/tests/BLite.Tests/VacuumTests.cs b/tests/BLite.Tests/VacuumTests.cs index 60174cd..8d9ed28 100644 --- a/tests/BLite.Tests/VacuumTests.cs +++ b/tests/BLite.Tests/VacuumTests.cs @@ -3,6 +3,7 @@ using BLite.Core.Collections; using BLite.Core.Indexing; using BLite.Core.Storage; +using BLite.Core.Transactions; using BLite.Shared; namespace BLite.Tests; @@ -690,11 +691,99 @@ uint SpatialRoot() => db.Storage.GetAllCollectionsMetadata() public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIndexAndFreesCopy() { using var db = new TestDbContext(_dbPath); + var people = await InsertPeopleAsync(db); + var queryAge = await CorruptLastDocumentAsync(db); + + await Assert.ThrowsAnyAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); + + // The old index still answers queries. + Assert.Equal(people.Count(p => p.Age == queryAge), (await db.PeopleV2.QueryIndexAsync("idx_Age", queryAge, queryAge).ToListAsync()).Count); + await AssertCopyWasFreedAsync(db); + } + + /// + /// If the abort record of a failed rebuild cannot be written either, the rebuild's own failure + /// is what surfaces, and the half-built copy is still freed. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRollback_KeepsOriginalErrorAndFreesCopy() + { + using var db = new TestDbContext(_dbPath); + var people = await InsertPeopleAsync(db); + var queryAge = await CorruptLastDocumentAsync(db); + + var walField = typeof(StorageEngine).GetField("_wal", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!; + var wal = (IWriteAheadLog)walField.GetValue(db.Storage)!; + walField.SetValue(db.Storage, new AbortRecordFailingWal(wal)); + Exception ex; + try + { + ex = await Assert.ThrowsAnyAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); + } + finally + { + walField.SetValue(db.Storage, wal); + } + + Assert.IsNotType(ex); + Assert.Equal(people.Count(p => p.Age == queryAge), (await db.PeopleV2.QueryIndexAsync("idx_Age", queryAge, queryAge).ToListAsync()).Count); + await AssertCopyWasFreedAsync(db); + } + + /// + /// If the rebuilt copy committed but cannot be swapped in (here: its metadata cannot be saved), + /// the old index stays in place, in memory and on disk, and the copy's pages are freed. + /// + [Fact] + public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedSwap_KeepsIndexAndFreesCopy() + { + int expected; + using (var db = new TestDbContext(_dbPath)) + { + var people = await InsertPeopleAsync(db); + expected = people.Count(p => p.Age == 7); + uint StoredRoot() => db.Storage.GetCollectionMetadata("peoplev2_collection")! + .Indexes.Single(i => i.Name == "idx_Age").RootPageId; + var rootBefore = StoredRoot(); + + // Hold the metadata lock so saving the new root times out. + var metadataLock = (SemaphoreSlim)typeof(StorageEngine) + .GetField("_metadataLock", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! + .GetValue(db.Storage)!; + metadataLock.Wait(); + try + { + await Assert.ThrowsAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); + } + finally + { + metadataLock.Release(); + } + + Assert.Equal(rootBefore, StoredRoot()); + Assert.Equal(rootBefore, db.PeopleV2.IndexManager.GetIndex("idx_Age")!.RootPageId); + Assert.Equal(expected, (await db.PeopleV2.QueryIndexAsync("idx_Age", 7, 7).ToListAsync()).Count); + await AssertCopyWasFreedAsync(db); + } + + using (var db = new TestDbContext(_dbPath)) + Assert.Equal(expected, (await db.PeopleV2.QueryIndexAsync("idx_Age", 7, 7).ToListAsync()).Count); + } + + private static async Task> InsertPeopleAsync(TestDbContext db) + { var people = Enumerable.Range(0, 3000).Select(i => new PersonV2 { Name = $"p{i}", Age = i % 90 }).ToList(); await db.PeopleV2.InsertBulkAsync(people); await db.CheckpointAsync(); + return people; + } - // Corrupt the last document in primary-key order, so the rebuild fails after most inserts. + /// + /// Corrupts the last document in primary-key order, so a rebuild fails after most inserts. + /// Returns an age that the corrupted document does not have, to query afterwards. + /// + private static async Task CorruptLastDocumentAsync(TestDbContext db) + { var storage = db.Storage; var primary = (BTreeIndex)typeof(DocumentCollection) .GetField("_primaryIndex", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! @@ -702,22 +791,23 @@ public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIn var lastEntry = primary.Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Backward).First(); var location = lastEntry.Location; var corruptedAge = (await db.PeopleV2.FindByIdAsync(lastEntry.Key.As()))!.Age; - var queryAge = (corruptedAge + 1) % 90; var page = new byte[storage.PageSize]; storage.ReadPage(location.PageId, null, page); var slot = SlotEntry.ReadFrom(page.AsSpan(SlottedPageHeader.Size + location.SlotIndex * SlotEntry.Size)); page.AsSpan(slot.Offset + 4, 12).Fill(0xFF); // invalid element types storage.WritePageImmediate(location.PageId, page); + return (corruptedAge + 1) % 90; + } - await Assert.ThrowsAnyAsync(() => db.PeopleV2.VacuumAsync(new VacuumOptions { RebuildIndexes = true, TruncateFile = false })); - - // The old index still answers queries. - Assert.Equal(people.Count(p => p.Age == queryAge), (await db.PeopleV2.QueryIndexAsync("idx_Age", queryAge, queryAge).ToListAsync()).Count); - - // The half-built copy of a 3000-entry index spans dozens of pages, not just its root. Each - // is either back on the free list (handed out next, below the high-water mark) or still - // queued for the next WAL truncation (a recycled page with a WAL image; a background - // checkpoint can make the explicit one below return without truncating). + /// + /// The copy of a 3000-entry index spans dozens of pages, not just its root. Each is either + /// back on the free list (handed out next, below the high-water mark) or still queued for the + /// next WAL truncation (a recycled page with a WAL image; a background checkpoint can make the + /// explicit one below return without truncating). + /// + private static async Task AssertCopyWasFreedAsync(TestDbContext db) + { + var storage = db.Storage; await db.CheckpointAsync(); var flags = System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance; int queued = ((HashSet)typeof(StorageEngine).GetField("_pendingFreePages", flags)!.GetValue(storage)!).Count; @@ -727,6 +817,22 @@ public async Task Vacuum_RebuildIndexes_DocumentCollection_FailedRebuild_KeepsIn Assert.True(reused + queued > 20, $"only {reused} pages were returned to the free list and {queued} queued"); } + private sealed class AbortRecordFailedException() : IOException("abort record could not be written"); + + /// Delegates to the real WAL but fails to write abort records. + private sealed class AbortRecordFailingWal(IWriteAheadLog inner) : IWriteAheadLog + { + public ValueTask WriteBeginRecordAsync(ulong transactionId, CancellationToken ct = default) => inner.WriteBeginRecordAsync(transactionId, ct); + public ValueTask WriteCommitRecordAsync(ulong transactionId, CancellationToken ct = default) => inner.WriteCommitRecordAsync(transactionId, ct); + public ValueTask WriteAbortRecordAsync(ulong transactionId, CancellationToken ct = default) => throw new AbortRecordFailedException(); + public ValueTask WriteDataRecordAsync(ulong transactionId, uint pageId, ReadOnlyMemory afterImage, CancellationToken ct = default) => inner.WriteDataRecordAsync(transactionId, pageId, afterImage, ct); + public Task FlushAsync(CancellationToken ct = default) => inner.FlushAsync(ct); + public long GetCurrentSize() => inner.GetCurrentSize(); + public Task TruncateAsync(CancellationToken ct = default) => inner.TruncateAsync(ct); + public List ReadAll() => inner.ReadAll(); + public void Dispose() { } + } + /// /// A rebuilt vector index holds only current embeddings, so nearest-neighbour search sees /// the updated vector. 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 7/7] 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();