diff --git a/src/pager.zig b/src/pager.zig index dcd407f..ae129d9 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -13,9 +13,30 @@ //! 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 1,2 watermark slots A and B. Generation N writes slot (N & 1), +//! so a torn write can never leave zero valid slots -- writing +//! a new generation over the only copy of the old one could. +//! The newer slot whose hash validates is authoritative. +//! [0..8) u64 generation (monotonic, >= 1) +//! [8..16) u64 seq -- the log sequence this image covers +//! [16..24) u64 alloc_tail -- pages [0, alloc_tail) are allocated +//! [24..32) u64 file_pages +//! [32..40) u64 catalog_page -- first page of the catalog stream +//! [40..48) u64 catalog_len -- bytes of it +//! [48..56) u64 freelist_page +//! [56..64) u64 freelist_len +//! [64..72) u64 prev_generation -- kept intact for fallback +//! [72..80) u64 live_docs -- cached hint +//! [80..88) u64 dead_bytes -- cached, drives the rebuild trigger +//! [88..4088) reserved (zero) +//! [4088..4096) u64 xxhash3 over [0..4088) //! pages 3.. data, handed out by a tail-bump extent allocator //! +//! The catalog stream and the free list are written *wholesale into freshly +//! allocated pages* at every checkpoint, never mutated in place. That is what +//! makes them untearable: the previous copy stays intact and referenced by the +//! previous watermark until the new watermark switches over. +//! //! 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 @@ -37,9 +58,21 @@ //! develops on. Everything else here goes through std.Io.File. const std = @import("std"); +const builtin = @import("builtin"); const assert = @import("assert.zig").assert; const assert_msg = @import("assert.zig").assert_msg; +/// Test-only: record which pages have been written since the last sync, so a +/// test can simulate the one failure a `kill -9` cannot produce -- writeback +/// that never happened. +/// +/// This exists because the ordering it guards is the whole crash-recovery +/// argument and is otherwise untestable. Killing a process does not lose +/// page-cache writes, so removing the msync before the watermark leaves every +/// test green while making a real power loss unrecoverable. Compiled out +/// entirely outside tests, so `page_mut` keeps no branch in a real build. +const track_dirty = builtin.is_test; + pub const page_shift: u6 = 12; pub const page_size: usize = 1 << page_shift; @@ -78,6 +111,34 @@ comptime { assert(@import("builtin").cpu.arch.endian() == .little); } +/// A run of contiguous pages. +pub const Extent = struct { + first: u32, + pages: u32, +}; + +/// What a checkpoint publishes. Everything here is authoritative except the +/// two cached counters, which are hints the engine recomputes if they look +/// wrong. +pub const Watermark = struct { + generation: u64 = 0, + /// The log sequence this image covers. The crash-recovery invariant is that + /// this never exceeds the durable log tail (PLAN D6), so replay after open + /// is exactly the records above it. + seq: u64 = 0, + alloc_tail: u32 = page_first_data, + file_pages: u32 = page_first_data, + catalog_page: u32 = 0, + catalog_len: u64 = 0, + freelist_page: u32 = 0, + freelist_len: u64 = 0, + prev_generation: u64 = 0, + live_docs: u64 = 0, + dead_bytes: u64 = 0, +}; + +const wm_hashed_len: usize = page_size - 8; + pub const Error = error{ /// Not a data file, or a version this build cannot read. InvalidDataFile, @@ -119,6 +180,32 @@ pub const Pager = struct { /// True when this file was created by this open (no checkpoint to load). fresh: bool, + /// The watermark this pager opened from, or a zero one when fresh. The + /// engine reads `seq` to decide where replay starts. + loaded: Watermark, + /// The generation the next checkpoint will publish. + generation: u64, + + /// Pages below this belong to the last published image and must never be + /// stored into (PLAN amendment A1). Zero until a checkpoint publishes one, + /// which is why copy-on-write is inert before then. + stable_pages: u32, + + /// Free pages, in three stages. `ready` may be handed out now; `hold` was + /// freed one generation ago; `pending` was freed during this generation. + /// + /// The two-generation delay is not caution, it is what makes generation N-1 + /// remain a usable image: its pages are still allocated while N is current, + /// so a torn watermark or a bad catalog can fall back to it instead of + /// discarding the database. Releasing one generation early would silently + /// take that away. + free_ready: std.ArrayListUnmanaged(Extent), + free_hold: std.ArrayListUnmanaged(Extent), + free_pending: std.ArrayListUnmanaged(Extent), + + /// See `track_dirty`. Pages written since the last sync. + dirty: if (track_dirty) std.AutoHashMapUnmanaged(u32, void) else void, + pub fn open( gpa: std.mem.Allocator, io: std.Io, @@ -167,6 +254,13 @@ pub const Pager = struct { .alloc_tail = page_first_data, .reserved_tail = page_first_data, .fresh = created, + .loaded = .{}, + .generation = 0, + .stable_pages = 0, + .free_ready = .empty, + .free_hold = .empty, + .free_pending = .empty, + .dirty = if (track_dirty) .empty else {}, }; if (created) { @@ -175,15 +269,20 @@ pub const Pager = struct { } 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). + // Everything already in the file is allocated until a watermark + // narrows it down. self.alloc_tail = @max(page_first_data, self.file_pages); self.reserved_tail = self.alloc_tail; + try self.load_watermark(); } return self; } pub fn deinit(self: *Pager) void { + self.free_ready.deinit(self.gpa); + self.free_hold.deinit(self.gpa); + self.free_pending.deinit(self.gpa); + if (track_dirty) self.dirty.deinit(self.gpa); std.posix.munmap(self.reserve); self.file.close(self.io); self.gpa.free(self.path); @@ -206,6 +305,7 @@ pub const Pager = struct { /// 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"); + if (track_dirty) self.dirty.put(self.gpa, p, {}) catch {}; return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); } @@ -219,6 +319,11 @@ pub const Pager = struct { 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"); + if (track_dirty) { + var pg_i: u32 = @intCast(off >> page_shift); + const last: u32 = @intCast((off + len - 1) >> page_shift); + while (pg_i <= last) : (pg_i += 1) self.dirty.put(self.gpa, pg_i, {}) catch {}; + } return self.reserve[@intCast(off)..][0..len]; } @@ -281,6 +386,30 @@ pub const Pager = struct { try std.posix.msync(self.reserve[0..len], std.posix.MSF.SYNC); } try self.file.sync(self.io); + if (track_dirty) self.dirty.clearRetainingCapacity(); + } + + /// Test-only: overwrite, in the *file*, every page written since the last + /// sync -- the state a power loss can leave behind and a `kill -9` cannot. + /// Call it with the mapping already closed. + fn simulate_lost_writeback(gpa: std.mem.Allocator, io: std.Io, path: []const u8, lost: []const u32) !void { + var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); + defer f.close(io); + const junk = try gpa.alloc(u8, page_size); + defer gpa.free(junk); + // Not zeroes: zeroes are a plausible value, and a test that passes + // because the lost page happened to read as zero proves nothing. + @memset(junk, 0xDD); + for (lost) |p| try f.writePositionalAll(io, junk, @as(u64, p) << page_shift); + } + + /// Test-only: the pages `simulate_lost_writeback` would discard. + fn dirty_pages(self: *const Pager, gpa: std.mem.Allocator) ![]u32 { + var out: std.ArrayListUnmanaged(u32) = .empty; + errdefer out.deinit(gpa); + var it = self.dirty.keyIterator(); + while (it.next()) |k| try out.append(gpa, k.*); + return out.toOwnedSlice(gpa); } // -- growth ------------------------------------------------------------- @@ -334,6 +463,208 @@ pub const Pager = struct { return @as(u64, self.file_pages) << page_shift; } + // -- watermark ---------------------------------------------------------- + + /// Adopt the newest valid watermark, or none. Never fails on a damaged + /// watermark: a data file whose slots are both unreadable is treated as + /// having no checkpoint, which means a full replay -- the database must + /// always open (ground rule 4). The caller learns which happened from + /// `loaded.generation == 0`. + fn load_watermark(self: *Pager) !void { + const a = self.read_slot(page_watermark_a); + const b = self.read_slot(page_watermark_b); + // Validate first, compare generations second. The other order picks a + // torn slot whenever its garbage generation happens to be larger. + const pick: ?Watermark = if (a != null and b != null) + (if (a.?.generation >= b.?.generation) a else b) + else if (a != null) a else b; + + if (pick) |wm| { + self.loaded = wm; + self.generation = wm.generation; + self.alloc_tail = wm.alloc_tail; + self.reserved_tail = wm.alloc_tail; + // Everything the published image references is off limits to + // writes from here on. + self.stable_pages = wm.alloc_tail; + try self.read_freelist(wm); + } else if (self.file_pages > page_first_data) { + std.debug.print( + "multiforadb: WARNING: data file '{s}' has no valid watermark; " ++ + "treating it as having no checkpoint and replaying the log in full\n", + .{self.path}, + ); + } + } + + fn read_slot(self: *const Pager, p: u32) ?Watermark { + if (p >= self.mapped_pages) return null; + const b = self.page(p); + const want = std.mem.readInt(u64, b[wm_hashed_len..][0..8], .little); + if (header_hash(b[0..wm_hashed_len]) != want) return null; + const wm: Watermark = .{ + .generation = std.mem.readInt(u64, b[0..8], .little), + .seq = std.mem.readInt(u64, b[8..16], .little), + .alloc_tail = @intCast(std.mem.readInt(u64, b[16..24], .little)), + .file_pages = @intCast(std.mem.readInt(u64, b[24..32], .little)), + .catalog_page = @intCast(std.mem.readInt(u64, b[32..40], .little)), + .catalog_len = std.mem.readInt(u64, b[40..48], .little), + .freelist_page = @intCast(std.mem.readInt(u64, b[48..56], .little)), + .freelist_len = std.mem.readInt(u64, b[56..64], .little), + .prev_generation = std.mem.readInt(u64, b[64..72], .little), + .live_docs = std.mem.readInt(u64, b[72..80], .little), + .dead_bytes = std.mem.readInt(u64, b[80..88], .little), + }; + // A slot whose hash matches but whose contents are impossible is worse + // than a torn one, because it would be trusted. + // + // The generation check is belt-and-braces: an all-zero slot is already + // rejected by the hash, so no test reddens when it is removed. Kept + // because it costs nothing and it is the field every comparison below + // depends on. The alloc_tail check is *not* redundant -- a slot can be + // rehashed to validate with nonsense in it, which is what the + // "hash-valid but impossible" test does. + if (wm.generation == 0) return null; + if (wm.alloc_tail < page_first_data) return null; + return wm; + } + + /// Publish `wm` as the new durable image. + /// + /// The ordering here is the entire crash-recovery argument, so it is worth + /// stating plainly: every page the watermark describes is made durable + /// *before* the watermark that describes them. Reverse the two and a crash + /// in between leaves a watermark pointing at pages that were never written, + /// with the log already truncated below it -- unrecoverable. + /// + /// The slot alternates by generation parity, so the previous watermark + /// survives this write and a torn slot leaves the older one intact. + pub fn publish(self: *Pager, wm_in: Watermark) !void { + var wm = wm_in; + wm.generation = self.generation + 1; + wm.prev_generation = self.generation; + + // 1. every data page durable. + try self.sync(); + + // 2. the free list, into fresh pages so the old one stays intact. + const fl = try self.write_freelist(); + wm.freelist_page = fl.first; + wm.freelist_len = fl.len; + try self.sync(); + + // 3. only now record the extent, *after* the free list has taken its + // pages. Capturing alloc_tail before this left those pages outside + // the published image, so the next open would hand them out again + // while the watermark still pointed at them. + wm.alloc_tail = self.alloc_tail; + wm.file_pages = self.file_pages; + + // 4. the watermark itself, last. + const slot = if (wm.generation & 1 == 1) page_watermark_a else page_watermark_b; + const b = self.page_mut(slot); + @memset(b, 0); + std.mem.writeInt(u64, b[0..8], wm.generation, .little); + std.mem.writeInt(u64, b[8..16], wm.seq, .little); + std.mem.writeInt(u64, b[16..24], wm.alloc_tail, .little); + std.mem.writeInt(u64, b[24..32], wm.file_pages, .little); + std.mem.writeInt(u64, b[32..40], wm.catalog_page, .little); + std.mem.writeInt(u64, b[40..48], wm.catalog_len, .little); + std.mem.writeInt(u64, b[48..56], wm.freelist_page, .little); + std.mem.writeInt(u64, b[56..64], wm.freelist_len, .little); + std.mem.writeInt(u64, b[64..72], wm.prev_generation, .little); + std.mem.writeInt(u64, b[72..80], wm.live_docs, .little); + std.mem.writeInt(u64, b[80..88], wm.dead_bytes, .little); + std.mem.writeInt(u64, b[wm_hashed_len..][0..8], header_hash(b[0..wm_hashed_len]), .little); + + const off = @as(usize, slot) << page_shift; + const end = std.mem.alignForward(usize, off + page_size, map_align); + const start = std.mem.alignBackward(usize, off, map_align); + try std.posix.msync(@alignCast(self.reserve[start..end]), std.posix.MSF.SYNC); + try self.file.sync(self.io); + // Deliberately does *not* clear the dirty record. After a correct + // publish the only page left in it is this slot, which this msync just + // made durable -- so a test can discard everything else and assert the + // image still loads. Clearing it here would make that check vacuous. + + // 5. only now is the new image current: advance the stable mark and + // rotate the free lists by one generation. + self.generation = wm.generation; + self.loaded = wm; + self.stable_pages = self.alloc_tail; + try self.free_ready.appendSlice(self.gpa, self.free_hold.items); + self.free_hold.clearRetainingCapacity(); + try self.free_hold.appendSlice(self.gpa, self.free_pending.items); + self.free_pending.clearRetainingCapacity(); + } + + // -- free list ---------------------------------------------------------- + + /// Give back a run of pages. They become reusable two generations later -- + /// see the field comment on `free_ready`. + pub fn free_pages(self: *Pager, first: u32, pages: u32) !void { + if (pages == 0) return; + try self.free_pending.append(self.gpa, .{ .first = first, .pages = pages }); + } + + /// Pages available for immediate reuse. + pub fn free_ready_pages(self: *const Pager) u32 { + var n: u32 = 0; + for (self.free_ready.items) |e| n += e.pages; + return n; + } + + fn write_freelist(self: *Pager) !struct { first: u32, len: u64 } { + const count = self.free_ready.items.len + self.free_hold.items.len + self.free_pending.items.len; + const len: u64 = 8 + @as(u64, count) * 8 + 8; + const pages: u32 = @intCast((len + page_size - 1) / page_size); + const first = try self.alloc_pages(pages); + const buf = self.bytes_mut(@as(u64, first) << page_shift, @intCast(len)); + @memset(buf, 0); + std.mem.writeInt(u64, buf[0..8], count, .little); + var at: usize = 8; + for ([_][]const Extent{ + self.free_ready.items, + self.free_hold.items, + self.free_pending.items, + }) |list| { + for (list) |e| { + std.mem.writeInt(u32, buf[at..][0..4], e.first, .little); + std.mem.writeInt(u32, buf[at + 4 ..][0..4], e.pages, .little); + at += 8; + } + } + std.mem.writeInt(u64, buf[at..][0..8], header_hash(buf[0..at]), .little); + return .{ .first = first, .len = len }; + } + + /// Load the free list a watermark points at. A damaged one is dropped with a + /// warning rather than refused: losing it wastes space, refusing to open + /// loses the database. + fn read_freelist(self: *Pager, wm: Watermark) !void { + if (wm.freelist_len == 0 or wm.freelist_page == 0) return; + if (wm.freelist_page >= self.mapped_pages) return; + const buf = self.bytes(@as(u64, wm.freelist_page) << page_shift, @intCast(wm.freelist_len)); + const count = std.mem.readInt(u64, buf[0..8], .little); + const want_len = 8 + count * 8 + 8; + if (want_len != wm.freelist_len) return; + const at: usize = @intCast(8 + count * 8); + if (header_hash(buf[0..at]) != std.mem.readInt(u64, buf[at..][0..8], .little)) { + std.debug.print("multiforadb: WARNING: data file free list is corrupt; its pages stay in use\n", .{}); + return; + } + // Everything the previous generations freed is reusable now: this open + // *is* a new generation, and nothing from before it is referenced. + var i: usize = 0; + while (i < count) : (i += 1) { + const o = 8 + i * 8; + try self.free_ready.append(self.gpa, .{ + .first = std.mem.readInt(u32, buf[o..][0..4], .little), + .pages = std.mem.readInt(u32, buf[o + 4 ..][0..4], .little), + }); + } + } + // -- header ------------------------------------------------------------- fn write_header(self: *Pager) !void { @@ -592,6 +923,227 @@ test "a corrupt header is rejected rather than misread" { ); } +/// Reopen the same file, so a test can check what survived a publish. +fn reopen(io: std.Io, path: []const u8) !Pager { + return Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); +} + +test "a published watermark is what the next open loads" { + 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 p = try tp.pg().alloc_pages(4); + @memset(tp.pg().page_mut(p), 0xC3); + try tp.pg().publish(.{ .seq = 4242, .catalog_page = p, .catalog_len = 99, .live_docs = 7 }); + // After publish, not before: publish allocates the free list's own pages, + // and the watermark has to cover them. + const tail_at_publish = tp.pg().alloc_tail; + tp.close(); + + var again = try reopen(io, tp.path); + defer again.deinit(); + try testing.expect(!again.fresh); + try testing.expectEqual(@as(u64, 1), again.loaded.generation); + try testing.expectEqual(@as(u64, 4242), again.loaded.seq); + try testing.expectEqual(p, again.loaded.catalog_page); + try testing.expectEqual(@as(u64, 99), again.loaded.catalog_len); + try testing.expectEqual(@as(u64, 7), again.loaded.live_docs); + try testing.expectEqual(tail_at_publish, again.alloc_tail); + // Everything the image references is off limits to writes. + try testing.expectEqual(tail_at_publish, again.stable_pages); + try testing.expectEqual(@as(u8, 0xC3), again.page(p)[0]); +} + +test "slots alternate by generation, so a torn write leaves the older one" { + // Writing each new generation over the only copy of the previous one risks + // having *neither* valid after a crash. Alternating is what makes a torn + // slot survivable, so both halves are asserted: the parity, and that + // corrupting the newer slot falls back rather than failing. + // + // Mutation checks: always write slot A, and the fallback finds nothing; + // compare generations before validating the hash, and the torn slot wins. + 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(); + + // Three generations: A, B, A. + try tp.pg().publish(.{ .seq = 10 }); + try tp.pg().publish(.{ .seq = 20 }); + try tp.pg().publish(.{ .seq = 30 }); + try testing.expectEqual(@as(u64, 3), tp.pg().generation); + // Generation 3 is odd, so slot A. Generation 2 must still be in slot B. + const a_gen = std.mem.readInt(u64, tp.pg().page(page_watermark_a)[0..8], .little); + const b_gen = std.mem.readInt(u64, tp.pg().page(page_watermark_b)[0..8], .little); + try testing.expectEqual(@as(u64, 3), a_gen); + try testing.expectEqual(@as(u64, 2), b_gen); + tp.close(); + + // Tear the newer slot: the open must fall back to generation 2, not fail + // and not believe the garbage. + { + var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); + defer f.close(io); + var junk: [64]u8 = undefined; + @memset(&junk, 0xFF); + try f.writePositionalAll(io, &junk, page_size); + } + var again = try reopen(io, tp.path); + defer again.deinit(); + try testing.expectEqual(@as(u64, 2), again.loaded.generation); + try testing.expectEqual(@as(u64, 20), again.loaded.seq); +} + +test "both slots unreadable opens anyway, with no checkpoint" { + // Ground rule 4: the database must always open. Once the log is truncated + // this costs data, which is why it warns -- but refusing to start would + // cost all of it. + // + // Mutation check: return an error from load_watermark instead and this + // fails outright. + 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 tp.pg().publish(.{ .seq = 5 }); + try tp.pg().publish(.{ .seq = 6 }); + tp.close(); + + { + var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); + defer f.close(io); + var junk: [page_size * 2]u8 = undefined; + @memset(&junk, 0xA5); + try f.writePositionalAll(io, &junk, page_size); + } + var again = try reopen(io, tp.path); + defer again.deinit(); + try testing.expectEqual(@as(u64, 0), again.loaded.generation); + try testing.expectEqual(@as(u64, 0), again.loaded.seq); +} + +test "a hash-valid but impossible slot is refused" { + // A slot that validates is *trusted*, so it has to be checked for sense as + // well as integrity: a generation of 0 or an alloc_tail below the first data + // page would be believed and would describe a file that does not exist. + 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 tp.pg().publish(.{ .seq = 11 }); + tp.close(); + + { + var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); + defer f.close(io); + var b: [page_size]u8 = undefined; + _ = try f.readPositionalAll(io, &b, page_size); // slot A holds gen 1 + // alloc_tail = 1, below page_first_data, and re-hash so it validates. + std.mem.writeInt(u64, b[16..24], 1, .little); + std.mem.writeInt(u64, b[wm_hashed_len..][0..8], header_hash(b[0..wm_hashed_len]), .little); + try f.writePositionalAll(io, &b, page_size); + } + var again = try reopen(io, tp.path); + defer again.deinit(); + // Refused, so no checkpoint rather than a nonsensical one. + try testing.expectEqual(@as(u64, 0), again.loaded.generation); +} + +test "freed pages are withheld for two generations and survive a reopen" { + // The delay is what keeps generation N-1 usable as a fallback image: its + // pages must still be allocated while N is current. + // + // Mutation check: move free_pending straight into free_ready at publish and + // the withholding assertions go red. + 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 victim = try tp.pg().alloc_pages(2); + @memset(tp.pg().page_mut(victim), 0x11); + try tp.pg().publish(.{ .seq = 1 }); + + try tp.pg().free_pages(victim, 2); + try testing.expectEqual(@as(u32, 0), tp.pg().free_ready_pages()); + + try tp.pg().publish(.{ .seq = 2 }); // pending -> hold + try testing.expectEqual(@as(u32, 0), tp.pg().free_ready_pages()); + // Still intact, which is the point: generation 1 references it. + try testing.expectEqual(@as(u8, 0x11), tp.pg().page(victim)[0]); + + try tp.pg().publish(.{ .seq = 3 }); // hold -> ready + try testing.expectEqual(@as(u32, 2), tp.pg().free_ready_pages()); + tp.close(); + + // And the list itself is durable. + var again = try reopen(io, tp.path); + defer again.deinit(); + try testing.expectEqual(@as(u32, 2), again.free_ready_pages()); +} + +test "the watermark is never published before the pages it describes" { + // The load-bearing ordering of the whole design: every page a watermark + // describes is durable before the watermark that describes it. Reverse them + // and a crash in between leaves a watermark pointing at pages that were + // never written, with the log already truncated below it. + // + // A kill -9 cannot produce that state -- the page cache survives the + // process -- so this simulates the writeback that never happened, using the + // pager's own record of which pages were dirtied since the last sync. That + // is the only honest way to make the ordering mutation-checkable, and + // without it the mutation "delete the sync before the watermark" leaves + // every other test in this file green. + // + // Mutation checks, stated precisely because the obvious one is not a + // violation: publish syncs twice before the watermark (data pages, then the + // free list), so removing *either* call alone is harmless and leaves this + // green -- correctly. What it catches is removing both, or moving them after + // the watermark write. The invariant is "at least one full sync between the + // last data write and the watermark", not any particular call. + const gpa = testing.allocator; + 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(); + + // A page the checkpoint will reference, written and then published. + const cat = try tp.pg().alloc_pages(2); + @memset(tp.pg().page_mut(cat), 0x77); + @memset(tp.pg().page_mut(cat + 1), 0x77); + try tp.pg().publish(.{ .seq = 900, .catalog_page = cat, .catalog_len = 2 * page_size }); + + // After a correct publish the only page still pending is the watermark slot + // itself, which its own msync made durable. Everything else the watermark + // references was flushed first, so discarding whatever is still pending -- + // minus that slot -- must leave a loadable image. + const pending = try tp.pg().dirty_pages(gpa); + defer gpa.free(pending); + const slot: u32 = if (tp.pg().generation & 1 == 1) page_watermark_a else page_watermark_b; + var lost: std.ArrayListUnmanaged(u32) = .empty; + defer lost.deinit(gpa); + for (pending) |pg_no| if (pg_no != slot) try lost.append(gpa, pg_no); + tp.close(); + try Pager.simulate_lost_writeback(gpa, io, tp.path, lost.items); + + var again = try reopen(io, tp.path); + defer again.deinit(); + try testing.expectEqual(@as(u64, 1), again.loaded.generation); + try testing.expectEqual(@as(u64, 900), again.loaded.seq); + try testing.expectEqual(cat, again.loaded.catalog_page); + // The referenced page survived, because publish synced it first. + for (again.page(cat)) |byte| try testing.expectEqual(@as(u8, 0x77), byte); + for (again.page(cat + 1)) |byte| try testing.expectEqual(@as(u8, 0x77), byte); +} + 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.