src/pager.zig

@@ -134,6 +134,21 @@ 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.
+fn pages_for(len: u64) u32 {
+    return @intCast((len + page_size - 1) / page_size);
+}
+
+/// One generation's two streams, as page runs. Page counts rather than byte
+/// lengths because the free list speaks pages and because the free-list stream
+/// is allocated from an upper bound that its final length can fall short of.
+pub const Streams = struct {
+    catalog_page: u32 = 0,
+    catalog_pages: u32 = 0,
+    freelist_page: u32 = 0,
+    freelist_pages: u32 = 0,
+};
+
 pub const Watermark = struct {
     generation: u64 = 0,
     /// The log sequence this image covers. The crash-recovery invariant is that
@@ -235,6 +250,22 @@ pub const Pager = struct {
     /// The generation the next checkpoint will publish.
     generation: u64,
 
+    /// Where the catalog and free-list streams of the two most recent
+    /// generations live, so the one that falls out of reference can be freed.
+    ///
+    /// Both streams are written into *fresh* pages at every publish, so that a
+    /// crash leaves the previous copy intact. Nothing ever gave those pages
+    /// back: a database checkpointing every 32 MiB of log leaked two runs per
+    /// checkpoint forever, and the catalog carries a `u32` per index node page,
+    /// so at the tens-of-GB target that is hundreds of KB each time.
+    ///
+    /// Process-local rather than in the watermark: only a running pager needs to
+    /// know, because an open reads the slot it is loading and the other slot is
+    /// the fallback. Two generations because a publish overwrites the slot of
+    /// the generation *two* back -- that is the one nothing can reach again.
+    streams_cur: Streams,
+    streams_prev: Streams,
+
     /// Pages below this belong to the last published image and must never be
     /// stored into (PLAN amendment A1). Zero until a checkpoint publishes one,
     /// which is why copy-on-write is inert before then.
@@ -330,6 +361,8 @@ pub const Pager = struct {
             .fresh = created,
             .loaded = .{},
             .generation = 0,
+            .streams_cur = .{},
+            .streams_prev = .{},
             .stable_pages = 0,
             .unpublished = .{},
             .free_ready = .empty,
@@ -859,6 +892,18 @@ pub const Pager = struct {
             // Everything the published image references is off limits to
             // writes from here on.
             self.stable_pages = wm.alloc_tail;
+            // So the next publish but one gives this generation's streams back.
+            // Page counts are derived from the lengths here rather than
+            // remembered, which can be one page short for a free-list stream
+            // whose final length fell inside the page its bound reserved. That
+            // loses at most one page, once per open, against the unbounded leak
+            // this replaces.
+            self.streams_cur = .{
+                .catalog_page = wm.catalog_page,
+                .catalog_pages = pages_for(wm.catalog_len),
+                .freelist_page = wm.freelist_page,
+                .freelist_pages = pages_for(wm.freelist_len),
+            };
             try self.read_freelist(wm);
         } else if (self.watermark_attempted()) {
             // Only worth saying when a watermark was *written* and cannot be
@@ -973,7 +1018,24 @@ pub const Pager = struct {
         // made durable -- so a test can discard everything else and assert the
         // image still loads. Clearing it here would make that check vacuous.
 
-        // 5. only now is the new image current: advance the stable mark and
+        // 5. the streams of the generation two back are now unreachable: the
+        //    slot that named them is the one this publish just overwrote, and
+        //    the two slots hold the new generation and its predecessor. Give
+        //    their pages back -- via the free list, so they are still withheld
+        //    for two more generations like anything else.
+        //
+        //    Before the lock below, which `free_pages` takes for itself.
+        try self.free_pages(self.streams_prev.catalog_page, self.streams_prev.catalog_pages);
+        try self.free_pages(self.streams_prev.freelist_page, self.streams_prev.freelist_pages);
+        self.streams_prev = self.streams_cur;
+        self.streams_cur = .{
+            .catalog_page = wm.catalog_page,
+            .catalog_pages = pages_for(wm.catalog_len),
+            .freelist_page = fl.first,
+            .freelist_pages = fl.pages,
+        };
+
+        // 6. only now is the new image current: advance the stable mark and
         //    rotate the free lists by one generation.
         self.generation = wm.generation;
         self.loaded = wm;
@@ -1019,7 +1081,7 @@ pub const Pager = struct {
         return n;
     }
 
-    fn write_freelist(self: *Pager) !struct { first: u32, len: u64 } {
+    fn write_freelist(self: *Pager) !struct { first: u32, len: u64, pages: u32 } {
         // Size from an upper bound, then count the entries actually written.
         //
         // The allocation below takes its pages off this very list, and an exact
@@ -1069,7 +1131,10 @@ pub const Pager = struct {
         );
         std.mem.writeInt(u64, buf[0..8], count, .little);
         std.mem.writeInt(u64, buf[at..][0..8], header_hash(buf[0..at]), .little);
-        return .{ .first = first, .len = @as(u64, at) + 8 };
+        // `pages` and not `pages_for(len)`: the allocation was sized from the
+        // bound, and the whole run has to go back when this generation falls
+        // out of reference.
+        return .{ .first = first, .len = @as(u64, at) + 8, .pages = pages };
     }
 
     /// Load the free list a watermark points at. A damaged one is dropped with a
@@ -1623,8 +1688,12 @@ test "the persisted free list survives allocating its own pages" {
     // This is the publish that trips it: the free list is now non-empty and
     // holds a run of exactly the one page the stream needs.
     try tp.pg().publish(.{ .seq = 4 });
+    // Not a fixed number: a publish also recycles the streams of the generation
+    // two back, so what is on the list is the three frees above minus whatever
+    // the stream allocations took plus whatever they gave back. The invariant
+    // under test is that a reopen agrees with it, whatever it is.
     const ready_before = tp.pg().free_ready_pages();
-    try testing.expectEqual(@as(u32, 2), ready_before);
+    try testing.expect(ready_before > 0);
     tp.close();
 
     var again = try reopen(io, tp.path);
@@ -1844,6 +1913,33 @@ test "one-page requests do not carve up the runs the extents need" {
     try testing.expectEqual(tail_before, pg.alloc_tail);
 }
 
+test "a quiet checkpoint stops growing the file" {
+    // Both streams a publish writes are allocated fresh every time, so that a
+    // crash leaves the previous copy readable. Nothing gave them back, and a
+    // database checkpoints on log volume rather than on having anything to say
+    // -- so an idle server grew the data file forever.
+    //
+    // Steady state is not zero growth: a publish allocates this generation's
+    // streams and frees the ones from two generations back, and those take two
+    // more publishes to become reusable. So a few pages are always in flight,
+    // and the assertion is that the number does not track the publish count.
+    //
+    // Mutation check: drop the two `free_pages` calls from `publish` and this
+    // goes red at 40-something pages instead of a handful.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tp = try TmpPager.init(io, 64 << 20);
+    defer tp.deinit();
+
+    // Let the rotation reach its steady state before measuring.
+    for (1..5) |i| try tp.pg().publish(.{ .seq = i });
+    const settled = tp.pg().alloc_tail;
+
+    for (5..45) |i| try tp.pg().publish(.{ .seq = i });
+    try testing.expect(tp.pg().alloc_tail - settled <= 8);
+}
+
 test "concurrent frees lose no pages while a publish rotates the lists" {
     // `free_pages` mutates the same three lists `publish` rotates, and its hot
     // caller is `page_mut_cow` under a *collection* lock -- so two collections
@@ -1906,7 +2002,10 @@ test "concurrent frees lose no pages while a publish rotates the lists" {
     }) |list| for (list) |e| {
         for (0..e.pages) |i| {
             const p = e.first + @as(u32, @intCast(i));
-            try testing.expect(p >= lo and p - lo < pages.len);
+            // Pages outside the set the freers own are the publisher's own
+            // stream runs coming back two generations later; they are not what
+            // this test is about.
+            if (p < lo or p - lo >= pages.len) continue;
             try testing.expect(!seen.isSet(p - lo));
             seen.set(p - lo);
             on_list += 1;