src/db.zig

@@ -36,6 +36,67 @@ const slab_extent_pages: u32 = (8 * 1024 * 1024) / 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
+/// -- a whole system page is the smallest thing `mark_appendable` and
+/// `protect_stable` can hand back -- so a counter never exceeds `map_align`,
+/// and the width follows from that rather than being chosen.
+///
+/// Two bytes per window on the usual platforms. That is the entire memory cost
+/// of knowing where a collection's garbage is: 2.7 MB for a 21 GB slab on
+/// 16 KiB pages, 10.6 MB on 4 KiB ones. The alternative shapes -- an interval
+/// set, a free-run list -- cost memory proportional to the number of *dead
+/// documents*, which for 200-byte documents at that scale is gigabytes, and
+/// would make `evict_doc` allocate after the write is already committed.
+const WindowDead = if (pgr.map_align <= std.math.maxInt(u16)) u16 else u32;
+
+/// A run of pages a collection's slab owns, plus where its garbage is.
+///
+/// This replaced a bare `pgr.Extent` because an extent can only be given back
+/// whole, and a churning collection almost never empties one. A run is split
+/// instead: the windows inside it with nothing live left go to the pager and
+/// the run becomes two shorter ones. So the list is kept sorted by page number,
+/// which makes `run_of` a binary search and the sortedness itself an assert --
+/// allocation order stopped being meaningful once a run could be recycled to a
+/// *lower* address than one already owned.
+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
+    /// holds nothing live and can be handed back.
+    dead: []WindowDead,
+
+    /// Windows wholly inside the pages `[first, first+pages)`. The bytes
+    /// outside them -- below `window_first`, and the tail after the last whole
+    /// window -- are real slab that documents do live in; their garbage is
+    /// counted in `Collection.dead_unlocated` instead, because it can never be
+    /// reclaimed on its own.
+    fn window_count(first: u32, pages: u32) usize {
+        const from = @as(u64, first) << pgr.page_shift;
+        const to = from + (@as(u64, pages) << pgr.page_shift);
+        const wf = std.mem.alignForward(u64, from, pgr.map_align);
+        const we = std.mem.alignBackward(u64, to, pgr.map_align);
+        return if (we > wf) @intCast((we - wf) / pgr.map_align) else 0;
+    }
+
+    fn start(self: SlabRun) u64 {
+        return @as(u64, self.first) << pgr.page_shift;
+    }
+
+    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;
+    }
+};
+
 pub const Collection = struct {
     /// Documents live as canonical BSON bytes in the data file, in extents this
     /// collection owns; the map holds each document's offset. Those are
@@ -53,8 +114,9 @@ pub const Collection = struct {
     doc_count: u64,
     /// The data file this collection's documents live in.
     pager: *pgr.Pager,
-    /// Extents owned by this collection's slab, in allocation order.
-    slab_extents: std.ArrayListUnmanaged(pgr.Extent),
+    /// Page runs owned by this collection's slab, sorted by page number, each
+    /// carrying the map of where its dead bytes are. See `SlabRun`.
+    slab_runs: std.ArrayListUnmanaged(SlabRun),
     /// Absolute file offset of the next document write, and the end of the
     /// extent it falls in.
     slab_tail: u64,
@@ -72,6 +134,25 @@ pub const Collection = struct {
     /// collection so dropping one can move the right amount from the engine's
     /// live total to its dead total.
     live_bytes: u64,
+    /// Garbage this collection knows it has but cannot place in a window: the
+    /// edges of a run that fall outside any whole window, and -- the larger
+    /// share -- everything that died before the last restart, since the window
+    /// map is not persisted.
+    ///
+    /// It exists to keep one identity exact:
+    ///
+    ///     sum of every window counter + dead_unlocated == slab_used - live_bytes
+    ///
+    /// Without it the two halves of the accounting would drift apart at every
+    /// open, and there would be no assert that could tell drift from a lost
+    /// 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,
+    /// 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
@@ -116,11 +197,13 @@ pub const Collection = struct {
         var self: Collection = .{
             .doc_count = 0,
             .pager = pager,
-            .slab_extents = .empty,
+            .slab_runs = .empty,
             .slab_tail = 0,
             .slab_end = 0,
             .slab_used = 0,
             .live_bytes = 0,
+            .dead_unlocated = 0,
+            .reclaimed_bytes = 0,
             .hold = .{},
             .indexes = .empty,
             .id_index = undefined,
@@ -146,6 +229,117 @@ pub const Collection = struct {
         return null;
     }
 
+    /// Take ownership of a page run, keeping `slab_runs` sorted by page number.
+    /// The only place a run enters the list, so the sort order and the window
+    /// map are established together and cannot disagree.
+    fn insert_run(self: *Collection, gpa: std.mem.Allocator, first: u32, pages: u32) !void {
+        const dead = try gpa.alloc(WindowDead, SlabRun.window_count(first, pages));
+        errdefer gpa.free(dead);
+        @memset(dead, 0);
+        var at: usize = 0;
+        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.
+        if (at > 0) {
+            const prev = self.slab_runs.items[at - 1];
+            assert_msg(prev.first + prev.pages <= first, "a slab run overlaps the one below it");
+        }
+        if (at < self.slab_runs.items.len) {
+            assert_msg(first + pages <= self.slab_runs.items[at].first, "a slab run overlaps the one above it");
+        }
+        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,
+        });
+    }
+
+    /// The run holding `off`, or null if no run does. Binary search, which the
+    /// sorted list is for: `mark_dead` runs once per evicted document, and a
+    /// collection with a fragmented slab can own thousands of runs.
+    fn run_of(self: *const Collection, off: u64) ?usize {
+        const page: u32 = @intCast(off >> pgr.page_shift);
+        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 (page < r.first) {
+                hi = mid;
+            } else if (page >= r.first + r.pages) {
+                lo = mid + 1;
+            } else {
+                return mid;
+            }
+        }
+        return null;
+    }
+
+    /// Record that `[off, off+len)` of slab is garbage.
+    ///
+    /// Infallible, and that is the constraint the whole representation was
+    /// chosen around: the two callers are `evict_doc`, which runs after the
+    /// log record is already durable, and the appender's skip accounting. An
+    /// allocation here would be a failure with nowhere to report it.
+    ///
+    /// Bytes that fall outside a whole window -- the head of a run before its
+    /// first window boundary, and the tail after its last -- go to
+    /// `dead_unlocated`. They are not lost, only unreclaimable on their own.
+    fn mark_dead(self: *Collection, off: u64, len: u64) void {
+        if (len == 0) return;
+        const ri = self.run_of(off) orelse {
+            assert_msg(false, "dead slab bytes fall outside every run the collection owns");
+            unreachable;
+        };
+        const r = &self.slab_runs.items[ri];
+        const stop = off + len;
+        // A document is written inside one extent by construction
+        // (`slab_reserve` never lets an append cross `slab_end`), so a dead
+        // range that crosses a run boundary means an offset from a different
+        // 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");
+        var pos = off;
+        if (pos < r.window_first) {
+            const n = @min(stop, r.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_end = r.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");
+            r.dead[w] += @intCast(n);
+            pos += n;
+        }
+        if (pos < stop) self.dead_unlocated += stop - pos;
+    }
+
+    /// Garbage this collection has placed in windows. Walks every window, so it
+    /// belongs to the reclamation scan and to tests, not to a hot path.
+    fn dead_located(self: *const Collection) u64 {
+        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);
+        self.slab_runs.clearRetainingCapacity();
+    }
+
     /// Append `bytes` to the slab, returning its flat offset. The last
     /// segment holds up to `slab_segment_size`; a full one starts the next.
     /// Make room for a document of `len` bytes, so the append that follows
@@ -210,7 +404,7 @@ pub const Collection = struct {
         ));
         try self.pager.reserve_pages(&self.hold, want_pages);
         const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages);
