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, 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/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; + } +}