db: the slab counts what the appender skips

`slab_used` only ever grew by a document's length, so the two places the
appender writes slab off went uncounted: the gap left when a checkpoint
freezes the page the cursor points into and the cursor resumes at the next
system page, and the tail of an extent abandoned for a document that no
longer fits. Both are real garbage -- only a rebuild gets them back -- and
both were invisible to the trigger that decides whether a rebuild is worth
doing. An abandoned tail can be most of 8 MiB.

`note_skip` counts them into `slab_used` where they happen and hands the
number back for the caller to charge to the engine, which keeps the identity
the last commit established: `dead_bytes` is the sum of
`slab_used - live_bytes` over the collections that exist.

That identity is also why `compact` no longer zeroes `dead_bytes`. A repack
appends through the same slab, so it abandons a tail of its own whenever the
next document does not fit; zeroing was true only if a rebuild leaves nothing
behind, and it does not. `sum_dead_bytes` recomputes it from the collections,
each under its own lock.

Carried with it, because this commit is what exposed it: the four engine
counters get a lock of their own. They are the only engine-wide mutable state
a writer touches while holding nothing but its own collection's lock, so two
writers on different collections reach them with no lock in common -- and the
checkpoint's consistency check read them ordered against nothing, while the
per-collection figures it compares them to were read under each collection's
lock. Before this commit `dead_bytes` moved on nearly every write, so that
check was skipped almost every time; with skips counted it stands still
between checkpoints, the check runs, and it aborted three of eight ReleaseSafe
runs of "a checkpoint runs alongside writers on several collections".

Not mutation-checked, and worth saying so: reverting the lock did not
re-trigger the abort in 24 further runs, and neither did the exact pre-fix
revision in 10. The rate depends on machine load, and a mutation check that
cannot be relied on to go red is not a check. The lock stands on inspection
instead -- an unsynchronized read-modify-write on a counter shared by threads
holding no common lock is a defect whatever its rate -- and the concurrency
test now asserts the identity once everything is quiet, which is the half of
it that does not depend on a race being caught in the act.

Mutation-checked, each red on its own: drop either `note_skip` call in
`slab_reserve`; put `self.dead_bytes = 0;` back in `compact`. The `note_skip`
in `slab_append` is covered by the concurrency test, the only place a publish
lands between a reservation and its append.

165/165 unit tests in ReleaseFast and ReleaseSafe, 82/82 fuzz, e2e 49,
e2e2 concurrent 2 + crash pair, e2e3 16, e2e4 17, e2e6 72, e2e7 86,
crash-fuzz 60 cycles.

src/db.zig

