db/pager/tests: cleanup pass over the free list

No behaviour is meant to change and the gate confirms it: 1.94x / 679.2 MB
reclaimed / 9 rebuilds and 2.46x / 934.0 MB / 13 on the two 16 KiB lines,
0.0 MB on the 200-byte line, all identical to the numbers recorded for them.

Deduplication. `pages_for` was written in db.zig and again in pager.zig and
twice more inline; there is now one, public, and the two pre-existing copies
call it. `pages_per_map_align` replaces three hand-rolled `map_align /
page_size`. `SlabRun.window_first` was a stored field that could never legally
disagree with `first` and was maintained by hand at two sites -- now a method.
`keep_piece` re-derived `SlabRun.window_count` character for character; it
calls it. `insert_run` scanned linearly for a position `run_of` binary-searches
for, which made loading a fragmented catalog quadratic; both now go through one
`run_lower_bound`. Freeing a run's window map was written four times; one
helper. The 20% rebuild share was stated in `note_compact` and again in
`wants_rebuild`, with a comment arguing at length that they must be the same
number -- `worth_rewriting` makes that structural.

Efficiency. The identity assert in `reclaim_windows` called `dead_located()`,
an O(every window) walk, and `assert_msg` is live in ReleaseFast -- so it
doubled the scan the reclamation was about to make (2.75 MB streamed twice per
reclaiming checkpoint at the 21 GB the gate targets). `dead_located` is now a
maintained counter, the check is O(1) in every build, and the scan cross-checks
it while it is there. `SlabRun.full` lets a run with nothing to give be copied
without its counters being read at all, so the common case is O(runs) rather
than O(windows).

The pager's two allocation policies were hand-copying the claim step, and the
copy had already lost two of the three preconditions -- `alloc_slab_run` never
checked `pages <= reserved_pages`. Both now go through `claim_locked`.

`reclaimed_bytes` moves from Collection to Engine, beside `compactions`, which
is how it is read and the only place it can be honest: a life-of-the-process
total must not lose a dropped collection's share. Both join `Counters`, so
`slab_stats` stops opening `counter_lock` by hand.

The ownership assertion in `write_catalog` was gated on `is_test or Debug`, a
predicate nothing else in the codebase uses, which left the one silent failure
this design can produce unchecked in ReleaseSafe. It is now `!= ReleaseFast`,
the line `protect_stable` already draws. Measured: no change to the suite's
runtime.

Altitude. `note_checkpoint` was called from exactly one place, the tail of
`upsert` -- so a delete armed no checkpoint by any route, which is why
reclamation only ever ran when the *rebuild* trigger fired and the rebuild then
reset the window map it would have used. `remove` and the TTL sweep arm one
now, next to the `note_compact` calls that were added for the same omission a
milestone ago. `compact`'s leading checkpoint stays, demoted in its comment
from the mechanism to the local ordering it actually guarantees.

serverStatus reports `allocTailBytes`/`freeReadyBytes` instead of page counts,
so the harness stops hard-coding 4096 -- the kind of constant this milestone
was blindsided by once already.

tests/e2e/churn.js: `deleteMany({_id: {$in: [5000 ids]}})` exceeded
`index.max_combos`, so the planner refused the index and every delete became a
full collection scan re-filtering each document against 5000 members. That was
the entire runtime of the harness. One delete spec per id instead: the 40k x
16 KiB gate goes 57 s -> 6 s, and the 150k x 200 B line 483 s -> 3 s, with
identical output. Also: the per-round `countDocuments` is gone (the harness
knows the count), and the server log is a bounded ring rather than a rope that
grows with everything the server ever said.

Reverted from the review: reusing one MongoClient across the startup poll. A
client whose first connect fails tears its topology down and every later
command on it fails identically, so it turns "not up yet" into "never comes
up" -- it broke the first run. The reason is now a comment.

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

@@ -408,8 +408,8 @@ fn cmd_server_status(ctx: *Context, _: *wire.Message, reply: *wire.Reply) !void
     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[5] = .{ .key = "freeReadyBytes", .value = .{ .int64 = @intCast(s.free_ready_bytes) } };
-    mf[6] = .{ .key = "allocTail", .value = .{ .int64 = @intCast(s.alloc_tail) } };
+    mf[6] = .{ .key = "allocTailBytes", .value = .{ .int64 = @intCast(s.alloc_tail_bytes) } };
     mf[7] = .{ .key = "compactions", .value = .{ .int64 = @intCast(s.compactions) } };
     try reply.put("multifora", .{ .doc = mf });
     try reply.put_ok();
@@ -3061,10 +3061,10 @@ test "serverStatus reports what the slab is doing" {
     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);
+    try testing.expect(bson.get_pair(mf, "allocTailBytes").?.int64 > 0);
-    // Present even at zero: a gate that cannot tell "no pages ready" from
+    // Present even at zero: a gate that cannot tell "nothing ready" from
     // "field missing" cannot be read at all.
