src/pager.zig

@@ -998,13 +998,23 @@ pub const Pager = struct {
     }
 
     fn write_freelist(self: *Pager) !struct { first: u32, len: u64 } {
-        const count = self.free_ready.items.len + self.free_hold.items.len + self.free_pending.items.len;
-        const len: u64 = 8 + @as(u64, count) * 8 + 8;
-        const pages: u32 = @intCast((len + page_size - 1) / page_size);
+        // Size from an upper bound, then count the entries actually written.
+        //
+        // The allocation below takes its pages off this very list, and an exact
+        // fit removes the entry it took (`take_free`). A count captured
+        // beforehand therefore claims one entry more than the loop writes: the
+        // hash lands eight bytes short of where `read_freelist` looks for it and
+        // the whole list is dropped as corrupt on the next open. The stream is
+        // one page and a one-page run is the commonest thing on the list, so
+        // that is the ordinary case rather than a corner.
+        //
+        // `take_free` never *adds* an entry, so the bound holds and one
+        // allocation is enough.
+        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 first = try self.alloc_pages(pages);
-        const buf = self.bytes_mut(@as(u64, first) << page_shift, @intCast(len));
+        const buf = self.bytes_mut(@as(u64, first) << page_shift, @as(usize, pages) << page_shift);
         @memset(buf, 0);
-        std.mem.writeInt(u64, buf[0..8], count, .little);
         var at: usize = 8;
         for ([_][]const Extent{
             self.free_ready.items,
@@ -1017,8 +1027,14 @@ pub const Pager = struct {
                 at += 8;
             }
         }
+        const count = (at - 8) / 8;
+        assert_msg(
+            count == bound or count + 1 == bound,
+            "the free list changed size while it was being written",
+        );
+        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 = len };
+        return .{ .first = first, .len = @as(u64, at) + 8 };
     }
 
     /// Load the free list a watermark points at. A damaged one is dropped with a
@@ -1536,6 +1552,51 @@ test "freed pages are withheld for two generations and survive a reopen" {
     try testing.expectEqual(@as(u32, 2), again.free_ready_pages());
 }
 
+test "the persisted free list survives allocating its own pages" {
+    // `write_freelist` allocates the pages it is about to write into, and that
+    // allocation goes through `take_free` like any other. On an exact fit the
+    // entry is removed, so a count captured beforehand describes one entry more
+    // than the loop writes, the hash lands short of where the reader looks, and
+    // the whole list is dropped as corrupt on the next open.
+    //
+    // The stream is one page and a one-page run is the commonest thing on the
+    // list, so this is the normal case, not a corner. The two-generation test
+    // above misses it because its free run is two pages and the stream asks for
+    // one: shrinking an entry keeps the count right, only removing it does not.
+    //
+    // Mutation check: compute `count` before `alloc_pages` again and the reopen
+    // assertion goes red with "data file free list is corrupt".
+    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();
+
+    // Five pages, of which three go back one at a time with a gap between each
+    // -- adjacent runs would be coalesced back into one and the list would be
+    // too short for a lost entry to show.
+    const base = try tp.pg().alloc_pages(5);
+    try tp.pg().publish(.{ .seq = 1 });
+    try tp.pg().free_pages(base, 1);
+    try tp.pg().free_pages(base + 2, 1);
+    try tp.pg().free_pages(base + 4, 1);
+
+    try tp.pg().publish(.{ .seq = 2 }); // pending -> hold
+    try tp.pg().publish(.{ .seq = 3 }); // hold -> ready
+    try testing.expectEqual(@as(u32, 3), tp.pg().free_ready_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 });
+    const ready_before = tp.pg().free_ready_pages();
+    try testing.expectEqual(@as(u32, 2), ready_before);
+    tp.close();
+
+    var again = try reopen(io, tp.path);
+    defer again.deinit();
+    try testing.expectEqual(ready_before, again.free_ready_pages());
+}
+
 test "the watermark is never published before the pages it describes" {
     // The load-bearing ordering of the whole design: every page a watermark
     // describes is durable before the watermark that describes it. Reverse them