db/pager: a document append cannot land in the published image

`slab_reserve` asks `is_unpublished_at` whether the append cursor is still
writable; `slab_append` copies the bytes there. Between them sits the log append
and its fsync, and `publish` clears the entire unpublished set and mprotects the
image. So the answer was routinely stale by the time it was used, and the copy
stored into the durable image.

In ReleaseSafe that is a bus error. In ReleaseFast, where `protect_stable` is
compiled out, there is no fault at all: the store simply overwrites bytes the
last checkpoint published, and the damage surfaces later as a document that
reads back as something else. ReleaseFast is the mode the server ships in.

Present since M0 -- reproduced on f2844e7 with the same test -- and invisible
because nothing paired concurrent writers with a checkpoint. The existing
concurrent suites run against ReleaseFast, where the corruption is silent, and
the unit tests that do run under the protection had no checkpoint racing them.
It has stayed harmless in practice only because a checkpoint fires once per
32 MiB of log; the churn workloads M1 is about to measure change that.

The pager gains an `append_lock`. Appenders hold it shared, so writers on
different collections still proceed concurrently and the lock decomposition
ROADMAP item 5 measured is not given back; `publish` holds it exclusively for
the step that freezes the image, which happens once per checkpoint. Order is
append lock then allocation lock, which is what `mark_appendable` already used.

`slab_append` re-asks the question under that lock and re-arms the cursor if a
checkpoint has published since the reservation. Re-arming has to be infallible,
because this runs after the log record is durable, so `slab_reserve` now
measures its room from the rounded-up cursor rather than the raw one -- under a
system page per extent, and the two share one `appendable_end` so they cannot
disagree about what "there is room" means.

Measured, since this is on the write path: concurrent durable writes, 8 clients
x 1500 inserts at `{w:1, j:true}`, two runs each -- 23779 and 24879 docs/s
before, 23823 and 24452 after. No regression outside run-to-run spread.

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

src/db.zig

