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