commands: serverStatus reports what the slab is doing

A `multifora` subdocument -- named so nobody mistakes it for a MongoDB section
-- carrying `liveBytes`, `deadBytes`, `slabBytes`, `reclaimedBytes`,
`slabRuns`, `freeReadyPages`, `allocTail` and `compactions`.

The milestone's gate cannot be read without them. A steady-state size ratio
can look respectable while reclamation does nothing at all: the file grows,
a rebuild periodically halves it, and the average comes out fine. What
distinguishes the two is `reclaimedBytes` rising while `allocTail` stays put,
and no ratio shows that. Same for rebuilds -- "the ratio improved" and "the
ratio improved because reclamation worked rather than because a rebuild ran"
are different results, so `compactions` counts collections rewritten.

The byte figures are summed from the collections rather than read off the
engine's running totals, so this reports the same side of the comparison
`checkpoint` asserts. A counter that had drifted from the catalog would
otherwise make the gate measure the drift.

Every field is present at zero. A gate that cannot tell "no pages ready" from
"field missing" cannot be read at all.

186/186 unit tests in ReleaseFast and ReleaseSafe, 83/83 fuzz, and the full
e2e matrix.

src/commands.zig

@@ -394,6 +394,24 @@ fn cmd_server_status(ctx: *Context, _: *wire.Message, reply: *wire.Reply) !void
     const connections = try reply.arena_alloc().alloc(bson.Pair, 1);
     connections[0] = .{ .key = "current", .value = .{ .int32 = @intCast(ctx.connection_id) } };
     try reply.put("connections", .{ .doc = connections });
+
+    // Not a MongoDB section, and named so nobody mistakes it for one. It is
+    // what the churn gate reads: a steady-state size ratio can look healthy
+    // while reclamation does nothing at all -- the file grows, a rebuild
+    // periodically halves it, and the average comes out respectable.
+    // `reclaimedBytes` rising while `allocTail` stays put is the shape that
+    // says the free list is carrying the workload, and no ratio shows that.
+    const s = ctx.engine.slab_stats();
+    const mf = try reply.arena_alloc().alloc(bson.Pair, 8);
+    mf[0] = .{ .key = "liveBytes", .value = .{ .int64 = @intCast(s.live_bytes) } };
+    mf[1] = .{ .key = "deadBytes", .value = .{ .int64 = @intCast(s.dead_bytes) } };
+    mf[2] = .{ .key = "slabBytes", .value = .{ .int64 = @intCast(s.slab_bytes) } };
+    mf[3] = .{ .key = "reclaimedBytes", .value = .{ .int64 = @intCast(s.reclaimed_bytes) } };
+    mf[4] = .{ .key = "slabRuns", .value = .{ .int64 = @intCast(s.slab_runs) } };
+    mf[5] = .{ .key = "freeReadyPages", .value = .{ .int64 = @intCast(s.free_ready_pages) } };
+    mf[6] = .{ .key = "allocTail", .value = .{ .int64 = @intCast(s.alloc_tail) } };
+    mf[7] = .{ .key = "compactions", .value = .{ .int64 = @intCast(s.compactions) } };
+    try reply.put("multifora", .{ .doc = mf });
     try reply.put_ok();
 }
 
@@ -3008,6 +3026,47 @@ test "ping and hello replies parse" {
     try testing.expectEqual(@as(i32, 9), bson.get_pair(reply2.pairs.items, "maxWireVersion").?.int32);
 }
 