@@ -165,7 +165,13 @@ pub const Collection = struct {
         // an extent recycled off the free list starts *below* the mark and is
         // still writable. Asking the mark meant every recycled extent was thrown
         // away after one document, so churn never reused anything.
-        if (self.pager.is_unpublished_at(self.slab_tail) and self.slab_tail + len <= self.slab_end) return;
+        //
+        // Room is checked from the *rounded-up* cursor rather than the cursor
+        // itself, so the round-up `slab_append` may have to do is guaranteed to
+        // fit. Without that the append's own re-check could discover it needs a
+        // fresh extent, which is fallible, after the log record is already
+        // durable. Costs under one system page per extent.
+        if (self.pager.is_unpublished_at(self.slab_tail) and self.appendable_end(len)) return;
         // The page holding the tail is frozen, but the *rest* of the extent is
         // not: nothing above the live cursor is referenced by the image or by an
         // index. So skip to the next system page and keep the extent, instead of
@@ -179,8 +185,8 @@ pub const Collection = struct {
         // collections: the data file reached 11.8x the live data and grew by
         // ~335 MB per checkpoint, heading for DatabaseTooLarge at around 6 GB of
         // real data.
-        const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align);
+        if (self.appendable_end(len)) {
-        if (resumed + len <= self.slab_end) {
+            const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align);
             self.pager.mark_appendable(resumed, self.slab_end);
             self.slab_tail = resumed;
             return;
@@ -198,9 +204,34 @@ pub const Collection = struct {
         self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
     }
 
+    /// Whether a document of `len` bytes fits in this extent even if the cursor
+    /// first has to be rounded up to a system page. The reservation and the
+    /// append both ask this, so they agree on what "there is room" means.
+    fn appendable_end(self: *const Collection, len: usize) bool {
+        return std.mem.alignForward(u64, self.slab_tail, pgr.map_align) + len <= self.slab_end;
+    }
+
     /// Copy `bytes` into the slab and return its absolute file offset.
     /// Infallible: slab_reserve must have run for at least this many bytes.
     fn slab_append(self: *Collection, bytes: []const u8) u64 {
+        // The cursor was checked in `slab_reserve`, but a checkpoint can have
+        // published since -- the reservation runs before the log append and this
+        // runs after it, with an fsync in between. `publish` clears the whole
+        // unpublished set, so a cursor that was writable then can be inside the
+        // frozen image now, and the copy below would store into it: a bus error
+        // where the protection is compiled in, and a silent overwrite of durable
+        // data in ReleaseFast, where it is not.
+        //
+        // Re-arming is infallible because `slab_reserve` measured its room from
+        // the rounded-up cursor. The pager's append lock holds off the next
+        // publish for the rest of this function, so the answer stays true.
+        self.pager.lock_append();
+        defer self.pager.unlock_append();
+        if (!self.pager.is_unpublished_at(self.slab_tail)) {
+            const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align);
+            self.pager.mark_appendable(resumed, self.slab_end);
+            self.slab_tail = resumed;
+        }
         assert_msg(
             self.slab_tail + bytes.len <= self.slab_end,
             "document append overran the slab reservation",

src/pager.zig

@@ -241,6 +241,21 @@ pub const Pager = struct {
     /// the log append -- that is what per-consumer reservations buy.
     alloc_lock: std.Io.Mutex,
 
+    /// Separates a publish from the appends that are mid-flight.
+    ///
+    /// An appender asks `is_unpublished_at` whether its cursor is still
+    /// writable, and copies bytes there afterwards. `publish` clears the whole
+    /// unpublished set and mprotects the image between those two steps, so the
+    /// answer was stale by the time it was used and the copy landed in the
+    /// published image: SIGBUS where the protection is compiled in, a silent
+    /// store into the durable image in ReleaseFast, where it is not.
+    ///
+    /// Shared by appenders so writers on different collections still run
+    /// concurrently -- the decomposition ROADMAP item 5 measured is not given
+    /// back. Exclusive only for the tail of a publish, which happens once per
+    /// checkpoint.
+    append_lock: std.Io.RwLock,
+
     /// True when this file was created by this open (no checkpoint to load).
     fresh: bool,
 
@@ -358,6 +373,7 @@ pub const Pager = struct {
             .alloc_tail = page_first_data,
             .reserved_pages = 0,
             .alloc_lock = .init,
+            .append_lock = .init,
             .fresh = created,
             .loaded = .{},
             .generation = 0,
@@ -529,6 +545,18 @@ pub const Pager = struct {
         return self.is_unpublished(@intCast(off >> page_shift));
     }
 
+    /// Hold off the next publish while an append decides where to put its bytes
+    /// and puts them there. Uncancelable and infallible: the append runs after
+    /// the log record is durable, where there is nowhere to report a failure.
+    /// See `append_lock`.
+    pub fn lock_append(self: *Pager) void {
+        self.append_lock.lockSharedUncancelable(self.io);
+    }
+
+    pub fn unlock_append(self: *Pager) void {
+        self.append_lock.unlockShared(self.io);
+    }
+
     /// The one page a write may legitimately land on below the stable mark: a
     /// watermark slot. Overwriting the *inactive* slot is the whole mechanism --
     /// alternating by generation parity is what makes it safe, where every other
@@ -1037,6 +1065,15 @@ pub const Pager = struct {
 
         // 6. only now is the new image current: advance the stable mark and
         //    rotate the free lists by one generation.
+        //
+        //    Exclusive against the appenders for this step alone. Freezing the
+        //    image while one of them is between "is my cursor still writable"
+        //    and the copy that relies on the answer is what put documents into
+        //    the durable image; holding them off here is what makes the answer
+        //    still true when it is used. Taken before `alloc_lock`, the order
+        //    `mark_appendable` uses on the appender's side.
+        self.append_lock.lockUncancelable(self.io);
+        defer self.append_lock.unlock(self.io);
         self.generation = wm.generation;
         self.loaded = wm;
         // The free lists and the unpublished set are allocator state, so the