src/db.zig

@@ -121,9 +121,16 @@ pub const Collection = struct {
     /// extent it falls in.
     slab_tail: u64,
     slab_end: u64,
-    /// Document bytes written into this collection's slab since the last
+    /// Slab this collection has consumed and not yet given back. `slab_tail`
-    /// rebuild. `slab_tail` cannot answer that -- it is an absolute file offset,
+    /// cannot answer that -- it is an absolute file offset, so it jumps forward
-    /// so it jumps forward whenever a fresh extent is taken.
+    /// whenever a fresh extent is taken.
+    ///
+    /// It used to mean "bytes ever appended since the last rebuild", which was
+    /// the same thing while a rebuild was the only way to get slab back. Window
+    /// reclamation subtracts from it, and that is what keeps
+    /// `slab_used - live_bytes` equal to the garbage the collection still has
+    /// -- with no new persistent field, since both halves are already in the
+    /// catalog.
     slab_used: u64,
     /// This collection's outstanding page promise, for the document slab. Per
     /// collection because concurrent writers must not release each other's --
@@ -148,6 +155,15 @@ 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,
+    /// Windows whose counter has reached `map_align`, i.e. how much there is
+    /// for the next checkpoint to give back.
+    ///
+    /// It exists so that a checkpoint costs nothing on a collection with
+    /// nothing to reclaim. Scanning would otherwise be O(slab) per checkpoint
+    /// whatever the workload -- and the workload this design is *known* not to
+    /// 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
@@ -203,6 +219,7 @@ pub const Collection = struct {
             .slab_used = 0,
             .live_bytes = 0,
             .dead_unlocated = 0,
+            .full_windows = 0,
             .reclaimed_bytes = 0,
             .hold = .{},
             .indexes = .empty,
@@ -316,7 +333,9 @@ pub const Collection = struct {
             // 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;
             pos += n;
         }
         if (pos < stop) self.dead_unlocated += stop - pos;
@@ -338,6 +357,122 @@ pub const Collection = struct {
     fn free_runs(self: *Collection, gpa: std.mem.Allocator) void {
         for (self.slab_runs.items) |r| gpa.free(r.dead);
         self.slab_runs.clearRetainingCapacity();
+        self.full_windows = 0;
+    }
+
+    /// One piece of a run that survives reclamation, with a window map of its
+    /// own copied out of the original.
+    ///
+    /// Every kept piece gets a fresh array, including a run nothing was taken
+    /// from. Moving the original array instead would save a copy and make the
+    /// failure path have to know which arrays it still owns -- the version that
+    /// tried it had a double free in the out-of-memory case, which is the one
+    /// case nothing exercises.
+    fn keep_piece(
+        out: *std.ArrayListUnmanaged(SlabRun),
+        gpa: std.mem.Allocator,
+        r: SlabRun,
+        p0: u32,
+        p1: u32,
+    ) !void {
+        const wf = std.mem.alignForward(u64, @as(u64, p0) << pgr.page_shift, pgr.map_align);
+        const we = std.mem.alignBackward(u64, @as(u64, p1) << pgr.page_shift, pgr.map_align);
+        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);
+        @memcpy(dead, r.dead[base..][0..count]);
+        try out.append(gpa, .{ .first = p0, .pages = p1 - p0, .window_first = wf, .dead = dead });
+    }
+
+    /// Give back every window with nothing live left in it, splitting the runs
+    /// around what is kept. Returns the bytes handed to the pager.
+    ///
+    /// Counting is the whole test: a window reaches `map_align` dead only once
+    /// every document with a byte in it has been through `evict_doc`, which
+    /// removes its index entries before marking it. So "no live bytes" and "no
+    /// reference to these bytes" are the same statement, and nothing has to be
+    /// scanned to establish it.
+    ///
+    /// Fallible, and arranged so a failure changes nothing: the replacement
+    /// list is built whole before the old one is touched. The garbage simply
+    /// stays and the next checkpoint tries again.
+    fn reclaim_windows(self: *Collection, gpa: std.mem.Allocator) !u64 {
+        assert_msg(
+            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.
+        assert_msg(
+            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);
+            out.deinit(gpa);
+        }
+        var give: std.ArrayListUnmanaged(pgr.Extent) = .empty;
+        defer give.deinit(gpa);
+
+        var freed: u64 = 0;
+        for (self.slab_runs.items) |r| {
+            var keep_from = r.first;
+            var i: usize = 0;
+            while (i < r.dead.len) {
+                if (r.dead[i] != pgr.map_align) {
+                    i += 1;
+                    continue;
+                }
+                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 to = r.window_first + j * pgr.map_align;
+                // The appender's own extent is off limits, and not by
+                // filtering: bytes above the cursor have never been written, so
+                // no window covering them can have reached `map_align` dead.
+                // Tripping this 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);
+                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) {
+                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;
     }
 
     /// Append `bytes` to the slab, returning its flat offset. The last
@@ -2314,6 +2449,62 @@ pub const Engine = struct {
         self.committed_seq = 0;
     }
 
+    /// Hand back every slab window with nothing live left in it, across every
+    /// collection. The first phase of a checkpoint.
+    ///
+    /// Inside the checkpoint rather than a hook after it, and that placement is
+    /// the whole safety argument. Reclamation changes two things: it splits
+    /// `slab_runs`, which the catalog describes, and it calls
+    /// `pager.free_pages`, which the free list describes. The checkpoint's
+    /// single `publish` makes both durable together, so a crash before it
+    /// leaves the old catalog and the old free list -- no reclamation happened
+    /// -- and a crash after leaves both describing the new ownership. There is
+    /// no order in between to get wrong, and no new record type or replay path.
+    ///
+    /// Batched with the checkpoint for a second reason: the write path pays
+    /// only for a counter update, and the cadence of actually returning pages
+    /// is the checkpoint threshold rather than per-delete.
+    ///
+    /// Catalog shared, then each collection exclusive, one at a time -- the
+    /// order `compact` and `write_catalog` both use.
+    fn reclaim_slabs(self: *Engine) void {
+        self.catalog_lock.lockSharedUncancelable(self.io);
+        defer self.catalog_lock.unlockShared(self.io);
+        var db_it = self.dbs.iterator();
+        while (db_it.next()) |db_entry| {
+            var coll_it = db_entry.value_ptr.collections.iterator();
+            while (coll_it.next()) |ce| self.reclaim_collection(ce.value_ptr.*);
+        }
+    }
+
+    fn reclaim_collection(self: *Engine, coll: *Collection) void {
+        coll.lock.lockUncancelable(self.io);
+        defer coll.lock.unlock(self.io);
+        // The common case, and the reason this is affordable at every
+        // checkpoint: a collection with no full window is not scanned at all.
+        if (coll.full_windows == 0) return;
+        // Out of memory here means the garbage stays where it is. Nothing is
+        // lost and the next checkpoint tries again.
+        const freed = coll.reclaim_windows(self.gpa) catch return;
+        if (freed == 0) return;
+        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.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,
+        // since the pages are only on the free list, so the answer would be
+        // plausible garbage rather than an error. `cursor_still_valid` compares
+        // this epoch and kills such a cursor with QueryPlanKilled, which is
+        // what a rebuild already does to it.
+        //
+        // Only when something was actually given back: a collection that
+        // reclaimed nothing must not have its cursors killed on the cadence of
+        // the checkpoint.
+        self.layout_epoch_seq += 1;
+        coll.layout_epoch = self.layout_epoch_seq;
+    }
+
     /// Publish the current state as a checkpoint.
     ///
     /// The watermark equals the sequence the log has already made durable, never
@@ -2323,6 +2514,7 @@ pub const Engine = struct {
     /// D6) reduced to an ordering.
     pub fn checkpoint(self: *Engine) !void {
         try self.commit();
+        self.reclaim_slabs();
 
         var buf: std.ArrayListUnmanaged(u8) = .empty;
         defer buf.deinit(self.gpa);
@@ -3350,6 +3542,134 @@ test "a restart forgets where the garbage is, not that there is any" {
     while (it.next()) |entry| try testing.expect(coll.run_of(entry.off) != null);
 }
 
+/// Whether `page` falls in one of the pager's free-list generations.
+fn in_extents(list: []const pgr.Extent, page: u32) bool {
+    for (list) |e| if (page >= e.first and page < e.first + e.pages) return true;
+    return false;
+}
+
+test "a slab window with one live document in it is never given back" {
+    // The load-bearing test of window reclamation. A window goes back when its
+    // counter reaches `map_align`, which is a statement about *bytes*, not
+    // about documents -- so the thing that must never happen is a window handed
+    // to the pager while a document still sits in it. The document would still
+    // read, because a freed page is only on a list, so the failure would be
+    // silent until the pages were handed out again and overwritten.
+    //
+    // Mutation check: relax the fullness test in `reclaim_windows` to
+    // `r.dead[i] + 2048 < pgr.map_align`, so a window 2 KiB short of empty
+    // qualifies. On its own that aborts the whole suite on the append-cursor
+    // assertion instead -- the appender's own window is the first thing a
+    // loosened test reaches, which is worth knowing. Drop that assertion too
+    // and this is the test that goes red, alone.
+    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 < 200) : (i += 1) {
+        var d = try make_padded(gpa, i, 2000);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    const coll = engine.get_collection("app", "c").?;
+
+    // Everything dies but one document, roughly in the middle of the slab.
+    i = 0;
+    while (i < 200) : (i += 1) {
+        if (i == 100) continue;
+        try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = i }));
+    }
+    try engine.commit();
+    const survivor_enc = try id_key_for(gpa, bson.Value{ .int32 = 100 });
+    defer gpa.free(survivor_enc);
+    const survivor = coll.id_index.lookup_exact(survivor_enc).?;
+
+    try engine.checkpoint();
+    // Most of the slab went back...
+    try testing.expect(coll.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);
+    // The accounting followed the pages: what is left is what is left.
+    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,
+    );
+
+    // And once the survivor is gone, its window goes too.
+    try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = 100 }));
+    try engine.commit();
+    try engine.checkpoint();
+    try testing.expect(coll.run_of(survivor) == null);
+}
+
+test "a reclaimed slab window is not reusable until two publishes later" {
+    // Reclamation hands pages to `free_pages`, which withholds them for two
+    // generations -- and it has to, because the image one generation back is
+    // still the fallback a crash would open, and its catalog still claims them.
+    // Handing them straight out would let a write land on pages the recovery
+    // path is about to read as documents.
+    //
+    // Asserted on the pager's own lists rather than on `free_ready_pages()`,
+    // whose total also moves for copy-on-write victims and the catalog stream.
+    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 i: i32 = 0;
+    while (i < 200) : (i += 1) {
+        var d = try make_padded(gpa, i, 2000);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    const coll = engine.get_collection("app", "c").?;
+    const owned_before = coll.slab_runs.items[0];
+
+    i = 0;
+    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);
+
+    // 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));
+    try testing.expect(coll.run_of(@as(u64, gone) << pgr.page_shift) == null);
+    try testing.expect(gone >= owned_before.first);
+
+    // One publish has happened, so it is held, not ready.
+    try testing.expect(!in_extents(engine.pager.free_ready.items, gone));
+    try testing.expect(in_extents(engine.pager.free_hold.items, gone));
+
+    // The second publish is what makes it allocatable.
+    try engine.checkpoint();
+    try testing.expect(in_extents(engine.pager.free_ready.items, gone));
+}
+
 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
