fix(vacuum): rebuild typed-collection indexes into fresh trees (#167)

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<T> QueryIndexAsync(
         }
     }
 
-    private async Task RebuildIndexAsync(CollectionSecondaryIndex<TId, T> index, CancellationToken ct = default, bool skipUnreadableDocuments = true)
+    /// <summary>
+    /// Populates an empty copy of index <paramref name="name"/> 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.
+    /// </summary>
+    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<uint> pages = Array.Empty<uint>();
+            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<uint> pages = Array.Empty<uint>();
+                try { pages = fresh.CollectAllPages(0); }
+                catch { /* best-effort: never mask the original failure */ }
+                FreeUnregisteredCopy(fresh, pages);
             }
+            throw;
+        }
+    }
+
+    private void FreeUnregisteredCopy(CollectionSecondaryIndex<TId, T> copy, IReadOnlyList<uint> 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<TId, T> 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<TId, T> index, Can
         }
     }
 
+    private async Task FillIndexAsync(CollectionSecondaryIndex<TId, T> 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

src/BLite.Core/Indexing/BTreeIndex.cs

@@ -1658,9 +1658,8 @@ private void MergeWithSibling(uint nodeId, uint parentId, int childIndex, List<I
         WriteInternalNode(parentId, p0, parentEntries, transactionId);
 
         // Free the empty Right Node
-        _storage.FreePage(rightNodeId); // Need to verify this works safely with Txn logic?
-                                         // Actually, FreePage is immediate in current impl. Might need TransactionalFreePage.
-                                         // Or just leave it allocated but unused for now.
+        _storage.FreePage(rightNodeId); // Not transactional: the page is released even if this
+                                         // transaction rolls back. Might need TransactionalFreePage.
 
         // Recursive Underflow Check on Parent
         int minInternal = MaxEntriesPerNode / 2;
@@ -1968,30 +1967,36 @@ public async IAsyncEnumerable<IndexEntry> LikeAsync(
     /// Returns all physical page IDs owned by this B+Tree.
     /// Used when dropping an index to free storage space.
     /// </summary>
-    public IReadOnlyList<uint> CollectAllPages()
+    public IReadOnlyList<uint> CollectAllPages() => CollectAllPages(0);
+
+    /// <summary>
+    /// Like <see cref="CollectAllPages()"/>, but walks the tree as <paramref name="transactionId"/>
+    /// sees it, including nodes that transaction has written but not committed.
+    /// </summary>
+    internal IReadOnlyList<uint> CollectAllPages(ulong transactionId)
     {
         var result = new List<uint>();
         var root = RootPageId;
         if (root != 0)
-            CollectPagesRecursive(root, result);
+            CollectPagesRecursive(root, result, transactionId);
         return result;
     }
 
-    private void CollectPagesRecursive(uint pageId, List<uint> result)
+    private void CollectPagesRecursive(uint pageId, List<uint> result, ulong transactionId)
     {
         if (pageId == 0) return;
         result.Add(pageId);
         var pageBuffer = System.Buffers.ArrayPool<byte>.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

src/BLite.Core/Indexing/CollectionIndexManager.cs

@@ -301,19 +301,65 @@ internal CollectionSecondaryIndex<TId, T> RecreateIndex(string name)
             if (!_indexes.TryGetValue(name, out var old))
                 throw new InvalidOperationException($"Index '{name}' does not exist");
 
-            var fresh = new CollectionSecondaryIndex<TId, T>(old.Definition, _storage, _mapper)
-            {
-                KeyFormat = old.KeyFormat
-            };
+            var fresh = CreateUnregisteredCopy(name);
+            fresh.KeyFormat = old.KeyFormat;
+            ReplaceIndex(name, fresh);
+            return fresh;
+        }
+    }
+
+    /// <summary>
+    /// Creates an empty index with the same definition as <paramref name="name"/> without
+    /// registering it, so queries keep using the existing index while the copy is populated.
+    /// The copy owns its root page: hand it to <see cref="ReplaceIndex"/> or release it with
+    /// <see cref="CollectionSecondaryIndex{TId,T}.FreeAllPages"/>.
+    /// </summary>
+    internal CollectionSecondaryIndex<TId, T> CreateUnregisteredCopy(string name)
+    {
+        lock (_lock)
+        {
+            if (_disposed)
+                throw new ObjectDisposedException(nameof(CollectionIndexManager<TId, T>));
+            if (!_indexes.TryGetValue(name, out var old))
+                throw new InvalidOperationException($"Index '{name}' does not exist");
+
+            return new CollectionSecondaryIndex<TId, T>(old.Definition, _storage, _mapper);
+        }
+    }
+
+    /// <summary>
+    /// Registers <paramref name="fresh"/> under <paramref name="name"/>, 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 <paramref name="fresh"/> remains the caller's to release.
+    /// </summary>
+    internal void ReplaceIndex(string name, CollectionSecondaryIndex<TId, T> fresh)
+    {
+        lock (_lock)
+        {
+            if (_disposed)
+                throw new ObjectDisposedException(nameof(CollectionIndexManager<TId, T>));
+            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;
         }
     }
 

src/BLite.Core/Indexing/CollectionSecondaryIndex.cs

@@ -638,12 +638,18 @@ public void Dispose()
     /// </summary>
     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);
+    }
+
+    /// <summary>
+    /// The pages of this index as <paramref name="transactionId"/> sees them (0: committed state).
+    /// Empty for a spatial index: <c>RTreeIndex</c> cannot enumerate its pages yet.
+    /// </summary>
+    internal IReadOnlyList<uint> CollectAllPages(ulong transactionId)
+    {
+        if (_btreeIndex != null) return _btreeIndex.CollectAllPages(transactionId);
+        if (_vectorIndex != null) return _vectorIndex.CollectAllPages(transactionId);
+        return Array.Empty<uint>();
     }
 }

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.
     /// </summary>
-    public IReadOnlyList<uint> CollectAllPages()
+    public IReadOnlyList<uint> CollectAllPages() => CollectAllPages(0);
+
+    /// <summary>
+    /// Like <see cref="CollectAllPages()"/>, but follows the page chain as
+    /// <paramref name="transactionId"/> sees it, including pages it has linked but not committed.
+    /// </summary>
+    internal IReadOnlyList<uint> 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<uint> 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;
             }

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

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