db: a rebuild copies only the collections that have garbage

`compact` walked every collection unconditionally, so garbage in one paid for
a full copy of the other thirty-nine. A copy is not free even where it
reclaims nothing: it rewrites every document and every index, and it bumps the
layout epoch, which kills every open cursor on a collection that had no reason
to be touched.

The gate is the share `note_compact` already applies to the engine's totals,
and using the same one is what keeps them from disagreeing. If no collection
passes it then `dead_i < live_i / 4` for every one, so `sum(dead) <
sum(live) / 4` and the engine's trigger could not have fired either -- a
compaction that runs always rebuilds at least one collection and cannot spin
re-arming itself over garbage no rebuild will take. An absolute floor per
collection would break exactly that: forty collections each under the floor
can sum to well over it.

Two existing tests needed real garbage, which is the change working. "a
rebuild kills an offsets cursor and spares a streaming one" and "the epochs
that invalidate a cursor move exactly when they must" both called `compact` on
a clean collection and relied on it rewriting anyway. The first now rewrites
all 60 documents (a replace, not a delete, so the drain still checks that the
stream yields exactly 60 once each -- and with a changed field, since an
identical replace is deliberately not a write).

The new test asserts both halves: the dirty collection's epoch moves, the
clean one's does not, and the clean one's `slab_used` is unchanged -- a repack
starts the slab over, so that number could not survive one.

Mutation: delete the guard, and the clean collection's epoch moves too.

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

src/commands.zig

