index: route arena access through accessors; tighten reserve_for's bound

Groundwork for M0: the node arena and overflow slab are about to move into an
mmap'd data file where a write to a page belonging to the last durable
checkpoint has to copy that page first (PLAN amendment A1). Two changes make
that a small commit rather than a sixty-site one, plus the reformat of this
file (see the preceding style commit for why it rides along here).

Accessors. Every read of a node page now goes through page(), every write
through page_mut(), and every overflow read through ovf(); nothing else touches
nodes.items or overflow.items. Which of the 55 sites mutate was decided by the
compiler rather than by inspection -- page() returns *const Node, so every
mutating site failed to compile until flipped -- and the result is that the
copy-on-write hook has exactly one home. Records the rule COW will impose
(never hold a *Node across a page_mut of the same id) and the audit showing
today's callers already comply.

Comptime layout asserts. These structures are about to become an on-disk
format, and nothing pinned them. Pinning also surfaced that @sizeOf(Slot) is
32, not the 20 its 160 declared bits suggest -- the backing integer's 16-byte
alignment rounds it up, so 12 of every 32 slot bytes are padding and a node
holds 127 slots where 203 would fit. Pinned, deliberately not fixed: narrowing
the slot changes the fanout and so the on-disk shape of every index, which
belongs in the commit that reshapes leaf records.

reserve_for. The old bound stood in for "levels a batch can add" with n/8,
which is ~125 levels for a 1000-entry batch and demands ~528 MiB of headroom.
Growing by g levels needs at least 2^g entries, so log2_ceil(n+1)+1 bounds it,
giving ~70 MiB for that batch. Harmless as ArrayList capacity; real file growth
once the arena is file-backed. Overrunning the reservation is a buffer overrun
on a path that has already appended to the log and cannot report failure, so
alloc_node and store_record now assert, using assert.zig so the checks survive
ReleaseFast. Mutation-checked by dropping the reservation entirely: six tests
go red with the new message. Worth noting the assert guards the allocation, not
the arithmetic -- ensureUnusedCapacity over-allocates, so a slightly-too-small
bound is masked until the reservation becomes exact.

build.zig gains a `fuzz` step. spill, spill2, stress and fuzz_split were in no
build step and are not in lib.zig's test block, so `zig build test` could not
see an API break in the only coverage for records past the inline limit and for
randomized split/remove interleavings -- exactly what this work puts at risk.

build.zig

@@ -46,4 +46,33 @@ pub fn build(b: *std.Build) void {
     const run_tests = b.addRunArtifact(test_step);
     const test_help = b.step("test", "Run unit tests");
     test_help.dependOn(&run_tests.step);
+
+    // The B+tree harnesses were in no build step, so `zig build test` -- which
+    // only compiles src/lib.zig's test block -- could not see an API break in
+    // them. They are also the only coverage for records past the inline limit
+    // and for randomized split/remove interleavings, i.e. exactly what M0's
+    // arena work puts at risk. Wire them up so they cannot rot unnoticed.
+    //
+    // Kept out of `test` because stress.zig runs for seconds and the three
+    // main harnesses print rather than assert-and-exit; `zig build fuzz` is
+    // the gate to run alongside the e2e matrix on any index change.
+    const fuzz_step = b.step("fuzz", "Run the B+tree stress and fuzz harnesses");
+
+    const fuzz_split_mod = b.createModule(.{
+        .root_source_file = b.path("src/fuzz_split.zig"),
+        .target = target,
+        .optimize = optimize,
+    });
+    const fuzz_split = b.addTest(.{ .root_module = fuzz_split_mod });
+    fuzz_step.dependOn(&b.addRunArtifact(fuzz_split).step);
+
+    for ([_][]const u8{ "spill", "spill2", "stress" }) |name| {
+        const mod = b.createModule(.{
+            .root_source_file = b.path(b.fmt("src/{s}.zig", .{name})),
+            .target = target,
+            .optimize = optimize,
+        });
+        const harness = b.addExecutable(.{ .name = name, .root_module = mod });
+        fuzz_step.dependOn(&b.addRunArtifact(harness).step);
+    }
 }

src/index.zig

@@ -41,6 +41,11 @@
 const std = @import("std");
 const bson = @import("bson.zig");
 const query = @import("query.zig");
+// Always active, including in the default ReleaseFast build. The tree's hot
+// inner loops keep std.debug.assert (see assert.zig's module comment); these
+// guard the reservation bounds, whose violation is a buffer overrun on a path
+// that has already appended to the log and cannot report failure.
+const assert_msg = @import("assert.zig").assert_msg;
 
 /// MongoDB's compound index field limit.
 pub const max_index_keys: usize = 32;
@@ -115,6 +120,33 @@ const Node = extern struct {
     buf: [page_data]u8,
 };
 
