diff --git a/src/lib.zig b/src/lib.zig index 15eb2d2..406c10e 100644 --- a/src/lib.zig +++ b/src/lib.zig @@ -10,6 +10,7 @@ pub const db = @import("db.zig"); pub const query = @import("query.zig"); pub const update = @import("update.zig"); pub const index = @import("index.zig"); +pub const pager = @import("pager.zig"); test { _ = @import("assert.zig"); @@ -22,4 +23,5 @@ test { _ = @import("query.zig"); _ = @import("update.zig"); _ = @import("index.zig"); + _ = @import("pager.zig"); } diff --git a/src/pager.zig b/src/pager.zig new file mode 100644 index 0000000..dcd407f --- /dev/null +++ b/src/pager.zig @@ -0,0 +1,615 @@ +//! The data file: a checkpoint of the engine's structures, mapped rather than +//! parsed. The log (storage.zig) remains the WAL and the source of truth; this +//! file is a snapshot at some sequence number, and on open the log replays +//! whatever happened after it (PLAN D4). +//! +//! File layout — an array of 4 KiB pages: +//! page 0 header. Written at create, never rewritten. +//! [0..4) u32 magic "MFDB" (0x4D464442) +//! [4] u8 format version (1) +//! [5] u8 page_shift (12) +//! [6..8) u16 reserved (0) +//! [8..24) u128 database uuid — must match the log's, so a data file +//! can never be paired with a foreign log +//! [24..32) u64 xxhash3 over [0..24) +//! [32..4096) zero +//! pages 1,2 watermark double buffer (commit 9) +//! pages 3.. data, handed out by a tail-bump extent allocator +//! +//! Every persisted reference is a page number (u32) or an absolute file byte +//! offset (u64) — never a pointer — so where the file happens to be mapped is +//! irrelevant. Node pages are raw host memory (D4: on-disk format == in-memory +//! format, no serialization on page-in), which makes the data file +//! little-endian-only; the log, framed field by field, stays portable. +//! +//! PLAN D6.1 called for a region table. It cannot be that: there is one node +//! arena per index and one document slab per collection, so the region count is +//! dynamic and unbounded, and N contiguous regions cannot all grow at the tail. +//! With one page array and extent allocation there is exactly one growth path, +//! therefore exactly one place where the write-then-extend discipline lives, +//! and `ls`/`du` stay honest for the backup story (D6.6). +//! +//! Why this reaches for std.posix instead of std.Io, against the house style: +//! `std.Io.File.MemoryMap` prefaults by default (`populate = true`), which is +//! the opposite of "RSS = working set"; it exposes no NORESERVE, FIXED or +//! address hint, so it cannot express an address-space reservation; and its +//! `setLength` is mremap on Linux and unsupported on darwin, which this project +//! develops on. Everything else here goes through std.Io.File. + +const std = @import("std"); +const assert = @import("assert.zig").assert; +const assert_msg = @import("assert.zig").assert_msg; + +pub const page_shift: u6 = 12; +pub const page_size: usize = 1 << page_shift; + +/// Page numbers of the fixed-position pages. +pub const page_header: u32 = 0; +pub const page_watermark_a: u32 = 1; +pub const page_watermark_b: u32 = 2; +/// The first page the allocator may hand out. +pub const page_first_data: u32 = 3; + +const magic: u32 = 0x4D464442; // "MFDB" +const format_version: u8 = 1; +const header_hashed_len: usize = 24; + +/// The alignment mmap requires, which is the *system* page size and not ours. +/// 16 KiB on Apple Silicon against a 4 KiB logical page, so four logical pages +/// share one system page there — which is why the checkpoint rounds its append +/// cursors up to this rather than to page_size. +pub const map_align = std.heap.page_size_min; + +/// Growth granularity. Large enough that growth is rare and each `setLength` +/// covers many allocations, and a multiple of every supported system page size. +const grow_chunk_pages: u32 = 2048; // 8 MiB + +/// Address space reserved by default. Reserved, not committed: a PROT_NONE +/// anonymous NORESERVE mapping is one VMA and zero pages. Sized for the +/// tens-of-GB target (D3) with room to spare. +pub const default_reserve_bytes: usize = 64 << 30; + +comptime { + assert(page_size == 4096); + assert(page_first_data == 3); + assert(grow_chunk_pages * page_size % map_align == 0); + assert(map_align % page_size == 0 or page_size % map_align == 0); + // Node pages are raw host memory in this file (see the header comment). + assert(@import("builtin").cpu.arch.endian() == .little); +} + +pub const Error = error{ + /// Not a data file, or a version this build cannot read. + InvalidDataFile, + /// The file belongs to a different database than the log beside it. + DatabaseMismatch, + /// Growth would exceed the reserved address space. + DatabaseTooLarge, +}; + +pub const OpenOptions = struct { + /// Must match the log's uuid. Zero means "adopt whatever the file has", + /// which is only for tests that do not care. + uuid: u128 = 0, + reserve_bytes: usize = default_reserve_bytes, +}; + +pub const Pager = struct { + gpa: std.mem.Allocator, + io: std.Io, + file: std.Io.File, + path: []u8, + uuid: u128, + + /// The address-space reservation. Never moves for the life of the process, + /// so a pointer obtained from `page`/`page_mut` stays valid across growth — + /// which the old ArrayList-backed arena could not promise, and which is the + /// reason a whole class of dangling-pointer bugs disappears here. + reserve: []align(map_align) u8, + /// Pages actually mapped to the file. Always >= alloc_tail. + mapped_pages: u32, + /// Pages the file is long enough to hold. + file_pages: u32, + /// Pages [0, alloc_tail) have been handed out. + alloc_tail: u32, + /// Headroom promised by `reserve_pages`, so allocation after it is + /// infallible. Never below alloc_tail. + reserved_tail: u32, + + /// True when this file was created by this open (no checkpoint to load). + fresh: bool, + + pub fn open( + gpa: std.mem.Allocator, + io: std.Io, + path: []const u8, + opts: OpenOptions, + ) !Pager { + // Absolute, for the same reason storage.Log resolves its own path: the + // rebuild renames this file and must not depend on the caller's working + // directory. + const owned_path = blk: { + if (path.len > 0 and path[0] == '/') break :blk try gpa.dupe(u8, path); + const cwd = try std.process.currentPathAlloc(io, gpa); + defer gpa.free(cwd); + break :blk try std.fmt.allocPrint(gpa, "{s}/{s}", .{ cwd, path }); + }; + errdefer gpa.free(owned_path); + + const dir = std.Io.Dir.cwd(); + var created = false; + var file = dir.openFile(io, owned_path, .{ .mode = .read_write }) catch |err| switch (err) { + error.FileNotFound => blk: { + created = true; + break :blk try dir.createFile(io, owned_path, .{ .read = true }); + }, + else => return err, + }; + errdefer file.close(io); + + // A zero-length file is as good as absent: a create that died before + // its header landed must not look like a corrupt database. + const existing_len = try file.length(io); + if (existing_len == 0) created = true; + + const reserve = try reserve_address_space(opts.reserve_bytes); + errdefer std.posix.munmap(reserve); + + var self: Pager = .{ + .gpa = gpa, + .io = io, + .file = file, + .path = owned_path, + .uuid = opts.uuid, + .reserve = reserve, + .mapped_pages = 0, + .file_pages = @intCast(existing_len / page_size), + .alloc_tail = page_first_data, + .reserved_tail = page_first_data, + .fresh = created, + }; + + if (created) { + try self.grow_to(page_first_data); + try self.write_header(); + } else { + try self.grow_to(@max(page_first_data, self.file_pages)); + try self.read_header(); + // Everything already in the file is allocated as far as this + // process knows until a watermark says otherwise (commit 9). + self.alloc_tail = @max(page_first_data, self.file_pages); + self.reserved_tail = self.alloc_tail; + } + return self; + } + + pub fn deinit(self: *Pager) void { + std.posix.munmap(self.reserve); + self.file.close(self.io); + self.gpa.free(self.path); + } + + // -- page access -------------------------------------------------------- + + /// The page's bytes, for reading. + pub inline fn page(self: *const Pager, p: u32) *align(page_size) const [page_size]u8 { + assert_msg(p < self.mapped_pages, "read of a page past the mapped end of the data file"); + return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); + } + + /// The page's bytes, for writing. + /// + /// This is where copy-on-write will hook in (commit 12): once a checkpoint + /// has published a stable prefix, a write to a page inside it must copy the + /// page first, or a crash finds a half-updated durable image. Funnelling + /// every write through one function is what makes that a change of one body + /// rather than of every caller. + 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"); + return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); + } + + /// Bytes at an absolute file offset, for the consumers whose references are + /// byte offsets rather than page numbers: the document slab and the + /// overflow slab. + pub inline fn bytes(self: *const Pager, off: u64, len: usize) []const u8 { + assert_msg(off + len <= @as(u64, self.mapped_pages) << page_shift, "read past the mapped end of the data file"); + return self.reserve[@intCast(off)..][0..len]; + } + + 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"); + return self.reserve[@intCast(off)..][0..len]; + } + + // -- allocation --------------------------------------------------------- + + /// Hand out `n` contiguous pages, growing the file if needed. + pub fn alloc_pages(self: *Pager, n: u32) !u32 { + assert(n > 0); + try self.grow_to(self.alloc_tail + n); + // Growing satisfies the promise `alloc_pages_assume_reserved` checks; + // without this it would assert against a reservation nobody made. + self.reserved_tail = @max(self.reserved_tail, self.alloc_tail + n); + return self.alloc_pages_assume_reserved(n); + } + + /// Guarantee that the next `n` pages can be handed out without failing. + /// Fallible, and meant to run *before* the log append on a write path, so + /// that publishing afterwards cannot fail — the invariant that keeps a + /// document from ever being live but unindexed. + /// + /// Nothing is dirtied here, so nothing is allocated on disk: `setLength` + /// leaves a sparse file, `ls -l` grows and `du` does not. That is what makes + /// a generous reservation cheap. + pub fn reserve_pages(self: *Pager, n: u32) !void { + try self.grow_to(self.alloc_tail + n); + self.reserved_tail = self.alloc_tail + n; + } + + /// Hand out `n` pages against a previous `reserve_pages`. Infallible. + pub fn alloc_pages_assume_reserved(self: *Pager, n: u32) u32 { + assert(n > 0); + assert_msg( + self.alloc_tail + n <= self.reserved_tail, + "page allocation overran reserve_pages' promise", + ); + assert_msg( + self.alloc_tail + n <= self.mapped_pages, + "page allocation past the mapped end of the data file", + ); + const first = self.alloc_tail; + self.alloc_tail += n; + return first; + } + + /// Bytes currently allocated, i.e. the extent of the live image. + pub fn allocated_bytes(self: *const Pager) u64 { + return @as(u64, self.alloc_tail) << page_shift; + } + + // -- durability --------------------------------------------------------- + + /// Flush every allocated page and the inode. Only the checkpoint calls + /// this: between checkpoints dirty pages may sit in the page cache + /// indefinitely, because recovery is `image + replay(seq > watermark)` and + /// the image's pages are never written *differently*. That is what keeps + /// the write path at exactly one fsync — the WAL's (PLAN amendment A1). + pub fn sync(self: *Pager) !void { + const len = std.mem.alignForward(usize, @intCast(self.allocated_bytes()), map_align); + if (len > 0) { + try std.posix.msync(self.reserve[0..len], std.posix.MSF.SYNC); + } + try self.file.sync(self.io); + } + + // -- growth ------------------------------------------------------------- + + /// Make at least `want_pages` pages mapped and file-backed. + /// + /// The order is the whole point: extend the file, *then* map the new + /// suffix. A store into a mapped page past end-of-file raises SIGBUS, which + /// no Zig error path can catch, so the file must be long enough before any + /// page in the range is reachable. The accessors assert against + /// `mapped_pages` so a violation is a panic with a message rather than a + /// signal. + fn grow_to(self: *Pager, want_pages: u32) !void { + if (want_pages <= self.mapped_pages) return; + + // Round up to a growth chunk, and to the system page size, so a + // 16 KiB-page host never gets a partial mapping request. + const chunk = @max(grow_chunk_pages, self.mapped_pages / 8); + var new_pages = std.mem.alignForward(u32, want_pages, chunk); + const sys_pages: u32 = @intCast(map_align / page_size); + if (sys_pages > 1) new_pages = std.mem.alignForward(u32, new_pages, sys_pages); + + if (@as(u64, new_pages) << page_shift > self.reserve.len) { + // One more try at exactly what was asked for: a reservation the + // chunking would overshoot is still usable up to its end. + new_pages = want_pages; + if (sys_pages > 1) new_pages = std.mem.alignForward(u32, new_pages, sys_pages); + if (@as(u64, new_pages) << page_shift > self.reserve.len) return Error.DatabaseTooLarge; + } + + const new_len: u64 = @as(u64, new_pages) << page_shift; + if (new_len > self.file_pages_bytes()) { + try self.file.setLength(self.io, new_len); + self.file_pages = new_pages; + } + + const off: usize = @as(usize, self.mapped_pages) << page_shift; + const len: usize = @intCast(new_len - (@as(u64, self.mapped_pages) << page_shift)); + _ = try std.posix.mmap( + @alignCast(self.reserve.ptr + off), + len, + .{ .READ = true, .WRITE = true }, + .{ .TYPE = .SHARED, .FIXED = true }, + self.file.handle, + off, + ); + self.mapped_pages = new_pages; + } + + fn file_pages_bytes(self: *const Pager) u64 { + return @as(u64, self.file_pages) << page_shift; + } + + // -- header ------------------------------------------------------------- + + fn write_header(self: *Pager) !void { + const h = self.page_mut(page_header); + @memset(h, 0); + std.mem.writeInt(u32, h[0..4], magic, .little); + h[4] = format_version; + h[5] = page_shift; + std.mem.writeInt(u128, h[8..24], self.uuid, .little); + std.mem.writeInt(u64, h[24..32], header_hash(h[0..header_hashed_len]), .little); + } + + fn read_header(self: *Pager) !void { + const h = self.page(page_header); + if (std.mem.readInt(u32, h[0..4], .little) != magic) return Error.InvalidDataFile; + if (h[4] != format_version) return Error.InvalidDataFile; + if (h[5] != page_shift) return Error.InvalidDataFile; + const want = std.mem.readInt(u64, h[24..32], .little); + if (header_hash(h[0..header_hashed_len]) != want) return Error.InvalidDataFile; + const file_uuid = std.mem.readInt(u128, h[8..24], .little); + if (self.uuid == 0) { + self.uuid = file_uuid; + } else if (file_uuid != self.uuid) { + return Error.DatabaseMismatch; + } + } +}; + +fn header_hash(b: []const u8) u64 { + return std.hash.XxHash3.hash(0, b); +} + +/// Reserve address space without committing memory: PROT_NONE, anonymous and +/// NORESERVE is one VMA and zero pages, and is not charged even under strict +/// overcommit. Nothing is mapped to the file yet, so nothing here can fault. +/// +/// Halves on refusal rather than failing outright, so a container with a tight +/// address-space limit still opens — with a smaller ceiling, which `grow_to` +/// reports as DatabaseTooLarge if it is ever actually reached. +fn reserve_address_space(want: usize) ![]align(map_align) u8 { + var len = std.mem.alignForward(usize, want, map_align); + while (true) { + if (std.posix.mmap( + null, + len, + .{}, // PROT_NONE + .{ .TYPE = .PRIVATE, .ANONYMOUS = true, .NORESERVE = true }, + -1, + 0, + )) |m| { + return m; + } else |err| { + if (len <= 1 << 30) return err; + len /= 2; + } + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +const testing = std.testing; + +/// A pager over a temp file, with a reservation small enough that the growth +/// ceiling is reachable in a test. +const TmpPager = struct { + tmp: std.testing.TmpDir, + path: []u8, + /// Optional so a test can close it early -- several deliberately break the + /// file and must not have to reopen one just to satisfy teardown. + pager: ?Pager, + + fn init(io: std.Io, reserve_bytes: usize) !TmpPager { + const tmp = std.testing.tmpDir(.{}); + const path = try std.fmt.allocPrint( + testing.allocator, + ".zig-cache/tmp/{s}/data", + .{tmp.sub_path}, + ); + return .{ + .tmp = tmp, + .path = path, + .pager = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = reserve_bytes }), + }; + } + + fn deinit(self: *TmpPager) void { + if (self.pager) |*p| p.deinit(); + self.pager = null; + self.tmp.cleanup(); + testing.allocator.free(self.path); + } + + fn close(self: *TmpPager) void { + if (self.pager) |*p| p.deinit(); + self.pager = null; + } + + fn pg(self: *TmpPager) *Pager { + return &self.pager.?; + } +}; + +test "pages survive growth and the mapping base never moves" { + // The reason the fixed reservation exists: a pointer handed out before a + // growth must still be valid after it. The ArrayList-backed arena this + // replaces could not promise that, and the workarounds for it (copying + // promoted keys into a scratch buffer) exist only because of it. + // + // Mutation check: make grow_to munmap and re-map the whole prefix at a + // kernel-chosen address and the base assertion goes red. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 256 << 20); + defer tp.deinit(); + const pg = tp.pg(); + + const base = @intFromPtr(pg.reserve.ptr); + const first = try pg.alloc_pages(1); + const first_ptr = pg.page_mut(first); + @memset(first_ptr, 0xAB); + + // Grow well past several chunk boundaries. + var written: std.ArrayListUnmanaged(u32) = .empty; + defer written.deinit(testing.allocator); + for (0..6000) |i| { + const p = try pg.alloc_pages(1); + const bytes = pg.page_mut(p); + @memset(bytes, @truncate(i)); + try written.append(testing.allocator, p); + } + + try testing.expectEqual(base, @intFromPtr(pg.reserve.ptr)); + // The pointer taken before all that growth still addresses its page. + for (first_ptr) |b| try testing.expectEqual(@as(u8, 0xAB), b); + for (written.items, 0..) |p, i| { + const want: u8 = @truncate(i); + try testing.expectEqual(want, pg.page(p)[0]); + try testing.expectEqual(want, pg.page(p)[page_size - 1]); + } +} + +test "the file is extended before any page in the range is reachable" { + // Write-then-extend, asserted as an ordering rather than by provoking the + // failure: a store past end-of-file is SIGBUS, which cannot be caught + // in-process, so the check is that the file is always long enough for + // everything mapped, and everything allocated is always mapped. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 256 << 20); + defer tp.deinit(); + const pg = tp.pg(); + + for (0..40) |_| { + _ = try pg.alloc_pages(97); // not a chunk divisor, so growth is ragged + try testing.expect(pg.alloc_tail <= pg.mapped_pages); + try testing.expect(pg.mapped_pages <= pg.file_pages); + const on_disk = try pg.file.length(io); + try testing.expect(on_disk >= @as(u64, pg.mapped_pages) << page_shift); + } +} + +test "a reservation makes the following allocation infallible" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 256 << 20); + defer tp.deinit(); + const pg = tp.pg(); + + try pg.reserve_pages(64); + const before = pg.alloc_tail; + // Exactly the promised amount, one page at a time. + for (0..64) |_| _ = pg.alloc_pages_assume_reserved(1); + try testing.expectEqual(before + 64, pg.alloc_tail); +} + +test "growth past the reservation is an error, not a crash" { + // A bounded reservation must fail cleanly at its ceiling: this is the path + // a container with a tight address-space limit takes. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 1 << 30); + defer tp.deinit(); + const pg = tp.pg(); + + const capacity: u32 = @intCast(pg.reserve.len / page_size); + try testing.expectError(Error.DatabaseTooLarge, pg.alloc_pages(capacity + 1)); + // And the pager is still usable afterwards. + const p = try pg.alloc_pages(1); + @memset(pg.page_mut(p), 1); +} + +test "header round-trips and rejects a foreign database" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + + var tp = try TmpPager.init(io, 64 << 20); + defer tp.deinit(); + try testing.expect(tp.pg().fresh); + const p = try tp.pg().alloc_pages(3); + @memset(tp.pg().page_mut(p), 0x5A); + try tp.pg().sync(); + const path = try testing.allocator.dupe(u8, tp.path); + defer testing.allocator.free(path); + tp.close(); + + // Reopening finds the same header and is not fresh. + var again = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); + try testing.expect(!again.fresh); + try testing.expectEqual(@as(u128, 7), again.uuid); + try testing.expectEqual(@as(u8, 0x5A), again.page(p)[0]); + again.deinit(); + + // A different uuid is a different database; pairing them would silently + // replay one database's log onto another's checkpoint. + // Mutation check: drop the uuid comparison in read_header and this passes. + try testing.expectError( + Error.DatabaseMismatch, + Pager.open(testing.allocator, io, path, .{ .uuid = 8, .reserve_bytes = 64 << 20 }), + ); +} + +test "a corrupt header is rejected rather than misread" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + + var tp = try TmpPager.init(io, 64 << 20); + defer tp.deinit(); + const path = try testing.allocator.dupe(u8, tp.path); + defer testing.allocator.free(path); + try tp.pg().sync(); + tp.close(); + + // Flip a byte inside the hashed region. + { + var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); + defer f.close(io); + var b: [1]u8 = undefined; + _ = try f.readPositionalAll(io, &b, 6); + b[0] ^= 0xFF; + try f.writePositionalAll(io, &b, 6); + } + // Mutation check: drop the hash comparison in read_header and this passes, + // which is worse than failing — a header believed on faith describes where + // every structure in the file lives. + try testing.expectError( + Error.InvalidDataFile, + Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }), + ); +} + +test "an empty file is treated as absent, not as corruption" { + // A create that died before its header landed must not look like a corrupt + // database — the engine has to be able to open. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + + var tp = try TmpPager.init(io, 64 << 20); + defer tp.deinit(); + const path = try testing.allocator.dupe(u8, tp.path); + defer testing.allocator.free(path); + tp.close(); + + { + var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); + defer f.close(io); + try f.setLength(io, 0); + } + tp.pager = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); + try testing.expect(tp.pg().fresh); +}