db: a dropped collection is reclaimed, not deadened

`free_collection` charged the engine's `dead_bytes` with the dropped
collection's live bytes, and then, three lines down, handed every page
that collection owned back to the pager. A drop therefore asked for a
rebuild -- a full copy of every collection that was left -- to reclaim
space that had already been reclaimed. Its own garbage was wrong the
other way: bytes that died before the drop stayed on the engine's books
after the pages holding them were freed.

Both halves of that are the same statement: `dead_bytes` is the sum of
`slab_used - live_bytes` over the collections that still exist. Make it
so on the drop path, then stop storing it separately at all --
`read_catalog` recomputes it from the collections the catalog lists, so
the watermark's copy is now a hint for anything inspecting the header
rather than a second source of truth. It would be wrong in one specific
way if it stayed one: a collection dropped after the last checkpoint is
gone from the catalog but still charged for in the hint.

`write_catalog` now returns the dead sum beside the live one and the
checkpoint asserts it the same way, under the same quiescence condition.
That is what makes the accounting checkable rather than merely intended.

Mutation-checked three ways, each red on its own: charge the drop again;
delete the subtraction of the collection's own garbage; delete the
accumulation in `read_catalog`.

163/163 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

@@ -438,9 +438,14 @@ pub const Engine = struct {
                 engine.seq = replay_from;
                 engine.committed_seq = replay_from;
                 engine.live_docs = engine.pager.loaded.live_docs;
-                // Without this a restart forgets its garbage, and a churned
+                // `dead_bytes` is *not* restored from the watermark. It is
-                // database would never compact again.
+                // derived, not stored: `read_catalog` has already summed
-                engine.dead_bytes = engine.pager.loaded.dead_bytes;
+                // `slab_used - live_bytes` over the collections the catalog
+                // still lists. The watermark's copy is a hint for anything
+                // inspecting the header without parsing the catalog, and it
+                // would be wrong here in one specific way -- a collection
+                // dropped after the last checkpoint takes its garbage with it,
+                // and the hint would keep charging the engine for it.
             }
         }
 