@@ -4291,7 +4291,9 @@ test "a rebuild kills an offsets cursor and spares a streaming one" {
     //
     // A rebuild is triggered directly rather than through churn, because whether
     // churn crosses the compaction threshold is not something a test should have
-    // to guess at.
+    // to guess at. The garbage below is still needed: `compact` now skips a
+    // collection with nothing to reclaim, so a clean one is not rewritten at
+    // all and there would be no rebuild to observe.
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();
     const io = threaded.io();
@@ -4315,6 +4317,25 @@ test "a rebuild kills an offsets cursor and spares a streaming one" {
     try testing.expectEqual(@as(i32, 0), try dispatch_get_more(&ctx, "c", walk.id));
     try testing.expectEqual(@as(i32, 0), try dispatch_get_more(&ctx, "c", narrowed.id));
 
+    // Rewrite every document, so the collection has as many dead bytes as live
+    // ones and is worth rebuilding. A replace rather than a delete: the count
+    // stays at 60, which is what the drain below checks.
+    try ctx.engine.lock();
+    var again: i32 = 1;
+    while (again <= 60) : (again += 1) {
+        const pairs = try testing.allocator.alloc(bson.Pair, 3);
+        defer testing.allocator.free(pairs);
+        pairs[0] = .{ .key = "_id", .value = .{ .int32 = again } };
+        // A different value: an identical replace is deliberately not a write.
+        pairs[1] = .{ .key = "a", .value = .{ .int32 = @mod(again, 5) + 100 } };
+        pairs[2] = .{ .key = "pad", .value = .{ .string = seed_pad } };
+        var doc: bson.Document = .{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = pairs };
+        defer doc.arena.deinit();
+        _ = try ctx.engine.replace("test", "c", &doc, ctx.oid_gen);
+    }
+    try ctx.engine.commit();
+    ctx.engine.unlock();
+
     try ctx.engine.compact();
 
     // The stream remembers key bytes, which a repack does not change.

src/db.zig

@@ -1930,6 +1930,32 @@ pub const Engine = struct {
         try self.checkpoint();
     }
 
+    /// Whether rewriting this collection would pay for itself. The caller holds
+    /// its lock.
+    ///
+    /// A rebuild copies a collection's live bytes to reclaim its dead ones, so
+    /// the one thing it must not do is copy a collection that has none. It used
+    /// to: `compact` walked every collection unconditionally, so garbage in one
+    /// paid for a full copy of the other thirty-nine.
+    ///
+    /// The share is the same one `note_compact` applies to the engine's totals,
+    /// and that is what keeps the two from disagreeing. If no collection passes
+    /// this test then `dead_i < live_i / 4` for every one of them, so
+    /// `sum(dead) < sum(live) / 4` and the engine's trigger could not have fired
+    /// either. So a compaction that runs always rebuilds at least one
+    /// collection, and cannot spin re-arming itself over garbage no rebuild will
+    /// take. An absolute floor per collection would break exactly that: forty
+    /// collections each under the floor can sum to well over it.
+    fn wants_rebuild(coll: *const Collection) bool {
+        assert_msg(
+            coll.slab_used >= coll.live_bytes,
+            "a collection cannot hold more live bytes than it ever appended",
+        );
+        const dead = coll.slab_used - coll.live_bytes;
+        if (dead == 0) return false;
+        return dead * 4 >= coll.live_bytes;
+    }
+
     /// Copy one collection's live documents into fresh extents and rebuild every
     /// index against the new offsets.
     ///
@@ -1940,6 +1966,7 @@ pub const Engine = struct {
     fn rebuild_collection(self: *Engine, coll: *Collection) !void {
         try coll.lock.lock(self.io);
         defer coll.lock.unlock(self.io);
+        if (!wants_rebuild(coll)) return;
 
         var old_extents = try self.gpa.alloc(pgr.Extent, coll.slab_runs.items.len);
         defer self.gpa.free(old_extents);
@@ -3752,6 +3779,62 @@ test "a churning collection reuses its slab instead of growing the file" {
     try testing.expect(grew < 3 * per_round * doc_size / pgr.page_size / 4);
 }
 
+test "a rebuild copies only the collections that have garbage" {
+    // `compact` walked every collection unconditionally, so garbage in one paid
+    // for a full copy of all the others -- and a copy is not free even when it
+    // reclaims nothing: it rewrites every document and every index, and it
+    // bumps the layout epoch, which kills every open cursor on a collection
+    // that had no reason to be touched.
+    //
+    // The epoch is the observable, and it is also the user-visible harm: the
+    // clean collection's cursors survive.
+    //
+    // Mutation check: delete the `wants_rebuild` guard. Red -- the clean
+    // collection's epoch moves too.
+    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();
+
+    var i: i32 = 0;
+    while (i < 40) : (i += 1) {
+        var d = try make_padded(gpa, i, 3000);
+        defer d.deinit();
+        try engine.insert("app", "dirty", &d, &env.gen);
+        var c = try make_padded(gpa, i, 3000);
+        defer c.deinit();
+        try engine.insert("app", "clean", &c, &env.gen);
+    }
+    // Only one of them loses anything.
+    i = 0;
+    while (i < 40) : (i += 2) _ = try engine.remove_by_id("app", "dirty", .{ .int32 = i });
+    try engine.commit();
+
+    const dirty = engine.get_collection("app", "dirty").?;
+    const clean = engine.get_collection("app", "clean").?;
+    const dirty_epoch = dirty.layout_epoch;
+    const clean_epoch = clean.layout_epoch;
+    const clean_used = clean.slab_used;
+    engine.unlock();
+
+    try engine.compact();
+
+    try testing.expect(dirty.layout_epoch != dirty_epoch);
+    try testing.expectEqual(clean_epoch, clean.layout_epoch);
+    // And it was not rewritten: a repack starts the slab over, so its byte
+    // count would not survive one unchanged.
+    try testing.expectEqual(clean_used, clean.slab_used);
+    try testing.expectEqual(@as(u64, 0), clean.slab_used - clean.live_bytes);
+}
+
 test "a replace that changes nothing is not a write" {
     // Mutation check: delete the byte comparison in `upsert`'s `.replace` arm.
     // Red on all three: the log grows, the document is superseded so the engine
@@ -5513,6 +5596,11 @@ test "the epochs that invalidate a cursor move exactly when they must" {
         defer d.deinit();
         try engine.insert("app", "c", &d, &env.gen);
     }
+    // Garbage, so the collection is worth rewriting: `compact` skips a
+    // collection with nothing to reclaim.
+    i = 0;
+    while (i < 20) : (i += 1) _ = try engine.remove_by_id("app", "c", .{ .int32 = i });
+    try engine.commit();
     const before = engine.get_collection("app", "c").?.layout_epoch;
     engine.unlock();
     try testing.expect(before != 0);