db/pager: the catalog may not claim a page that is on the free list

Reclamation runs as a checkpoint phase, so it frees pages concurrently with
everything else -- and the one failure that arrangement can produce is silent.
A catalog that claims a page already handed to the pager gets that page back
two generations later, written over by somebody else; the crash that falls
back to that generation then reads a document which is no longer there.
Nothing fails at the time, and the `seq` retry cannot see it because neither a
reclamation nor a rebuild appends a log record.

So `write_catalog` now asserts it, per run, in test and Debug builds -- three
list scans where every run is being walked anyway. It is proven to fire: have
`reclaim_windows` free the pages and keep the old run list, and the suite
panics on it. This is the double-ownership detector the plan said an enlarged
free list deserves.

And `checkpoint` takes a lock of its own. Two can be in flight -- a writer's
epilogue claims the pending flag while another is inside `compact`, which
checkpoints of its own. The publish was always safe, since it runs under
`log_lock`; the phase in front of it is new. One checkpoint's `reclaim_slabs`
frees pages under a collection's lock that the other's `write_catalog` may
already have serialized, and that is exactly the shape above.

Stated plainly: the argument for the lock is by construction, and no test
reproduces the interleaving -- removing it leaves the new concurrency test
green. What that test does do is run reclamation under two checkpointers and a
writer with the ownership assertion armed, which is the harness that would
catch the argument being wrong.

This is the second instance of the shape PLAN records as still open (a rebuild
frees pages under only the collection's lock while a checkpoint may have
snapshotted a catalog claiming them). Reclamation is now excluded from it;
`compact`'s rebuild walk still is not, and that remains recorded rather than
fixed here.

187/187 unit tests in ReleaseFast and ReleaseSafe, 83/83 fuzz, the full e2e
matrix, crash-fuzz 60 cycles.

src/db.zig

