pager: the free lists are read and written under the allocation lock

`free_pages` appended to `free_pending` with no lock at all, and `write_freelist`
walked all three lists the same way -- while `publish` rotated them under
`alloc_lock`. Both are reachable concurrently in production: the hot caller of
`free_pages` is `page_mut_cow`, which runs under a *collection* lock, and a
checkpoint holds only the shared catalog lock, so copy-on-write in one collection
races a checkpoint and a second collection's copy-on-write freely.

The append race loses or duplicates entries. The read race is worse: an append
that reallocates leaves `write_freelist`'s loop walking freed memory, and it is
walking it to decide which pages are safe to hand out again.

Both now take `alloc_lock`. `write_freelist` holds it across reading the lists
*and* allocating the pages it writes them into, which the mutex being
non-reentrant makes awkward, so `reserve_pages` and `alloc_pages_assume_reserved`
grow `_locked` bodies and thin locking wrappers. A free that lands while the
stream is being written simply waits for the next generation's list -- the page
stays allocated one generation longer, which is the safe direction.

The read race is what the new test actually caught: written to assert only the
append side, it tripped the size assertion added in the previous commit on its
first run, because a concurrent free had grown the list between the bound and the
loop. That is a bug no reading of `free_pages` alone would have found.

The test asserts page *identity* rather than a total, because a publish allocates
its stream off this very list and a plain count is short by however many publishes
found a fit. Every page left on the lists must be one a freer put there, exactly
once. Probabilistic, as any test of a data race is -- it is evidence only when
red. Mutation-checked per the repo's second ground rule: dropping the lock from
`free_pages` crashes it in roughly two runs out of three; five consecutive runs
with the lock in place are green.

Verified: `zig build test` 160/160 in ReleaseFast and ReleaseSafe, e2e 49, e2e2
concurrent 2 + crash pair 1/3, e2e3 16, e2e4 17, e2e6 72, e2e7 86, `crash-fuzz.js`
60 cycles.

src/pager.zig

