diff --git a/.gitignore b/.gitignore index b398b00..9a72088 100644 --- a/.gitignore +++ b/.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/ diff --git a/src/db.zig b/src/db.zig index b1bd678..56e2827 100644 --- a/src/db.zig +++ b/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 -/// would hold up to 2x its contents after doubling). -const slab_segment_size = 8 * 1024 * 1024; +/// Pages in a standard slab extent: 8 MiB, as the old in-memory segments were. +/// Slack is bounded by one extent per collection. +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 - /// fixed segments; the map holds each document's flat slab offset. - /// Offsets stay valid forever: segments are append-only and never move, - /// so a segment's bytes are stable even when the segment list reallocates. - /// Segmenting (instead of one geometric-growth array) keeps the slab's - /// capacity slack under one segment — a single array would hold up to - /// 2x its contents after doubling. Removed documents leave garbage bytes - /// until compaction rewrites. + /// Documents live as canonical BSON bytes in the data file, in extents this + /// collection owns; the map holds each document's offset. Those are + /// *absolute file offsets* now, which is what makes doc_bytes a single add + /// rather than a binary search over segment starts -- and what removes the + /// dangling-pointer hazard the old segment list had, since the mapping's + /// base never moves. + /// + /// 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)), - /// Flat offset where each segment begins; doc_bytes binary-searches it. - seg_starts: std.ArrayListUnmanaged(u64), + /// The data file this collection's documents live in. + pager: *pgr.Pager, + /// 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 { - var self: Collection = .{ .docs = .empty, .slab = .empty, .seg_starts = .empty, .indexes = .empty, .id_index = undefined }; + fn init(gpa: std.mem.Allocator, pager: *pgr.Pager) !Collection { + 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 { - if (self.slab.items.len == 0) { - try self.slab.append(gpa, .empty); - try self.seg_starts.append(gpa, 0); - } - // Length of the last segment as a value, never as a pointer into - // slab.items: appending the next segment below may reallocate that - // list, which would dangle a pointer taken before the append and - // corrupt the new segment's start offset (and with it every - // doc_bytes lookup in that segment — reads that surfaced as - // InvalidBson, or a crash in Debug builds). - const last_len = self.slab.items[self.slab.items.len - 1].items.len; - if (last_len + bytes.len > slab_segment_size) { - try self.slab.append(gpa, .empty); - try self.seg_starts.append(gpa, self.seg_starts.items[self.seg_starts.items.len - 1] + last_len); - return self.slab_append(gpa, bytes); - } - const last = &self.slab.items[self.slab.items.len - 1]; - const off = self.seg_starts.items[self.seg_starts.items.len - 1] + last.items.len; - try last.appendSlice(gpa, bytes); + /// Make room for a document of `len` bytes, so the append that follows + /// cannot fail. + /// + /// Separated from the append because the append runs *after* the log + /// record is durable, where failure has nowhere to go: the write is already + /// committed and reporting an error for it would be a lie the next open + /// contradicts. Reserving first keeps the fallible half before the log. + fn slab_reserve(self: *Collection, gpa: std.mem.Allocator, len: usize) !void { + if (self.slab_tail + len <= self.slab_end) return; + // A document larger than the standard extent gets one of its own; BSON + // reaches 16 MB and the extent is 8 MiB. + const want_pages: u32 = @intCast(@max( + slab_extent_pages, + (len + pgr.page_size - 1) / pgr.page_size, + )); + try self.pager.reserve_pages(want_pages); + const first = self.pager.alloc_pages_assume_reserved(want_pages); + try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages }); + self.slab_tail = @as(u64, first) << pgr.page_shift; + 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). - var lo: usize = 0; - var hi: usize = self.slab.items.len; - while (lo + 1 < hi) { - 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]; + // An absolute file offset, so this is base + off. The length comes from + // the document's own BSON int32 prefix, as it always has. + const len: usize = std.mem.readInt(u32, self.pager.bytes(off, 4)[0..4], .little); + return self.pager.bytes(off, 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); - coll.slab.deinit(self.gpa); - coll.seg_starts.deinit(self.gpa); + // Give the slab's pages back. They become reusable two generations + // later, so a fallback to the previous image still finds them intact. + 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 - // insertion after the log append is infallible. + // 4. Reserve everything the publish step needs -- tree capacity and + // 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. - const off = try coll.slab_append(self.gpa, doc_bytes); + // slab and record the offset. Infallible from here. + 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; diff --git a/tests/e2e/big.js b/tests/e2e/big.js index bb7fda8..e76e505 100644 --- a/tests/e2e/big.js +++ b/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 =='); diff --git a/tests/e2e/compare-run.sh b/tests/e2e/compare-run.sh index 58562fb..40687c4 100644 --- a/tests/e2e/compare-run.sh +++ b/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 ------------------------------------------------------------- diff --git a/tests/e2e/e2e6.js b/tests/e2e/e2e6.js index b7571fa..6d19642 100644 --- a/tests/e2e/e2e6.js +++ b/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);