@@ -765,6 +765,27 @@ pub const Engine = struct {
     /// A leaf: nothing else is taken while it is held, and it is never held
     /// across an append, an fsync, or an allocation.
     counter_lock: std.Io.Mutex = .init,
+    /// One checkpoint at a time.
+    ///
+    /// Two can be in flight without it -- a writer's epilogue claims the
+    /// pending flag while another writer's epilogue is inside `compact`, which
+    /// checkpoints of its own. The publish itself was always safe, because it
+    /// runs under `log_lock`; what is not is the phase in front of it, which is
+    /// new. `reclaim_slabs` frees pages under a collection's lock, and the
+    /// other checkpoint's `write_catalog` may already have serialized that
+    /// collection's runs. It then publishes a catalog claiming pages that are
+    /// on the free list, and two generations later they are handed out and
+    /// written over -- so the crash that falls back to that generation reads a
+    /// document that is no longer there. The `seq` retry cannot see it, because
+    /// neither a reclamation nor a rebuild appends a log record.
+    ///
+    /// The argument is by construction; no test reproduces the interleaving.
+    /// What would catch it is the ownership assertion in `write_catalog`, which
+    /// is armed in test and Debug builds and is proven to fire.
+    ///
+    /// Taken before `catalog_lock`, so the order is checkpoint -> catalog ->
+    /// collection, the same descent everything else makes.
+    checkpoint_lock: std.Io.Mutex = .init,
     /// The checkpoint's own page promise, for the catalog and free-list pages it
     /// writes. Separate from any collection's for the same reason those are
     /// separate from each other.
@@ -2335,6 +2356,17 @@ pub const Engine = struct {
                 // no second read path to keep working.
                 try put_u32(gpa, out, @intCast(coll.slab_runs.items.len));
                 for (coll.slab_runs.items) |r| {
+                    // A catalog that claims a page already on the free list is
+                    // the one failure this whole design can produce silently:
+                    // the page is handed out two generations later, written
+                    // over, and the crash that falls back to this generation
+                    // reads a document that is no longer there. Nothing else
+                    // notices. Checked where every run is walked anyway, in the
+                    // builds that can afford three list scans.
+                    if (builtin.is_test or builtin.mode == .Debug) assert_msg(
+                        !self.pager.owns_freed(r.first, r.pages),
+                        "the catalog claims a slab run that is already on the free list",
+                    );
                     try put_u32(gpa, out, r.first);
                     try put_u32(gpa, out, r.pages);
                 }
@@ -2629,6 +2661,8 @@ pub const Engine = struct {
     /// compaction has always used. That is the crash-recovery invariant (PLAN
     /// D6) reduced to an ordering.
     pub fn checkpoint(self: *Engine) !void {
+        try self.checkpoint_lock.lock(self.io);
+        defer self.checkpoint_lock.unlock(self.io);
         try self.commit();
         self.reclaim_slabs();
 
@@ -4584,6 +4618,83 @@ test "a checkpoint runs alongside writers on several collections" {
     try testing.expectEqual(docs_sum, engine.live_docs);
 }
 
+test "checkpoints reclaim under concurrent writers without losing a page" {
+    // Reclamation is now a checkpoint phase, so it runs concurrently with
+    // writers and with a second checkpoint -- two are reachable without
+    // contrivance, since a writer's epilogue can claim the pending flag while
+    // another is inside `compact`, which checkpoints of its own.
+    //
+    // The ownership assertion in `write_catalog` is armed here: no collection
+    // may claim a run the pager has already been given. That is the one failure
+    // this design can produce silently, and it is proven to fire -- have
+    // `reclaim_windows` free the pages and keep the old run list, and this test
+    // panics on it.
+    //
+    // What it does *not* do is reproduce the interleaving `checkpoint_lock`
+    // exists for; removing that lock leaves this green. The lock is there by
+    // argument, and this is the harness that would catch the argument being
+    // wrong. Said plainly rather than left to be assumed.
+    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();
+    engine.compact_threshold = std.math.maxInt(u64); // no rebuild may intervene
+
+    var done = std.atomic.Value(usize).init(1);
+    const Worker = struct {
+        fn writer(e: *Engine, left: *std.atomic.Value(usize), alloc: std.mem.Allocator) error{Canceled}!void {
+            defer _ = left.fetchSub(1, .release);
+            for (0..300) |i| {
+                var doc = make_padded(alloc, @intCast(i), 4000) catch return error.Canceled;
+                defer doc.deinit();
+                {
+                    e.lock_catalog(false) catch return error.Canceled;
+                    defer e.unlock_catalog(false);
+                    const coll = (e.lock_collection("app", "c", true, true) catch
+                        return error.Canceled) orelse return error.Canceled;
+                    defer e.unlock_collection(coll, true);
+                    e.insert("app", "c", &doc, undefined) catch return error.Canceled;
+                    if (i % 3 == 2) _ = e.remove_by_id("app", "c", .{ .int32 = @intCast(i - 1) }) catch
+                        return error.Canceled;
+                }
+                e.commit() catch return error.Canceled;
+            }
+        }
+
+        fn checkpointer(e: *Engine, left: *std.atomic.Value(usize)) error{Canceled}!void {
+            while (left.load(.acquire) > 0) e.checkpoint() catch {};
+        }
+    };
+
+    var group: std.Io.Group = .init;
+    defer group.cancel(io);
+    group.async(io, Worker.writer, .{ &engine, &done, gpa });
+    group.async(io, Worker.checkpointer, .{ &engine, &done });
+    group.async(io, Worker.checkpointer, .{ &engine, &done });
+    try group.await(io);
+
+    // Checkpoints actually happened, and the last one published cleanly: the
+    // watermark the pager loaded is the generation it just wrote.
+    const generation = engine.pager.generation;
+    try testing.expect(generation > 1);
+    try engine.checkpoint();
+    try testing.expectEqual(generation + 1, engine.pager.generation);
+    try testing.expectEqual(generation + 1, engine.pager.loaded.generation);
+
+    // And the accounting the two of them were racing on still adds up.
+    const coll = engine.get_collection("app", "c").?;
+    try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
+    try testing.expectEqual(
+        coll.slab_used - coll.live_bytes,
+        coll.dead_located() + coll.dead_unlocated,
+    );
+}
+
 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

src/pager.zig

@@ -1216,6 +1216,29 @@ pub const Pager = struct {
         try self.free_pending.append(self.gpa, .{ .first = first, .pages = pages });
     }
 
+    /// Whether any page of `[first, first+pages)` has been handed to the free
+    /// list. A consumer that still claims one is claiming a page the pager is
+    /// about to give to somebody else, and the symptom is a document quietly
+    /// overwritten rather than anything failing -- so this is the detector the
+    /// enlarged free list deserves, and it is why the free lists are readable
+    /// from outside at all.
+    ///
+    /// Walks three lists, so it is for assertions in test and Debug builds.
+    pub fn owns_freed(self: *Pager, first: u32, pages: u32) bool {
+        self.alloc_lock.lockUncancelable(self.io);
+        defer self.alloc_lock.unlock(self.io);
+        for ([_][]const Extent{
+            self.free_pending.items,
+            self.free_hold.items,
+            self.free_ready.items,
+        }) |list| {
+            for (list) |e| {
+                if (first < e.first + e.pages and e.first < first + pages) return true;
+            }
+        }
+        return false;
+    }
+
     /// Pages available for immediate reuse.
     pub fn free_ready_pages(self: *const Pager) u32 {
         var n: u32 = 0;