diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 19d77c2..238994f 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()) - await RebuildIndexAsync(index, ct).ConfigureAwait(false); + { + // 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; @@ -1594,29 +1599,72 @@ public async IAsyncEnumerable QueryIndexAsync( } } - private async Task RebuildIndexAsync(CollectionSecondaryIndex index, CancellationToken ct = default, bool skipUnreadableDocuments = true) + /// + /// 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); 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)) + await FillIndexAsync(fresh, transaction, skipUnreadableDocuments: false, ct).ConfigureAwait(false); + await transaction.CommitAsync(ct); + } + catch + { + // 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 */ } + + try { await transaction.RollbackAsync(); } + catch { /* best-effort: never mask the original failure */ } + + 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)) { - 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 - } + IReadOnlyList pages = Array.Empty(); + try { pages = fresh.CollectAllPages(0); } + catch { /* best-effort: never mask the original failure */ } + FreeUnregisteredCopy(fresh, pages); } + throw; + } + } + + 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) + { + var transaction = _storage.BeginTransaction(IsolationLevel.ReadCommitted); + try + { + await FillIndexAsync(index, transaction, skipUnreadableDocuments, ct).ConfigureAwait(false); await transaction.CommitAsync(ct); } catch @@ -1626,6 +1674,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 e657c24..db20278 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, List 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 d465365..1077ea4 100644 --- a/src/BLite.Core/Indexing/CollectionIndexManager.cs +++ b/src/BLite.Core/Indexing/CollectionIndexManager.cs @@ -301,19 +301,65 @@ 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 (_disposed) + throw new ObjectDisposedException(nameof(CollectionIndexManager)); + 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. 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) + { + lock (_lock) + { + if (_disposed) + throw new ObjectDisposedException(nameof(CollectionIndexManager)); + if (!_indexes.TryGetValue(name, out var old)) + throw new InvalidOperationException($"Index '{name}' does not exist"); + _indexes[name] = fresh; _cachedIndexInfo = null; // invalidate cache // 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(); - return fresh; } } 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/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index 20559b7..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) @@ -193,9 +205,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 _); } } @@ -221,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.Memory.cs b/src/BLite.Core/Storage/StorageEngine.Memory.cs index c2d369b..ca1f740 100644 --- a/src/BLite.Core/Storage/StorageEngine.Memory.cs +++ b/src/BLite.Core/Storage/StorageEngine.Memory.cs @@ -68,10 +68,126 @@ public uint AllocateCollectionPage(string collectionName) return CollectionPageMarker | ((uint)(slot & 0x3F) << CollSlotShift) | (localId & CollLocalMask); } - /// Frees 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) + { + return _walIndex.ContainsKey(pageId) + || _checkpointedWalPages.ContainsKey(pageId) + || IsInOpenTransaction(pageId); + } + + /// + /// 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() + { + _checkpointedWalPages.Clear(); + + HashSet pages; + lock (_pendingFreeLock) + { + if (_pendingFreePages.Count == 0) return; + pages = _pendingFreePages; + _pendingFreePages = new HashSet(); + } + + var keep = new List(); + foreach (var pageId in pages) + { + if (IsInOpenTransaction(pageId)) + { + keep.Add(pageId); + continue; + } + + try + { + GetPageFile(pageId, out var physId).FreePage(physId); + } + catch + { + keep.Add(pageId); + } + } + + if (keep.Count > 0) + { + lock (_pendingFreeLock) + foreach (var pageId in keep) + _pendingFreePages.Add(pageId); + } + } + + private bool IsInOpenTransaction(uint pageId) + { + foreach (var txnPages in _walCache.Values) + if (txnPages.ContainsKey(pageId)) + return true; + return false; + } + + /// 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 +277,9 @@ public void DropCollectionFile(string collectionName) { if (_collectionFiles == null) return; + if (_collectionNameToSlot != null && _collectionNameToSlot.TryGetValue(collectionName, out var slot)) + DiscardPendingFreePages(slot); + if (_collectionFiles.TryRemove(collectionName, out var lazy)) { if (lazy.IsValueCreated) lazy.Value.Dispose(); @@ -369,13 +488,9 @@ internal IReadOnlyCollection FreeCollectionPagesCore(string collectionName beforeFree?.Invoke(toFree); - // Free all collected pages and remove stale WAL-index entries so that - // a re-allocated page is not served stale data from the WAL index. + // FreePage defers reuse until the WAL holds no stale image of these pages. foreach (var pageId in toFree) - { FreePage(pageId); - _walIndex.TryRemove(pageId, out _); - } return toFree; } @@ -391,11 +506,15 @@ internal IReadOnlyCollection 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 +525,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..98366a3 100644 --- a/src/BLite.Core/Storage/StorageEngine.Recovery.cs +++ b/src/BLite.Core/Storage/StorageEngine.Recovery.cs @@ -38,7 +38,17 @@ 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. + // 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); } + 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 +80,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 +123,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).ConfigureAwait(false); } else if (_shm != null && !_walIndex.IsEmpty) { @@ -201,6 +172,64 @@ 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; + } + + // 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. + 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 +267,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 +381,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.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index ef40b21..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. @@ -219,12 +232,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 +382,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/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/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/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/FreePageReuseTests.cs b/tests/BLite.Tests/FreePageReuseTests.cs new file mode 100644 index 0000000..a621a1e --- /dev/null +++ b/tests/BLite.Tests/FreePageReuseTests.cs @@ -0,0 +1,325 @@ +using BLite.Bson; +using BLite.Core; +using BLite.Core.Indexing; +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")); + } + + [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 */ } + } + } + + [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(); + 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; + } +} diff --git a/tests/BLite.Tests/VacuumTests.cs b/tests/BLite.Tests/VacuumTests.cs index ed5303d..8d9ed28 100644 --- a/tests/BLite.Tests/VacuumTests.cs +++ b/tests/BLite.Tests/VacuumTests.cs @@ -1,7 +1,9 @@ using BLite.Bson; using BLite.Core; using BLite.Core.Collections; +using BLite.Core.Indexing; using BLite.Core.Storage; +using BLite.Core.Transactions; using BLite.Shared; namespace BLite.Tests; @@ -345,7 +347,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 +401,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 +588,271 @@ 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); + } + } + + /// + /// 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_IsKept() + { + using var db = new TestDbContext(_dbPath); + 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 }); + + 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 = 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; + } + + /// + /// 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)! + .GetValue(db.PeopleV2)!; + 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 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; + } + + /// + /// 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; + var highWater = ((IPageStorage)typeof(StorageEngine).GetField("_pageFile", flags)!.GetValue(storage)!).NextPageId; + int reused = 0; + while (storage.AllocatePage() < highWater) reused++; + 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. + /// + [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 // ======================================================================