db: documents live in the data file

Collection's slab of malloc'd 8 MiB segments becomes extents in the data file,
and a document's offset becomes an absolute file offset. That makes `doc_bytes`
base + off instead of a binary search over segment starts, and it removes the
hazard the segment list carried: the mapping's base never moves, so a slice into
it cannot be invalidated by growth. phase8 records a dangling-slab-pointer bug
of exactly the shape this deletes.

`slab_reserve` now runs before the log append and `slab_append` after it, and is
infallible. It had no reservation before because appending to an ArrayList could
only fail on OOM; a file-backed slab can also fail on growth, and failing after
the record is durable would report an error for a write the next open produces
anyway.

The data file is still recreated empty on every open and the log still replays in
full, so nothing durable depends on it yet and open/close semantics are
byte-for-byte what they were. The watermark that turns it into a checkpoint comes
next.

--

One bug, and it is worth reading because the milestone keeps producing this
shape. Collection holds a `*Pager`, and `Engine.open` builds an Engine on the
*stack* and returns it by value -- so every pointer taken during replay dangled
the moment it was moved. It surfaced as SIGBUS, then as a corrupt hashmap in the
*second* engine of an unrelated test: nothing resembling its cause. The pager is
heap-allocated now, as `Index` already was, for the same reason.

--

Measured on one harness, 512 MB / 16 KB docs, before and after:

  bulk insert throughput      742.6 MB/s -> 746.7 MB/s
  createIndex({k: 1})         26.8 ms    -> 16.2 ms
  countDocuments({})          2.1 ms     -> 1.1 ms
  findOne({k: 500}) indexed   0.75 ms    -> 0.53 ms
  find({p: range}).count()    6.6 ms     -> 4.1 ms
  aggregate $group by k       5.8 ms     -> 3.7 ms
  insertOne (sequential)      0.20 ms    -> 0.20 ms

Every row equal or faster. The regression this commit was scheduled early to
catch -- document bytes now reaching disk uncompressed on top of the LZ4 log --
did not appear at this size; bulk insert is flat and the reads gain from one
contiguous mapping instead of separately allocated segments.

What did *not* improve, stated plainly because it is the milestone's headline
claim: RSS is unchanged, 552 MB against 563 MB for 512 MB of data. It cannot
improve yet. Every open still replays the whole log and rewrites the whole slab,
so every page is touched and resident regardless of where it lives. "RSS =
working set" only becomes measurable once an open loads a checkpoint instead of
rebuilding, and the honest test for it is the multi-GB reopen in big.js, not this
microbenchmark.

.gitignore

@@ -1,6 +1,7 @@
 .zig-cache/
 zig-out/
 *.log
+*.log.data
 node_modules/
 # Pinned upstream spec suites, fetched by tests/spec/fetch.sh (PLAN D2).
 tests/spec/specifications/

src/db.zig

@@ -9,6 +9,7 @@ const std = @import("std");
 const bson = @import("bson.zig");
 const storage = @import("storage.zig");
 const index = @import("index.zig");
+const pgr = @import("pager.zig");
 // Always active, including in the default ReleaseFast build -- see assert.zig
 // for why std.debug.assert is the wrong tool for these invariants.
 const assert = @import("assert.zig").assert;
@@ -16,25 +17,32 @@ const assert = @import("assert.zig").assert;
 // message is all an operator gets.
 const assert_msg = @import("assert.zig").assert_msg;
 