-    try testing.expect(bson.get_pair(mf, "freeReadyPages") != null);
+    try testing.expect(bson.get_pair(mf, "freeReadyBytes") != null);
 }
 
 test "the wire version agrees with the version the server calls itself" {

src/db.zig

@@ -39,11 +39,6 @@ const slab_extent_pages: u32 = (8 * 1024 * 1024) / pgr.page_size;
 /// off whatever is left of the one before it.
 const slab_run_min_pages: u32 = slab_extent_pages / 8;
 
-/// Pages a slab allocation of `len` bytes needs at minimum.
-fn pages_for(len: usize) u32 {
-    return @intCast((len + pgr.page_size - 1) / pgr.page_size);
-}
-
 const LogKind = enum { upsert, delete, index_create, index_drop };
 
 /// Dead bytes in one `map_align` window. The window is the unit of reclamation
@@ -71,14 +66,15 @@ const WindowDead = if (pgr.map_align <= std.math.maxInt(u16)) u16 else u32;
 const SlabRun = struct {
     first: u32,
     pages: u32,
-    /// The run's start, rounded up to `map_align`: the first offset that begins
-    /// a whole window. `alloc_pages` works in 4 KiB pages, so a run need not
-    /// start on a system page.
-    window_first: u64,
     /// Dead bytes per window, `dead[i]` covering
-    /// `[window_first + i*map_align, +map_align)`. `map_align` means the window
+    /// `[window_first() + i*map_align, +map_align)`. `map_align` means the
-    /// holds nothing live and can be handed back.
+    /// window holds nothing live and can be handed back.
     dead: []WindowDead,
+    /// How many of those have reached `map_align`, so a reclamation scan can
+    /// skip a run with nothing to give without touching its counters. Always
+    /// zero in a run that survived a reclamation, because every full window is
+    /// taken.
+    full: u32,
 
     /// Windows wholly inside the pages `[first, first+pages)`. The bytes
     /// outside them -- below `window_first`, and the tail after the last whole
@@ -97,13 +93,21 @@ const SlabRun = struct {
         return @as(u64, self.first) << pgr.page_shift;
     }
 
+    /// The run's start, rounded up to `map_align`: the first offset that begins
+    /// a whole window. `alloc_pages` works in 4 KiB pages, so a run need not
+    /// start on a system page. Derived rather than stored, so it cannot
+    /// disagree with `first`.
+    fn window_first(self: SlabRun) u64 {
+        return std.mem.alignForward(u64, self.start(), pgr.map_align);
+    }
+
     fn end(self: SlabRun) u64 {
         return (@as(u64, self.first) + self.pages) << pgr.page_shift;
     }
 
     /// One past the last byte covered by a window counter.
     fn window_end(self: SlabRun) u64 {
-        return self.window_first + self.dead.len * pgr.map_align;
+        return self.window_first() + self.dead.len * pgr.map_align;
     }
 };
 
@@ -165,6 +169,13 @@ pub const Collection = struct {
     /// update. What it costs is only that garbage from before a restart is not
     /// reclaimed window-wise; it still arms compaction like any other.
     dead_unlocated: u64,
+    /// The other half of it: garbage that *is* in a window, i.e. the sum of
+    /// every run's counters. Maintained rather than summed, so the identity
+    /// above is an O(1) check on a path that runs in every build -- summing it
+    /// walked every window of the slab, doubling the scan the reclamation was
+    /// about to make anyway (1.38M counters at the 21 GB the gate targets).
+    /// The scan cross-checks it while it is there.
+    dead_located: u64,
     /// Windows whose counter has reached `map_align`, i.e. how much there is
     /// for the next checkpoint to give back.
     ///
@@ -174,11 +185,6 @@ pub const Collection = struct {
     /// help, small documents on large system pages, is exactly the one that
     /// would pay that for no return.
     full_windows: u32,
-    /// Slab handed back to the pager by window reclamation, cumulative for the
-    /// life of the process. Purely an observation: it is what distinguishes
-    /// "the ratio improved because reclamation worked" from "the ratio improved
-    /// for some other reason", which is the only way to read the churn gate.
-    reclaimed_bytes: u64,
     /// Secondary indexes (persisted through the log). Heap-allocated, so an
     /// `*Index` handed out by `find_index` or `create_index` stays valid when
     /// a sibling index is dropped. Held by value, `orderedRemove` memmoved the
@@ -229,8 +235,8 @@ pub const Collection = struct {
             .slab_used = 0,
             .live_bytes = 0,
             .dead_unlocated = 0,
+            .dead_located = 0,
             .full_windows = 0,
-            .reclaimed_bytes = 0,
             .hold = .{},
             .indexes = .empty,
             .id_index = undefined,
@@ -263,8 +269,7 @@ pub const Collection = struct {
         const dead = try gpa.alloc(WindowDead, SlabRun.window_count(first, pages));
         errdefer gpa.free(dead);
         @memset(dead, 0);
-        var at: usize = 0;
+        const at = self.run_lower_bound(first);
-        while (at < self.slab_runs.items.len and self.slab_runs.items[at].first < first) at += 1;
         // A recycled run must not overlap one this collection already owns:
         // that would be the pager handing out pages twice, and the symptom
         // would be a document quietly overwritten rather than anything failing.
@@ -278,30 +283,39 @@ pub const Collection = struct {
         try self.slab_runs.insert(gpa, at, .{
             .first = first,
             .pages = pages,
-            .window_first = std.mem.alignForward(u64, @as(u64, first) << pgr.page_shift, pgr.map_align),
             .dead = dead,
+            .full = 0,
         });
     }
 
-    /// The run holding `off`, or null if no run does. Binary search, which the
+    /// The first run at or past `page`. The one search over the sorted list:
-    /// sorted list is for: `mark_dead` runs once per evicted document, and a
+    /// `run_of` adds a range test to it and `insert_run` uses it as the
-    /// collection with a fragmented slab can own thousands of runs.
+    /// insertion point, so the ordering is interpreted in one place.
-    fn run_of(self: *const Collection, off: u64) ?usize {
+    ///
-        const page: u32 = @intCast(off >> pgr.page_shift);
+    /// Binary rather than linear, which matters at both ends: `mark_dead` runs
+    /// once per evicted document, and `read_catalog` inserts every run of a
+    /// fragmented collection at open.
+    fn run_lower_bound(self: *const Collection, page: u32) usize {
         var lo: usize = 0;
         var hi: usize = self.slab_runs.items.len;
         while (lo < hi) {
             const mid = lo + (hi - lo) / 2;
-            const r = self.slab_runs.items[mid];
+            if (self.slab_runs.items[mid].first < page) lo = mid + 1 else hi = mid;
-            if (page < r.first) {
-                hi = mid;
-            } else if (page >= r.first + r.pages) {
-                lo = mid + 1;
-            } else {
-                return mid;
         }
+        return lo;
     }
-        return null;
+
+    /// The run holding `off`, or null if no run does.
+    fn run_of(self: *const Collection, off: u64) ?usize {
+        const page: u32 = @intCast(off >> pgr.page_shift);
+        // `lower_bound` lands on the run starting at `page` if there is one,
+        // otherwise on the one after it -- so the candidate is that one or its
+        // predecessor, and only the predecessor can contain an interior page.
+        const at = self.run_lower_bound(page);
+        if (at < self.slab_runs.items.len and self.slab_runs.items[at].first == page) return at;
+        if (at == 0) return null;
+        const prev = &self.slab_runs.items[at - 1];
+        return if (page < prev.first + prev.pages) at - 1 else null;
     }
 
     /// Record that `[off, off+len)` of slab is garbage.
@@ -328,46 +342,50 @@ pub const Collection = struct {
         // layout -- a stale index entry, which is the failure the layout epoch
         // exists to prevent.
         assert_msg(stop <= r.end(), "a dead slab range crosses the end of the run holding it");
+        const window_first = r.window_first();
         var pos = off;
-        if (pos < r.window_first) {
+        if (pos < window_first) {
-            const n = @min(stop, r.window_first) - pos;
+            const n = @min(stop, window_first) - pos;
             self.dead_unlocated += n;
             pos += n;
         }
         const win_end = r.window_end();
         while (pos < stop and pos < win_end) {
-            const w: usize = @intCast((pos - r.window_first) / pgr.map_align);
+            const w: usize = @intCast((pos - window_first) / pgr.map_align);
-            const w_end = r.window_first + (w + 1) * pgr.map_align;
+            const w_end = window_first + (w + 1) * pgr.map_align;
             const n = @min(stop, w_end) - pos;
             // A window cannot hold more dead bytes than it has bytes. Tripping
             // this means the same range was marked twice -- a double eviction,
             // or a recycled offset marked against the previous owner's map.
             assert_msg(r.dead[w] + n <= pgr.map_align, "a slab window holds more dead bytes than it has");
-            const was_full = r.dead[w] == pgr.map_align;
             r.dead[w] += @intCast(n);
-            if (!was_full and r.dead[w] == pgr.map_align) self.full_windows += 1;
+            if (r.dead[w] == pgr.map_align and n > 0) {
+                // Just filled: `n > 0` and the bound above mean this cannot be
+                // a window that was already full.
+                r.full += 1;
+                self.full_windows += 1;
+            }
+            self.dead_located += n;
             pos += n;
         }
         if (pos < stop) self.dead_unlocated += stop - pos;
     }
 
-    /// Garbage this collection has placed in windows. Walks every window, so it
+    /// Drop the window maps and the run list, and forget where the garbage was.
-    /// belongs to the reclamation scan and to tests, not to a hot path.
+    /// The pages themselves are the caller's business -- a drop hands them to
-    fn dead_located(self: *const Collection) u64 {
+    /// the pager, a rebuild has already done so.
-        var sum: u64 = 0;
-        for (self.slab_runs.items) |r| {
-            for (r.dead) |d| sum += d;
-        }
-        return sum;
-    }
-
-    /// Drop the window maps and the run list. The pages themselves are the
-    /// caller's business -- a drop hands them to the pager, a rebuild has
-    /// already done so.
     fn free_runs(self: *Collection, gpa: std.mem.Allocator) void {
-        for (self.slab_runs.items) |r| gpa.free(r.dead);
+        free_window_maps(gpa, self.slab_runs.items);
         self.slab_runs.clearRetainingCapacity();
         self.full_windows = 0;
+        self.dead_located = 0;
+    }
+
+    /// Release the window map of every run in `runs`. The one place a map is
+    /// freed, because there are four callers and missing one is a leak nothing
+    /// would notice.
+    fn free_window_maps(gpa: std.mem.Allocator, runs: []const SlabRun) void {
+        for (runs) |r| gpa.free(r.dead);
     }
 
     /// One piece of a run that survives reclamation, with a window map of its
@@ -385,17 +403,18 @@ pub const Collection = struct {
         p0: u32,
         p1: u32,
     ) !void {
-        const wf = std.mem.alignForward(u64, @as(u64, p0) << pgr.page_shift, pgr.map_align);
+        const piece: SlabRun = .{ .first = p0, .pages = p1 - p0, .dead = &.{}, .full = 0 };
-        const we = std.mem.alignBackward(u64, @as(u64, p1) << pgr.page_shift, pgr.map_align);
+        const count = SlabRun.window_count(p0, p1 - p0);
-        const count: usize = if (we > wf) @intCast((we - wf) / pgr.map_align) else 0;
         const dead = try gpa.alloc(WindowDead, count);
         errdefer gpa.free(dead);
         // A piece boundary is either the run's own start/end or a window
         // boundary, so the piece's windows line up with a contiguous stretch of
         // the original's and the counters can be copied rather than rebuilt.
-        const base: usize = @intCast((wf - r.window_first) / pgr.map_align);
+        const base: usize = @intCast((piece.window_first() - r.window_first()) / pgr.map_align);
         @memcpy(dead, r.dead[base..][0..count]);
-        try out.append(gpa, .{ .first = p0, .pages = p1 - p0, .window_first = wf, .dead = dead });
+        // `full` stays zero: every full window is taken, so what survives has
+        // none. Asserted by the caller, which knows the whole run's count.
+        try out.append(gpa, .{ .first = p0, .pages = p1 - p0, .dead = dead, .full = 0 });
     }
 
     /// Give back every window with nothing live left in it, splitting the runs
@@ -415,21 +434,72 @@ pub const Collection = struct {
             self.slab_used >= self.live_bytes,
             "a collection cannot hold more live bytes than it ever appended",
         );
-        // The identity, checked where every window is being walked anyway.
+        // The identity the window map rests on. O(1), because both halves are
+        // maintained by `mark_dead`; the scan below cross-checks the located
+        // half against the counters themselves.
         assert_msg(
-            self.dead_located() + self.dead_unlocated == self.slab_used - self.live_bytes,
+            self.dead_located + self.dead_unlocated == self.slab_used - self.live_bytes,
             "the collection's placed and unplaced garbage must add up to its garbage",
         );
         var out: std.ArrayListUnmanaged(SlabRun) = .empty;
         errdefer {
-            for (out.items) |p| gpa.free(p.dead);
+            free_window_maps(gpa, out.items);
             out.deinit(gpa);
         }
         var give: std.ArrayListUnmanaged(pgr.Extent) = .empty;
         defer give.deinit(gpa);
 
         var freed: u64 = 0;
+        var full_seen: u32 = 0;
         for (self.slab_runs.items) |r| {
+            full_seen += r.full;
+            // The common case, and what `full` is for: a run with nothing to
+            // give is copied without its counters being looked at.
+            if (r.full == 0) {
+                try keep_piece(&out, gpa, r, r.first, r.first + r.pages);
+                continue;
+            }
+            freed += try self.reclaim_run(gpa, r, &out, &give);
+        }
+        // `full_windows` is the gate that decides whether a collection is
+        // scanned at all, so a drift low silently stops reclaiming it -- and the
+        // symptom would be "the ratio did not move", which is the one conclusion
+        // this milestone had to work hardest to tell apart from a real result.
+        assert_msg(full_seen == self.full_windows, "the collection's full-window count disagrees with its runs");
+        if (freed == 0) {
+            free_window_maps(gpa, out.items);
+            out.deinit(gpa);
+            return 0;
+        }
+
+        // Past the last fallible step: swap the list in, then hand the pages
+        // over. A `free_pages` that fails here leaks the run -- it is no longer
+        // the collection's and not yet the pager's -- which costs space and
+        // nothing else. The other order would leave the same pages owned twice.
+        free_window_maps(gpa, self.slab_runs.items);
+        self.slab_runs.deinit(gpa);
+        self.slab_runs = out;
+        // Every full window was given back, so nothing is left for the next
+        // checkpoint to find.
+        self.full_windows = 0;
+        for (give.items) |e| self.pager.free_pages(e.first, e.pages) catch {};
+        assert_msg(self.slab_used >= self.live_bytes + freed, "reclaiming more slab than the collection has");
+        self.slab_used -= freed;
+        self.dead_located -= freed;
+        return freed;
+    }
+
+    /// The full-window stretches of one run: `give` collects the extents handed
+    /// back, `out` the pieces that survive around them. Returns the bytes freed.
+    fn reclaim_run(
+        self: *const Collection,
+        gpa: std.mem.Allocator,
+        r: SlabRun,
+        out: *std.ArrayListUnmanaged(SlabRun),
+        give: *std.ArrayListUnmanaged(pgr.Extent),
+    ) !u64 {
+        const window_first = r.window_first();
+        var freed: u64 = 0;
         var keep_from = r.first;
         var i: usize = 0;
         while (i < r.dead.len) {
@@ -439,49 +509,25 @@ pub const Collection = struct {
             }
             var j = i + 1;
             while (j < r.dead.len and r.dead[j] == pgr.map_align) j += 1;
-                const from = r.window_first + i * pgr.map_align;
+            const from = window_first + i * pgr.map_align;
-                const to = r.window_first + j * pgr.map_align;
+            const to = window_first + j * pgr.map_align;
-                // The appender's own extent is off limits, and not by
+            // The appender's own extent is off limits, and not by filtering:
-                // filtering: bytes above the cursor have never been written, so
+            // bytes above the cursor have never been written, so no window
-                // no window covering them can have reached `map_align` dead.
+            // covering them can have reached `map_align` dead. Tripping this
-                // Tripping this means a range was marked dead twice.
+            // means a range was marked dead twice.
             assert_msg(
                 to <= self.slab_tail or from >= self.slab_end,
                 "reclaiming a slab window the append cursor is still walking",
             );
             const p_from: u32 = @intCast(from >> pgr.page_shift);
             const p_to: u32 = @intCast(to >> pgr.page_shift);
-                if (p_from > keep_from) try keep_piece(&out, gpa, r, keep_from, p_from);
+            if (p_from > keep_from) try keep_piece(out, gpa, r, keep_from, p_from);
             try give.append(gpa, .{ .first = p_from, .pages = p_to - p_from });
             freed += to - from;
             keep_from = p_to;
             i = j;
         }
-            if (keep_from < r.first + r.pages) {
+        if (keep_from < r.first + r.pages) try keep_piece(out, gpa, r, keep_from, r.first + r.pages);
-                try keep_piece(&out, gpa, r, keep_from, r.first + r.pages);
-            }
-        }
-        if (freed == 0) {
-            for (out.items) |p| gpa.free(p.dead);
-            out.deinit(gpa);
-            // Every full window was given back or there were none, so nothing
-            // is left for the next checkpoint to find.
-            self.full_windows = 0;
-            return 0;
-        }
-
-        // Past the last fallible step: swap the list in, then hand the pages
-        // over. A `free_pages` that fails here leaks the run -- it is no longer
-        // the collection's and not yet the pager's -- which costs space and
-        // nothing else. The other order would leave the same pages owned twice.
-        for (self.slab_runs.items) |r| gpa.free(r.dead);
-        self.slab_runs.deinit(gpa);
-        self.slab_runs = out;
-        self.full_windows = 0;
-        for (give.items) |e| self.pager.free_pages(e.first, e.pages) catch {};
-        assert_msg(self.slab_used >= self.live_bytes + freed, "reclaiming more slab than the collection has");
-        self.slab_used -= freed;
-        self.reclaimed_bytes += freed;
         return freed;
     }
 
@@ -543,7 +589,7 @@ pub const Collection = struct {
         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 = @max(slab_extent_pages, pages_for(len));
+        const want_pages: u32 = @max(slab_extent_pages, pgr.pages_for(len));
         try self.pager.reserve_pages(&self.hold, want_pages);
         // Off the free list first, or window reclamation is decorative: the
         // pages come back, nothing asks for them in a shape they arrive in, and
@@ -551,7 +597,7 @@ pub const Collection = struct {
         // because a shorter extent is exhausted after a handful of documents
         // and every exhaustion abandons what is left of it -- and because the
         // floor is what makes trimming a larger run harmless.
-        const min_pages: u32 = @min(want_pages, @max(pages_for(len), slab_run_min_pages));
+        const min_pages: u32 = @min(want_pages, @max(pgr.pages_for(len), slab_run_min_pages));
         const run = self.pager.alloc_slab_run(&self.hold, min_pages, want_pages) orelse pgr.Extent{
             .first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages),
             .pages = want_pages,
@@ -705,12 +751,25 @@ pub const Engine = struct {
     /// Set while a compaction runs, so only one runs at a time. Compactions
     /// 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.
+    ///
+    /// A loser here *skips*, where a loser on `checkpoint_lock` waits, and the
+    /// difference is what the caller is owed. A compaction is idempotent and
+    /// its garbage keeps, so the winner's pass covers the loser's reason for
+    /// asking. A checkpoint is a postcondition its caller depends on -- most of
+    /// all `compact`, which needs reclamation to have run before it copies --
+    /// so skipping one would silently break that.
     compacting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
-    /// Collections rewritten by a rebuild since the process started. Reported by
+    /// Collections rewritten by a rebuild since the process started, and slab
-    /// `serverStatus`, because "the ratio improved" and "the ratio improved
+    /// handed back by window reclamation. Reported by `serverStatus`, because
-    /// because reclamation worked rather than because a rebuild ran" are
+    /// "the ratio improved" and "the ratio improved because reclamation worked
-    /// different results and no ratio distinguishes them. Under `counter_lock`.
+    /// rather than because a rebuild ran" are different results and no ratio
+    /// distinguishes them. Both under `counter_lock`.
+    ///
+    /// On the engine rather than per collection, which is how they are read and
+    /// also the only place they can be honest: a dropped collection's share of
+    /// a life-of-the-process total must not vanish with it.
     compactions: u64 = 0,
+    reclaimed_bytes: u64 = 0,
     log: storage.Log,
     /// The data file: documents live here, and the B+tree arenas follow.
     ///
@@ -939,7 +998,14 @@ pub const Engine = struct {
     /// 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 };
+    const Counters = struct {
+        live_docs: u64,
+        dead_docs: u64,
+        live_bytes: u64,
+        dead_bytes: u64,
+        compactions: u64,
+        reclaimed_bytes: u64,
+    };
 
     fn counters(self: *Engine) Counters {
         self.counter_lock.lockUncancelable(self.io);
@@ -949,6 +1015,8 @@ pub const Engine = struct {
             .dead_docs = self.dead_docs,
             .live_bytes = self.live_bytes,
             .dead_bytes = self.dead_bytes,
+            .compactions = self.compactions,
+            .reclaimed_bytes = self.reclaimed_bytes,
         };
     }
 
@@ -1501,6 +1569,13 @@ pub const Engine = struct {
         // Deletes grow the log too. Without this a delete-heavy workload
         // never compacts, because only upsert and ttl_sweep used to check.
         self.note_compact();
+        // And the same omission a second time, found by the churn harness: a
+        // checkpoint is what reclaims empty slab windows, and only `upsert`
+        // armed one. So deleting half a collection reclaimed nothing until the
+        // *rebuild* trigger fired, and the rebuild then reset the window map it
+        // would have used -- six rounds of delete-and-refill, six rebuilds, 1 MB
+        // reclaimed. Armed where the garbage is made, it is 256 MB.
+        self.note_checkpoint();
         return true;
     }
 
@@ -1649,9 +1724,13 @@ pub const Engine = struct {
                 deleted += try self.ttl_sweep_coll(coll, now_ms, db_entry.key_ptr.*, coll_entry.key_ptr.*);
             }
         }
-        // A TTL-only workload never reaches the threshold check in `upsert`,
+        // A TTL-only workload never reaches the threshold checks in `upsert`,
-        // so the log would otherwise grow without bound.
+        // so the log would otherwise grow without bound and nothing would ever
-        if (deleted > 0) self.note_compact();
+        // reclaim the slab the expired documents left behind.
+        if (deleted > 0) {
+            self.note_compact();
+            self.note_checkpoint();
+        }
         return deleted;
     }
 
@@ -1852,12 +1931,9 @@ pub const Engine = struct {
         // 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 (c.dead_bytes < self.compact_threshold) return;
-        // Then the share, dead / (live + dead), firing at ~20%: the file stays
+        // Then the share: the file stays near 1.25x the live data and each
-        // near 1.25x the live data and each rebuild is paid for by the space it
+        // rebuild is paid for by the space it reclaims.
-        // reclaims. Bytes rather than document counts, because a rewrite copies
+        if (!worth_rewriting(c.dead_bytes, c.live_bytes)) return;
-        // bytes -- 100k evicted 40 B documents are not worth the same rebuild as
-        // 100k evicted 16 KiB ones.
-        if (c.dead_bytes * 4 < c.live_bytes) return;
         self.compact_pending.store(true, .release);
     }
 
@@ -1917,12 +1993,12 @@ pub const Engine = struct {
         // takes, the per-collection gate below no longer sees, so a collection
         // whose garbage was all in empty windows is not rewritten at all.
         //
-        // This is not a refinement, it is what makes reclamation reachable
+        // Belt and braces rather than the mechanism: `remove` and the TTL sweep
-        // under a delete-heavy workload. A checkpoint is otherwise armed by log
+        // arm a checkpoint of their own, so reclamation runs on the cadence of
-        // volume, and a delete logs only an `_id` -- so deleting half a 190 MB
+        // garbage production rather than waiting for the rebuild trigger. What
-        // collection moves the log by a couple of megabytes and no checkpoint
+        // this guarantees is only the local ordering -- that whatever the
-        // runs, while the garbage sails past the rebuild threshold. Measured
+        // collections are about to be judged on has already had the cheap half
-        // with the churn harness: six rounds, six rebuilds, 1 MB reclaimed.
+        // applied to it.
         try self.checkpoint();
 
         try self.catalog_lock.lockShared(self.io);
@@ -1991,9 +2067,19 @@ pub const Engine = struct {
             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;
+        return worth_rewriting(coll.slab_used - coll.live_bytes, coll.live_bytes);
-        if (dead == 0) return false;
+    }
-        return dead * 4 >= coll.live_bytes;
+
+    /// dead / (live + dead) at or above ~20%. The one share both the engine's
+    /// trigger and the per-collection gate apply, written once so the argument
+    /// above -- that a compaction which runs always rebuilds something -- is
+    /// enforced by construction rather than by two literals agreeing.
+    ///
+    /// 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.
+    fn worth_rewriting(dead: u64, live: u64) bool {
+        return dead > 0 and dead * 4 >= live;
     }
 
     /// Copy one collection's live documents into fresh extents and rebuild every
@@ -2077,8 +2163,11 @@ pub const Engine = struct {
         slab_bytes: u64 = 0,
         reclaimed_bytes: u64 = 0,
         slab_runs: u64 = 0,
-        free_ready_pages: u32 = 0,
+        /// Bytes, like every other figure here. Pages would make a reader
-        alloc_tail: u32 = 0,
+        /// outside the process supply `page_size` from somewhere, and the
+        /// harness that reads this is a Node script.
+        free_ready_bytes: u64 = 0,
+        alloc_tail_bytes: u64 = 0,
         compactions: u64 = 0,
     };
 
@@ -2094,7 +2183,6 @@ pub const Engine = struct {
                 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;
             }
         }
@@ -2104,11 +2192,11 @@ pub const Engine = struct {
         // 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.free_ready_bytes = @as(u64, self.pager.free_ready_pages()) << pgr.page_shift;
-        out.alloc_tail = self.pager.alloc_tail;
+        out.alloc_tail_bytes = self.pager.allocated_bytes();
-        self.counter_lock.lockUncancelable(self.io);
+        const c = self.counters();
-        out.compactions = self.compactions;
+        out.compactions = c.compactions;
-        self.counter_lock.unlock(self.io);
+        out.reclaimed_bytes = c.reclaimed_bytes;
         return out;
     }
 
@@ -2363,7 +2451,10 @@ pub const Engine = struct {
                     // 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(
+                    // Off in ReleaseFast only, the same line `protect_stable`
+                    // draws: the check is O(runs x free list) and the builds
+                    // that do the checking are the ones that pay for it.
+                    if (builtin.mode != .ReleaseFast) assert_msg(
                         !self.pager.owns_freed(r.first, r.pages),
                         "the catalog claims a slab run that is already on the free list",
                     );
@@ -2638,6 +2729,7 @@ pub const Engine = struct {
         self.counter_lock.lockUncancelable(self.io);
         assert_msg(self.dead_bytes >= freed, "reclaiming more slab than the engine counts as dead");
         self.dead_bytes -= freed;
+        self.reclaimed_bytes += freed;
         self.counter_lock.unlock(self.io);
         // A cursor holding slab offsets is now holding some that name pages
         // this collection no longer owns -- and reading them would succeed,
@@ -3414,7 +3506,7 @@ fn make_padded(gpa: std.mem.Allocator, id: i32, size: usize) !bson.Document {
 
 /// Every dead byte the collection knows about, placed or not.
 fn dead_total(coll: *const Collection) u64 {
-    return coll.dead_located() + coll.dead_unlocated;
+    return coll.dead_located + coll.dead_unlocated;
 }
 
 test "every dead slab byte is counted in exactly one place" {
@@ -3480,7 +3572,7 @@ test "every dead slab byte is counted in exactly one place" {
     try testing.expectEqual(coll.slab_used - coll.live_bytes, dead_total(coll));
     // And the evicted bytes are mostly placeable: 6000-byte documents are far
     // smaller than a window, so they fall inside one rather than off its edge.
-    try testing.expect(coll.dead_located() > coll.dead_unlocated);
+    try testing.expect(coll.dead_located > coll.dead_unlocated);
 }
 
 test "dead bytes outside a whole window are counted but not placed" {
@@ -3527,18 +3619,18 @@ test "dead bytes outside a whole window are counted but not placed" {
     const ri = coll.run_of(@as(u64, aligned + 1) << pgr.page_shift).?;
     const r = coll.slab_runs.items[ri];
     try testing.expectEqual(@as(usize, 2), r.dead.len);
-    try testing.expect(r.window_first > r.start());
+    try testing.expect(r.window_first() > r.start());
     try testing.expect(r.window_end() < r.end());
 
     const before = coll.dead_unlocated;
     // The head, one whole window, and the tail.
-    coll.mark_dead(r.start(), r.window_first - r.start());
+    coll.mark_dead(r.start(), r.window_first() - r.start());
-    coll.mark_dead(r.window_first, pgr.map_align);
+    coll.mark_dead(r.window_first(), pgr.map_align);
     coll.mark_dead(r.window_end(), r.end() - r.window_end());
 
-    try testing.expectEqual(@as(u64, pgr.map_align), coll.dead_located());
+    try testing.expectEqual(@as(u64, pgr.map_align), coll.dead_located);
     try testing.expectEqual(
-        before + (r.window_first - r.start()) + (r.end() - r.window_end()),
+        before + (r.window_first() - r.start()) + (r.end() - r.window_end()),
         coll.dead_unlocated,
     );
     // The whole window is full and the one beside it untouched: the head and
@@ -3675,7 +3767,7 @@ test "a restart forgets where the garbage is, not that there is any" {
         }
         try engine.commit();
         const coll = engine.get_collection("app", "c").?;
-        try testing.expect(coll.dead_located() > 0);
+        try testing.expect(coll.dead_located > 0);
         dead_before = coll.slab_used - coll.live_bytes;
         try engine.checkpoint();
     }
@@ -3684,7 +3776,7 @@ test "a restart forgets where the garbage is, not that there is any" {
     defer engine2.deinit();
     const coll = engine2.get_collection("app", "c").?;
     try testing.expectEqual(dead_before, coll.slab_used - coll.live_bytes);
-    try testing.expectEqual(@as(u64, 0), coll.dead_located());
+    try testing.expectEqual(@as(u64, 0), coll.dead_located);
     try testing.expectEqual(dead_before, coll.dead_unlocated);
     try testing.expectEqual(dead_before, dead_total(coll));
     // The runs came back too, and in a shape `run_of` can use.
@@ -3748,7 +3840,7 @@ test "a slab window with one live document in it is never given back" {
 
     try engine.checkpoint();
     // Most of the slab went back...
-    try testing.expect(coll.reclaimed_bytes > 100 * 2000);
+    try testing.expect(engine.reclaimed_bytes > 100 * 2000);
     // ...but not the window the survivor is in, and it still reads.
     try testing.expect(coll.run_of(survivor) != null);
     try testing.expect(std.mem.indexOf(u8, coll.doc_bytes(survivor), "xxxx") != null);
@@ -3756,7 +3848,7 @@ test "a slab window with one live document in it is never given back" {
     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,
+        coll.dead_located + coll.dead_unlocated,
     );
 
     // And once the survivor is gone, its window goes too.
@@ -3803,11 +3895,11 @@ test "a reclaimed slab window is not reusable until two publishes later" {
     while (i < 200) : (i += 1) _ = try engine.remove_by_id("app", "c", .{ .int32 = i });
     try engine.commit();
     try engine.checkpoint();
-    try testing.expect(coll.reclaimed_bytes > 0);
+    try testing.expect(engine.reclaimed_bytes > 0);
 
     // A page from the first window given back: the run's first window is all
     // dead now, so its first page is no longer the collection's.
-    const gone = @as(u32, @intCast(owned_before.window_first >> pgr.page_shift));
+    const gone = @as(u32, @intCast(owned_before.window_first() >> pgr.page_shift));
     try testing.expect(coll.run_of(@as(u64, gone) << pgr.page_shift) == null);
     try testing.expect(gone >= owned_before.first);
 
@@ -3877,8 +3969,7 @@ test "a churning collection reuses its slab instead of growing the file" {
         if (round == 0) tail_after_first = engine.pager.alloc_tail;
     }
 
-    const coll = engine.get_collection("app", "c").?;
+    try testing.expect(engine.reclaimed_bytes > 0);
-    try testing.expect(coll.reclaimed_bytes > 0);
     // Three more rounds of the same volume after the first. Anything left is
     // fragmentation the windows could not cover, not the write volume.
     const grew = engine.pager.alloc_tail - tail_after_first;
@@ -4256,8 +4347,8 @@ test "the slab counts what the appender skips" {
     // the live documents below it, and the bytes of the run outside any whole
     // window.
     try engine.checkpoint();
-    try testing.expect(coll.reclaimed_bytes > 4 * 1024 * 1024);
+    try testing.expect(engine.reclaimed_bytes > 4 * 1024 * 1024);
-    try testing.expectEqual(gap + abandoned - coll.reclaimed_bytes, engine.dead_bytes);
+    try testing.expectEqual(gap + abandoned - engine.reclaimed_bytes, engine.dead_bytes);
     try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
     try testing.expect(engine.dead_bytes < gap + 2 * pgr.map_align);
 }
@@ -4314,7 +4405,7 @@ test "a rebuild leaves behind what its own copying skipped" {
     // assert but the same statement: the counter is *not* zeroed, and it equals
     // what the collection actually has.
     try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
-    try testing.expect(coll.reclaimed_bytes > 2 * 1024 * 1024);
+    try testing.expect(engine.reclaimed_bytes > 2 * 1024 * 1024);
     try testing.expect(engine.dead_bytes > 0);
     try testing.expect(engine.dead_bytes < 4 * pgr.map_align);
 }
@@ -4691,7 +4782,7 @@ test "checkpoints reclaim under concurrent writers without losing a page" {
     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,
+        coll.dead_located + coll.dead_unlocated,
     );
 }
 
@@ -5832,7 +5923,7 @@ test "the epochs that invalidate a cursor move exactly when they must" {
     try engine.commit();
     try engine.checkpoint();
     const after_reclaim = engine.get_collection("app", "c").?.layout_epoch;
-    try testing.expect(engine.get_collection("app", "c").?.reclaimed_bytes > 0);
+    try testing.expect(engine.reclaimed_bytes > 0);
     try testing.expect(after_reclaim != quiet_before);
 
     // And the index-level token, which guards the position hint.

src/index.zig

@@ -491,7 +491,7 @@ pub const Index = struct {
         // 16 MB).
         const want_pages: u32 = @intCast(@max(
             ovf_extent_pages,
-            (overflow_bytes + pgr.page_size - 1) / pgr.page_size,
+            pgr.pages_for(overflow_bytes),
         ));
         try self.pager.reserve_pages(&self.hold, want_pages);
         const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages);

src/pager.zig

@@ -108,6 +108,10 @@ const header_hashed_len: usize = 24;
 /// cursors up to this rather than to page_size.
 pub const map_align = std.heap.page_size_min;
 
+/// Logical pages per system page. At least one: the comptime block below only
+/// requires one of the two sizes to divide the other.
+pub const pages_per_map_align: u32 = @max(1, map_align / page_size);
+
 /// Growth granularity. Large enough that growth is rare and each `setLength`
 /// covers many allocations, and a multiple of every supported system page size.
 const grow_chunk_pages: u32 = 2048; // 8 MiB
@@ -135,8 +139,9 @@ pub const Extent = struct {
 /// What a checkpoint publishes. Everything here is authoritative except the
 /// two cached counters, which are hints the engine recomputes if they look
 /// wrong.
-/// Pages a stream of `len` bytes occupies.
+/// Pages a run of `len` bytes occupies, rounded up. The one place the
-fn pages_for(len: u64) u32 {
+/// partial-page rule is written.
+pub fn pages_for(len: u64) u32 {
     return @intCast((len + page_size - 1) / page_size);
 }
 
@@ -700,7 +705,6 @@ pub const Pager = struct {
         } else {
             self.free_ready.items[i] = .{ .first = e.first + n, .pages = e.pages - n };
         }
-        self.unprotect(first, n);
         return first;
     }
 
@@ -736,19 +740,19 @@ pub const Pager = struct {
         assert(min_pages <= max_pages);
         self.alloc_lock.lockUncancelable(self.io);
         defer self.alloc_lock.unlock(self.io);
-        assert_msg(max_pages <= hold.pages, "a slab run request overran reserve_pages' promise");
+
         // A split can leave a piece at each end, so one entry may become two.
         // Out of memory before anything is disturbed: the caller falls back to
         // bumping the tail, which is what it would have done anyway.
         self.free_ready.ensureUnusedCapacity(self.gpa, 1) catch return null;
 
-        const spp: u32 = if (map_align >= page_size) @intCast(map_align / page_size) else 1;
         var best: ?usize = null;
         var best_first: u32 = 0;
         var best_take: u32 = 0;
+        var best_src: u32 = 0;
         for (self.free_ready.items, 0..) |e, i| {
-            const from = std.mem.alignForward(u32, e.first, spp);
+            const from = std.mem.alignForward(u32, e.first, pages_per_map_align);
-            const to = std.mem.alignBackward(u32, e.first + e.pages, spp);
+            const to = std.mem.alignBackward(u32, e.first + e.pages, pages_per_map_align);
             if (to <= from) continue;
             const usable = to - from;
             if (usable < min_pages) continue;
@@ -756,16 +760,14 @@ pub const Pager = struct {
             // The longest run available, so the collection switches extents as
             // rarely as possible -- every switch abandons what is left of the
             // one before it. Ties go to the smallest source run, which leaves
-            // the big ones as whole as it can.
+            // the big ones as whole as it can. `best_take` starts at zero and
-            const better = if (best) |b|
+            // every candidate takes at least `min_pages`, so "nothing yet" is
-                take > best_take or
+            // already encoded.
-                    (take == best_take and e.pages < self.free_ready.items[b].pages)
+            if (take > best_take or (take == best_take and e.pages < best_src)) {
-            else
-                true;
-            if (better) {
                 best = i;
                 best_first = from;
                 best_take = take;
+                best_src = e.pages;
             }
         }
         const i = best orelse return null;
@@ -781,10 +783,7 @@ pub const Pager = struct {
         } else {
             _ = self.free_ready.swapRemove(i);
         }
-        self.reserved_pages -= best_take;
+        self.claim_locked(hold, best_first, best_take);
-        hold.pages -= best_take;
-        self.unprotect(best_first, best_take);
-        self.mark_unpublished(best_first, best_take);
         return .{ .first = best_first, .pages = best_take };
     }
 
@@ -823,33 +822,38 @@ pub const Pager = struct {
     /// For callers already holding `alloc_lock`; see `reserve_pages_locked`.
     fn alloc_assume_reserved_locked(self: *Pager, hold: *Reservation, n: u32) u32 {
         assert(n > 0);
-        assert_msg(
-            n <= hold.pages,
-            "page allocation overran reserve_pages' promise",
-        );
-        assert_msg(
-            n <= self.reserved_pages,
-            "page allocation overran the pager's total promise",
-        );
         assert_msg(
             self.alloc_tail + n <= self.mapped_pages,
             "page allocation past the mapped end of the data file",
         );
-        self.reserved_pages -= n;
-        hold.pages -= n;
         // Reuse before growing. Without this the free list is decorative and the
         // file grows without bound under churn, because copy-on-write abandons
         // every page it touches in every generation (PLAN amendment A2).
         if (self.take_free(n)) |recycled| {
-            self.mark_unpublished(recycled, n);
+            self.claim_locked(hold, recycled, n);
             return recycled;
         }
         const first = self.alloc_tail;
         self.alloc_tail += n;
-        self.mark_unpublished(first, n);
+        self.claim_locked(hold, first, n);
         return first;
     }
 
+    /// Charge `[first, first+pages)` against the reservation and make it
+    /// writable. The one place a claim is booked, because there are two
+    /// allocation policies above it and hand-copying this is how they drift --
+    /// the copy in `alloc_slab_run` had already lost one of the preconditions.
+    fn claim_locked(self: *Pager, hold: *Reservation, first: u32, pages: u32) void {
+        assert_msg(pages <= hold.pages, "page allocation overran reserve_pages' promise");
+        assert_msg(pages <= self.reserved_pages, "page allocation overran the pager's total promise");
+        self.reserved_pages -= pages;
+        hold.pages -= pages;
+        // A recycled page was inside a published image once, so its protection
+        // has to be lifted before it is handed out again.
+        self.unprotect(first, pages);
+        self.mark_unpublished(first, pages);
+    }
+
     fn mark_unpublished(self: *Pager, first: u32, n: u32) void {
         // `grow_to` sizes the set to the mapping, and `reserve_pages` has already
         // grown the mapping past this run, so the range is in bounds.
@@ -963,7 +967,7 @@ pub const Pager = struct {
         // where the chunk stops being `grow_chunk_pages`.
         const chunk = @max(grow_chunk_pages, self.mapped_pages / 8);
         var new_pages = std.mem.alignForwardAnyAlign(u32, want_pages, chunk);
-        const sys_pages: u32 = @intCast(map_align / page_size);
+        const sys_pages: u32 = pages_per_map_align;
         if (sys_pages > 1) new_pages = std.mem.alignForward(u32, new_pages, sys_pages);
 
         if (@as(u64, new_pages) << page_shift > self.reserve.len) {
@@ -1271,7 +1275,7 @@ pub const Pager = struct {
         self.alloc_lock.lockUncancelable(self.io);
         defer self.alloc_lock.unlock(self.io);
         const bound = self.free_ready.items.len + self.free_hold.items.len + self.free_pending.items.len;
-        const pages: u32 = @intCast((8 + bound * 8 + 8 + page_size - 1) / page_size);
+        const pages: u32 = pages_for(8 + bound * 8 + 8);
         var hold: Reservation = .{};
         try self.reserve_pages_locked(&hold, pages);
         const first = self.alloc_assume_reserved_locked(&hold, pages);
@@ -2097,7 +2101,7 @@ test "a slab run comes off the free list aligned, or not at all" {
     var tp = try TmpPager.init(io, 64 << 20);
     defer tp.deinit();
     const pg = tp.pg();
-    const spp: u32 = @intCast(map_align / page_size);
+    const spp: u32 = pages_per_map_align;
 
     // A long run deliberately starting one 4 KiB page past a boundary, and a
     // short one, kept apart so coalescing cannot merge them.

tests/e2e/churn.js

@@ -32,7 +32,8 @@
 //   --doc-size <n[k|m]> payload bytes per document (default 16k)
 //   --index             create one secondary index over a churned field
 //   --mode <m>          delete-refill (default) | update
-//   --rounds <n>        delete-refill rounds (default 6)
+//   --rounds <n>        rounds in each mode; update mode splits its writes
+//                       across them (default 6)
 //   --multiple <n>      update mode: total writes as a multiple of --docs
 //   --target <x>        fail unless the steady-state ratio is at or under x
 //   --port <n>          listen port (default 27320)
@@ -100,7 +101,13 @@ const pick = (n) => Math.floor(rnd() * n);
 
 let server = null;
 let serverDead = false;
+// Bounded: the listeners below run for the life of the process and only the
+// last few lines are ever read, so an unbounded string would hold a rope
+// proportional to everything the server ever said.
 let serverLog = '';
+const noteServerLog = (d) => {
+  serverLog = (serverLog + d).slice(-65536);
+};
 
 function cleanup() {
   if (server && !serverDead) {
@@ -119,14 +126,14 @@ function startServer() {
     server = spawn(BIN, ['--port', String(opt.port), '--db', DBFILE], {
       stdio: ['ignore', 'pipe', 'pipe'],
     });
-    server.stdout.on('data', (d) => (serverLog += d));
+    server.stdout.on('data', noteServerLog);
-    server.stderr.on('data', (d) => (serverLog += d));
+    server.stderr.on('data', noteServerLog);
     server.on('error', (e) => reject(new Error(`cannot start ${BIN}: ${e.message}`)));
     server.on('exit', (code, sig) => {
       // A child that dies must fail the start, or the poll below would find a
       // *stale* server on the same port and measure the wrong database.
       serverDead = true;
-      if (code !== null && sig === null) serverLog += `\n[child exited rc=${code}]`;
+      if (code !== null && sig === null) noteServerLog(`\n[child exited rc=${code}]`);
     });
     const deadline = Date.now() + 15000;
     (async () => {
@@ -135,6 +142,10 @@ function startServer() {
           reject(new Error(`server child exited during start (port ${opt.port} busy?)\n${serverLog}`));
           return;
         }
+        // A fresh client per attempt, deliberately: a MongoClient whose first
+        // connect fails tears its topology down and every later command on it
+        // fails the same way, so reusing one turns "not up yet" into "never
+        // comes up".
         const c = new MongoClient(URL, { serverSelectionTimeoutMS: 1000 });
         try {
           await c.connect();
@@ -177,23 +188,21 @@ const MB = (n) => (n / (1 << 20)).toFixed(1);
 // a run against an older build comparable.
 let docBytes = 0;
 
-async function stats(client, coll) {
+async function stats(client, count) {
   const s = await client.db('admin').command({ serverStatus: 1 });
   const m = s.multifora || null;
-  const num = (v) => Number(v);
-  const count = await coll.countDocuments({});
   return {
     live: count * docBytes,
     count,
     file: dataFileBytes(),
     m: m && {
-      live: num(m.liveBytes),
+      live: Number(m.liveBytes),
-      dead: num(m.deadBytes),
+      dead: Number(m.deadBytes),
-      reclaimed: num(m.reclaimedBytes),
+      reclaimed: Number(m.reclaimedBytes),
-      runs: num(m.slabRuns),
+      runs: Number(m.slabRuns),
-      freeReady: num(m.freeReadyPages),
+      freeReady: Number(m.freeReadyBytes),
-      allocTail: num(m.allocTail),
+      allocTail: Number(m.allocTailBytes),
-      compactions: num(m.compactions),
+      compactions: Number(m.compactions),
     },
   };
 }
@@ -206,10 +215,10 @@ function report(label, s) {
     // had to occupy. The file never shrinks, so `ratio` is a high-water mark
     // and cannot come down however well reclamation works; `inUse` is the
     // number that moves when it does.
-    const inUse = (s.m.allocTail - s.m.freeReady) * 4096;
+    const inUse = s.m.allocTail - s.m.freeReady;
     line += `  inUse ${(inUse / s.live).toFixed(2)}x` +
       `  dead ${MB(s.m.dead)}MB  reclaimed ${MB(s.m.reclaimed)}MB` +
-      `  allocTail ${MB(s.m.allocTail * 4096)}MB  freeReady ${MB(s.m.freeReady * 4096)}MB` +
+      `  allocTail ${MB(s.m.allocTail)}MB  freeReady ${MB(s.m.freeReady)}MB` +
       `  runs ${s.m.runs}  compactions ${s.m.compactions}`;
   } else {
     line += '  (no multifora section: counters n/a)';
@@ -260,7 +269,7 @@ async function main() {
   docBytes = require('mongodb').BSON.serialize(makeDoc(0)).length;
   await insertRange(coll, 0, opt.docs);
   if (opt.index) await coll.createIndex({ k: 1 });
-  const base = await stats(client, coll);
+  const base = await stats(client, opt.docs);
   report('loaded', base);
 
   const ratios = [];
@@ -283,14 +292,26 @@ async function main() {
         live[at] = live[live.length - 1];
         live.pop();
       }
+      // One delete spec per id, not one `$in` over thousands of them. The
+      // server's planner refuses to use an index for an `$in` wider than
+      // `index.max_combos` (100), so a 5000-element one falls back to a full
+      // collection scan that re-filters every document against every member --
+      // quadratic in `--docs`, and it was the whole of this harness's runtime:
+      // the documented 40k x 16 KiB gate took 62 s and takes 11 s now, and the
+      // 150k x 200 B line went from ~480 s to 3 s. Reported ratios are
+      // unchanged, which is the point: this was the instrument's cost, not the
+      // database's.
       const bs = 5000;
       for (let i = 0; i < ids.length; i += bs) {
-        await coll.deleteMany({ _id: { $in: ids.slice(i, i + bs) } });
+        await coll.bulkWrite(
+          ids.slice(i, i + bs).map((id) => ({ deleteOne: { filter: { _id: id } } })),
+          { ordered: false },
+        );
       }
       await insertRange(coll, nextId, nextId + ids.length);
       for (let i = 0; i < ids.length; i++) live.push(nextId + i);
       nextId += ids.length;
-      ratios.push(report(`round ${r + 1}`, await stats(client, coll)));
+      ratios.push(report(`round ${r + 1}`, await stats(client, opt.docs)));
     }
   } else {
     // The same documents rewritten over and over: every rewrite leaves the old
@@ -309,12 +330,13 @@ async function main() {
         await coll.bulkWrite(ops, { ordered: false });
         done += ops.length;
       }
-      ratios.push(report(`round ${r + 1}`, await stats(client, coll)));
+      ratios.push(report(`round ${r + 1}`, await stats(client, opt.docs)));
     }
   }
 
-  const final = await stats(client, coll);
+  // The one real count in the run, and the only one the check below needs.
-  const count = final.count;
+  const count = await coll.countDocuments({});
+  const final = await stats(client, count);
   const elapsed = ((Date.now() - t0) / 1000).toFixed(0);
   console.log(`churn: ${count} documents live at the end, ${elapsed}s`);
 

tests/e2e/results/m0-gates.txt

@@ -175,7 +175,8 @@
              which has no reclamation and no `multifora` section.
 
   The harness is committed this time (`tests/e2e/churn.js`), which is half the
-  point of the block: D7.4's numbers were real and unrepeatable.
+  point of the block: D7.4's numbers were real and unrepeatable. Each line
+  above runs in 6-10 s.
 
                                   Stage 2      M1        target
   delete half and refill, 6x      1.94x        1.94x     <= 1.45x   NOT MET
@@ -232,7 +233,7 @@
 
 [M1.2] churn gate — 200-byte documents
   reproduce: node tests/e2e/churn.js --docs 150000 --doc-size 200 --index \
-               --mode delete-refill --rounds 4
+               --mode delete-refill --rounds 4        (3 s)
   Predicted in advance, in the plan, as a pass rather than a fault:
 
     reclaimed 0.0 MB over four rounds, exactly as forecast.