@@ -3656,9 +3976,19 @@ test "the slab counts what the appender skips" {
     try testing.expectEqual(gap + abandoned, engine.dead_bytes);
     try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes);
 
-    // And it survives the round trip, because it is in `slab_used`.
+    // And the next checkpoint gives most of it straight back. An abandoned
+    // extent tail is whole windows with nothing live in them, which is exactly
+    // what window reclamation is for -- so counting it was not bookkeeping for
+    // its own sake, it is what made this reclaimable at all.
+    //
+    // What stays is the edges: the round-up gap, which shares its window with
+    // the live documents below it, and the bytes of the run outside any whole
+    // window.
     try engine.checkpoint();
-    try testing.expectEqual(gap + abandoned, engine.dead_bytes);
+    try testing.expect(coll.reclaimed_bytes > 4 * 1024 * 1024);
+    try testing.expectEqual(gap + abandoned - coll.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);
 }
 
 test "a rebuild leaves behind what its own copying skipped" {
@@ -3706,11 +4036,16 @@ test "a rebuild leaves behind what its own copying skipped" {
 
     const coll = engine.get_collection("app", "c").?;
     try testing.expectEqual(@as(u64, 2), coll.doc_count);
-    // The deleted document is gone from the slab, but the gap the copy left
+    // The deleted document is gone from the slab, and the gap the copy left
-    // between the two survivors is not -- and the engine says so.
+    // between the two survivors is now mostly gone too -- `compact` ends in a
+    // checkpoint, and a checkpoint reclaims whole windows. What survives is the
+    // edges of that gap, which is a smaller number than this test used to
+    // 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(engine.dead_bytes > 2 * 1024 * 1024);
+    try testing.expect(coll.reclaimed_bytes > 2 * 1024 * 1024);
-    try testing.expect(engine.dead_bytes < 4 * 1024 * 1024);
+    try testing.expect(engine.dead_bytes > 0);
+    try testing.expect(engine.dead_bytes < 4 * pgr.map_align);
 }
 
 test "dropping a collection does not arm compaction" {
@@ -5119,8 +5454,35 @@ test "the epochs that invalidate a cursor move exactly when they must" {
     try testing.expect(after_recreate != after_rebuild);
     try testing.expect(after_recreate != before);
 
-    // And the index-level token, which guards the position hint.
+    // Reclamation does not move a live document, so a cursor's *live* offsets
+    // stay good -- but the pages it gives back can be handed out again, and a
+    // cursor's saved offset list may name one of them. Same remedy, and the
+    // same token.
+    //
+    // Both halves matter. A checkpoint that reclaims nothing must leave the
+    // epoch alone, or every open cursor on a busy collection dies on the
+    // checkpoint cadence for nothing.
     try engine.lock();
+    var j: i32 = 0;
+    while (j < 200) : (j += 1) {
+        var d = try make_padded(gpa, 1000 + j, 2000);
+        defer d.deinit();
+        try engine.insert("app", "c", &d, &env.gen);
+    }
+    try engine.commit();
+    const quiet_before = engine.get_collection("app", "c").?.layout_epoch;
+    try engine.checkpoint();
+    try testing.expectEqual(quiet_before, engine.get_collection("app", "c").?.layout_epoch);
+
+    j = 0;
+    while (j < 200) : (j += 1) _ = try engine.remove_by_id("app", "c", .{ .int32 = 1000 + j });
+    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(after_reclaim != quiet_before);
+
+    // And the index-level token, which guards the position hint.
     const coll = engine.get_collection("app", "c").?;
     const index_before = coll.id_index.epoch;
     try coll.id_index.reset_tree(gpa);