-/// One slab segment; slack is bounded by this (a geometric-growth array
+/// Pages in a standard slab extent: 8 MiB, as the old in-memory segments were.
-/// would hold up to 2x its contents after doubling).
+/// Slack is bounded by one extent per collection.
-const slab_segment_size = 8 * 1024 * 1024;
+const slab_extent_pages: u32 = (8 * 1024 * 1024) / pgr.page_size;
 
 const LogKind = enum { upsert, delete, index_create, index_drop };
 
 pub const Collection = struct {
-    /// Documents live as canonical BSON bytes in a per-collection slab of
+    /// Documents live as canonical BSON bytes in the data file, in extents this
-    /// fixed segments; the map holds each document's flat slab offset.
+    /// collection owns; the map holds each document's offset. Those are
-    /// Offsets stay valid forever: segments are append-only and never move,
+    /// *absolute file offsets* now, which is what makes doc_bytes a single add
-    /// so a segment's bytes are stable even when the segment list reallocates.
+    /// rather than a binary search over segment starts -- and what removes the
-    /// Segmenting (instead of one geometric-growth array) keeps the slab's
+    /// dangling-pointer hazard the old segment list had, since the mapping's
-    /// capacity slack under one segment — a single array would hold up to
+    /// base never moves.
-    /// 2x its contents after doubling. Removed documents leave garbage bytes
+    ///
-    /// until compaction rewrites.
+    /// Removed documents leave garbage bytes until a rebuild rewrites them. A
+    /// checkpoint must never renumber these offsets: every index leaf holds one
+    /// (PLAN amendment A3).
     docs: std.StringHashMapUnmanaged(u64),
-    slab: std.ArrayListUnmanaged(std.ArrayListUnmanaged(u8)),
+    /// The data file this collection's documents live in.
-    /// Flat offset where each segment begins; doc_bytes binary-searches it.
+    pager: *pgr.Pager,
-    seg_starts: std.ArrayListUnmanaged(u64),
+    /// Extents owned by this collection's slab, in allocation order.
+    slab_extents: std.ArrayListUnmanaged(pgr.Extent),
+    /// Absolute file offset of the next document write, and the end of the
+    /// extent it falls in.
+    slab_tail: u64,
+    slab_end: u64,
     /// Secondary indexes (persisted through the log). Heap-allocated, so an
     /// `*Index` handed out by `find_index` or `create_index` stays valid when
     /// a sibling index is dropped. Held by value, `orderedRemove` memmoved the
@@ -62,8 +70,16 @@ pub const Collection = struct {
     /// integer/string/etc. _id lookups.
     id_index: index.Index,
 
-    fn init(gpa: std.mem.Allocator) !Collection {
+    fn init(gpa: std.mem.Allocator, pager: *pgr.Pager) !Collection {
-        var self: Collection = .{ .docs = .empty, .slab = .empty, .seg_starts = .empty, .indexes = .empty, .id_index = undefined };
+        var self: Collection = .{
+            .docs = .empty,
+            .pager = pager,
+            .slab_extents = .empty,
+            .slab_tail = 0,
+            .slab_end = 0,
+            .indexes = .empty,
+            .id_index = undefined,
+        };
         const keys = [_]index.IndexKey{.{ .path = "_id", .descending = false }};
         // unique: the tree, not the docs map, is what enforces _id uniqueness
         // now (PLAN A3/A4). It is keyed on bson.encode_key, which is canonical
@@ -86,43 +102,48 @@ pub const Collection = struct {
 
     /// Append `bytes` to the slab, returning its flat offset. The last
     /// segment holds up to `slab_segment_size`; a full one starts the next.
-    fn slab_append(self: *Collection, gpa: std.mem.Allocator, bytes: []const u8) !u64 {
+    /// Make room for a document of `len` bytes, so the append that follows
-        if (self.slab.items.len == 0) {
+    /// cannot fail.
-            try self.slab.append(gpa, .empty);
+    ///
-            try self.seg_starts.append(gpa, 0);
+    /// Separated from the append because the append runs *after* the log
-        }
+    /// record is durable, where failure has nowhere to go: the write is already
-        // Length of the last segment as a value, never as a pointer into
+    /// committed and reporting an error for it would be a lie the next open
-        // slab.items: appending the next segment below may reallocate that
+    /// contradicts. Reserving first keeps the fallible half before the log.
-        // list, which would dangle a pointer taken before the append and
+    fn slab_reserve(self: *Collection, gpa: std.mem.Allocator, len: usize) !void {
-        // corrupt the new segment's start offset (and with it every
+        if (self.slab_tail + len <= self.slab_end) return;
-        // doc_bytes lookup in that segment — reads that surfaced as
+        // A document larger than the standard extent gets one of its own; BSON
-        // InvalidBson, or a crash in Debug builds).
+        // reaches 16 MB and the extent is 8 MiB.
-        const last_len = self.slab.items[self.slab.items.len - 1].items.len;
+        const want_pages: u32 = @intCast(@max(
-        if (last_len + bytes.len > slab_segment_size) {
+            slab_extent_pages,
-            try self.slab.append(gpa, .empty);
+            (len + pgr.page_size - 1) / pgr.page_size,
-            try self.seg_starts.append(gpa, self.seg_starts.items[self.seg_starts.items.len - 1] + last_len);
+        ));
-            return self.slab_append(gpa, bytes);
+        try self.pager.reserve_pages(want_pages);
-        }
+        const first = self.pager.alloc_pages_assume_reserved(want_pages);
-        const last = &self.slab.items[self.slab.items.len - 1];
+        try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages });
-        const off = self.seg_starts.items[self.seg_starts.items.len - 1] + last.items.len;
+        self.slab_tail = @as(u64, first) << pgr.page_shift;
-        try last.appendSlice(gpa, bytes);
+        self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
+    }
+
+    /// 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 {
+        assert_msg(
+            self.slab_tail + bytes.len <= self.slab_end,
+            "document append overran the slab reservation",
+        );
+        const off = self.slab_tail;
+        @memcpy(self.pager.bytes_mut(off, bytes.len), bytes);
+        self.slab_tail += bytes.len;
         return off;
     }
 
     /// The canonical bytes of the document stored at `off` — a slice into a
     /// segment, stable until the collection is freed or rebuilt.
     pub fn doc_bytes(self: *const Collection, off: u64) []const u8 {
-        // Last segment start <= off (binary search over the starts).
+        // An absolute file offset, so this is base + off. The length comes from
-        var lo: usize = 0;
+        // the document's own BSON int32 prefix, as it always has.
-        var hi: usize = self.slab.items.len;
+        const len: usize = std.mem.readInt(u32, self.pager.bytes(off, 4)[0..4], .little);
-        while (lo + 1 < hi) {
+        return self.pager.bytes(off, len);
-            const mid = lo + (hi - lo) / 2;
-            if (self.seg_starts.items[mid] <= off) lo = mid else hi = mid;
-        }
-        const seg = &self.slab.items[lo];
-        const in_seg: usize = @intCast(off - self.seg_starts.items[lo]);
-        const len: usize = std.mem.readInt(u32, seg.items[in_seg..][0..4], .little);
-        return seg.items[in_seg .. in_seg + len];
     }
 
     /// Remove and free the index with this name. Returns whether it existed.
@@ -187,6 +208,14 @@ pub const Engine = struct {
     /// once would publish one compaction's half-written file as the database.
     compacting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     log: storage.Log,
+    /// The data file: documents live here, and the B+tree arenas follow.
+    ///
+    /// Heap-allocated because `open` builds an Engine on the stack and returns
+    /// it by value: every Collection holds a `*Pager`, and those were taken
+    /// during replay, before the move. They all dangled -- which surfaced as a
+    /// corrupt docs hashmap on the *second* engine in a test, not as anything
+    /// resembling its cause.
+    pager: *pgr.Pager,
     dbs: std.StringHashMapUnmanaged(Db),
     seq: u64,
     /// Floor for the compaction trigger. The real trigger also scales with
@@ -204,17 +233,36 @@ pub const Engine = struct {
     dup_index: ?[]const u8 = null,
 
     pub fn open(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Engine {
+        var log = try storage.Log.open(gpa, io, path);
+        errdefer log.close();
+
+        // The data file sits beside the log. It is recreated empty on every
+        // open for now: the log is still replayed in full, so nothing durable
+        // depends on the file yet, and open/close semantics stay exactly what
+        // they were. The watermark that makes it a checkpoint comes later.
+        const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{log.path});
+        defer gpa.free(data_path);
+        std.Io.Dir.cwd().deleteFile(io, data_path) catch |err| switch (err) {
+            error.FileNotFound => {},
+            else => return err,
+        };
+
+        const pager_box = try gpa.create(pgr.Pager);
+        errdefer gpa.destroy(pager_box);
+        pager_box.* = try pgr.Pager.open(gpa, io, data_path, .{});
+
         var engine = Engine{
             .gpa = gpa,
             .io = io,
             .rwlock = .init,
-            .log = try storage.Log.open(gpa, io, path),
+            .log = log,
+            .pager = pager_box,
             .dbs = .empty,
             .seq = 0,
             .compact_threshold = 16 * 1024 * 1024,
         };
         errdefer {
-            engine.log.close();
+            engine.pager.deinit();
             engine.dbs.deinit(gpa);
         }
 
@@ -232,6 +280,8 @@ pub const Engine = struct {
             self.gpa.free(db_entry.key_ptr.*);
         }
         self.dbs.deinit(self.gpa);
+        self.pager.deinit();
+        self.gpa.destroy(self.pager);
         self.log.close();
     }
 
@@ -258,9 +308,12 @@ pub const Engine = struct {
             self.gpa.free(doc_entry.key_ptr.*);
         }
         coll.docs.deinit(self.gpa);
-        for (coll.slab.items) |*seg| seg.deinit(self.gpa);
+        // Give the slab's pages back. They become reusable two generations
-        coll.slab.deinit(self.gpa);
+        // later, so a fallback to the previous image still finds them intact.
-        coll.seg_starts.deinit(self.gpa);
+        for (coll.slab_extents.items) |e| {
+            self.pager.free_pages(e.first, e.pages) catch {};
+        }
+        coll.slab_extents.deinit(self.gpa);
         self.gpa.destroy(coll);
     }
 
@@ -605,11 +658,17 @@ pub const Engine = struct {
             };
         }
 
-        // 4. Reserve tree capacity — the last fallible step, so the entry
+        // 4. Reserve everything the publish step needs -- tree capacity and
-        //    insertion after the log append is infallible.
+        //    slab room -- as the last fallible work, so nothing after the log
+        //    append can fail. The slab reservation used to be absent because
+        //    appending to an in-memory ArrayList was the only failure mode; a
+        //    file-backed slab can also fail on growth, and failing *after* the
+        //    record is durable would report an error for a write the next open
+        //    would produce anyway.
         for (built_list.items) |*b| {
             try b.ix.reserve_for(self.gpa, b.built.entries.items);
         }
+        try coll.slab_reserve(self.gpa, doc_bytes.len);
 
         // 5. Log (and sync) before anything becomes visible. The append
         //    takes the log lock; durability (fsync) is the command's commit.
@@ -619,8 +678,8 @@ pub const Engine = struct {
         if (mode == .replace) self.evict_doc(coll, id_key);
 
         // 7. Publish the document and its entries: copy the bytes into the
-        //    slab and record the offset.
+        //    slab and record the offset. Infallible from here.
-        const off = try coll.slab_append(self.gpa, doc_bytes);
+        const off = coll.slab_append(doc_bytes);
         try coll.docs.put(self.gpa, id_key, off);
         self.live_docs += 1;
         for (built_list.items) |*b| {
@@ -908,7 +967,7 @@ pub const Engine = struct {
         errdefer self.gpa.free(coll_key);
         const new_coll = try self.gpa.create(Collection);
         errdefer self.gpa.destroy(new_coll);
-        new_coll.* = try Collection.init(self.gpa);
+        new_coll.* = try Collection.init(self.gpa, self.pager);
         errdefer new_coll.id_index.deinit(self.gpa);
         try db.collections.put(self.gpa, coll_key, new_coll);
         return new_coll;
@@ -1276,7 +1335,8 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
             self.evict_doc(coll, id_key);
             const doc_bytes = try serialize_doc(self.gpa, doc);
             defer self.gpa.free(doc_bytes);
-            const off = try coll.slab_append(self.gpa, doc_bytes);
+            try coll.slab_reserve(self.gpa, doc_bytes.len);
+            const off = coll.slab_append(doc_bytes);
             try coll.docs.put(self.gpa, id_key, off);
             self.live_docs += 1;
             key_owned = true;

tests/e2e/big.js

@@ -136,6 +136,7 @@ async function main() {
     process.exit(1);
   }
   fs.rmSync(DBFILE, { force: true });
+  fs.rmSync(DBFILE + '.data', { force: true });
   const fmt = (n) => (n >= 1e9 ? (n / 1e9).toFixed(2) + ' GB' : n >= 1e6 ? (n / 1e6).toFixed(1) + ' MB' : n >= 1e3 ? (n / 1e3).toFixed(1) + ' KB' : n + ' B');
   console.log(`multiforadb big-collection harness`);
   console.log(`  size ${fmt(SIZE)} · doc ~${fmt(DOC_SIZE)} · batch ${opt.batch} · index ${opt.index || 'none'} · ids ${opt.oid ? 'ObjectId' : 'int'} · compact-threshold ${opt.compactThreshold || '16m'} · db ${DBFILE}`);
@@ -327,7 +328,10 @@ async function main() {
   row('kill -9 after 200 committed writes', `${crashN}/200 survived (${crashN === 200 ? 'OK' : 'MISMATCH!'})`);
   await c3.close();
 
-  if (!opt.keep) fs.rmSync(DBFILE, { force: true });
+  if (!opt.keep) {
+    fs.rmSync(DBFILE, { force: true });
+    fs.rmSync(DBFILE + '.data', { force: true });
+  }
   await stopServer('SIGKILL');
 
   console.log('\n== summary ==');

tests/e2e/compare-run.sh

@@ -20,7 +20,7 @@ MFDB_OUT="$CMPDIR/mfdb-srv.out"
 MD_OUT="$CMPDIR/md-srv.out"
 MFDB_PORT=27019
 MD_PORT=27018
-rm -f "$MFDB_LOG"
+rm -f "$MFDB_LOG" "$MFDB_LOG.data"
 rm -rf "$CMPDIR/mongod" && mkdir -p "$CMPDIR/mongod"
 
 # ---- MongoDB -------------------------------------------------------------

tests/e2e/e2e6.js

@@ -423,7 +423,10 @@ async function main() {
   console.log('phase 3: kill -9 crash recovery');
   await phase3(client2);
 
-  if (process.env.E2E6_KEEP !== '1') fs.rmSync(DBFILE, { force: true });
+  if (process.env.E2E6_KEEP !== '1') {
+    fs.rmSync(DBFILE, { force: true });
+    fs.rmSync(DBFILE + '.data', { force: true });
+  }
   await stopServer('SIGTERM');
 
   const failed = results.filter((r) => !r.ok);