diff --git a/roadmap/v5/MULTI_PROCESS_WAL.md b/roadmap/v5/MULTI_PROCESS_WAL.md index cb220e9..2086baf 100644 --- a/roadmap/v5/MULTI_PROCESS_WAL.md +++ b/roadmap/v5/MULTI_PROCESS_WAL.md @@ -152,7 +152,9 @@ The WAL index is a double-buffered open-addressing hash table mapping `uint page **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)). +**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)). Group commit appends at the published `WalEndOffset` and cuts off whatever lies beyond it: the remains of a batch whose writer died before publishing it, never acknowledged. + +**Unreadable records.** A catch-up that cannot read a record inside the published range keeps its bookmark, so it never skips the records after it; it retries on the next transaction, commit or checkpoint. Meanwhile that process's checkpoints write nothing (reported as failed in the `Checkpoint` metric, `MetricsSnapshot.CheckpointsFailedTotal`), and a trace warning names the WAL and the offset of the record: the first failure, then every power of two in a row, and from 16 in a row it calls the replay stalled. --- diff --git a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs index b1622db..0399b0f 100644 --- a/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs +++ b/src/BLite.Core/Storage/StorageEngine.GroupCommit.cs @@ -161,13 +161,14 @@ private async Task ProcessBatchAsync(List batch) // 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). + // past this batch below never skips them, and append right after them instead of at + // this engine's own, stale, write position (which would overwrite them) or after the + // torn remains of a batch a crashed writer never published. bool caughtUp = false; if (wal != null) { caughtUp = TryCatchUpWithSharedWal(wal); - wal.SeekToEnd(); + wal.SeekToSharedEnd(_shm!.ReadWalEndOffset()); } foreach (var commit in batch) diff --git a/src/BLite.Core/Storage/StorageEngine.Transactions.cs b/src/BLite.Core/Storage/StorageEngine.Transactions.cs index b334377..ded7445 100644 --- a/src/BLite.Core/Storage/StorageEngine.Transactions.cs +++ b/src/BLite.Core/Storage/StorageEngine.Transactions.cs @@ -94,9 +94,11 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) if (shmEnd > lastKnown) { List<(uint pageId, byte[] data, long walOffset)> newPages; + bool complete; + long stoppedAt; try { - newPages = wal.ReadCommittedPagesSince(lastKnown, shmEnd); + newPages = wal.ReadCommittedPagesSince(lastKnown, shmEnd, out complete, out stoppedAt); } catch (Exception ex) { @@ -110,6 +112,20 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) if (_shm.ReadWalGeneration() != generation) return false; + // A record that could not be read ends the scan early: the records after it were not + // replayed. Moving the bookmark past them would let this engine skip them and truncate + // the WAL that still holds them (#198). Keep it, and with it the WAL, until a replay + // gets through. If the record stays unreadable, every catch-up stops there: reported + // so that the stall is visible. + if (!complete) + { + ReportFailedCatchUp(wal, lastKnown, shmEnd, + stoppedAt < shmEnd + ? $"the record at offset {stoppedAt} is unreadable (torn, corrupt, or past the end of the file)" + : $"the last record ends at offset {stoppedAt}, past the published end"); + return false; + } + foreach (var (pid, data, walOffset) in newPages) { _walIndex[pid] = data; @@ -124,6 +140,13 @@ private bool TryCatchUpWithSharedWal(Transactions.WriteAheadLog wal) return true; } + /// + /// Failed catch-ups in a row from which calls the replay + /// stalled. A record inside the published range that stays unreadable stops every catch-up + /// of every process whose bookmark is before it. + /// + internal const int CatchUpStallThreshold = 16; + /// /// 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 @@ -139,7 +162,10 @@ private void ReportFailedCatchUp(Transactions.WriteAheadLog wal, long fromOffset 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."); + "and checkpoints write nothing, so the WAL is not truncated." + + (failures >= CatchUpStallThreshold + ? " The replay has stalled: the WAL and this engine's memory use grow until the record can be read or another process truncates the WAL." + : "")); } /// diff --git a/src/BLite.Core/Transactions/WriteAheadLog.cs b/src/BLite.Core/Transactions/WriteAheadLog.cs index c8c563b..aa2b82c 100644 --- a/src/BLite.Core/Transactions/WriteAheadLog.cs +++ b/src/BLite.Core/Transactions/WriteAheadLog.cs @@ -474,19 +474,36 @@ 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. + /// Multi-process: moves the write position to , the end of the + /// last batch any process has published (the SHM WAL end), and cuts off whatever lies beyond + /// it. Another process may have appended since this instance last wrote, so the position left + /// by its own writes is stale. Bytes past the shared end belong to a batch whose writer died + /// before publishing it, so that batch was never acknowledged. Appending after them would put + /// a torn record in front of this batch, where every reader's replay stops (#198). + /// Call under the cross-process writer lock before appending. /// - internal void SeekToEnd() + /// + /// Falls back to the end of the file if lies beyond it, or, for an + /// encrypted WAL, inside the crypto file header, which only a truncation resets. + /// + internal void SeekToSharedEnd(long sharedEnd) { if (!_lock.Wait(_writeTimeoutMs)) throw new TimeoutException("Timed out acquiring WAL write lock."); try { - if (_walStream != null) - _walStream.Position = _walStream.Length; + if (_walStream == null) return; + + long length = _walStream.Length; + if (sharedEnd > length || (_crypto != null && sharedEnd < _crypto.FileHeaderSize)) + { + _walStream.Position = length; + return; + } + + if (length > sharedEnd) + _walStream.SetLength(sharedEnd); + _walStream.Position = sharedEnd; } finally { @@ -718,9 +735,19 @@ private int ReadViaSeek(long offset, byte[] buffer) /// write per page wins — for population into the local _walIndex and /// _walOffsets (Phase 7 replay). /// - internal List<(uint pageId, byte[] data, long walOffset)> ReadCommittedPagesSince(long fromOffset, long toOffset) + /// + /// true if the records ended exactly at . false if the + /// scan stopped earlier at a record it could not read (torn, truncated, corrupt or + /// undecryptable) or ran past : the records after that point were not + /// read, so the caller must not treat the range as replayed. + /// + /// Where the scan stopped: if complete. + internal List<(uint pageId, byte[] data, long walOffset)> ReadCommittedPagesSince(long fromOffset, long toOffset, out bool complete, out long stoppedAt) { + complete = true; + stoppedAt = toOffset; if (_walStream == null || fromOffset >= toOffset) return new List<(uint, byte[], long)>(); + long scanned = fromOffset; // Track writes per transaction as (walOffset, data) so we can determine which // version of a page is latest (highest walOffset) across concurrent transactions. @@ -771,6 +798,7 @@ private int ReadViaSeek(long offset, byte[] buffer) } pos += 4 + ciphertextSize; } + scanned = pos; } else { @@ -819,8 +847,11 @@ private int ReadViaSeek(long offset, byte[] buffer) default: goto done; } } - done: ; + done: + scanned = pos; } + complete = scanned == toOffset; + stoppedAt = scanned; // Emit one entry per page: the latest committed write (highest walOffset) wins. // Iterate all committed transactions' writes in WAL order so that later commits diff --git a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs index 6dd7d92..451739e 100644 --- a/tests/BLite.Tests/MultiProcessWalTruncationTests.cs +++ b/tests/BLite.Tests/MultiProcessWalTruncationTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using BLite.Core.Storage; using BLite.Core.Transactions; @@ -243,6 +244,199 @@ public async Task RandomInterleavings_KeepEveryCommit() } } + [Fact] + public async Task TornBatchOfACrashedWriter_IsCutOff_SoTheNextCommitIsReplayed() + { + 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(); + + await WriteAndCommitAsync(a, p1, 0x11); + using (b.BeginTransaction()) { } + + // A writer dies part-way through a batch: its bytes are in the file, but it never + // published a new WAL end. + long published = WalLength; + AppendToWal(TornWriteRecord()); + Assert.True(WalLength > published); + + // The next commit appends at the published end, so no reader meets the torn bytes. + await WriteAndCommitAsync(a, p2, 0x22); + + // B replays A's commit, checkpoints it and can truncate the WAL. Had the commit been + // appended after the torn bytes, B's replay would stop there on every attempt, and B could + // never truncate the WAL again. + using (b.BeginTransaction()) { } + await b.CheckpointAsync(); + Assert.Equal(0, WalLength); + + await WriteAndCommitAsync(b, p3, 0x33); + using (a.BeginTransaction()) { } + await a.CheckpointAsync(); + + AssertFreshEngineReads(p1, 0x11); + AssertFreshEngineReads(p2, 0x22); + AssertFreshEngineReads(p3, 0x33); + } + + [Fact] + public async Task UnreadableRecordInsideTheWal_StopsTheReplay_WithoutLettingTheReaderTruncate() + { + 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(); + + using (b.BeginTransaction()) { } // B's bookmark: the empty WAL + await WriteAndCommitAsync(a, p1, 0x11); + long secondRecord = WalLength; + await WriteAndCommitAsync(a, p2, 0x22); + + // The first of A's records becomes unreadable for B. + CorruptWalByte(0); + + // B's replay stops at it, so it must not count A's records as replayed: committing and + // checkpointing its own page must leave the WAL, and with it A's records, in place. + await WriteAndCommitAsync(b, p3, 0x33); + long walBeforeCheckpoint = WalLength; + await b.CheckpointAsync(); + Assert.Equal(walBeforeCheckpoint, WalLength); + Assert.True(walBeforeCheckpoint > secondRecord); + + // A still holds its pages and checkpoints them, together with B's. + await a.CheckpointAsync(); + AssertFreshEngineReads(p1, 0x11); + AssertFreshEngineReads(p2, 0x22); + AssertFreshEngineReads(p3, 0x33); + } + + [Fact] + public async Task CutOff_CannotDeleteCommitsOfTheDirectOrTwoPhasePaths() + { + // The repro from the review of #202: a commit through the direct path never publishes + // the WAL end, so the next group commit's cut-off would delete it. Those paths are + // rejected in multi-process mode, so every acknowledged commit lies within the published + // end and survives the cut-off. + uint p, q; + using (var engine = new StorageEngine(_dbPath, Config)) + { + p = engine.AllocatePage(); + q = engine.AllocatePage(); + using (var tx = engine.BeginTransaction()) + { + engine.WritePage(p, tx.TransactionId, Filled(engine.PageSize, 0x44)); + await Assert.ThrowsAsync(() => engine.CommitTransactionAsync(tx.TransactionId)); + await Assert.ThrowsAsync(() => engine.PrepareTransactionAsync(tx.TransactionId)); + await Assert.ThrowsAsync(() => engine.MarkTransactionCommittedAsync(tx.TransactionId)); + AssertNothingBeyondThePublishedEnd(); + await tx.CommitAsync(); + } + + await WriteAndCommitAsync(engine, q, 0x55); + AssertNothingBeyondThePublishedEnd(); + } + + using var reopened = new StorageEngine(_dbPath, Config); + using (reopened.BeginTransaction()) + { + AssertPage(reopened, p, 0x44); + AssertPage(reopened, q, 0x55); + } + } + + [Fact] + public async Task UnreadableRecordInsideThePublishedRange_IsReportedAsAStall() + { + using var a = new StorageEngine(_dbPath, Config); + using var b = new StorageEngine(_dbPath, Config); + var metrics = b.EnsureMetrics(); + var p1 = a.AllocatePage(); + var p2 = a.AllocatePage(); + var p3 = a.AllocatePage(); + + var warnings = new WarningCollector(Path.ChangeExtension(_dbPath, ".wal")); + Trace.Listeners.Add(warnings); + try + { + using (b.BeginTransaction()) { } // B's bookmark: the empty WAL + await WriteAndCommitAsync(a, p1, 0x11); + CorruptWalByte(0); // A's record: unreadable for B, whose bookmark lies before it + await WriteAndCommitAsync(b, p3, 0x33); // something for B to checkpoint + await WriteAndCommitAsync(a, p2, 0x22); + + for (int i = 0; i < 2 * StorageEngine.CatchUpStallThreshold; i++) + using (b.BeginTransaction()) { } + await b.CheckpointAsync(); + } + finally + { + Trace.Listeners.Remove(warnings); + } + + // Reported, but not on every attempt: the first, then every power of two in a row. + var messages = warnings.Messages; + Assert.NotEmpty(messages); + Assert.InRange(messages.Count, 2, 7); + Assert.All(messages, m => Assert.Contains("the record at offset 0 is unreadable", m)); + Assert.Contains(messages, m => m.Contains("stalled")); + Assert.DoesNotContain("stalled", messages[0]); + + // The checkpoint wrote nothing, and says so. + Assert.NotEqual(0, WalLength); + await WaitUntilAsync(() => metrics.GetSnapshot().CheckpointsFailedTotal >= 1); + Assert.Equal(0, metrics.GetSnapshot().CheckpointsTotal); + } + + private void AssertNothingBeyondThePublishedEnd() + { + using var shm = WalSharedMemory.Open(ShmPath, Config.PageSize); + Assert.Equal(WalLength, shm.ReadWalEndOffset()); + } + + /// Collects the trace warnings that name one WAL file. + private sealed class WarningCollector(string walPath) : TraceListener + { + private readonly List _messages = new(); + + public List Messages { get { lock (_messages) return _messages.ToList(); } } + + public override void Write(string? message) { } + + public override void WriteLine(string? message) + { + if (message != null && message.Contains(walPath)) + lock (_messages) _messages.Add(message); + } + } + + /// The start of a Write record whose page data never made it to disk. + private static byte[] TornWriteRecord() + { + var record = new byte[1 + 16 + 100]; + record[0] = 2; // WalRecordType.Write + BitConverter.GetBytes(999_999UL).CopyTo(record, 1); // transaction id + BitConverter.GetBytes(12345u).CopyTo(record, 9); // page id + BitConverter.GetBytes(16384).CopyTo(record, 13); // data length; only 100 bytes follow + return record; + } + + private void AppendToWal(byte[] bytes) + { + using var stream = new FileStream(Path.ChangeExtension(_dbPath, ".wal"), FileMode.Open, FileAccess.Write, FileShare.ReadWrite); + stream.Seek(0, SeekOrigin.End); + stream.Write(bytes); + } + + private void CorruptWalByte(long offset) + { + using var stream = new FileStream(Path.ChangeExtension(_dbPath, ".wal"), FileMode.Open, FileAccess.Write, FileShare.ReadWrite); + stream.Position = offset; + stream.WriteByte(0xFF); // not a record type + } + [Fact] public async Task DirectCommitPath_IsRejected_SoAnotherProcessesTruncationCannotLoseIt() {