@@ -557,6 +557,13 @@ pub const Pager = struct {
     pub fn reserve_pages(self: *Pager, hold: *Reservation, n: u32) !void {
         self.alloc_lock.lockUncancelable(self.io);
         defer self.alloc_lock.unlock(self.io);
+        return self.reserve_pages_locked(hold, n);
+    }
+
+    /// For callers already holding `alloc_lock`. The lock is not reentrant, so
+    /// the split is what lets `write_freelist` hold it across reading the lists
+    /// *and* allocating the pages it writes them into.
+    fn reserve_pages_locked(self: *Pager, hold: *Reservation, n: u32) !void {
         // Additive: room for every promise outstanding anywhere *plus* this one.
         // Two consumers reserving before the same log append must both be able to
         // rely on their promise.
@@ -646,9 +653,14 @@ pub const Pager = struct {
     }
 
     pub fn alloc_pages_assume_reserved(self: *Pager, hold: *Reservation, n: u32) u32 {
-        assert(n > 0);
         self.alloc_lock.lockUncancelable(self.io);
         defer self.alloc_lock.unlock(self.io);
+        return self.alloc_assume_reserved_locked(hold, n);
+    }
+
+    /// 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",
@@ -985,8 +997,18 @@ pub const Pager = struct {
 
     /// Give back a run of pages. They become reusable two generations later --
     /// see the field comment on `free_ready`.
+    ///
+    /// Under the allocation lock, like every other mutation of the free lists.
+    /// It was not, and the hot caller is `page_mut_cow`, which runs under a
+    /// *collection* lock: two collections doing copy-on-write concurrently
+    /// appended to the same list, and `publish` rotated all three lists
+    /// underneath them. No caller holds the lock already -- `page_mut_cow` takes
+    /// it inside `alloc_pages` and has released it by here -- so this cannot
+    /// recurse.
     pub fn free_pages(self: *Pager, first: u32, pages: u32) !void {
         if (pages == 0) return;
+        self.alloc_lock.lockUncancelable(self.io);
+        defer self.alloc_lock.unlock(self.io);
         try self.free_pending.append(self.gpa, .{ .first = first, .pages = pages });
     }
 
@@ -1010,9 +1032,22 @@ pub const Pager = struct {
         //
         // `take_free` never *adds* an entry, so the bound holds and one
         // allocation is enough.
+        //
+        // Under `alloc_lock` for the whole of it, allocation included. Reading
+        // the three lists is as much a use of them as appending is: a concurrent
+        // `free_pages` -- copy-on-write in some collection, which a checkpoint
+        // does not exclude -- grows `free_pending` while the loop below walks it,
+        // and a growth that reallocates leaves the loop on freed memory. A free
+        // that lands after this point simply waits for the next generation's
+        // list; the page stays allocated one generation longer, which is the
+        // safe direction.
+        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 first = try self.alloc_pages(pages);
+        var hold: Reservation = .{};
+        try self.reserve_pages_locked(&hold, pages);
+        const first = self.alloc_assume_reserved_locked(&hold, pages);
         const buf = self.bytes_mut(@as(u64, first) << page_shift, @as(usize, pages) << page_shift);
         @memset(buf, 0);
         var at: usize = 8;
@@ -1809,6 +1844,79 @@ test "one-page requests do not carve up the runs the extents need" {
     try testing.expectEqual(tail_before, pg.alloc_tail);
 }
 
+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
+    // copying nodes concurrently were appending to one `ArrayList` unserialized
+    // while a checkpoint moved it out from under them.
+    //
+    // Pages are conserved across the rotation and across coalescing, so the sum
+    // over all three lists is the invariant to assert. Probabilistic by nature,
+    // as any test of a data race is: it says nothing when green and is only
+    // evidence when red. Mutation check: drop the lock from `free_pages` and
+    // this fails or crashes within a few runs.
+    const gpa = testing.allocator;
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tp = try TmpPager.init(io, 64 << 20);
+    defer tp.deinit();
+
+    const freers = 4;
+    const per_freer = 200;
+    // One page each, allocated up front so no fiber is also growing the file.
+    var pages: [freers * per_freer]u32 = undefined;
+    for (&pages) |*p| p.* = try tp.pg().alloc_pages(1);
+    try tp.pg().publish(.{ .seq = 1 });
+
+    const Worker = struct {
+        fn freer(p: *Pager, run: []const u32) error{Canceled}!void {
+            for (run) |page| p.free_pages(page, 1) catch return error.Canceled;
+        }
+        fn publisher(p: *Pager, seq: *std.atomic.Value(u64)) error{Canceled}!void {
+            for (0..8) |_| {
+                p.publish(.{ .seq = seq.fetchAdd(1, .monotonic) }) catch return error.Canceled;
+            }
+        }
+    };
+
+    var seq = std.atomic.Value(u64).init(2);
+    var group: std.Io.Group = .init;
+    defer group.cancel(io);
+    for (0..freers) |i| {
+        group.async(io, Worker.freer, .{ tp.pg(), pages[i * per_freer ..][0..per_freer] });
+    }
+    group.async(io, Worker.publisher, .{ tp.pg(), &seq });
+    try group.await(io);
+
+    // Page identity, not a total: the publisher's own free-list streams are
+    // allocated *off this list*, so a plain count would be short by however many
+    // publishes found a fit. Every page still on the list must therefore be one
+    // the freers put there, exactly once -- a lost or half-written append shows
+    // up as a duplicate or as a page nobody freed, neither of which recycling
+    // can produce.
+    const lo = pages[0];
+    var seen = try std.DynamicBitSetUnmanaged.initEmpty(gpa, pages.len);
+    defer seen.deinit(gpa);
+    var on_list: usize = 0;
+    for ([_][]const Extent{
+        tp.pg().free_ready.items,
+        tp.pg().free_hold.items,
+        tp.pg().free_pending.items,
+    }) |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);
+            try testing.expect(!seen.isSet(p - lo));
+            seen.set(p - lo);
+            on_list += 1;
+        }
+    };
+    // The only pages missing are the ones a publish recycled into its stream,
+    // and there were nine publishes at one page each.
+    try testing.expect(pages.len - on_list <= 9);
+}
+
 test "freed pages that touch merge back into a usable run" {
     // Mutation check: drop the `coalesce_free_ready()` call from `publish`. Red
     // -- the four one-page frees below stay four separate holes and the run of