src/db.zig

@@ -1612,14 +1612,20 @@ pub const Engine = struct {
     const catalog_magic: u32 = 0x4D464354; // "MFCT"
     const catalog_version: u32 = 1;
 
-    fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !void {
+    /// Serialize the catalog and return the live-byte total it observed.
+    ///
+    /// The engine's own total is by definition the sum over collections, and
+    /// `read_catalog` rebuilds it that way, so a divergence means some path
+    /// published or evicted bytes at one level and not the other -- with a
+    /// compaction trigger that fires never or always as the visible symptom.
+    /// The check is worth making and this is where every collection is walked
+    /// anyway, but it cannot be made *here*: the sum is accumulated across
+    /// collections over time while the engine's total moves under it, so a
+    /// writer landing mid-walk would trip it on a database that is perfectly
+    /// consistent. The caller asserts it after the `seq` check has established
+    /// that no writer landed at all.
+    fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !u64 {
         const gpa = self.gpa;
-        // The engine's live-byte total is by definition the sum over
-        // collections, and `read_catalog` rebuilds it that way. Check it here,
-        // where every collection is being walked regardless: a divergence means
-        // some path published or evicted bytes at one level and not the other,
-        // and the visible symptom would be a compaction trigger that fires
-        // never or always.
         var live_sum: u64 = 0;
         try put_u32(gpa, out, catalog_magic);
         try put_u32(gpa, out, catalog_version);
@@ -1653,11 +1659,8 @@ pub const Engine = struct {
                 for (coll.indexes.items) |ix| try write_index_catalog(gpa, out, ix);
             }
         }
-        assert_msg(
-            live_sum == self.live_bytes,
-            "the engine's live-byte total must equal the sum over collections",
-        );
         try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items));
