fix(storage): drain only the flushed WAL image, atomically, after a checkpoint (#311)

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

@@ -141,13 +141,17 @@ private async Task CheckpointCoreAsync(bool commitLockHeld, CancellationToken ct
             // Drain _walIndex (and _walOffsets) entries that we successfully flushed.
             // Track the highest checkpointed WAL offset in the same loop (avoids a
             // second O(n) pass over toCheckpoint).
-            // ReferenceEquals check ensures we only remove the exact version we flushed.
+            // Only the exact image we flushed is removed, in one atomic step: commits do not wait
+            // for this loop, and a check followed by a separate remove could drop an image a commit
+            // published in between, leaving the older flushed one on disk (the commit is lost, and
+            // an insert reading the page back reuses its slot) (#311).
             long maxCheckpointedOffset = 0;
             foreach (var kvp in toCheckpoint)
             {
-                if (_walIndex.TryGetValue(kvp.Key, out var current) && ReferenceEquals(current, kvp.Value))
+                // byte[] values compare by reference, so this removes the entry only if it still
+                // holds the flushed image.
+                if (((ICollection<KeyValuePair<uint, byte[]>>)_walIndex).Remove(kvp))
                 {
-                    _walIndex.TryRemove(kvp.Key, out _);
                     // Track max offset while removing from _walOffsets.
                     if (_walOffsets != null && _walOffsets.TryRemove(kvp.Key, out long removedOffset)
                         && removedOffset > maxCheckpointedOffset)

tests/BLite.Tests/ConcurrentConsistencyTests.cs

@@ -239,9 +239,16 @@ public async Task HighContention_ManySmallTransactions_ConsistentFinalState()
             {
                 var found = await readCol.FindByIdAsync(id);
                 Assert.NotNull(found);
-                Assert.True(found.TryGetInt32("t", out var ft) && ft == thread);
-                Assert.True(found.TryGetInt32("tx", out var ftx) && ftx == txn);
-                Assert.True(found.TryGetInt32("d", out var fd) && fd == doc);
+                // Name the document that came back, so a failure says what was wrong.
+                string Mismatch() =>
+                    $"contention_{thread}: id {id} written as t={thread} tx={txn} d={doc} read back as " +
+                    string.Join(", ", found.EnumerateFields().Select(f => $"{f.Name}={f.Value}"));
+                var tOk = found.TryGetInt32("t", out var ft) && ft == thread;
+                Assert.True(tOk, tOk ? null : Mismatch());
+                var txOk = found.TryGetInt32("tx", out var ftx) && ftx == txn;
+                Assert.True(txOk, txOk ? null : Mismatch());
+                var dOk = found.TryGetInt32("d", out var fd) && fd == doc;
+                Assert.True(dOk, dOk ? null : Mismatch());
             }
         }
     }