@@ -498,7 +503,28 @@ pub const Engine = struct {
         self.dead_docs += coll.doc_count;
         assert_msg(self.live_bytes >= coll.live_bytes, "dropping a collection would underflow the engine's live bytes");
         self.live_bytes -= coll.live_bytes;
-        self.dead_bytes += coll.live_bytes;
+        // A drop *reclaims*, it does not deaden. The loop below hands every page
+        // this collection owned back to the pager, so its live bytes are not
+        // garbage waiting for a rebuild -- they are already gone. Adding them to
+        // `dead_bytes` armed a compaction for space that had just been returned,
+        // and a rebuild costs a full copy of every *other* collection.
+        //
+        // Its garbage goes the other way, for the same reason: the bytes this
+        // collection had already lost to eviction were counted in `dead_bytes`
+        // when they died, and those pages are being freed too. That keeps
+        // `dead_bytes` exactly the sum of `slab_used - live_bytes` over the
+        // collections that still exist, which is what `read_catalog` recomputes
+        // on open and what `write_catalog` asserts.
+        assert_msg(
+            coll.slab_used >= coll.live_bytes,
+            "a collection cannot hold more live bytes than it ever appended",
+        );
+        const coll_dead = coll.slab_used - coll.live_bytes;
+        assert_msg(
+            self.dead_bytes >= coll_dead,
+            "dropping a collection would underflow the engine's dead bytes",
+        );
+        self.dead_bytes -= coll_dead;
         coll.id_index.deinit(self.gpa);
         for (coll.indexes.items) |ix| {
             ix.deinit(self.gpa);
@@ -1643,21 +1669,26 @@ pub const Engine = struct {
     const catalog_magic: u32 = 0x4D464354; // "MFCT"
     const catalog_version: u32 = 1;
 
-    /// Serialize the catalog and return the live-byte total it observed.
+    /// What a catalog walk observed, for the caller to check the engine's own
+    /// running totals against.
+    const CatalogSums = struct { live: u64, dead: u64 };
+
+    /// Serialize the catalog and return the byte totals it observed.
     ///
-    /// The engine's own total is by definition the sum over collections, and
+    /// The engine's own totals are by definition the sums over collections, and
-    /// `read_catalog` rebuilds it that way, so a divergence means some path
+    /// `read_catalog` rebuilds them 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
+    /// anyway, but it cannot be made *here*: the sums are accumulated across
-    /// collections over time while the engine's total moves under it, so a
+    /// collections over time while the engine's totals move under them, 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
+    /// consistent. The caller asserts them after the `seq` check has established
     /// that no writer landed at all.
-    fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !u64 {
+    fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !CatalogSums {
         const gpa = self.gpa;
         var live_sum: u64 = 0;
+        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);
@@ -1688,6 +1719,7 @@ pub const Engine = struct {
                     coll.live_bytes <= coll.slab_used,
                     "a collection cannot hold more live bytes than it ever appended",
                 );
+                dead_sum += coll.slab_used - coll.live_bytes;
                 try put_u32(gpa, out, @intCast(coll.slab_extents.items.len));
                 for (coll.slab_extents.items) |e| {
                     try put_u32(gpa, out, e.first);
@@ -1699,7 +1731,7 @@ pub const Engine = struct {
             }
         }
         try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items));
-        return live_sum;
+        return .{ .live = live_sum, .dead = dead_sum };
     }
 
     fn write_index_catalog(
@@ -1764,9 +1796,14 @@ pub const Engine = struct {
                 coll.slab_end = try r.read_u64();
                 coll.slab_used = try r.read_u64();
                 coll.live_bytes = try r.read_u64();
-                // The engine's total is the sum over collections rather than a
+                // Both engine totals are sums over collections rather than
-                // separately stored field, so the two cannot disagree.
+                // separately stored fields, so neither can disagree with the
+                // catalog. `slab_used - live_bytes` is this collection's slab
+                // garbage by definition -- bytes it appended and no longer
+                // reaches -- which is exactly what the rebuild trigger counts.
+                if (coll.slab_used < coll.live_bytes) return error.CorruptCatalog;
                 self.live_bytes += coll.live_bytes;
+                self.dead_bytes += coll.slab_used - coll.live_bytes;
                 const nex = try r.read_u32();
                 var e: u32 = 0;
                 while (e < nex) : (e += 1) {
@@ -1926,7 +1963,8 @@ pub const Engine = struct {
             try self.catalog_lock.lockShared(self.io);
             const snapshot_seq = self.seq;
             const live_before = self.live_bytes;
-            const live_sum = self.write_catalog(&buf) catch |err| {
+            const dead_before = self.dead_bytes;
+            const sums = self.write_catalog(&buf) catch |err| {
                 self.catalog_lock.unlockShared(self.io);
                 return err;
             };
@@ -1973,9 +2011,16 @@ pub const Engine = struct {
             // 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,
+                sums.live == live_before,
                 "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(
+                sums.dead == dead_before,
+                "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);
             const first = self.pager.alloc_pages(pages) catch |err| {
                 self.log_lock.unlock(self.io);
@@ -2697,8 +2742,9 @@ test "compaction still triggers after a checkpoint has truncated the log" {
     // then never fires and the doc slab grows without bound. This test exists
     // because the churn gate measured exactly that: 4.1x live data.
     //
-    // Second mutation: drop `dead_bytes` from the watermark, or from the restore
+    // Second mutation: drop the `dead_bytes` accumulation from `read_catalog`.
-    // beside `live_docs` in `open`. Red on the reopened engine below.
+    // Red in "reopen carries the garbage counter across a restart", which is
+    // where the counter has to survive a restart.
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();
     var env = test_env(&threaded);
@@ -2807,6 +2853,14 @@ test "a rebuild leaves the space it reclaimed ready to reuse" {
 }
 
 test "reopen carries the garbage counter across a restart" {
+    // It carries it by *recomputing* it: `read_catalog` sums
+    // `slab_used - live_bytes` over the collections the catalog lists, rather
+    // than trusting the watermark's cached copy. That is a stronger claim than
+    // the hint was -- a collection dropped since the last checkpoint is simply
+    // not in the sum, where the hint kept charging the engine for it.
+    //
+    // Mutation: drop the accumulation in `read_catalog`; `engine2.dead_bytes`
+    // reads zero below.
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();
     var env = test_env(&threaded);
@@ -2844,6 +2898,87 @@ test "reopen carries the garbage counter across a restart" {
     defer engine2.deinit();
     try testing.expectEqual(dead_before, engine2.dead_bytes);
     try testing.expectEqual(live_before, engine2.live_bytes);
+    // And it is the sum over collections on both sides of the restart, not a
+    // number kept beside them.
+    const reopened = engine2.get_collection("app", "c").?;
+    try testing.expectEqual(reopened.slab_used - reopened.live_bytes, engine2.dead_bytes);
+}
+
+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
+    // the pager on the following line. A drop of a large collection therefore
+    // asked for a rebuild -- a full copy of every collection that was left --
+    // to reclaim space that had already been reclaimed. Its own garbage was
+    // wrong the other way: it stayed on the engine's books after the pages
+    // holding it were gone.
+    //
+    // Mutation: restore `self.dead_bytes += coll.live_bytes;`, or delete the
+    // subtraction of `coll.slab_used - coll.live_bytes`. Either one is red on
+    // the equality below, and the first also re-arms the trigger.
+    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 while setting up
+    try engine.lock();
+    defer engine.unlock();
+
+    // A small collection that survives, and a large one that does not. Both
+    // hold garbage, so the drop has to keep one collection's and discard the
+    // other's.
+    for (0..40) |i| {
+        var d = try make_doc(gpa, @intCast(i), "x");
+        defer d.deinit();
+        try engine.insert("app", "keep", &d, &env.gen);
+    }
+    for (0..200) |i| {
+        var d = try make_doc(gpa, @intCast(i), "x");
+        defer d.deinit();
+        try engine.insert("app", "gone", &d, &env.gen);
+    }
+    for (0..10) |i| {
+        var d = try make_doc(gpa, @intCast(i), "yy");
+        defer d.deinit();
+        _ = try engine.replace("app", "keep", &d, &env.gen);
+    }
+    for (0..200) |i| {
+        var d = try make_doc(gpa, @intCast(i), "yy");
+        defer d.deinit();
+        _ = try engine.replace("app", "gone", &d, &env.gen);
+    }
+    try engine.commit();
+
+    const keep = engine.get_collection("app", "keep").?;
+    const keep_dead = keep.slab_used - keep.live_bytes;
+    const keep_live = keep.live_bytes;
+    try testing.expect(keep_dead > 0);
+    try testing.expect(engine.dead_bytes > keep_dead);
+
+    // Arm the trigger for what the two of them hold between them.
+    engine.compact_threshold = keep_dead + 1;
+    engine.note_compact();
+    try testing.expect(engine.take_compact());
+
+    try testing.expect(try engine.drop_collection("app", "gone"));
+    try testing.expectEqual(keep_dead, engine.dead_bytes);
+    try testing.expectEqual(keep_live, engine.live_bytes);
+
+    // The next write reconsiders the trigger and finds nothing worth a rebuild:
+    // what the drop reclaimed is not garbage, it is free.
+    engine.note_compact();
+    try testing.expect(!engine.take_compact());
+
+    // The catalog agrees, which is what the reopened engine will read.
+    try engine.checkpoint();
+    try testing.expectEqual(keep_dead, engine.dead_bytes);
 }
 
 test "reopen replays log" {

src/pager.zig

@@ -27,7 +27,8 @@
 //!     [56..64)    u64 freelist_len
 //!     [64..72)    u64 prev_generation -- kept intact for fallback
 //!     [72..80)    u64 live_docs    -- cached hint
-//!     [80..88)    u64 dead_bytes   -- cached, drives the rebuild trigger
+//!     [80..88)    u64 dead_bytes   -- cached hint; the engine recomputes it
+//!                                     from the catalog on open
 //!     [88..4088)  reserved (zero)
 //!     [4088..4096) u64 xxhash3 over [0..4088)
 //!   pages 3..     data, handed out by a tail-bump extent allocator