fix(storage): don't skip WAL records past a torn one in multi-process replay (#198)

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.
 
 ---
 

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

@@ -161,13 +161,14 @@ private async Task ProcessBatchAsync(List<PendingCommit> 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)

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;
     }
 
+    /// <summary>
+    /// Failed catch-ups in a row from which <see cref="ReportFailedCatchUp"/> 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.
+    /// </summary>
+    internal const int CatchUpStallThreshold = 16;
+
     /// <summary>
     /// 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 (<see cref="Trace"/>): 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."
+                : ""));
     }
 
     /// <summary>

src/BLite.Core/Transactions/WriteAheadLog.cs

@@ -474,19 +474,36 @@ public long GetCurrentSize()
     internal long GetCurrentPositionNoLock() => _walStream?.Position ?? 0;
 
     /// <summary>
-    /// 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 <paramref name="sharedEnd"/>, 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.
     /// </summary>
-    internal void SeekToEnd()
+    /// <remarks>
+    /// Falls back to the end of the file if <paramref name="sharedEnd"/> lies beyond it, or, for an
+    /// encrypted WAL, inside the crypto file header, which only a truncation resets.
+    /// </remarks>
+    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 <c>_walIndex</c> and
     /// <c>_walOffsets</c> (Phase 7 replay).
     /// </summary>
-    internal List<(uint pageId, byte[] data, long walOffset)> ReadCommittedPagesSince(long fromOffset, long toOffset)
+    /// <param name="complete">
+    /// <c>true</c> if the records ended exactly at <paramref name="toOffset"/>. <c>false</c> if the
+    /// scan stopped earlier at a record it could not read (torn, truncated, corrupt or
+    /// undecryptable) or ran past <paramref name="toOffset"/>: the records after that point were not
+    /// read, so the caller must not treat the range as replayed.
+    /// </param>
+    /// <param name="stoppedAt">Where the scan stopped: <paramref name="toOffset"/> if complete.</param>
+    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

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<NotSupportedException>(() => engine.CommitTransactionAsync(tx.TransactionId));
+                await Assert.ThrowsAsync<NotSupportedException>(() => engine.PrepareTransactionAsync(tx.TransactionId));
+                await Assert.ThrowsAsync<NotSupportedException>(() => 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());
+    }
+
+    /// <summary>Collects the trace warnings that name one WAL file.</summary>
+    private sealed class WarningCollector(string walPath) : TraceListener
+    {
+        private readonly List<string> _messages = new();
+
+        public List<string> 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);
+        }
+    }
+
+    /// <summary>The start of a Write record whose page data never made it to disk.</summary>
+    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()
     {