+// These layouts are about to become an on-disk format: M0 maps the node arena
+// straight out of the data file, with no serialization on page-in, so a page
+// written by one build must be readable by the next. Nothing pinned them
+// before, which meant a field added to Node or a change in how Zig lays out a
+// packed Slot would silently reshape the file. `@sizeOf(Slot)` in particular
+// is not obvious from its declaration -- a packed struct's size depends on the
+// alignment of its backing integer, so the 160 declared bits round up.
+comptime {
+    std.debug.assert(page_size == 4096);
+    std.debug.assert(@sizeOf(Node) == page_size);
+    std.debug.assert(@alignOf(Node) <= page_size);
+    std.debug.assert(page_data == page_size - 32);
+    std.debug.assert(@offsetOf(Node, "buf") == 32);
+    std.debug.assert(@bitSizeOf(Slot) == 160);
+    // 160 declared bits is 20 bytes, but the backing integer's 16-byte
+    // alignment rounds @sizeOf up to 32 -- so 12 of every 32 slot bytes are
+    // padding, and a node holds 127 slots where a 20-byte slot would give it
+    // 203. Pinned rather than fixed: narrowing the slot changes the fanout
+    // and therefore the on-disk shape of every index, which belongs in the
+    // commit that reshapes leaf records, not in a refactor.
+    std.debug.assert(slot_size == 32);
+    std.debug.assert(max_slots * slot_size <= page_data);
+    // Node pages are raw host memory in the data file, so the file is
+    // little-endian-only (the log, which is framed field by field, is not).
+    std.debug.assert(@import("builtin").cpu.arch.endian() == .little);
+}
+
 /// A view of one stored entry yielded by iteration. Both slices alias the
 /// tree and are valid only while the tree is not mutated.
 pub const EntryRef = struct {
@@ -157,7 +189,14 @@ pub const Index = struct {
     /// copied here so the reference survives node-array growth.
     promo: [inline_limit]u8,
 
-    pub fn init(gpa: std.mem.Allocator, name: []const u8, keys: []const IndexKey, unique: bool, sparse: bool, ttl: ?i64) !Index {
+    pub fn init(
+        gpa: std.mem.Allocator,
+        name: []const u8,
+        keys: []const IndexKey,
+        unique: bool,
+        sparse: bool,
+        ttl: ?i64,
+    ) !Index {
         var self: Index = .{
             .name = undefined,
             .keys = undefined,
@@ -220,7 +259,12 @@ pub const Index = struct {
     /// equality on `{tags: ["a","b"]}` are covered. Returns an empty list
     /// for a sparse index when a path yields no values (the document is
     /// skipped); a non-sparse index indexes missing fields as null.
-    pub fn build_entries(self: *const Index, gpa: std.mem.Allocator, doc: []const u8, id: []const u8) !BuiltEntries {
+    pub fn build_entries(
+        self: *const Index,
+        gpa: std.mem.Allocator,
+        doc: []const u8,
+        id: []const u8,
+    ) !BuiltEntries {
         // One arena for the whole call: the collected values and any nested
         // spines the byte walker materializes (whole-array/document values)
         // live here, so nothing leaks. The finished keys are still
@@ -303,12 +347,36 @@ pub const Index = struct {
     /// and the exact overflow bytes (spilled records are copied once).
     pub fn reserve_for(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void {
         const n: u64 = entries.len;
-        // One entry splits at most one node per level plus a new root. A
+        // One entry splits at most one node per level, plus a new root when
-        // batch can also deepen the tree as it goes, and every level it
+        // the old root is the level that splits: `levels + 1` nodes, where
-        // adds costs one more node per remaining entry; a level needs a
+        // `levels == depth + 1`.
-        // full root and the smallest root holds three ~1 KiB separators, so
+        //
-        // n/8 levels per batch is well clear of the worst case.
+        // A batch can also deepen the tree as it goes, and each added level
-        const extra_nodes: u64 = n * (self.depth + 2 + n / 8) + 4;
+        // costs one more node per remaining entry -- so the bound has to use
+        // the *final* depth. Growing by one level means splitting the root,
+        // which means filling it first, and the smallest root a split can
+        // leave holds one separator: refilling it to a split takes at least
+        // two more arrivals from below, each of which needs a split one
+        // level down. So g added levels take at least 2^g entries, and
+        // `log2_ceil(n+1) + 1` bounds g.
+        //
+        // This replaces an earlier `n/8` stand-in for g, which was ~125
+        // levels for a 1000-entry batch -- harmless as ArrayList capacity,
+        // but it becomes real file growth once the arena is file-backed
+        // (~528 MiB of demanded headroom for that batch, against ~70 MiB
+        // here). Tightening it further needs an amortized argument rather
+        // than this per-entry one, since no single insertion can split a
+        // full path twice in a row.
+        //
+        // Note what does and does not guard this arithmetic: `alloc_node`'s
+        // assert catches an overrun of the *actual* capacity, and
+        // ensureUnusedCapacity over-allocates geometrically, so a bound
+        // that is slightly too small is usually masked here. (Mutation-
+        // checked by dropping the reservation entirely, which does fire it
+        // -- six tests.) Once the arena is file-backed and the reservation
+        // is exact, that assert becomes a real check on this expression.
+        const growth: u64 = std.math.log2_int_ceil(u64, n + 1) + 1;
+        const extra_nodes: u64 = n * (self.depth + 2 + growth) + 4;
         try self.nodes.ensureUnusedCapacity(gpa, @intCast(extra_nodes));
         try self.reserve_overflow(gpa, entries);
     }
@@ -344,7 +412,13 @@ pub const Index = struct {
     /// With `enforce_unique` false a duplicate is tolerated rather than
     /// rejected (the rebuild path keeps the index and warns); the return
     /// value reports whether that happened.
-    pub fn add_doc(self: *Index, gpa: std.mem.Allocator, doc: []const u8, id: []const u8, enforce_unique: bool) !bool {
+    pub fn add_doc(
+        self: *Index,
+        gpa: std.mem.Allocator,
+        doc: []const u8,
+        id: []const u8,
+        enforce_unique: bool,
+    ) !bool {
         var built = try self.build_entries(gpa, doc, id);
         // Runs on success too: the batch's keys are copied into the tree,
         // so deinit frees exactly what this call allocated.
@@ -370,7 +444,12 @@ pub const Index = struct {
     /// every entry, so building an index over n documents would move O(n²)
     /// bytes — that was the whole cost of createIndex on a large
     /// collection. Staging and packing is O(n log n) and no memmove.
-    pub fn append_doc_entries(self: *Index, gpa: std.mem.Allocator, doc: []const u8, id: []const u8) !void {
+    pub fn append_doc_entries(
+        self: *Index,
+        gpa: std.mem.Allocator,
+        doc: []const u8,
+        id: []const u8,
+    ) !void {
         var built = try self.build_entries(gpa, doc, id);
         // Runs on success too: the append below moves the keys into the
         // staging array, leaving only the (now empty) ArrayList buffer.
@@ -387,7 +466,11 @@ pub const Index = struct {
     /// With `enforce_unique` false a duplicate is tolerated rather than
     /// rejected, matching `add_doc`; the return value reports whether that
     /// happened.
-    pub fn finish_bulk(self: *Index, gpa: std.mem.Allocator, enforce_unique: bool) error{ DuplicateKeyIndex, OutOfMemory }!bool {
+    pub fn finish_bulk(
+        self: *Index,
+        gpa: std.mem.Allocator,
+        enforce_unique: bool,
+    ) error{ DuplicateKeyIndex, OutOfMemory }!bool {
         std.mem.sort(Entry, self.staging.items, {}, entry_less);
         var duplicate = false;
         if (self.unique and self.staging.items.len >= 2) {
@@ -418,7 +501,7 @@ pub const Index = struct {
         _ = gpa;
         var leaf_id = self.first_leaf;
         while (leaf_id != 0) {
-            const node = &self.nodes.items[leaf_id];
+            const node = self.page(leaf_id);
             const next = node.next;
             // Slots are removed high-to-low so the indices stay valid.
             var i = node.count;
@@ -458,7 +541,11 @@ pub const Index = struct {
     /// Reject when any of `new_entries` has a key already present under a
     /// different id. Entries with `exclude_id` (the replacing document's
     /// own old entries) are allowed.
-    pub fn check_unique(self: *const Index, new_entries: []const Entry, exclude_id: []const u8) error{DuplicateKeyIndex}!void {
+    pub fn check_unique(
+        self: *const Index,
+        new_entries: []const Entry,
+        exclude_id: []const u8,
+    ) error{DuplicateKeyIndex}!void {
         for (new_entries) |e| {
             var it = self.seek(e.key);
             while (it.next()) |have| {
@@ -472,7 +559,12 @@ pub const Index = struct {
 
     /// All ids whose key equals `key` (component-wise). For a partial key
     /// (fewer components than the index has) this is a prefix search.
-    pub fn lookup_eq(self: *const Index, gpa: std.mem.Allocator, key: []const bson.Value, out: *std.ArrayListUnmanaged([]const u8)) !void {
+    pub fn lookup_eq(
+        self: *const Index,
+        gpa: std.mem.Allocator,
+        key: []const bson.Value,
+        out: *std.ArrayListUnmanaged([]const u8),
+    ) !void {
         var enc: std.ArrayListUnmanaged(u8) = .empty;
         defer enc.deinit(gpa);
         for (key) |v| try bson.encode_key(v, gpa, &enc);
@@ -549,7 +641,7 @@ pub const Index = struct {
         pub fn next(self: *Iter) ?EntryRef {
             const ix = self.ix;
             while (self.leaf != 0) {
-                const node = &ix.nodes.items[self.leaf];
+                const node = ix.page(self.leaf);
                 if (self.slot < node.count) {
                     const key = ix.key_of(self.leaf, self.slot);
                     const id = ix.id_of(self.leaf, self.slot);
@@ -580,7 +672,11 @@ pub const Index = struct {
     /// The canonical spec document bytes
     /// ({v, key, name, unique?, sparse?, expireAfterSeconds?}) stored in the
     /// log and used to rebuild the index on replay.
-    pub fn write_spec(self: *const Index, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) !void {
+    pub fn write_spec(
+        self: *const Index,
+        gpa: std.mem.Allocator,
+        out: *std.ArrayListUnmanaged(u8),
+    ) !void {
         var arena = std.heap.ArenaAllocator.init(gpa);
         defer arena.deinit();
         var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
@@ -590,7 +686,11 @@ pub const Index = struct {
 
     /// The spec as pairs in `arena` (values alias this index's own storage,
     /// which outlives any reply). Used by listIndexes.
-    pub fn spec_pairs(self: *const Index, arena: std.mem.Allocator, out: *std.ArrayListUnmanaged(bson.Pair)) !void {
+    pub fn spec_pairs(
+        self: *const Index,
+        arena: std.mem.Allocator,
+        out: *std.ArrayListUnmanaged(bson.Pair),
+    ) !void {
         try out.append(arena, .{ .key = "v", .value = .{ .int32 = 2 } });
         const key_pairs = try arena.alloc(bson.Pair, self.keys.len);
         for (self.keys, 0..) |k, i| {
@@ -661,7 +761,16 @@ pub const Index = struct {
 
     /// Allocate a node id. Infallible: insert paths reserve capacity first;
     /// the pack path reserves via reserve_for before packing.
+    ///
+    /// The assert is the tripwire for `reserve_for`'s bound. Overrunning the
+    /// reservation is not a graceful failure: `appendAssumeCapacity` writes
+    /// past the buffer, and in ReleaseFast (this project's default) nothing
+    /// else checks. It also cannot be reported to the caller -- this runs
+    /// after the log append, on the path whose whole point is that a
+    /// document can never be live but unindexed -- so panicking is the only
+    /// honest response.
     fn alloc_node(self: *Index) u32 {
+        assert_msg(self.nodes.items.len < self.nodes.capacity, "node allocation overran reserve_for's bound");
         self.nodes.appendAssumeCapacity(empty_node(0));
         return @intCast(self.nodes.items.len - 1);
     }
@@ -674,35 +783,78 @@ pub const Index = struct {
         return @intCast(self.nodes.items.len - 1);
     }
 
-    fn get_slot(node: *const Node, i: u32) Slot {
+    // -- arena access -------------------------------------------------------
-        return std.mem.bytesToValue(Slot, node.buf[i * slot_size ..][0..slot_size]);
+    //
+    // Every read of a node page goes through `page`, every write through
+    // `page_mut`, and every overflow-slab read through `ovf`. Nothing else
+    // touches `nodes.items` or `overflow.items`. That is not style: M0 moves
+    // both stores into an mmap'd data file, where a write to a page belonging
+    // to the last durable checkpoint has to copy the page first (PLAN
+    // Amendment A1). Funnelling writes through one function is what makes
+    // that a change of three bodies instead of sixty call sites, and what
+    // lets a debug build enforce "no store below the stable mark".
+    //
+    // The rule that copy-on-write will impose, worth honouring already: do
+    // not hold a `*Node` across a `page_mut` of *the same* id. Under COW the
+    // second call can move that id to a fresh page, leaving the first
+    // pointer aimed at a page nothing will ever read again. Holding pointers
+    // to two *different* ids at once stays fine.
+    //
+    // The helpers that take a node id rather than a `*Node` -- store_record,
+    // remove_record, repack_keep_prefix -- each re-acquire the page, so they
+    // are where a caller could break the rule. Every caller currently
+    // complies, and where it is not obvious it is because a value was read
+    // out first: `insert_rec` passes `node.count` to repack_keep_prefix and
+    // never touches `node` again afterwards. Audited as part of introducing
+    // these accessors; re-audit when COW lands, and consider passing the
+    // `*Node` down so the copy happens once at the top of the call.
+
+    /// The page holding node `id`, for reading.
+    inline fn page(self: *const Index, id: u32) *const Node {
+        return &self.nodes.items[id];
     }
 
-    fn set_slot(node: *Node, i: u32, s: Slot) void {
+    /// The page holding node `id`, for writing.
-        std.mem.bytesAsValue(Slot, node.buf[i * slot_size ..][0..slot_size]).* = s;
+    inline fn page_mut(self: *Index, id: u32) *Node {
+        return &self.nodes.items[id];
+    }
+
+    /// Overflow-slab bytes in `[from, to)`. Slot offsets are u64 because a
+    /// spilled record can sit anywhere in the slab; the casts live here so
+    /// the callers read as plain slicing.
+    inline fn ovf(self: *const Index, from: u64, to: u64) []const u8 {
+        return self.overflow.items[@intCast(from)..@intCast(to)];
+    }
+
+    fn get_slot(node_page: *const Node, i: u32) Slot {
+        return std.mem.bytesToValue(Slot, node_page.buf[i * slot_size ..][0..slot_size]);
+    }
+
+    fn set_slot(node_page: *Node, i: u32, s: Slot) void {
+        std.mem.bytesAsValue(Slot, node_page.buf[i * slot_size ..][0..slot_size]).* = s;
     }
 
     /// The key bytes of slot `i`, either in the node's page or the slab.
     fn key_of(self: *const Index, node_id: u32, i: u32) []const u8 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         const s = get_slot(node, i);
-        if (s.spill) return self.overflow.items[@intCast(s.off) .. @intCast(s.off + s.key_len)];
+        if (s.spill) return self.ovf(s.off, s.off + s.key_len);
-        return node.buf[@intCast(s.off) .. @intCast(s.off + s.key_len)];
+        return node.buf[@intCast(s.off)..@intCast(s.off + s.key_len)];
     }
 
     /// The id bytes of leaf slot `i`.
     fn id_of(self: *const Index, node_id: u32, i: u32) []const u8 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         const s = get_slot(node, i);
         const start = s.off + s.key_len;
-        if (s.spill) return self.overflow.items[@intCast(start) .. @intCast(start + s.extra)];
+        if (s.spill) return self.ovf(start, start + s.extra);
-        return node.buf[@intCast(start) .. @intCast(start + s.extra)];
+        return node.buf[@intCast(start)..@intCast(start + s.extra)];
     }
 
     /// Whether a record of `rec_len` bytes fits `node`: the slot plus, when
     /// it stays inline, its bytes. Oversized records spill (slot only).
     fn fits(self: *const Index, node_id: u32, rec_len: u64) bool {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         const inline_bytes: u64 = if (rec_len > inline_limit) 0 else rec_len;
         return (@as(u64, node.count) + 1) * slot_size + inline_bytes <= node.data_start;
     }
@@ -710,7 +862,7 @@ pub const Index = struct {
     /// Write `rec` into `node` at slot position `pos` (append when pos ==
     /// count), shifting later slots right to make room. Assumes fit.
     fn store_record(self: *Index, node_id: u32, pos: u32, rec: Record) void {
-        const node = &self.nodes.items[node_id];
+        const node = self.page_mut(node_id);
         const rec_len: u64 = rec.key.len + rec.id.len;
         var s: Slot = .{
             .off = 0,
@@ -723,6 +875,10 @@ pub const Index = struct {
             s.off = off;
             s.spill = true;
         } else if (rec_len > inline_limit) {
+            // Tripwire for reserve_overflow, for the same reason as
+            // alloc_node's: this append runs after the log append and cannot
+            // fail back to the caller.
+            assert_msg(self.overflow.items.len + rec_len <= self.overflow.capacity, "spilled record overran reserve_overflow's bound");
             s.off = self.overflow.items.len;
             self.overflow.appendSliceAssumeCapacity(rec.key);
             if (rec.id.len > 0) self.overflow.appendSliceAssumeCapacity(rec.id);
@@ -746,7 +902,7 @@ pub const Index = struct {
     /// Remove slot `i`, closing the hole its inline bytes leave and
     /// adjusting surviving inline offsets. Infallible.
     fn remove_record(self: *Index, node_id: u32, i: u32) void {
-        const node = &self.nodes.items[node_id];
+        const node = self.page_mut(node_id);
         const s = get_slot(node, i);
         const is_leaf = node.is_leaf == 1;
         const rec_len: u64 = s.key_len + (if (is_leaf) s.extra else 0);
@@ -776,7 +932,7 @@ pub const Index = struct {
     /// the scratch and written back packed from the top of the page; slots
     /// are rewritten in place with their new offsets.
     fn repack_keep_prefix(self: *Index, node_id: u32, k: u32) void {
-        const node = &self.nodes.items[node_id];
+        const node = self.page_mut(node_id);
         const is_leaf = node.is_leaf == 1;
         // Pass 1: surviving inline records to the scratch, in slot order.
         var scratch_len: usize = 0;
@@ -785,7 +941,7 @@ pub const Index = struct {
             const s = get_slot(node, j);
             if (s.spill) continue;
             const rec_len: usize = @intCast(s.key_len + (if (is_leaf) s.extra else 0));
-            const src = node.buf[@intCast(s.off) .. @intCast(s.off + rec_len)];
+            const src = node.buf[@intCast(s.off)..@intCast(s.off + rec_len)];
             @memcpy(self.scratch[scratch_len .. scratch_len + rec_len], src);
             scratch_len += rec_len;
         }
@@ -810,9 +966,9 @@ pub const Index = struct {
 
     /// The key at (node, i) in a form stable across node-array growth.
     fn stable_key(self: *Index, node_id: u32, i: u32) StableKey {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         const s = get_slot(node, i);
-        if (s.spill) return .{ .key = self.overflow.items[@intCast(s.off) .. @intCast(s.off + s.key_len)], .spill_off = s.off };
+        if (s.spill) return .{ .key = self.ovf(s.off, s.off + s.key_len), .spill_off = s.off };
         const k = self.key_of(node_id, i);
         @memcpy(self.promo[0..k.len], k);
         return .{ .key = self.promo[0..k.len], .spill_off = null };
@@ -826,7 +982,7 @@ pub const Index = struct {
 
     /// `record_cost` of the record already in slot `i`.
     fn slot_cost(self: *const Index, node_id: u32, i: u32) u32 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         const s = get_slot(node, i);
         if (s.spill) return slot_size;
         return slot_size + s.key_len + (if (node.is_leaf == 1) s.extra else 0);
@@ -852,7 +1008,7 @@ pub const Index = struct {
 
     /// Entry position in a leaf, by (key, id).
     fn leaf_pos(self: *const Index, leaf_id: u32, key: []const u8, id: []const u8) u32 {
-        const node = &self.nodes.items[leaf_id];
+        const node = self.page(leaf_id);
         var lo: u32 = 0;
         var hi: u32 = node.count;
         while (lo < hi) {
@@ -867,7 +1023,7 @@ pub const Index = struct {
     /// Separator position in an internal node: after any equal keys, so the
     /// "last separator <= key" descent lands on the newest right child.
     fn separator_pos(self: *const Index, node_id: u32, key: []const u8) u32 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         var lo: u32 = 0;
         var hi: u32 = node.count;
         while (lo < hi) {
@@ -880,7 +1036,7 @@ pub const Index = struct {
     /// The child holding the range `key` sorts into: right of the last
     /// separator that is <= key (equal keys live right of equal separators).
     fn descend_insert(self: *const Index, node_id: u32, key: []const u8) u32 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         var lo: u32 = 0;
         var hi: u32 = node.count;
         while (lo < hi) {
@@ -899,7 +1055,7 @@ pub const Index = struct {
     /// the first separator not less than the prefix (prefix semantics), or
     /// the rightmost child when every separator is less.
     fn descend_lower(self: *const Index, node_id: u32, prefix: []const u8) u32 {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         var lo: u32 = 0;
         var hi: u32 = node.count;
         while (lo < hi) {
@@ -919,7 +1075,7 @@ pub const Index = struct {
     /// Position in a leaf of the first slot whose key is not less than
     /// `prefix` (prefix semantics).
     fn leaf_lower(self: *const Index, leaf_id: u32, prefix: []const u8) u32 {
-        const node = &self.nodes.items[leaf_id];
+        const node = self.page(leaf_id);
         var lo: u32 = 0;
         var hi: u32 = node.count;
         while (lo < hi) {
@@ -933,7 +1089,7 @@ pub const Index = struct {
     /// `prefix`.
     fn lower_bound(self: *const Index, prefix: []const u8) struct { leaf: u32, slot: u32 } {
         var node_id = self.root;
-        while (self.nodes.items[node_id].is_leaf == 0) node_id = self.descend_lower(node_id, prefix);
+        while (self.page(node_id).is_leaf == 0) node_id = self.descend_lower(node_id, prefix);
         return .{ .leaf = node_id, .slot = self.leaf_lower(node_id, prefix) };
     }
 
@@ -942,10 +1098,10 @@ pub const Index = struct {
         if (self.insert_rec(self.root, key, id)) |up| {
             // The root split: a new root with the two halves as children.
             const new_root = self.alloc_node();
-            self.nodes.items[new_root].first_child = self.root;
+            self.page_mut(new_root).first_child = self.root;
-            self.nodes.items[self.root].parent = new_root;
+            self.page_mut(self.root).parent = new_root;
             self.store_record(new_root, 0, .{ .key = up.key, .child = up.right, .spill_off = up.spill_off });
-            self.nodes.items[up.right].parent = new_root;
+            self.page_mut(up.right).parent = new_root;
             self.root = new_root;
             self.depth += 1;
         }
@@ -954,7 +1110,7 @@ pub const Index = struct {
     /// Descend and insert; return the split to promote at the level above,
     /// or null when the subtree absorbed the record.
     fn insert_rec(self: *Index, node_id: u32, key: []const u8, id: []const u8) ?Split {
-        const node = &self.nodes.items[node_id];
+        const node = self.page(node_id);
         if (node.is_leaf == 1) {
             if (self.fits(node_id, key.len + id.len)) {
                 self.store_record(node_id, self.leaf_pos(node_id, key, id), .{ .key = key, .id = id });
@@ -990,7 +1146,7 @@ pub const Index = struct {
         // insert_rec reclaims dead bytes before giving up on a page, so the
         // page is genuinely full here and holds at least two records --
         // which is what makes both halves below non-empty.
-        const old_count = self.nodes.items[leaf_id].count;
+        const old_count = self.page(leaf_id).count;
         std.debug.assert(old_count >= 2);
         const pos = self.leaf_pos(leaf_id, key, id);
         const n = old_count + 1;
@@ -1006,8 +1162,8 @@ pub const Index = struct {
 
         const right_id = self.alloc_node();
         {
-            const node = &self.nodes.items[leaf_id];
+            const node = self.page_mut(leaf_id);
-            const right = &self.nodes.items[right_id];
+            const right = self.page_mut(right_id);
             right.is_leaf = 1;
             right.next = node.next;
             right.prev = leaf_id;
@@ -1016,13 +1172,13 @@ pub const Index = struct {
         // Move the merged tail; spilled records keep their slab reference.
         var m: u32 = mid;
         while (m < n) : (m += 1) {
-            const at: u32 = self.nodes.items[right_id].count;
+            const at: u32 = self.page(right_id).count;
             if (m == pos) {
                 self.store_record(right_id, at, .{ .key = key, .id = id });
                 continue;
             }
             const src: u32 = if (m < pos) m else m - 1;
-            const slot = get_slot(&self.nodes.items[leaf_id], src);
+            const slot = get_slot(self.page(leaf_id), src);
             self.store_record(right_id, at, .{
                 .key = self.key_of(leaf_id, src),
                 .id = self.id_of(leaf_id, src),
@@ -1038,8 +1194,8 @@ pub const Index = struct {
             self.repack_keep_prefix(leaf_id, mid);
         }
         // Link the chain.
-        const left = &self.nodes.items[leaf_id];
+        const left = self.page_mut(leaf_id);
-        if (left.next != 0) self.nodes.items[left.next].prev = right_id;
+        if (left.next != 0) self.page_mut(left.next).prev = right_id;
         left.next = right_id;
         self.leaf_count += 1;
         self.entry_count += 1;
@@ -1064,7 +1220,7 @@ pub const Index = struct {
         // in the page, so reclaim before believing the page is full. This
         // is also what guarantees at least two separators at the split.
         if (!self.fits(node_id, key.len)) {
-            self.repack_keep_prefix(node_id, self.nodes.items[node_id].count);
+            self.repack_keep_prefix(node_id, self.page(node_id).count);
         }
         if (self.fits(node_id, key.len)) {
             self.store_record(node_id, self.separator_pos(node_id, key), .{
@@ -1072,7 +1228,7 @@ pub const Index = struct {
                 .child = split.right,
                 .spill_off = split.spill_off,
             });
-            self.nodes.items[split.right].parent = node_id;
+            self.page_mut(split.right).parent = node_id;
             return null;
         }
         return self.split_internal(node_id, key, split.spill_off, split.right);
@@ -1083,8 +1239,14 @@ pub const Index = struct {
     /// the halves come closest to equal cost (see split_leaf), and the
     /// separator at the cut is promoted -- its child becoming the right
     /// node's first child.
-    fn split_internal(self: *Index, node_id: u32, key: []const u8, spill_off: ?u64, child: u32) Split {
+    fn split_internal(
-        const old_count = self.nodes.items[node_id].count;
+        self: *Index,
+        node_id: u32,
+        key: []const u8,
+        spill_off: ?u64,
+        child: u32,
+    ) Split {
+        const old_count = self.page(node_id).count;
         std.debug.assert(old_count >= 2);
         const pos = self.separator_pos(node_id, key);
         const n = old_count + 1;
@@ -1113,37 +1275,37 @@ pub const Index = struct {
         };
 
         const right_id = self.alloc_node();
-        self.nodes.items[right_id].is_leaf = 0;
+        self.page_mut(right_id).is_leaf = 0;
-        self.nodes.items[right_id].parent = self.nodes.items[node_id].parent;
+        self.page_mut(right_id).parent = self.page(node_id).parent;
         // The promoted separator's child heads the right node.
         const mid_child: u32 = if (mid == pos)
             child
         else
-            get_slot(&self.nodes.items[node_id], if (mid < pos) mid else mid - 1).extra;
+            get_slot(self.page(node_id), if (mid < pos) mid else mid - 1).extra;
-        self.nodes.items[right_id].first_child = mid_child;
+        self.page_mut(right_id).first_child = mid_child;
-        self.nodes.items[mid_child].parent = right_id;
+        self.page_mut(mid_child).parent = right_id;
 
         var m: u32 = mid + 1;
         while (m < n) : (m += 1) {
-            const at: u32 = self.nodes.items[right_id].count;
+            const at: u32 = self.page(right_id).count;
             if (m == pos) {
                 self.store_record(right_id, at, .{ .key = key, .child = child, .spill_off = spill_off });
-                self.nodes.items[child].parent = right_id;
+                self.page_mut(child).parent = right_id;
                 continue;
             }
             const src: u32 = if (m < pos) m else m - 1;
-            const slot = get_slot(&self.nodes.items[node_id], src);
+            const slot = get_slot(self.page(node_id), src);
             self.store_record(right_id, at, .{
                 .key = self.key_of(node_id, src),
                 .child = slot.extra,
                 .spill_off = if (slot.spill) slot.off else null,
             });
-            self.nodes.items[slot.extra].parent = right_id;
+            self.page_mut(slot.extra).parent = right_id;
         }
         if (pos < mid) {
             self.repack_keep_prefix(node_id, mid - 1);
             self.store_record(node_id, pos, .{ .key = key, .child = child, .spill_off = spill_off });
-            self.nodes.items[child].parent = node_id;
+            self.page_mut(child).parent = node_id;
         } else {
             self.repack_keep_prefix(node_id, mid);
         }
@@ -1172,20 +1334,20 @@ pub const Index = struct {
     fn leaf_remove(self: *Index, leaf_id: u32, slot_idx: u32) void {
         self.remove_record(leaf_id, slot_idx);
         self.entry_count -= 1;
-        const node = &self.nodes.items[leaf_id];
+        const node = self.page(leaf_id);
         if (node.count > 0) return;
         // Empty leaf: unlink and drop from the parent (unless it is the
         // root, which stays as the empty root leaf).
         if (leaf_id == self.root) return;
-        if (node.prev != 0) self.nodes.items[node.prev].next = node.next;
+        if (node.prev != 0) self.page_mut(node.prev).next = node.next;
-        if (node.next != 0) self.nodes.items[node.next].prev = node.prev;
+        if (node.next != 0) self.page_mut(node.next).prev = node.prev;
         if (leaf_id == self.first_leaf) self.first_leaf = node.next;
         self.leaf_count -= 1;
         var child = leaf_id;
         var parent = node.parent;
         while (parent != 0) {
             self.drop_child(parent, child);
-            const pnode = &self.nodes.items[parent];
+            const pnode = self.page(parent);
             if (pnode.count == 0 and pnode.first_child == 0) {
                 if (parent == self.root) {
                     self.replace_root_with_leaf();
@@ -1204,7 +1366,7 @@ pub const Index = struct {
     /// again — node ids are append-only, so this is a leak of at most the
     /// peak tree size, exactly what the old entry array's capacity was).
     fn drop_child(self: *Index, parent_id: u32, child_id: u32) void {
-        const pnode = &self.nodes.items[parent_id];
+        const pnode = self.page_mut(parent_id);
         if (pnode.first_child == child_id) {
             if (pnode.count > 0) {
                 const s0 = get_slot(pnode, 0);
@@ -1228,7 +1390,7 @@ pub const Index = struct {
     fn replace_root_with_leaf(self: *Index) void {
         // Removal reserves no capacity, so this must not allocate a node:
         // re-use the emptied root page as the empty root leaf.
-        self.nodes.items[self.root] = empty_node(1);
+        self.page_mut(self.root).* = empty_node(1);
         self.first_leaf = self.root;
         self.leaf_count = 1;
         self.depth = 0;
@@ -1260,7 +1422,7 @@ pub const Index = struct {
         var lit = self.staging.items;
         while (lit.len > 0) {
             const leaf = try self.alloc_node_grow(gpa);
-            self.nodes.items[leaf].is_leaf = 1;
+            self.page_mut(leaf).is_leaf = 1;
             const first_key = lit[0].key;
             // Fill until the next record would not fit.
             var used: usize = 0;
@@ -1274,14 +1436,14 @@ pub const Index = struct {
                 slots += 1;
                 used += inline_bytes;
             }
-            self.nodes.items[leaf].prev = prev;
+            self.page_mut(leaf).prev = prev;
-            if (prev != 0) self.nodes.items[prev].next = leaf;
+            if (prev != 0) self.page_mut(prev).next = leaf;
             prev = leaf;
             self.leaf_count += 1;
             try level.append(gpa, .{ .id = leaf, .first_key = first_key });
             lit = lit[n..];
         }
-        self.nodes.items[prev].next = 0;
+        self.page_mut(prev).next = 0;
         self.first_leaf = level.items[0].id;
 
         // Interior levels: group the level below into internal nodes whose
@@ -1292,8 +1454,8 @@ pub const Index = struct {
             var i: usize = 0;
             while (i < level.items.len) {
                 const node = try self.alloc_node_grow(gpa);
-                self.nodes.items[node].first_child = level.items[i].id;
+                self.page_mut(node).first_child = level.items[i].id;
-                self.nodes.items[level.items[i].id].parent = node;
+                self.page_mut(level.items[i].id).parent = node;
                 const first_key = level.items[i].first_key;
                 var used: usize = 0;
                 var slots: usize = 0;
@@ -1306,7 +1468,7 @@ pub const Index = struct {
                         .key = level.items[j].first_key,
                         .child = level.items[j].id,
                     });
-                    self.nodes.items[level.items[j].id].parent = node;
+                    self.page_mut(level.items[j].id).parent = node;
                     slots += 1;
                     used += inline_bytes;
                 }
@@ -1319,7 +1481,7 @@ pub const Index = struct {
             self.depth += 1;
         }
         self.root = level.items[0].id;
-        self.nodes.items[self.root].parent = 0;
+        self.page_mut(self.root).parent = 0;
         self.entry_count = self.staging.items.len;
     }
 };
@@ -1507,7 +1669,11 @@ const Clause = struct {
 /// Flatten top-level pairs and $and members into AND-ed predicates. Every
 /// other top-level operator ($or, $nor, ...) is skipped: the full filter is
 /// re-applied later, so a usable sibling still yields a valid superset.
-fn flatten_clauses(gpa: std.mem.Allocator, pairs: []const bson.Pair, out: *std.ArrayListUnmanaged(Clause)) !void {
+fn flatten_clauses(
+    gpa: std.mem.Allocator,
+    pairs: []const bson.Pair,
+    out: *std.ArrayListUnmanaged(Clause),
+) !void {
     for (pairs) |p| {
         if (p.key.len > 0 and p.key[0] == '$') {
             if (std.mem.eql(u8, p.key, "$and")) {
@@ -1623,7 +1789,11 @@ pub const Plan = struct {
     /// several entries) or from several lookup keys (whose ranges can be the
     /// same key repeated, as in {$in: [1, 1]}); the common single-key lookup
     /// on a non-multikey index skips the pass entirely.
-    pub fn search(self: *const Plan, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged([]const u8)) !void {
+    pub fn search(
+        self: *const Plan,
+        gpa: std.mem.Allocator,
+        out: *std.ArrayListUnmanaged([]const u8),
+    ) !void {
         for (self.lookup_keys.items) |key| {
             if (self.lo == null and self.hi == null) {
                 try self.index.lookup_eq(gpa, key, out);
@@ -1733,7 +1903,12 @@ fn index_provides_sort(ix: *const Index, run: usize, sort: []const query.SortKey
     return backward;
 }
 
-fn evaluate_index(gpa: std.mem.Allocator, ix: *const Index, clauses: []const Clause, sort: []const query.SortKey) !?Plan {
+fn evaluate_index(
+    gpa: std.mem.Allocator,
+    ix: *const Index,
+    clauses: []const Clause,
+    sort: []const query.SortKey,
+) !?Plan {
     const n = ix.keys.len;
     var infos: [max_index_keys]CompInfo = undefined;
     for (0..n) |i| {
@@ -1856,7 +2031,12 @@ fn doc_of(pairs: []const bson.Pair) bson.Document {
     return .{ .arena = undefined, .pairs = pairs };
 }
 
-fn simple_index(gpa: std.mem.Allocator, paths: []const []const u8, unique: bool, sparse: bool) !Index {
+fn simple_index(
+    gpa: std.mem.Allocator,
+    paths: []const []const u8,
+    unique: bool,
+    sparse: bool,
+) !Index {
     var keys: [max_index_keys]IndexKey = undefined;
     for (paths, 0..) |p, i| keys[i] = .{ .path = p, .descending = false };
     return Index.init(gpa, "test", keys[0..paths.len], unique, sparse, null);
@@ -1864,7 +2044,12 @@ fn simple_index(gpa: std.mem.Allocator, paths: []const []const u8, unique: bool,
 
 /// Look up ids and compare with the expected set. Entry ids alias the
 /// caller's storage, so tests pass stable static byte strings as ids.
-fn expect_ids(gpa: std.mem.Allocator, ix: *const Index, key: []const bson.Value, expected: []const []const u8) !void {
+fn expect_ids(
+    gpa: std.mem.Allocator,
+    ix: *const Index,
+    key: []const bson.Value,
+    expected: []const []const u8,
+) !void {
     var out: std.ArrayListUnmanaged([]const u8) = .empty;
     defer out.deinit(gpa);
     try ix.lookup_eq(gpa, key, &out);
@@ -1945,7 +2130,7 @@ test "multikey expansion indexes the array and its elements" {
     var ix = try simple_index(gpa, &.{"tags"}, false, false);
     defer ix.deinit(gpa);
 
-    const d = try bytes_of(gpa, &.{.{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } }});
+    const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } } });
     defer gpa.free(d);
     _ = try ix.add_doc(gpa, d, "mk1", true);
 
@@ -1963,7 +2148,7 @@ test "per-document dedup keeps {a: [1,1]} under a unique index" {
     const gpa = testing.allocator;
     var ix = try simple_index(gpa, &.{"a"}, true, false);
     defer ix.deinit(gpa);
-    const d = try bytes_of(gpa, &.{.{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 1 } } } }});
+    const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 1 } } } } });
     defer gpa.free(d);
     _ = try ix.add_doc(gpa, d, "d1", true);
     // Entries after dedup: the array itself and one element.
@@ -2029,7 +2214,7 @@ test "range bounds inclusive and exclusive" {
     };
     for (docs) |s| {
         const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = s.a } }, .{ .key = "a", .value = .{ .int32 = s.a } } });
-    defer gpa.free(d);
+        defer gpa.free(d);
         _ = try ix.add_doc(gpa, d, s.id, true);
     }
 
@@ -2088,7 +2273,7 @@ test "compound index prefix search and range on the next key" {
             .{ .key = "a", .value = .{ .int32 = s.a } },
             .{ .key = "b", .value = .{ .int32 = s.b } },
         });
-    defer gpa.free(d);
+        defer gpa.free(d);
         _ = try ix.add_doc(gpa, d, s.id, true);
     }
 
@@ -2234,7 +2419,7 @@ test "incremental inserts and removals stay identical to a brute-force model" {
             .{ .key = "a", .value = .{ .int32 = a } },
             .{ .key = "b", .value = .{ .int32 = b } },
         });
-    defer gpa.free(d);
+        defer gpa.free(d);
         _ = try ix.add_doc(gpa, d, id, false);
         try model.append(gpa, .{ .a = a, .b = b, .id = id });
         try live.append(gpa, true);
@@ -2254,7 +2439,7 @@ test "incremental inserts and removals stay identical to a brute-force model" {
             .{ .key = "a", .value = .{ .int32 = m.a } },
             .{ .key = "b", .value = .{ .int32 = m.b } },
         });
-    defer gpa.free(d);
+        defer gpa.free(d);
         ix.remove_doc(gpa, d, m.id);
         live.items[i] = false;
         try verify_model(gpa, &ix, model.items, live.items, rand);
@@ -2264,7 +2449,13 @@ test "incremental inserts and removals stay identical to a brute-force model" {
 /// One document's facts in the incremental-mutation differential.
 const ModelFact = struct { a: i32, b: i32, id: []const u8 };
 
-fn verify_model(gpa: std.mem.Allocator, ix: *const Index, model: []const ModelFact, live: []const bool, rand: std.Random) !void {
+fn verify_model(
+    gpa: std.mem.Allocator,
+    ix: *const Index,
+    model: []const ModelFact,
+    live: []const bool,
+    rand: std.Random,
+) !void {
     // lookup_eq over a random value.
     const a = rand.intRangeAtMost(i32, 0, 30);
     var out: std.ArrayListUnmanaged([]const u8) = .empty;
@@ -2327,7 +2518,7 @@ test "lookup_range matches a brute-force filter over random data" {
             .{ .key = "a", .value = .{ .int32 = a } },
             .{ .key = "b", .value = .{ .int32 = b } },
         });
-    defer gpa.free(d);
+        defer gpa.free(d);
         try ix.append_doc_entries(gpa, d, id);
     }
     _ = try ix.finish_bulk(gpa, false);
@@ -2591,7 +2782,6 @@ test "a split with lopsided record sizes keeps the new record inside its page" {
     try testing.expectEqual(docs.items.len, seen);
 }
 
-
 test "the _id index plan covers equality, ranges and _id sort order" {
     // The implicit _id_ index is a normal Index (keys = [_id: 1]) passed to
     // plan separately from the secondaries. Its encoded keys are canonical,
@@ -2606,7 +2796,7 @@ test "the _id index plan covers equality, ranges and _id sort order" {
             .{ .key = "_id", .value = .{ .int32 = @intCast(i + 1) } },
             .{ .key = "v", .value = .{ .int32 = @intCast(i) } },
         });
-    defer gpa.free(d);
+        defer gpa.free(d);
         const id = try std.fmt.allocPrint(gpa, "d{d}", .{i + 1});
         defer gpa.free(id);
         _ = try ix.add_doc(gpa, d, id, false);