-        try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages });
+        try self.insert_run(gpa, first, want_pages);
         self.slab_tail = @as(u64, first) << pgr.page_shift;
         self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
         return skipped;
@@ -230,8 +424,15 @@ pub const Collection = struct {
     /// Two collections churning against a checkpoint every 32 MiB skip up to a
     /// system page each per checkpoint, and an abandoned extent tail can be
     /// most of 8 MiB. Counted here, that garbage arms compaction like any other.
+    ///
+    /// Skipped slab always starts at the cursor -- all three callers write off
+    /// the bytes in front of it and then move it -- so this is also where the
+    /// window map learns about it. That matters more for the abandoned tail
+    /// than for the round-up: most of 8 MiB of a run is whole windows, dead on
+    /// arrival, and reclaiming them is free.
     fn note_skip(self: *Collection, bytes: u64) u64 {
         self.slab_used += bytes;
+        self.mark_dead(self.slab_tail, bytes);
         return bytes;
     }
 
@@ -619,10 +820,11 @@ pub const Engine = struct {
         coll.indexes.deinit(self.gpa);
         // Give the slab's pages back. They become reusable two generations
         // later, so a fallback to the previous image still finds them intact.
-        for (coll.slab_extents.items) |e| {
-            self.pager.free_pages(e.first, e.pages) catch {};
+        for (coll.slab_runs.items) |r| {
+            self.pager.free_pages(r.first, r.pages) catch {};
         }
-        coll.slab_extents.deinit(self.gpa);
+        coll.free_runs(self.gpa);
+        coll.slab_runs.deinit(self.gpa);
         self.gpa.destroy(coll);
     }
 
@@ -700,6 +902,11 @@ pub const Engine = struct {
         coll.doc_count -= 1;
         assert_msg(coll.live_bytes >= old_bytes.len, "evicting a document would underflow the collection's live bytes");
         coll.live_bytes -= old_bytes.len;
+        // Last, and after every index entry naming these bytes is gone. That
+        // ordering is what makes window reclamation safe to do by counting
+        // alone: a window only reaches `map_align` dead once every document
+        // touching it has been through here, so nothing reachable is inside it.
+        coll.mark_dead(off, old_bytes.len);
     }
 
     /// `bson.encode_key` of a stored document's `_id`, owned by the caller.
@@ -1582,16 +1789,21 @@ pub const Engine = struct {
         try coll.lock.lock(self.io);
         defer coll.lock.unlock(self.io);
 
-        const old_extents = try self.gpa.dupe(pgr.Extent, coll.slab_extents.items);
+        var old_extents = try self.gpa.alloc(pgr.Extent, coll.slab_runs.items.len);
         defer self.gpa.free(old_extents);
+        for (coll.slab_runs.items, 0..) |r, i| old_extents[i] = .{ .first = r.first, .pages = r.pages };
 
         // Fresh slab. The old extents stay allocated until the free list
         // releases them, two generations on.
-        coll.slab_extents.clearRetainingCapacity();
+        coll.free_runs(self.gpa);
         coll.slab_tail = 0;
         coll.slab_end = 0;
         coll.slab_used = 0;
         coll.live_bytes = 0;
+        // The rebuild is the one place the located and unlocated halves are
+        // both reset: every byte it copies is live, so a fresh slab has no
+        // garbage to place. Anything the copy skips is marked as it happens.
+        coll.dead_unlocated = 0;
 
         // Walk in _id order, which is also the order the new slab ends up in --
         // so a later scan reads it sequentially.
@@ -1848,7 +2060,7 @@ pub const Engine = struct {
             while (coll_it.next()) |ce| {
                 const coll = ce.value_ptr.*;
                 // Everything below this line is written by a collection's own
-                // writer under its own lock, and `slab_extents` is an ArrayList
+                // writer under its own lock, and `slab_runs` is an ArrayList
                 // that `slab_reserve` appends to -- so reading it under only the
                 // shared catalog lock could walk a slice a concurrent append had
                 // already reallocated. Lock order is catalog then collection,
@@ -1866,10 +2078,14 @@ pub const Engine = struct {
                     "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);
-                    try put_u32(gpa, out, e.pages);
+                // Runs, not extents, but the same two u32s: only the window map
+                // is new and it is deliberately not persisted (see
+                // `dead_unlocated`), so `catalog_version` stays 1 and there is
+                // no second read path to keep working.
+                try put_u32(gpa, out, @intCast(coll.slab_runs.items.len));
+                for (coll.slab_runs.items) |r| {
+                    try put_u32(gpa, out, r.first);
+                    try put_u32(gpa, out, r.pages);
                 }
                 try put_u32(gpa, out, @intCast(coll.indexes.items.len + 1));
                 try write_index_catalog(gpa, out, &coll.id_index);
@@ -1950,12 +2166,21 @@ pub const Engine = struct {
                 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;
+                // An open knows *that* the collection has garbage but not
+                // *where*: the window map is rebuilt empty, and the whole
+                // amount starts out unlocated. The consequence runs one way --
+                // a forgotten dead byte is a window that is not handed back,
+                // never a live window that is.
+                coll.dead_unlocated = coll.slab_used - coll.live_bytes;
                 const nex = try r.read_u32();
                 var e: u32 = 0;
                 while (e < nex) : (e += 1) {
                     const first = try r.read_u32();
                     const pages = try r.read_u32();
-                    try coll.slab_extents.append(self.gpa, .{ .first = first, .pages = pages });
+                    // Sorted insert rather than append: a catalog written
+                    // before runs were address-ordered holds them in
+                    // allocation order, and `run_of` is a binary search.
+                    try coll.insert_run(self.gpa, first, pages);
                 }
                 const nix = try r.read_u32();
                 // Index 0 is the implicit _id_, already created by
@@ -2787,8 +3012,8 @@ test "an append after a checkpoint keeps its extent instead of abandoning it" {
     try engine.commit();
 
     const coll = engine.get_collection("app", "users").?;
-    try testing.expectEqual(@as(usize, 1), coll.slab_extents.items.len);
-    const extent_start = @as(u64, coll.slab_extents.items[0].first) << pgr.page_shift;
+    try testing.expectEqual(@as(usize, 1), coll.slab_runs.items.len);
+    const extent_start = coll.slab_runs.items[0].start();
 
     try engine.checkpoint();
     const tail_at_checkpoint = coll.slab_tail;
@@ -2801,7 +3026,7 @@ test "an append after a checkpoint keeps its extent instead of abandoning it" {
     try engine.insert("app", "users", &second, &env.gen);
     try engine.commit();
 
-    try testing.expectEqual(@as(usize, 1), coll.slab_extents.items.len);
+    try testing.expectEqual(@as(usize, 1), coll.slab_runs.items.len);
     // A page or two for the tree's copy-on-write is expected; a whole slab
     // extent is the regression this guards against.
     try testing.expect(engine.pager.alloc_tail < tail_before + slab_extent_pages);
@@ -2831,6 +3056,300 @@ test "an append after a checkpoint keeps its extent instead of abandoning it" {
     }
 }
 
+/// A document of roughly `size` bytes, so a test can fill extents without
+/// writing tens of thousands of records.
+fn make_padded(gpa: std.mem.Allocator, id: i32, size: usize) !bson.Document {
+    var arena = std.heap.ArenaAllocator.init(gpa);
+    errdefer arena.deinit();
+    const a = arena.allocator();
+    const pad = try a.alloc(u8, size);
+    @memset(pad, 'x');
+    const pairs = try a.alloc(bson.Pair, 2);
+    pairs[0] = .{ .key = try a.dupe(u8, "_id"), .value = .{ .int32 = id } };
+    pairs[1] = .{ .key = try a.dupe(u8, "pad"), .value = .{ .string = pad } };
+    return .{ .arena = arena, .pairs = pairs };
+}
+
+/// Every dead byte the collection knows about, placed or not.
+fn dead_total(coll: *const Collection) u64 {
+    return coll.dead_located() + coll.dead_unlocated;
+}
+
+test "every dead slab byte is counted in exactly one place" {
+    // The identity the window map rests on:
+    //
+    //     sum of window counters + dead_unlocated == slab_used - live_bytes
+    //
+    // The left side is where the garbage is, the right side is how much there
+    // is; reclamation reads the first and the compaction trigger reads the
+    // second, so a drift between them is a rebuild that fires on a clean
+    // database or a window that is handed back with a document in it.
+    //
+    // Both kinds of death are exercised: evicted documents, and the slab the
+    // appender writes off when a checkpoint freezes the page its cursor points
+    // into.
+    //
+    // Mutation check: drop the `mark_dead` call from `note_skip`, or the
+    // `mark_dead` call from `evict_doc` -- each removes one of the two ways
+    // slab dies and the sides part company by that amount. The run *edges* are
+    // covered separately, by the test below, because whether a run's start is
+    // `map_align`-aligned is up to the allocator and not something an
+    // engine-level test can arrange.
+    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 may intervene
+    try engine.lock();
+    defer engine.unlock();
+
+    var i: i32 = 0;
+    while (i < 40) : (i += 1) {
+        var d = try make_padded(gpa, i, 6000);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+        // A checkpoint every few documents, so the appender keeps having to
+        // round its cursor up to a system page and skipping the bytes between.
+        if (@mod(i, 7) == 6) {
+            try engine.commit();
+            try engine.checkpoint();
+        }
+    }
+    try engine.commit();
+
+    const coll = engine.get_collection("app", "c").?;
+    // Some skipping must actually have happened, or the test proves only the
+    // easy half. 6000-byte documents never land flush against a page boundary.
+    try testing.expect(coll.slab_used > coll.live_bytes);
+    try testing.expectEqual(coll.slab_used - coll.live_bytes, dead_total(coll));
+
+    // Now the other kind: evictions.
+    i = 0;
+    while (i < 40) : (i += 2) {
+        try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = i }));
+    }
+    try engine.commit();
+    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);
+}
+
+test "dead bytes outside a whole window are counted but not placed" {
+    // A run is allocated in 4 KiB pages but reclaimed in `map_align` windows,
+    // so unless the allocator happens to hand back an aligned run there is a
+    // head below its first window boundary and a tail above its last. Bytes
+    // that die there can never be given back on their own -- but they are
+    // still garbage, and if they were simply dropped the amount of garbage the
+    // collection reports would fall short of the amount it has, which is a
+    // compaction that never fires.
+    //
+    // Driven against `mark_dead` directly: the alignment of a real slab extent
+    // is the allocator's business and an engine-level test cannot arrange for
+    // an unaligned one.
+    //
+    // Mutation checks, each red on its own: drop the trailing `if (pos <
+    // stop)` and the tail bytes go uncounted; drop the leading `if (pos <
+    // r.window_first)` and the window index underflows instead, which takes
+    // down half the suite.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const gpa = testing.allocator;
+
+    var tmp = try TmpLog.init(gpa);
+    defer tmp.deinit(gpa);
+    var engine = try Engine.open(gpa, env.io, tmp.path);
+    defer engine.deinit();
+    try engine.lock();
+    defer engine.unlock();
+
+    var d = try make_doc(gpa, 1, "alice");
+    defer d.deinit();
+    try engine.insert("app", "c", &d, &env.gen);
+    const coll = engine.get_collection("app", "c").?;
+
+    // A run deliberately starting one 4 KiB page past a window boundary, long
+    // enough to hold two whole windows plus a partial one at each end.
+    const pages_per_window: u32 = @intCast(pgr.map_align / pgr.page_size);
+    if (pages_per_window < 2) return error.SkipZigTest; // no edges to test
+    const owned = coll.slab_runs.items[0];
+    const aligned = std.mem.alignForward(u32, owned.first + owned.pages + 8, pages_per_window);
+    try coll.insert_run(gpa, aligned + 1, 3 * pages_per_window);
+    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_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.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(
+        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
+    // tail bytes did not leak into a counter that would hand a window back.
+    try testing.expectEqual(@as(WindowDead, pgr.map_align), r.dead[0]);
+    try testing.expectEqual(@as(WindowDead, 0), r.dead[1]);
+}
+
+test "a slab run holds whole documents" {
+    // `mark_dead` charges a document to the run holding its first byte and
+    // asserts the rest is in the same run. That is only sound because
+    // `slab_reserve` never lets an append cross `slab_end` -- so check it
+    // against a collection that owns several runs, including one taken for a
+    // single oversized document.
+    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);
+    try engine.lock();
+    defer engine.unlock();
+
+    var small = try make_padded(gpa, 1, 1000);
+    defer small.deinit();
+    try engine.insert("app", "c", &small, &env.gen);
+    // Larger than the standard 8 MiB extent, so it gets a run of its own and
+    // the next document forces a third.
+    var huge = try make_padded(gpa, 2, 9 * 1024 * 1024);
+    defer huge.deinit();
+    try engine.insert("app", "c", &huge, &env.gen);
+    var after = try make_padded(gpa, 3, 1000);
+    defer after.deinit();
+    try engine.insert("app", "c", &after, &env.gen);
+    try engine.commit();
+
+    const coll = engine.get_collection("app", "c").?;
+    try testing.expect(coll.slab_runs.items.len >= 2);
+    // Sorted by page number, and non-overlapping.
+    for (coll.slab_runs.items[1..], 0..) |r, k| {
+        const prev = coll.slab_runs.items[k];
+        try testing.expect(prev.first + prev.pages <= r.first);
+    }
+    var it = coll.id_index.iter();
+    while (it.next()) |entry| {
+        const ri = coll.run_of(entry.off) orelse return error.TestUnexpectedResult;
+        const r = coll.slab_runs.items[ri];
+        try testing.expect(entry.off + coll.doc_bytes(entry.off).len <= r.end());
+    }
+}
+
+test "a slab run recycled to a lower address keeps the list sorted" {
+    // Runs used to be held in allocation order, which was fine while an extent
+    // could only be appended. Reclamation makes a recycled run arrive at an
+    // address *below* one the collection already owns, and `run_of` is a binary
+    // search -- so the list has to be ordered by page, not by age.
+    //
+    // Exercised on the structure directly: producing a lower-addressed
+    // allocation through the engine needs the free list to be primed, which is
+    // 3.3's business, and this invariant should hold before then.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const gpa = testing.allocator;
+
+    var tmp = try TmpLog.init(gpa);
+    defer tmp.deinit(gpa);
+    var engine = try Engine.open(gpa, env.io, tmp.path);
+    defer engine.deinit();
+    try engine.lock();
+    defer engine.unlock();
+
+    var d = try make_doc(gpa, 1, "alice");
+    defer d.deinit();
+    try engine.insert("app", "c", &d, &env.gen);
+    const coll = engine.get_collection("app", "c").?;
+
+    // Three more runs, arriving out of order and clear of the one the insert
+    // took. They are never written to, so no pages need to exist.
+    const base: u32 = coll.slab_runs.items[0].first + coll.slab_runs.items[0].pages + 16;
+    try coll.insert_run(gpa, base + 200, 8);
+    try coll.insert_run(gpa, base, 8);
+    try coll.insert_run(gpa, base + 100, 8);
+    try testing.expectEqual(@as(usize, 4), coll.slab_runs.items.len);
+    for (coll.slab_runs.items[1..], 0..) |r, k| {
+        try testing.expect(coll.slab_runs.items[k].first < r.first);
+    }
+    // And every one of them is findable at its own address, which is the point
+    // of the ordering.
+    for ([_]u32{ base, base + 100, base + 200 }) |first| {
+        const off = @as(u64, first) << pgr.page_shift;
+        const ri = coll.run_of(off) orelse return error.TestUnexpectedResult;
+        try testing.expectEqual(first, coll.slab_runs.items[ri].first);
+    }
+    try testing.expect(coll.run_of(@as(u64, base + 8) << pgr.page_shift) == null);
+}
+
+test "a restart forgets where the garbage is, not that there is any" {
+    // The window map is deliberately not persisted: it would be a new catalog
+    // field, a new version, and a second read path, to save re-deriving
+    // something the collection can live without. What must survive is the
+    // *amount*, because that is what arms compaction -- so an open puts the
+    // whole of it into `dead_unlocated` and the identity still holds with every
+    // window counter at zero.
+    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 dead_before: u64 = 0;
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        engine.compact_threshold = std.math.maxInt(u64);
+        try engine.lock();
+        defer engine.unlock();
+        var i: i32 = 0;
+        while (i < 20) : (i += 1) {
+            var d = try make_padded(gpa, i, 4000);
+            defer d.deinit();
+            try engine.insert("app", "c", &d, &env.gen);
+        }
+        i = 0;
+        while (i < 20) : (i += 2) {
+            _ = try engine.remove_by_id("app", "c", .{ .int32 = i });
+        }
+        try engine.commit();
+        const coll = engine.get_collection("app", "c").?;
+        try testing.expect(coll.dead_located() > 0);
+        dead_before = coll.slab_used - coll.live_bytes;
+        try engine.checkpoint();
+    }
+
+    var engine2 = try Engine.open(gpa, io, tmp.path);
+    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(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.
+    var it = coll.id_index.iter();
+    while (it.next()) |entry| try testing.expect(coll.run_of(entry.off) != null);
+}
+
 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
@@ -3401,7 +3920,7 @@ test "a checkpoint runs alongside writers on several collections" {
     // counters while holding only the *shared catalog* lock -- and a writer
     // holds that same lock shared, taking the collection's lock exclusively.
     // So the snapshot walked structures its owner was free to mutate, and
-    // `slab_extents` is an ArrayList a new extent appends to: a reallocation
+    // `slab_runs` is an ArrayList a new extent inserts into: a reallocation
     // mid-walk leaves the serializer reading freed memory.
     //
     // Several collections rather than one, because the interesting overlap is a
@@ -4485,12 +5004,7 @@ fn id_key_for(gpa: std.mem.Allocator, v: bson.Value) ![]u8 {
 /// index from one still holding pre-rebuild offsets -- the old bytes are on the
 /// free list rather than overwritten, so reading them still succeeds.
 fn offset_in_extents(coll: *const Collection, off: u64) bool {
-    for (coll.slab_extents.items) |e| {
-        const first = @as(u64, e.first) << pgr.page_shift;
-        const end = first + (@as(u64, e.pages) << pgr.page_shift);
-        if (off >= first and off < end) return true;
-    }
-    return false;
+    return coll.run_of(off) != null;
 }
 
 /// Document bytes at an absolute file offset, without needing the Collection.