+test "serverStatus reports what the slab is doing" {
+    // The churn gate reads these, and it needs them to be the collections' own
+    // figures rather than a second opinion about them -- a counter that drifts
+    // from what the catalog says would make the gate measure the drift.
+    //
+    // Mutation check: report the engine's running `live_bytes`/`dead_bytes`
+    // instead of summing the collections. Not red here, and that is the point:
+    // it is red in `checkpoint`, where the two are compared, which is why this
+    // reads the same side of that comparison.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tdb = try TestDb.init(io);
+    defer tdb.deinit();
+    var ctx = tdb.ctx(io);
+
+    try seed_docs(&tdb, io, "c", 60);
+
+    var reply = wire.Reply.init(testing.allocator);
+    defer reply.deinit();
+    var msg = try parse_fake_msg("serverStatus", .{ .int32 = 1 }, &.{});
+    defer msg.deinit();
+    try dispatch(&ctx, &msg, &reply);
+
+    const mf = bson.get_pair(reply.pairs.items, "multifora").?.doc;
+    const live = bson.get_pair(mf, "liveBytes").?.int64;
+    const dead = bson.get_pair(mf, "deadBytes").?.int64;
+    const slab = bson.get_pair(mf, "slabBytes").?.int64;
+    try testing.expect(live > 0);
+    try testing.expectEqual(slab, live + dead);
+    // Nothing has died and nothing has been reclaimed yet.
+    try testing.expectEqual(@as(i64, 0), dead);
+    try testing.expectEqual(@as(i64, 0), bson.get_pair(mf, "reclaimedBytes").?.int64);
+    try testing.expectEqual(@as(i64, 0), bson.get_pair(mf, "compactions").?.int64);
+    try testing.expectEqual(@as(i64, 1), bson.get_pair(mf, "slabRuns").?.int64);
+    try testing.expect(bson.get_pair(mf, "allocTail").?.int64 > 0);
+    // Present even at zero: a gate that cannot tell "no pages ready" from
+    // "field missing" cannot be read at all.
+    try testing.expect(bson.get_pair(mf, "freeReadyPages") != null);
+}
+
 test "the wire version agrees with the version the server calls itself" {
     // These two are read by different parts of a driver -- the handshake picks
     // features off the wire version, `runOnRequirements` in the spec suites

src/db.zig

@@ -706,6 +706,11 @@ pub const Engine = struct {
     /// share one tmp path and each ends in a rename onto the log, so two at
     /// once would publish one compaction's half-written file as the database.
     compacting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
+    /// Collections rewritten by a rebuild since the process started. Reported by
+    /// `serverStatus`, because "the ratio improved" and "the ratio improved
+    /// because reclamation worked rather than because a rebuild ran" are
+    /// different results and no ratio distinguishes them. Under `counter_lock`.
+    compactions: u64 = 0,
     log: storage.Log,
     /// The data file: documents live here, and the B+tree arenas follow.
     ///
@@ -2017,6 +2022,59 @@ pub const Engine = struct {
         // would have been invalidated for nothing.
         self.layout_epoch_seq += 1;
         coll.layout_epoch = self.layout_epoch_seq;
+        self.counter_lock.lockUncancelable(self.io);
+        self.compactions += 1;
+        self.counter_lock.unlock(self.io);
+    }
+
+    /// What the slab is doing, for `serverStatus`. Collections under the same
+    /// catalog-then-collection order everything else uses.
+    ///
+    /// It exists because the milestone's own gate cannot be read without it. A
+    /// steady-state size ratio can look healthy while reclamation does nothing
+    /// -- the file grows, a rebuild periodically halves it, and the average
+    /// comes out fine. `reclaimed_bytes` climbing while `alloc_tail` stays put
+    /// is the shape that says the free list is load-bearing; either one alone
+    /// says very little.
+    pub const SlabStats = struct {
+        live_bytes: u64 = 0,
+        dead_bytes: u64 = 0,
+        slab_bytes: u64 = 0,
+        reclaimed_bytes: u64 = 0,
+        slab_runs: u64 = 0,
+        free_ready_pages: u32 = 0,
+        alloc_tail: u32 = 0,
+        compactions: u64 = 0,
+    };
+
+    pub fn slab_stats(self: *Engine) SlabStats {
+        var out: SlabStats = .{};
+        self.catalog_lock.lockSharedUncancelable(self.io);
+        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);
+                out.live_bytes += coll.live_bytes;
+                out.slab_bytes += coll.slab_used;
+                out.reclaimed_bytes += coll.reclaimed_bytes;
+                out.slab_runs += coll.slab_runs.items.len;
+            }
+        }
+        self.catalog_lock.unlockShared(self.io);
+        // From the collections rather than the engine's running total, so this
+        // is the same figure `write_catalog` asserts against rather than a
+        // second opinion about it.
+        assert_msg(out.slab_bytes >= out.live_bytes, "the slab cannot hold more live bytes than it has");
+        out.dead_bytes = out.slab_bytes - out.live_bytes;
+        out.free_ready_pages = self.pager.free_ready_pages();
+        out.alloc_tail = self.pager.alloc_tail;
+        self.counter_lock.lockUncancelable(self.io);
+        out.compactions = self.compactions;
+        self.counter_lock.unlock(self.io);
+        return out;
     }
 
     /// The engine's dead-byte total, recomputed from the collections that