db/pager: the concurrency test writes the way the server does

"a checkpoint runs alongside writers on several collections" drove its
writers through `Engine.lock()` -- the legacy whole-engine lock, which
the server has not used since the locks were decomposed. That serialized
the writers against each other, so the overlap the test is named for
never happened: `write_catalog` takes each collection's lock shared, and
nothing it was racing against took that lock at all.

Drive them the way `commands.zig` dispatch does instead: catalog shared,
then the target collection exclusive. The test then does what it says,
and immediately found something -- two appenders on different
collections calling `bytes_mut` at the same time corrupt the pager's
`dirty` set, which is an unsynchronized hash map. ReleaseSafe aborts in
`getOrPutContextAdapted`; three runs in five.

`dirty` is test-only instrumentation (`track_dirty = builtin.is_test`),
so this is a harness bug rather than a server one -- but it is the one
shared structure on a write path whose writers are otherwise kept apart
by owning different pages, and it needs a lock of its own. Six ReleaseSafe
runs clean afterwards.

163/163 unit tests in ReleaseFast and ReleaseSafe, 82/82 fuzz.

src/db.zig

@@ -3146,8 +3146,19 @@ test "a checkpoint runs alongside writers on several collections" {
                 var doc = make_doc(alloc, @intCast(i), "user") catch return error.Canceled;
                 defer doc.deinit();
                 {
-                    e.lock() catch return error.Canceled;
+                    // The server's discipline, not the legacy whole-engine
-                    defer e.unlock();
+                    // lock: catalog shared, then the target collection
+                    // exclusive (commands.zig dispatch). `write_catalog` takes
+                    // the same two in the same order, and that is the whole
+                    // reason its walk of a collection's counters and extents is
+                    // safe -- a writer that skipped the collection lock would
+                    // not be excluded by it, and the test would be checking
+                    // nothing.
+                    e.lock_catalog(false) catch return error.Canceled;
+                    defer e.unlock_catalog(false);
+                    const coll = (e.lock_collection("app", name, true, true) catch
+                        return error.Canceled) orelse return error.Canceled;
+                    defer e.unlock_collection(coll, true);
                     e.insert("app", name, &doc, undefined) catch return error.Canceled;
                 }
                 // As the dispatch epilogue does (commands.zig): the append bumps

src/pager.zig

@@ -321,6 +321,13 @@ pub const Pager = struct {
 
     /// See `track_dirty`. Pages written since the last sync.
     dirty: if (track_dirty) std.AutoHashMapUnmanaged(u32, void) else void,
+    /// Guards `dirty`, and only `dirty`. Writers to the file itself are kept
+    /// apart by the locks their *owners* hold -- a collection's, an index's --
+    /// and the pages two of them touch never overlap. This set is the one thing
+    /// they share: two appenders on different collections record into the same
+    /// hash map at the same time, which is a torn map rather than a torn page.
+    /// Test-only instrumentation, so it costs the server nothing.
+    dirty_lock: if (track_dirty) std.Io.Mutex else void,
     /// Whether the last `protect_image` actually took effect. Checked by a test:
     /// an mprotect that silently fails would leave the belt looking present and
     /// doing nothing, which is worse than not having it.
@@ -386,6 +393,7 @@ pub const Pager = struct {
             .free_hold = .empty,
             .free_pending = .empty,
             .dirty = if (track_dirty) .empty else {},
+            .dirty_lock = if (track_dirty) .init else {},
             .protect_ok = false,
         };
         // The bit set is the one heap allocation `self` owns before `deinit` can
@@ -467,7 +475,7 @@ pub const Pager = struct {
     /// recycling hands back.
     pub inline fn page_mut(self: *Pager, p: u32) *align(page_size) [page_size]u8 {
         assert_msg(p < self.mapped_pages, "write to a page past the mapped end of the data file");
-        if (track_dirty) self.dirty.put(self.gpa, p, {}) catch {};
+        self.note_dirty(p, p);
         return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift)));
     }
 
@@ -569,7 +577,7 @@ pub const Pager = struct {
     inline fn page_mut_slot(self: *Pager, p: u32) *align(page_size) [page_size]u8 {
         assert(p == page_watermark_a or p == page_watermark_b);
         assert_msg(p < self.mapped_pages, "write to a watermark slot past the mapped end");
-        if (track_dirty) self.dirty.put(self.gpa, p, {}) catch {};
+        self.note_dirty(p, p);
         return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift)));
     }
 
@@ -589,13 +597,21 @@ pub const Pager = struct {
     pub inline fn bytes_mut(self: *Pager, off: u64, len: usize) []u8 {
         assert_msg(off + len <= @as(u64, self.mapped_pages) << page_shift, "write past the mapped end of the data file");
         if (track_dirty) {
-            var pg_i: u32 = @intCast(off >> page_shift);
+            self.note_dirty(@intCast(off >> page_shift), @intCast((off + len - 1) >> page_shift));
-            const last: u32 = @intCast((off + len - 1) >> page_shift);
-            while (pg_i <= last) : (pg_i += 1) self.dirty.put(self.gpa, pg_i, {}) catch {};
         }
         return self.reserve[@intCast(off)..][0..len];
     }
 
+    /// Record pages `first..=last` as written since the last sync. See
+    /// `dirty_lock` for why this is the one shared structure on the write path.
+    inline fn note_dirty(self: *Pager, first: u32, last: u32) void {
+        if (!track_dirty) return;
+        self.dirty_lock.lockUncancelable(self.io);
+        defer self.dirty_lock.unlock(self.io);
+        var p = first;
+        while (p <= last) : (p += 1) self.dirty.put(self.gpa, p, {}) catch {};
+    }
+
     // -- allocation ---------------------------------------------------------
 
     /// Hand out `n` contiguous pages, growing the file if needed. For callers
@@ -778,7 +794,11 @@ pub const Pager = struct {
             try std.posix.msync(self.reserve[0..len], std.posix.MSF.SYNC);
         }
         try self.file.sync(self.io);
-        if (track_dirty) self.dirty.clearRetainingCapacity();
+        if (track_dirty) {
+            self.dirty_lock.lockUncancelable(self.io);
+            defer self.dirty_lock.unlock(self.io);
+            self.dirty.clearRetainingCapacity();
+        }
     }
 
     /// Make the published image read-only at the hardware level. See