+        return live_sum;
     }
 
     fn write_index_catalog(
@@ -1883,7 +1886,8 @@ pub const Engine = struct {
             buf.clearRetainingCapacity();
             try self.catalog_lock.lockShared(self.io);
             const snapshot_seq = self.seq;
-            self.write_catalog(&buf) catch |err| {
+            const live_before = self.live_bytes;
+            const live_sum = self.write_catalog(&buf) catch |err| {
                 self.catalog_lock.unlockShared(self.io);
                 return err;
             };
@@ -1896,9 +1900,42 @@ pub const Engine = struct {
                 self.log_lock.unlock(self.io);
                 continue;
             }
-            assert_msg(
-                snapshot_seq <= self.committed_seq,
-                "checkpoint watermark past the durable log tail",
+            if (snapshot_seq > self.committed_seq) {
+                // A writer appended before the snapshot and its commit has not
+                // landed yet -- it is between `insert` and `commit`, or inside
+                // one, waiting on the leader's fsync. The seq check above does
+                // not catch this: nothing appended *during* the walk, the
+                // append was already there when it started.
+                //
+                // Publishing here would claim durability for a record that is
+                // still in the log's buffer, and the truncation that follows a
+                // checkpoint would then throw it away. That is the one thing
+                // the whole watermark ordering exists to prevent (PLAN D6), and
+                // it used to be an assertion -- so the failure mode was a
+                // server abort under exactly the load that makes checkpoints
+                // frequent. Reproduced in seconds by four writers against a
+                // checkpoint loop, and the window is as wide as an fsync.
+                //
+                // Seal it and take the snapshot again rather than spinning:
+                // `commit` covers every append made so far, so one more round
+                // is enough. Outside `log_lock`, which `commit` takes itself.
+                self.log_lock.unlock(self.io);
+                try self.commit();
+                continue;
+            }
+            // Only when the walk was quiet. An unchanged `seq` is not enough on
+            // its own: a writer bumps it when it appends the log record and
+            // updates the byte counters afterwards, so it can be past the seq
+            // the snapshot captured and still be about to move `live_bytes`
+            // under a collection the walk has already been through. Requiring
+            // the engine total to be unmoved across the whole walk closes that,
+            // at the cost of skipping the check under sustained writes -- which
+            // is the right trade, because what it guards against is a code path
+            // that updates one level and not the other, and that is
+            // deterministic wherever it exists.
+            if (self.live_bytes == live_before) assert_msg(
+                live_sum == live_before,
+                "the engine's live-byte total must equal the sum over collections",
             );
             const pages: u32 = @intCast((buf.items.len + pgr.page_size - 1) / pgr.page_size);
             const first = self.pager.alloc_pages(pages) catch |err| {
@@ -1906,6 +1943,12 @@ pub const Engine = struct {
                 return err;
             };
             @memcpy(self.pager.bytes_mut(@as(u64, first) << pgr.page_shift, buf.items.len), buf.items);
+            // The invariant, on the line that would break it: `log_lock` has
+            // been held since the check above and `committed_seq` only grows.
+            assert_msg(
+                snapshot_seq <= self.committed_seq,
+                "checkpoint watermark past the durable log tail",
+            );
             self.pager.publish(.{
                 .seq = snapshot_seq,
                 .catalog_page = first,
@@ -2889,6 +2932,85 @@ test "compaction rewrites log and keeps data" {
     engine3.unlock();
 }
 
+test "a checkpoint runs alongside writers on several collections" {
+    // `write_catalog` reads each collection's slab extents, indexes and byte
+    // counters while holding only the *shared catalog* lock -- and a writer
+    // holds that same lock shared, taking the collection's lock exclusively.
+    // So the snapshot walked structures its owner was free to mutate, and
+    // `slab_extents` is an ArrayList a new extent appends to: a reallocation
+    // mid-walk leaves the serializer reading freed memory.
+    //
+    // Several collections rather than one, because the interesting overlap is a
+    // writer on collection B while the catalog is serializing collection A.
+    //
+    // Mutation check: drop the `lockShared` from `write_catalog`'s collection
+    // loop. Not reliably red -- a data race never is -- but it runs under
+    // ReleaseSafe, where the reads it makes are bounds-checked.
+    const gpa = testing.allocator;
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var tmp = try TmpLog.init(gpa);
+    defer tmp.deinit(gpa);
+    var engine = try Engine.open(gpa, io, tmp.path);
+    defer engine.deinit();
+
+    const colls = [_][]const u8{ "a", "b", "c", "d" };
+    const per_coll: i32 = 150;
+    var done = std.atomic.Value(usize).init(colls.len);
+
+    const Worker = struct {
+        fn writer(
+            e: *Engine,
+            name: []const u8,
+            left: *std.atomic.Value(usize),
+            alloc: std.mem.Allocator,
+        ) error{Canceled}!void {
+            defer _ = left.fetchSub(1, .release);
+            for (1..per_coll + 1) |i| {
+                var doc = make_doc(alloc, @intCast(i), "user") catch return error.Canceled;
+                defer doc.deinit();
+                {
+                    e.lock() catch return error.Canceled;
+                    defer e.unlock();
+                    e.insert("app", name, &doc, undefined) catch return error.Canceled;
+                }
+                // As the dispatch epilogue does (commands.zig): the append bumps
+                // `seq`, the commit is what makes it durable, and a checkpoint
+                // may only describe what is durable.
+                e.commit() catch return error.Canceled;
+            }
+        }
+
+        fn checkpointer(e: *Engine, left: *std.atomic.Value(usize)) error{Canceled}!void {
+            while (left.load(.acquire) > 0) {
+                // Errors are the point of the retry loop inside `checkpoint`,
+                // not a failure of this test; a checkpoint that gives up under
+                // sustained writes has still not corrupted anything.
+                e.checkpoint() catch {};
+            }
+        }
+    };
+
+    var group: std.Io.Group = .init;
+    defer group.cancel(io);
+    for (colls) |name| group.async(io, Worker.writer, .{ &engine, name, &done, gpa });
+    group.async(io, Worker.checkpointer, .{ &engine, &done });
+    try group.await(io);
+
+    // Every write is still there, and the catalog the checkpoints wrote agrees
+    // with the engine -- the second half is what `write_catalog`'s own assertion
+    // checks on the way through.
+    try engine.checkpoint();
+    try engine.lock_read();
+    defer engine.unlock_read();
+    for (colls) |name| {
+        const coll = engine.get_collection("app", name) orelse return error.TestUnexpectedResult;
+        try testing.expectEqual(@as(usize, @intCast(per_coll)), coll.id_index.count());
+    }
+}
+
 test "concurrent readers and writers on a threaded Io" {
     // Real worker threads: writers hold the exclusive lock, readers the
     // shared lock. Proves the RwLock split keeps committed writes visible