@@ -155,7 +155,10 @@ pub const Collection = struct {
     /// record is durable, where failure has nowhere to go: the write is already
     /// committed and reporting an error for it would be a lie the next open
     /// contradicts. Reserving first keeps the fallible half before the log.
-    fn slab_reserve(self: *Collection, gpa: std.mem.Allocator, len: usize) !void {
+    ///
+    /// Returns the slab bytes it wrote off along the way, for the caller to
+    /// charge to the engine's dead total. See `note_skip`.
+    fn slab_reserve(self: *Collection, gpa: std.mem.Allocator, len: usize) !u64 {
         // A checkpoint can land in the middle of an extent, which freezes the
         // page the tail points into. Appending there would store inside the
         // durable image, so abandon the rest of the extent and start a fresh
@@ -171,7 +174,7 @@ pub const Collection = struct {
         // fit. Without that the append's own re-check could discover it needs a
         // fresh extent, which is fallible, after the log record is already
         // durable. Costs under one system page per extent.
-        if (self.pager.is_unpublished_at(self.slab_tail) and self.appendable_end(len)) return;
+        if (self.pager.is_unpublished_at(self.slab_tail) and self.appendable_end(len)) return 0;
         // The page holding the tail is frozen, but the *rest* of the extent is
         // not: nothing above the live cursor is referenced by the image or by an
         // index. So skip to the next system page and keep the extent, instead of
@@ -187,10 +190,18 @@ pub const Collection = struct {
         // real data.
         if (self.appendable_end(len)) {
             const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align);
+            const skipped = self.note_skip(resumed - self.slab_tail);
             self.pager.mark_appendable(resumed, self.slab_end);
             self.slab_tail = resumed;
-            return;
+            return skipped;
         }
+        // Nothing will ever be written between the cursor and the end of the
+        // extent this collection is walking away from -- the extent stays
+        // allocated to it and every byte above the cursor is unreachable. That
+        // is garbage, in the whole 8 MiB, and the reservation is the only place
+        // that knows about it.
+        assert_msg(self.slab_tail <= self.slab_end, "the slab cursor is past the end of its extent");
+        const skipped = self.note_skip(self.slab_end - self.slab_tail);
         // A document larger than the standard extent gets one of its own; BSON
         // reaches 16 MB and the extent is 8 MiB.
         const want_pages: u32 = @intCast(@max(
@@ -202,6 +213,26 @@ pub const Collection = struct {
         try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages });
         self.slab_tail = @as(u64, first) << pgr.page_shift;
         self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
+        return skipped;
+    }
+
+    /// Count `bytes` of slab that no document will ever occupy, and hand the
+    /// same number back so the caller can charge the engine's dead total.
+    ///
+    /// The appender skips slab in two places -- rounding the cursor up to a
+    /// system page after a checkpoint froze the page it pointed into, and
+    /// abandoning the tail of an extent that no longer has room. Neither used to
+    /// be counted anywhere: not in `slab_used`, which only ever grew by a
+    /// document's length, and so not in the engine's `dead_bytes` either. It is
+    /// garbage all the same -- only a rebuild gets it back -- and it was
+    /// invisible to the trigger that decides whether a rebuild is worth doing.
+    ///
+    /// Two collections churning against a checkpoint every 32 MiB skip up to a
+    /// system page each per checkpoint, and an abandoned extent tail can be
+    /// most of 8 MiB. Counted here, that garbage arms compaction like any other.
+    fn note_skip(self: *Collection, bytes: u64) u64 {
+        self.slab_used += bytes;
+        return bytes;
     }
 
     /// Whether a document of `len` bytes fits in this extent even if the cursor
@@ -211,9 +242,13 @@ pub const Collection = struct {
         return std.mem.alignForward(u64, self.slab_tail, pgr.map_align) + len <= self.slab_end;
     }
 
+    /// Where a document landed, and what the landing cost besides its own
+    /// length. `skipped` is `note_skip`'s tally for this append.
+    const Appended = struct { off: u64, skipped: u64 };
+
     /// Copy `bytes` into the slab and return its absolute file offset.
     /// Infallible: slab_reserve must have run for at least this many bytes.
-    fn slab_append(self: *Collection, bytes: []const u8) u64 {
+    fn slab_append(self: *Collection, bytes: []const u8) Appended {
         // The cursor was checked in `slab_reserve`, but a checkpoint can have
         // published since -- the reservation runs before the log append and this
         // runs after it, with an fsync in between. `publish` clears the whole
@@ -227,8 +262,10 @@ pub const Collection = struct {
         // publish for the rest of this function, so the answer stays true.
         self.pager.lock_append();
         defer self.pager.unlock_append();
+        var skipped: u64 = 0;
         if (!self.pager.is_unpublished_at(self.slab_tail)) {
             const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align);
+            skipped = self.note_skip(resumed - self.slab_tail);
             self.pager.mark_appendable(resumed, self.slab_end);
             self.slab_tail = resumed;
         }
@@ -243,7 +280,7 @@ pub const Collection = struct {
         // Here rather than at the call site: a rebuild appends through this same
         // path, and its copies are live by definition.
         self.live_bytes += bytes.len;
-        return off;
+        return .{ .off = off, .skipped = skipped };
     }
 
     /// The canonical bytes of the document stored at `off` — a slice into a
@@ -352,6 +389,24 @@ pub const Engine = struct {
     /// 16 KiB documents and one of 40 B documents look identical.
     live_bytes: u64 = 0,
     dead_bytes: u64 = 0,
+    /// Guards the four counters above -- `live_docs`, `dead_docs`,
+    /// `live_bytes`, `dead_bytes` -- and nothing else.
+    ///
+    /// They are the only engine-wide mutable state a writer touches while
+    /// holding nothing but its own collection's lock, so two writers on
+    /// different collections reach them with no lock in common. The lost update
+    /// that allows is the smaller half of the problem. The larger half is that a
+    /// reader had no way to see them consistently with the per-collection totals
+    /// they are supposed to equal: `checkpoint` reads each collection's counters
+    /// under that collection's lock -- which orders it against that
+    /// collection's writer -- and then read these with no lock at all, so it
+    /// could see a total that predated a write it had just serialized. Its own
+    /// assertion then aborted the server, correctly, about a database that was
+    /// consistent.
+    ///
+    /// 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,
     /// 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.
@@ -489,10 +544,40 @@ pub const Engine = struct {
         self.log.close();
     }
 
+    /// Slab a reservation wrote off, on the engine's books. Its own acquisition
+    /// rather than the reservation's caller adding to the field: skips cluster
+    /// at a checkpoint -- `publish` freezes every collection's append cursor at
+    /// once, so the next write to each of them skips -- which is precisely when
+    /// several writers reach this counter at the same moment.
+    fn count_slab_skip(self: *Engine, bytes: u64) void {
+        if (bytes == 0) return;
+        self.counter_lock.lockUncancelable(self.io);
+        defer self.counter_lock.unlock(self.io);
+        self.dead_bytes += bytes;
+    }
+
+    /// A snapshot of the four counters, taken together. Both readers reason
+    /// about a *relation* -- the compaction trigger about the ratio of two of
+    /// them, the checkpoint about how they compare to the sum over collections
+    /// -- so reading them one at a time would be comparing two moments.
+    const Counters = struct { live_docs: u64, dead_docs: u64, live_bytes: u64, dead_bytes: u64 };
+
+    fn counters(self: *Engine) Counters {
+        self.counter_lock.lockUncancelable(self.io);
+        defer self.counter_lock.unlock(self.io);
+        return .{
+            .live_docs = self.live_docs,
+            .dead_docs = self.dead_docs,
+            .live_bytes = self.live_bytes,
+            .dead_bytes = self.dead_bytes,
+        };
+    }
+
     /// Free every document in a collection along with its owned _id keys
     /// and secondary indexes (whose entries alias the documents — freed
     /// first).
     fn free_collection(self: *Engine, coll: *Collection) void {
+        self.counter_lock.lockUncancelable(self.io);
         // Dropping a collection turns all of its records into garbage. The
         // engine's live count includes every collection's documents, so it can
         // never be smaller than this one's -- and a u64 underflow here would
@@ -525,6 +610,7 @@ pub const Engine = struct {
             "dropping a collection would underflow the engine's dead bytes",
         );
         self.dead_bytes -= coll_dead;
+        self.counter_lock.unlock(self.io);
         coll.id_index.deinit(self.gpa);
         for (coll.indexes.items) |ix| {
             ix.deinit(self.gpa);
@@ -575,10 +661,21 @@ pub const Engine = struct {
         for (coll.indexes.items) |ix| self.pager.release_reservation(&ix.hold);
     }
 
+    /// A document becoming live: its bytes into the slab, and every engine
+    /// counter that describes. The document count moved here from the two call
+    /// sites so that one document costs one acquisition of `counter_lock` and
+    /// leaves the totals agreeing at every moment a reader could look.
     fn publish_doc_bytes(self: *Engine, coll: *Collection, bytes: []const u8) u64 {
-        const off = coll.slab_append(bytes);
+        const appended = coll.slab_append(bytes);
+        self.counter_lock.lockUncancelable(self.io);
+        defer self.counter_lock.unlock(self.io);
         self.live_bytes += bytes.len;
-        return off;
+        self.live_docs += 1;
+        // Slab the append had to write off. Charged here, under the same
+        // collection lock the append ran under, so a checkpoint's catalog walk
+        // never sees the collection's total moved and the engine's not.
+        self.dead_bytes += appended.skipped;
+        return appended.off;
     }
 
     fn evict_doc(self: *Engine, coll: *Collection, id_enc: []const u8) void {
@@ -589,12 +686,16 @@ pub const Engine = struct {
         coll.id_index.remove_doc(self.gpa, old_bytes, off);
         for (coll.indexes.items) |ix| ix.remove_doc(self.gpa, old_bytes, off);
         // This document's log record (and its slab bytes) just became garbage.
-        assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count");
+        {
-        self.live_docs -= 1;
+            self.counter_lock.lockUncancelable(self.io);
-        self.dead_docs += 1;
+            defer self.counter_lock.unlock(self.io);
-        assert_msg(self.live_bytes >= old_bytes.len, "evicting a document would underflow the engine's live bytes");
+            assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count");
-        self.live_bytes -= old_bytes.len;
+            self.live_docs -= 1;
-        self.dead_bytes += old_bytes.len;
+            self.dead_docs += 1;
+            assert_msg(self.live_bytes >= old_bytes.len, "evicting a document would underflow the engine's live bytes");
+            self.live_bytes -= old_bytes.len;
+            self.dead_bytes += old_bytes.len;
+        }
         assert_msg(coll.doc_count >= 1, "evicting a document would underflow the collection's count");
         coll.doc_count -= 1;
         assert_msg(coll.live_bytes >= old_bytes.len, "evicting a document would underflow the collection's live bytes");
@@ -951,7 +1052,7 @@ pub const Engine = struct {
         for (built_list.items) |*b| {
             try b.ix.reserve_for(self.gpa, b.built.entries.items);
         }
-        try coll.slab_reserve(self.gpa, doc_bytes.len);
+        self.count_slab_skip(try coll.slab_reserve(self.gpa, doc_bytes.len));
 
         // 5. Log (and sync) before anything becomes visible. The append
         //    takes the log lock; durability (fsync) is the command's commit.
@@ -963,7 +1064,6 @@ pub const Engine = struct {
         // 7. Publish the document and its entries: copy the bytes into the
         //    slab and record the offset. Infallible from here.
         const off = self.publish_doc_bytes(coll, doc_bytes);
-        self.live_docs += 1;
         coll.doc_count += 1;
         for (built_list.items) |*b| {
             if (b.built.multikey) b.ix.multikey = true;
@@ -1359,15 +1459,20 @@ pub const Engine = struct {
         // the live data, with a trigger that had been dead since the log started
         // being reclaimed.
         //
+        // One snapshot, because the second gate is a ratio: reading the two
+        // totals separately compares a live figure from one moment against a
+        // dead figure from another, and under concurrent writers that is how a
+        // rebuild fires on a database that does not want one.
+        const c = self.counters();
         // Absolute volume first: a rewrite costs a full copy of the live data,
         // so it is not worth doing for a few kilobytes however bad the ratio.
-        if (self.dead_bytes < self.compact_threshold) return;
+        if (c.dead_bytes < self.compact_threshold) return;
         // Then the share, dead / (live + dead), firing at ~20%: the file stays
         // near 1.25x the live data and each rebuild is paid for by the space it
         // reclaims. Bytes rather than document counts, because a rewrite copies
         // bytes -- 100k evicted 40 B documents are not worth the same rebuild as
         // 100k evicted 16 KiB ones.
-        if (self.dead_bytes * 4 < self.live_bytes) return;
+        if (c.dead_bytes * 4 < c.live_bytes) return;
         self.compact_pending.store(true, .release);
     }
 
@@ -1433,12 +1538,21 @@ pub const Engine = struct {
                 };
             }
         }
+        // Before releasing the catalog, and whether the walk finished or gave up
+        // part way: a rebuild resets a collection's totals, so an incremental
+        // `dead_bytes` now describes collections that no longer exist in that
+        // shape. It used to be zeroed here, which was true only if a repack
+        // leaves nothing behind -- it does not. A checkpoint landing mid-rebuild
+        // forces the copy's own append cursor up to a system page, and those
+        // skipped bytes are as dead as the ones being reclaimed.
+        const dead_after = self.sum_dead_bytes();
         self.catalog_lock.unlockShared(self.io);
+        self.counter_lock.lockUncancelable(self.io);
+        self.dead_bytes = dead_after;
+        self.dead_docs = 0;
+        self.counter_lock.unlock(self.io);
         if (rebuild_err) |err| return err;
 
-        self.dead_docs = 0;
-        // Every collection's slab was just repacked to hold only live bytes.
-        self.dead_bytes = 0;
         // Publish the rebuilt layout, which is also what reclaims the log. Until
         // this lands the old pages are still referenced by the previous
         // watermark, so a crash mid-rebuild simply loses the rebuild.
@@ -1463,6 +1577,7 @@ pub const Engine = struct {
     /// Documents and indexes have to move together: an index leaf holds a
     /// physical offset, so a document that moves without its indexes being
     /// rebuilt is a stale entry pointing at whatever now occupies those bytes.
+    ///
     fn rebuild_collection(self: *Engine, coll: *Collection) !void {
         try coll.lock.lock(self.io);
         defer coll.lock.unlock(self.io);
@@ -1489,10 +1604,13 @@ pub const Engine = struct {
         var it = coll.id_index.iter();
         while (it.next()) |entry| {
             const bytes = doc_bytes_in(self.pager, entry.off);
-            try coll.slab_reserve(self.gpa, bytes.len);
+            // The skips are dropped rather than charged: this collection's
-            const new_off = coll.slab_append(bytes);
+            // totals were reset above and `compact` recomputes the engine's from
+            // what the rebuild leaves behind.
+            _ = try coll.slab_reserve(self.gpa, bytes.len);
+            const appended = coll.slab_append(bytes);
             self.pager.release_reservation(&coll.hold);
-            try moved.append(self.gpa, .{ .off = new_off });
+            try moved.append(self.gpa, .{ .off = appended.off });
         }
         // Republish the offsets.
 
@@ -1510,6 +1628,34 @@ pub const Engine = struct {
         coll.layout_epoch = self.layout_epoch_seq;
     }
 
+    /// The engine's dead-byte total, recomputed from the collections that
+    /// exist. Everywhere else the counter is incremental; a rebuild is the one
+    /// place that has to reset it, and the answer after a rebuild is not zero --
+    /// the copying skips slab of its own whenever a checkpoint lands mid-walk.
+    ///
+    /// Each collection is read under its own lock, catalog then collection: the
+    /// order `compact` and `write_catalog` both use. Uncancelable, because the
+    /// caller has already rewritten the collections and the counter describing
+    /// them cannot be left behind.
+    fn sum_dead_bytes(self: *Engine) u64 {
+        var sum: u64 = 0;
+        var db_it = self.dbs.iterator();
+        while (db_it.next()) |db_entry| {
+            var coll_it = db_entry.value_ptr.collections.iterator();
+            while (coll_it.next()) |ce| {
+                const coll = ce.value_ptr.*;
+                coll.lock.lockSharedUncancelable(self.io);
+                defer coll.lock.unlockShared(self.io);
+                assert_msg(
+                    coll.slab_used >= coll.live_bytes,
+                    "a collection cannot hold more live bytes than it ever appended",
+                );
+                sum += coll.slab_used - coll.live_bytes;
+            }
+        }
+        return sum;
+    }
+
     fn repack_index(
         self: *Engine,
         coll: *Collection,
@@ -1691,7 +1837,7 @@ pub const Engine = struct {
         var dead_sum: u64 = 0;
         try put_u32(gpa, out, catalog_magic);
         try put_u32(gpa, out, catalog_version);
-        try put_u64(gpa, out, self.live_docs);
+        try put_u64(gpa, out, self.counters().live_docs);
         try put_u32(gpa, out, @intCast(self.dbs.count()));
         var db_it = self.dbs.iterator();
         while (db_it.next()) |db_entry| {
@@ -1962,8 +2108,7 @@ pub const Engine = struct {
             buf.clearRetainingCapacity();
             try self.catalog_lock.lockShared(self.io);
             const snapshot_seq = self.seq;
-            const live_before = self.live_bytes;
+            const before = self.counters();
-            const dead_before = self.dead_bytes;
             const sums = self.write_catalog(&buf) catch |err| {
                 self.catalog_lock.unlockShared(self.io);
                 return err;
@@ -2010,15 +2155,24 @@ pub const Engine = struct {
             // 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(
+            //
-                sums.live == live_before,
+            // Both reads go through `counters`, which is the only way the
+            // comparison means anything: the sums were gathered under each
+            // collection's lock, which orders them against that collection's
+            // writer, and an unsynchronized read of the engine's own totals is
+            // ordered against nothing at all -- so it could return a figure from
+            // before a write the walk had already serialized, and the assertion
+            // would abort a server whose accounting was correct.
+            const after = self.counters();
+            if (after.live_bytes == before.live_bytes) assert_msg(
+                sums.live == before.live_bytes,
                 "the engine's live-byte total must equal the sum over collections",
             );
             // The same argument, for the total the rebuild trigger reads. This
             // is what makes a drop's accounting checkable: charge the engine for
             // a dropped collection's bytes and the two sides part company here.
-            if (self.dead_bytes == dead_before) assert_msg(
+            if (after.dead_bytes == before.dead_bytes) assert_msg(
-                sums.dead == dead_before,
+                sums.dead == before.dead_bytes,
                 "the engine's dead-byte total must equal the sum over collections",
             );
             const pages: u32 = @intCast((buf.items.len + pgr.page_size - 1) / pgr.page_size);
@@ -2037,8 +2191,8 @@ pub const Engine = struct {
                 .seq = snapshot_seq,
                 .catalog_page = first,
                 .catalog_len = buf.items.len,
-                .live_docs = self.live_docs,
+                .live_docs = after.live_docs,
-                .dead_bytes = self.dead_bytes,
+                .dead_bytes = after.dead_bytes,
             }) catch |err| {
                 self.log_lock.unlock(self.io);
                 return err;
@@ -2195,9 +2349,8 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
             self.evict_doc(coll, id_enc);
             const doc_bytes = try serialize_doc(self.gpa, doc);
             defer self.gpa.free(doc_bytes);
-            try coll.slab_reserve(self.gpa, doc_bytes.len);
+            self.count_slab_skip(try coll.slab_reserve(self.gpa, doc_bytes.len));
             const off = self.publish_doc_bytes(coll, doc_bytes);
-            self.live_docs += 1;
             coll.doc_count += 1;
             // The `_id_` entry is added *now*, not after replay: it is the only
             // way the next record can find this document to supersede it. The
@@ -2787,7 +2940,14 @@ test "compaction still triggers after a checkpoint has truncated the log" {
     try engine.commit();
 
     try testing.expectEqual(live_after_load + 200, engine.live_bytes);
-    try testing.expectEqual(live_after_load, engine.dead_bytes);
+    // Every superseded document, plus what the first replace after the
+    // checkpoint had to skip: that checkpoint froze the page the append cursor
+    // pointed into, so the cursor moved up to the next system page and the gap
+    // it stepped over is garbage too. Both are in the same total, which is the
+    // point -- the trigger reads one number.
+    const superseded = live_after_load;
+    try testing.expect(engine.dead_bytes >= superseded);
+    try testing.expect(engine.dead_bytes - superseded < pgr.map_align);
     try testing.expect(engine.dead_bytes >= engine.compact_threshold);
     try testing.expect(engine.take_compact());
 
@@ -2904,6 +3064,136 @@ test "reopen carries the garbage counter across a restart" {
     try testing.expectEqual(reopened.slab_used - reopened.live_bytes, engine2.dead_bytes);
 }
 
+test "the slab counts what the appender skips" {
+    // `slab_used` only ever grew by a document's length, so the two places the
+    // appender writes slab off went uncounted: the gap left when a checkpoint
+    // freezes the page the cursor points into and the cursor moves up to the
+    // next system page, and the tail of an extent abandoned for a document that
+    // no longer fits. Real garbage -- only a rebuild gets it back -- and
+    // invisible to the trigger that decides whether a rebuild is worth doing.
+    //
+    // Mutation: drop either `note_skip` call in `slab_reserve`. Red on the
+    // matching half below. Dropping the one in `slab_append` is not covered
+    // here: it needs a publish between the reservation and the append, which is
+    // what "a checkpoint runs alongside writers on several collections"
+    // arranges, and the accounting assertion in `checkpoint` is what catches it.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const io = env.io;
+    const gpa = testing.allocator;
+
+    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 mid-test
+    try engine.lock();
+    defer engine.unlock();
+
+    for (0..200) |i| {
+        var d = try make_doc(gpa, @intCast(i), "x");
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    const coll = engine.get_collection("app", "c").?;
+    // Nothing is dead yet: every document inserted is still live and the slab
+    // has been walked straight through.
+    try testing.expectEqual(@as(u64, 0), engine.dead_bytes);
+    try testing.expectEqual(coll.slab_used, coll.live_bytes);
+
+    // 1. The round-up. The checkpoint freezes the page the cursor is in, so the
+    //    next append resumes at the next system page and the bytes in between
+    //    are never written.
+    try engine.checkpoint();
+    const tail_before = coll.slab_tail;
+    const gap = std.mem.alignForward(u64, tail_before, pgr.map_align) - tail_before;
+    try testing.expect(gap > 0);
+    {
+        var d = try make_doc(gpa, 1000, "x");
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    try testing.expectEqual(gap, engine.dead_bytes);
+    try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
+
+    // 2. The abandoned tail. A document bigger than the standard extent takes
+    //    one of its own, and everything left in the extent being walked away
+    //    from is unreachable -- the extent stays allocated to this collection.
+    const abandoned = coll.slab_end - coll.slab_tail;
+    try testing.expect(abandoned > 4 * 1024 * 1024);
+    {
+        const big = try gpa.alloc(u8, 9 * 1024 * 1024);
+        defer gpa.free(big);
+        @memset(big, 'z');
+        var d = try make_doc(gpa, 1001, big);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    try testing.expectEqual(gap + abandoned, engine.dead_bytes);
+    try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
+
+    // And it survives the round trip, because it is in `slab_used`.
+    try engine.checkpoint();
+    try testing.expectEqual(gap + abandoned, engine.dead_bytes);
+}
+
+test "a rebuild leaves behind what its own copying skipped" {
+    // `compact` used to set `dead_bytes = 0`, on the reasoning that a repack
+    // holds only live bytes. It does not: the repack appends through the same
+    // slab, so it abandons an extent tail whenever the next document no longer
+    // fits. Zeroing the counter there broke the identity the checkpoint asserts
+    // -- and understated the garbage, so a collection that fragments on every
+    // rebuild would never be rebuilt again.
+    //
+    // Mutation: put `self.dead_bytes = 0;` back. Red below, and the checkpoint
+    // inside `compact` aborts on the accounting assertion before it gets there.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const io = env.io;
+    const gpa = testing.allocator;
+
+    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); // rebuild only when told to
+    try engine.lock();
+    defer engine.unlock();
+
+    // Documents big enough that two of them do not fit in one 8 MiB extent, so
+    // the copying itself has to abandon a tail.
+    const big = try gpa.alloc(u8, 5 * 1024 * 1024);
+    defer gpa.free(big);
+    @memset(big, 'z');
+    for (0..3) |i| {
+        var d = try make_doc(gpa, @intCast(i), big);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+
+    // One of them becomes garbage, which is what the rebuild is for.
+    try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = 1 }));
+    try engine.commit();
+
+    try engine.compact();
+
+    const coll = engine.get_collection("app", "c").?;
+    try testing.expectEqual(@as(u64, 2), coll.doc_count);
+    // The deleted document is gone from the slab, but the gap the copy left
+    // between the two survivors is not -- and the engine says so.
+    try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
+    try testing.expect(engine.dead_bytes > 2 * 1024 * 1024);
+    try testing.expect(engine.dead_bytes < 4 * 1024 * 1024);
+}
+
 test "dropping a collection does not arm compaction" {
     // `free_collection` charged the engine's `dead_bytes` with the dropped
     // collection's *live* bytes, having just handed every page it owned back to
@@ -3146,14 +3436,6 @@ test "a checkpoint runs alongside writers on several collections" {
                 var doc = make_doc(alloc, @intCast(i), "user") catch return error.Canceled;
                 defer doc.deinit();
                 {
-                    // The server's discipline, not the legacy whole-engine
-                    // lock: catalog shared, then the target collection
-                    // exclusive (commands.zig dispatch). `write_catalog` takes
-                    // the same two in the same order, and that is the whole
-                    // reason its walk of a collection's counters and extents is
-                    // safe -- a writer that skipped the collection lock would
-                    // not be excluded by it, and the test would be checking
-                    // nothing.
                     e.lock_catalog(false) catch return error.Canceled;
                     defer e.unlock_catalog(false);
                     const coll = (e.lock_collection("app", name, true, true) catch
@@ -3190,10 +3472,25 @@ test "a checkpoint runs alongside writers on several collections" {
     try engine.checkpoint();
     try engine.lock_read();
     defer engine.unlock_read();
+    var live_sum: u64 = 0;
+    var dead_sum: u64 = 0;
+    var docs_sum: u64 = 0;
     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());
+        live_sum += coll.live_bytes;
+        dead_sum += coll.slab_used - coll.live_bytes;
+        docs_sum += coll.doc_count;
     }
+    // The engine's counters are the sums over collections, checked once
+    // everything is quiet rather than left to `checkpoint`'s own assertion --
+    // which only runs on a checkpoint that happened to fall in a gap between
+    // writes, so under sustained load it can go a whole run without firing.
+    // Every writer here held nothing but its own collection's lock while moving
+    // these, so a lost update lands exactly as a mismatch on one of these three.
+    try testing.expectEqual(live_sum, engine.live_bytes);
+    try testing.expectEqual(dead_sum, engine.dead_bytes);
+    try testing.expectEqual(docs_sum, engine.live_docs);
 }
 
 test "concurrent readers and writers on a threaded Io" {