lib/sql/src/wal.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

   1 const std = @import("std");
   2 const alloc_phase = @import("alloc_phase");
   3 const page = @import("page.zig");
   4 const trace = @import("trace.zig");
   5 
   6 const Allocator = std.mem.Allocator;
   7 
   8 pub const header_size: usize = 32;
   9 pub const frame_header_size: usize = 24;
  10 pub const frame_size: usize = frame_header_size + page.size;
  11 const controlled_copy_chunk_bytes: usize = 1 * 1024 * 1024;
  12 
  13 pub const ReadError = error{
  14     InvalidChecksum,
  15     InvalidWal,
  16     UnsupportedPageSize,
  17 };
  18 
  19 pub const ControlledReadError = ReadError || error{Interrupted};
  20 
  21 pub const Error = Allocator.Error || ReadError || error{
  22     CapacityOverflow,
  23     WalFull,
  24 };
  25 
  26 pub const Salt = struct {
  27     first: u32,
  28     second: u32,
  29 };
  30 
  31 pub const Header = struct {
  32     sequence: u32,
  33     salt: Salt,
  34 };
  35 
  36 pub const Frame = struct {
  37     index: usize,
  38     page_id: u32,
  39     db_page_count: u32,
  40     image: []const u8,
  41 
  42     pub fn committed(self: Frame) bool {
  43         return self.db_page_count != 0;
  44     }
  45 };
  46 
  47 pub const FrameRef = extern struct {
  48     page_id: u32,
  49     frame: u32,
  50     checksum_before: [2]u32,
  51     checksum_after: [2]u32,
  52 };
  53 
  54 comptime {
  55     std.debug.assert(@sizeOf(FrameRef) == 24);
  56 }
  57 
  58 pub const Control = struct {
  59     context: ?*anyopaque = null,
  60     interrupted_fn: *const fn (context: ?*anyopaque) bool = neverInterrupted,
  61 
  62     pub fn check(self: Control) error{Interrupted}!void {
  63         if (self.interrupted_fn(self.context)) return error.Interrupted;
  64     }
  65 
  66     fn neverInterrupted(_: ?*anyopaque) bool {
  67         return false;
  68     }
  69 };
  70 
  71 pub const Extent = union(enum) {
  72     partial_header: u8,
  73     admitted: Admitted,
  74     over_capacity: OverCapacity,
  75 
  76     pub const Admitted = struct {
  77         complete_frames: u32,
  78         partial_tail_bytes: u16,
  79     };
  80 
  81     pub const OverCapacity = struct {
  82         complete_frames: u64,
  83         partial_tail_bytes: u16,
  84     };
  85 
  86     pub fn classify(
  87         file_bytes: u64,
  88         frame_capacity: usize,
  89         control: Control,
  90     ) error{ CapacityOverflow, Interrupted }!Extent {
  91         try control.check();
  92         if (frame_capacity > std.math.maxInt(u32)) return error.CapacityOverflow;
  93         if (file_bytes < header_size) {
  94             return .{ .partial_header = @intCast(file_bytes) };
  95         }
  96         const body_bytes = file_bytes - header_size;
  97         const complete_frames = body_bytes / frame_size;
  98         const partial_tail_bytes: u16 = @intCast(body_bytes % frame_size);
  99         const capacity: u64 = @intCast(frame_capacity);
 100         if (complete_frames > capacity or
 101             (complete_frames == capacity and partial_tail_bytes > 0))
 102         {
 103             return .{ .over_capacity = .{
 104                 .complete_frames = complete_frames,
 105                 .partial_tail_bytes = partial_tail_bytes,
 106             } };
 107         }
 108         return .{ .admitted = .{
 109             .complete_frames = @intCast(complete_frames),
 110             .partial_tail_bytes = partial_tail_bytes,
 111         } };
 112     }
 113 };
 114 
 115 pub const Scanner = struct {
 116     pub const Workspace = struct {
 117         scratch: *[frame_size]u8,
 118         refs: []FrameRef,
 119     };
 120 
 121     pub const Capacity = struct {
 122         frames: u32,
 123         storage_bytes: usize,
 124 
 125         pub fn derive(frame_capacity: usize) error{CapacityOverflow}!Capacity {
 126             if (frame_capacity > std.math.maxInt(u32)) return error.CapacityOverflow;
 127             return .{
 128                 .frames = @intCast(frame_capacity),
 129                 .storage_bytes = try storageBytes(
 130                     frame_capacity,
 131                     frame_size,
 132                     @sizeOf(FrameRef),
 133                 ),
 134             };
 135         }
 136     };
 137 
 138     pub const State = enum {
 139         uninitialized,
 140         scanning,
 141         ready,
 142         invalidated,
 143         teardown,
 144     };
 145 
 146     pub const MalformedFrame = enum {
 147         invalid_page_id,
 148         salt_mismatch,
 149         checksum_mismatch,
 150         database_page_count_regressed,
 151         page_outside_database,
 152     };
 153 
 154     pub const FrameOutcome = union(enum) {
 155         staged,
 156         committed,
 157         malformed_full_frame: MalformedFrame,
 158     };
 159 
 160     pub const Outcome = union(enum) {
 161         complete,
 162         partial_uncommitted_tail: u16,
 163     };
 164 
 165     pub const Progress = struct {
 166         state: State,
 167         header: [header_size]u8,
 168         tail_checksum: [2]u32,
 169         target_frames: u32,
 170         covered_frames: u32,
 171         committed_end_mark: u32,
 172         logical_database_page_count: u32,
 173         committed_ref_count: u32,
 174         staged_transaction_start: u32,
 175         staged_ref_count: u32,
 176         partial_tail_bytes: u16,
 177     };
 178 
 179     const ValidatedFrame = struct {
 180         ref: FrameRef,
 181         checksum: Checksum,
 182         database_page_count: u32,
 183     };
 184 
 185     const Validation = union(enum) {
 186         valid: ValidatedFrame,
 187         malformed: MalformedFrame,
 188     };
 189 
 190     workspace: Workspace,
 191     capacity: Capacity,
 192     state: State = .uninitialized,
 193     header: [header_size]u8 = @splat(0),
 194     salt: Salt = .{ .first = 0, .second = 0 },
 195     tail_checksum: Checksum = .{},
 196     target_frames: u32 = 0,
 197     covered_frames: u32 = 0,
 198     committed_end_mark: u32 = 0,
 199     logical_database_page_count: u32 = 0,
 200     committed_ref_count: u32 = 0,
 201     staged_transaction_start: u32 = 0,
 202     staged_ref_count: u32 = 0,
 203     partial_tail_bytes: u16 = 0,
 204 
 205     pub fn init(workspace: Workspace) error{CapacityOverflow}!Scanner {
 206         return .{
 207             .workspace = workspace,
 208             .capacity = try Capacity.derive(workspace.refs.len),
 209         };
 210     }
 211 
 212     pub fn deinit(self: *Scanner) Workspace {
 213         std.debug.assert(self.state != .teardown);
 214         const workspace = self.workspace;
 215         self.state = .teardown;
 216         self.workspace.refs = &.{};
 217         return workspace;
 218     }
 219 
 220     pub fn reset(
 221         self: *Scanner,
 222         header: *const [header_size]u8,
 223         extent: Extent.Admitted,
 224         database_page_floor: u32,
 225         control: Control,
 226     ) (ReadError || error{ CapacityExceeded, Interrupted, InvalidExtent, ScannerNotReady })!void {
 227         if (self.state == .teardown) return error.ScannerNotReady;
 228         if (extent.partial_tail_bytes >= frame_size) return error.InvalidExtent;
 229         if (extent.complete_frames > self.capacity.frames) {
 230             self.state = .invalidated;
 231             return error.CapacityExceeded;
 232         }
 233         const decoded = decodeHeaderControlled(header, control) catch |err| switch (err) {
 234             error.Interrupted => return error.Interrupted,
 235             else => {
 236                 self.state = .invalidated;
 237                 return err;
 238             },
 239         };
 240         try control.check();
 241         self.header = header.*;
 242         self.salt = decoded.salt;
 243         self.tail_checksum = decoded.checksum;
 244         self.target_frames = extent.complete_frames;
 245         self.covered_frames = 0;
 246         self.committed_end_mark = 0;
 247         self.logical_database_page_count = database_page_floor;
 248         self.committed_ref_count = 0;
 249         self.staged_transaction_start = 0;
 250         self.staged_ref_count = 0;
 251         self.partial_tail_bytes = extent.partial_tail_bytes;
 252         self.state = if (extent.complete_frames == 0) .ready else .scanning;
 253         self.assertBounds();
 254     }
 255 
 256     pub fn extend(
 257         self: *Scanner,
 258         extent: Extent.Admitted,
 259         control: Control,
 260     ) error{ CapacityExceeded, Interrupted, InvalidExtent, RewindRequired, ScannerNotReady }!void {
 261         if (self.state != .ready) return error.ScannerNotReady;
 262         if (extent.partial_tail_bytes >= frame_size) return error.InvalidExtent;
 263         try control.check();
 264         if (extent.complete_frames > self.capacity.frames) {
 265             self.state = .invalidated;
 266             return error.CapacityExceeded;
 267         }
 268         const rewound = extent.complete_frames < self.covered_frames or
 269             (extent.complete_frames == self.covered_frames and
 270                 extent.partial_tail_bytes < self.partial_tail_bytes);
 271         if (rewound) {
 272             self.state = .invalidated;
 273             return error.RewindRequired;
 274         }
 275         self.target_frames = extent.complete_frames;
 276         self.partial_tail_bytes = extent.partial_tail_bytes;
 277         if (self.covered_frames < self.target_frames) self.state = .scanning;
 278         self.assertBounds();
 279     }
 280 
 281     pub fn nextFrame(self: *Scanner) error{ScannerNotReady}!?FrameRequest {
 282         if (self.state != .scanning and self.state != .ready) {
 283             return error.ScannerNotReady;
 284         }
 285         self.assertBounds();
 286         if (self.state == .ready) return null;
 287         std.debug.assert(self.covered_frames < self.target_frames);
 288         const frame = self.covered_frames + 1;
 289         return .{
 290             .frame = frame,
 291             .offset = frameOffset(frame),
 292             .bytes = self.workspace.scratch,
 293         };
 294     }
 295 
 296     pub fn consumeFrame(
 297         self: *Scanner,
 298         control: Control,
 299     ) error{ Interrupted, ScannerNotReady }!FrameOutcome {
 300         if (self.state != .scanning) return error.ScannerNotReady;
 301         self.assertBounds();
 302         const validation = try self.validateFrame(control);
 303         const validated = switch (validation) {
 304             .valid => |value| value,
 305             .malformed => |reason| return self.invalidate(reason),
 306         };
 307         if (validated.database_page_count != 0) {
 308             if (try self.invalidMarker(validated, control)) |reason| {
 309                 return self.invalidate(reason);
 310             }
 311         }
 312         try control.check();
 313         self.appendValidated(validated);
 314         if (validated.database_page_count == 0) {
 315             self.finishPhysicalFrame();
 316             return .staged;
 317         }
 318         self.foldStaged(control) catch |err| {
 319             self.state = .invalidated;
 320             return err;
 321         };
 322         self.committed_end_mark = validated.ref.frame;
 323         self.logical_database_page_count = validated.database_page_count;
 324         self.staged_transaction_start = 0;
 325         self.staged_ref_count = 0;
 326         self.finishPhysicalFrame();
 327         return .committed;
 328     }
 329 
 330     pub fn outcome(self: *const Scanner) error{ScannerNotReady}!Outcome {
 331         if (self.state != .ready) return error.ScannerNotReady;
 332         self.assertBounds();
 333         if (self.partial_tail_bytes == 0) return .complete;
 334         return .{ .partial_uncommitted_tail = self.partial_tail_bytes };
 335     }
 336 
 337     pub fn progress(self: *const Scanner) Progress {
 338         return .{
 339             .state = self.state,
 340             .header = self.header,
 341             .tail_checksum = self.tail_checksum.pair(),
 342             .target_frames = self.target_frames,
 343             .covered_frames = self.covered_frames,
 344             .committed_end_mark = self.committed_end_mark,
 345             .logical_database_page_count = self.logical_database_page_count,
 346             .committed_ref_count = self.committed_ref_count,
 347             .staged_transaction_start = self.staged_transaction_start,
 348             .staged_ref_count = self.staged_ref_count,
 349             .partial_tail_bytes = self.partial_tail_bytes,
 350         };
 351     }
 352 
 353     pub fn findCommitted(
 354         self: *const Scanner,
 355         page_id: u32,
 356         control: Control,
 357     ) error{ Interrupted, ScannerNotReady }!?FrameRef {
 358         if (self.state != .ready) return error.ScannerNotReady;
 359         self.assertBounds();
 360         try control.check();
 361         if (page_id == 0 or page_id > self.logical_database_page_count) return null;
 362         var left: u32 = 0;
 363         var right = self.committed_ref_count;
 364         while (left < right) {
 365             try control.check();
 366             const middle = left + (right - left) / 2;
 367             const ref = self.workspace.refs[middle];
 368             if (ref.page_id < page_id) {
 369                 left = middle + 1;
 370             } else {
 371                 right = middle;
 372             }
 373         }
 374         if (left == self.committed_ref_count) return null;
 375         const ref = self.workspace.refs[left];
 376         return if (ref.page_id == page_id) ref else null;
 377     }
 378 
 379     pub fn selectedFrame(
 380         self: *Scanner,
 381         ref: FrameRef,
 382         control: Control,
 383     ) error{ Interrupted, ScannerNotReady, StaleFrameRef }!FrameRequest {
 384         const current = (try self.findCommitted(ref.page_id, control)) orelse
 385             return error.StaleFrameRef;
 386         if (!std.meta.eql(current, ref)) return error.StaleFrameRef;
 387         return .{
 388             .frame = ref.frame,
 389             .offset = frameOffset(ref.frame),
 390             .bytes = self.workspace.scratch,
 391         };
 392     }
 393 
 394     pub fn verifySelected(
 395         self: *const Scanner,
 396         ref: FrameRef,
 397         destination: *[page.size]u8,
 398         control: Control,
 399     ) error{ Interrupted, ScannerNotReady, SelectedFrameMismatch, StaleFrameRef }!void {
 400         if (self.state != .ready) return error.ScannerNotReady;
 401         std.debug.assert(disjoint(self.workspace.scratch, destination));
 402         const current = (try self.findCommitted(ref.page_id, control)) orelse
 403             return error.StaleFrameRef;
 404         if (!std.meta.eql(current, ref)) return error.StaleFrameRef;
 405         const frame = self.workspace.scratch;
 406         try control.check();
 407         if (readU32(frame[0..4]) != ref.page_id) return error.SelectedFrameMismatch;
 408         if (readU32(frame[8..12]) != self.salt.first) return error.SelectedFrameMismatch;
 409         if (readU32(frame[12..16]) != self.salt.second) return error.SelectedFrameMismatch;
 410         var checksum = Checksum.fromPair(ref.checksum_before);
 411         try checksum.updateControlled(frame[0..8], control);
 412         try checksum.updateControlled(frame[frame_header_size..], control);
 413         const on_disk = [2]u32{
 414             readU32(frame[16..20]),
 415             readU32(frame[20..24]),
 416         };
 417         if (!std.mem.eql(u32, &on_disk, &ref.checksum_after)) {
 418             return error.SelectedFrameMismatch;
 419         }
 420         const computed = checksum.pair();
 421         if (!std.mem.eql(u32, &on_disk, &computed)) {
 422             return error.SelectedFrameMismatch;
 423         }
 424         try control.check();
 425         @memcpy(destination, frame[frame_header_size..]);
 426     }
 427 
 428     fn validateFrame(
 429         self: *const Scanner,
 430         control: Control,
 431     ) error{Interrupted}!Validation {
 432         try control.check();
 433         const frame = self.workspace.scratch;
 434         const page_id = readU32(frame[0..4]);
 435         const database_page_count = readU32(frame[4..8]);
 436         if (page_id == 0) return .{ .malformed = .invalid_page_id };
 437         if (readU32(frame[8..12]) != self.salt.first or
 438             readU32(frame[12..16]) != self.salt.second)
 439         {
 440             return .{ .malformed = .salt_mismatch };
 441         }
 442         var checksum = self.tail_checksum;
 443         try checksum.updateControlled(frame[0..8], control);
 444         try checksum.updateControlled(frame[frame_header_size..], control);
 445         if (readU32(frame[16..20]) != checksum.first or
 446             readU32(frame[20..24]) != checksum.second)
 447         {
 448             return .{ .malformed = .checksum_mismatch };
 449         }
 450         const frame_index = self.covered_frames + 1;
 451         return .{ .valid = .{
 452             .ref = .{
 453                 .page_id = page_id,
 454                 .frame = frame_index,
 455                 .checksum_before = self.tail_checksum.pair(),
 456                 .checksum_after = checksum.pair(),
 457             },
 458             .checksum = checksum,
 459             .database_page_count = database_page_count,
 460         } };
 461     }
 462 
 463     fn invalidMarker(
 464         self: *const Scanner,
 465         validated: ValidatedFrame,
 466         control: Control,
 467     ) error{Interrupted}!?MalformedFrame {
 468         const page_count = validated.database_page_count;
 469         if (page_count < self.logical_database_page_count) {
 470             return .database_page_count_regressed;
 471         }
 472         if (validated.ref.page_id > page_count) return .page_outside_database;
 473         var index: u32 = 0;
 474         while (index < self.staged_ref_count) : (index += 1) {
 475             try control.check();
 476             const staged = self.workspace.refs[self.committed_ref_count + index];
 477             if (staged.page_id > page_count) return .page_outside_database;
 478         }
 479         return null;
 480     }
 481 
 482     fn appendValidated(self: *Scanner, validated: ValidatedFrame) void {
 483         const index = self.committed_ref_count + self.staged_ref_count;
 484         std.debug.assert(index < self.capacity.frames);
 485         std.debug.assert(validated.ref.frame == self.covered_frames + 1);
 486         self.workspace.refs[index] = validated.ref;
 487         self.tail_checksum = validated.checksum;
 488         self.covered_frames += 1;
 489         if (self.staged_ref_count == 0) {
 490             self.staged_transaction_start = validated.ref.frame;
 491         }
 492         self.staged_ref_count += 1;
 493         std.debug.assert(self.covered_frames <= self.target_frames);
 494         std.debug.assert(
 495             @as(u64, self.committed_ref_count) + self.staged_ref_count <=
 496                 self.capacity.frames,
 497         );
 498         std.debug.assert(
 499             @as(u64, self.committed_ref_count) + self.staged_ref_count <=
 500                 self.covered_frames,
 501         );
 502     }
 503 
 504     fn foldStaged(self: *Scanner, control: Control) error{Interrupted}!void {
 505         var committed = self.committed_ref_count;
 506         var staged = self.staged_ref_count;
 507         while (staged > 0) {
 508             const candidate = self.workspace.refs[committed];
 509             var insertion: u32 = 0;
 510             while (insertion < committed and
 511                 self.workspace.refs[insertion].page_id < candidate.page_id)
 512             {
 513                 try self.checkFold(control);
 514                 insertion += 1;
 515             }
 516             const replaces = insertion < committed and
 517                 self.workspace.refs[insertion].page_id == candidate.page_id;
 518             if (replaces) {
 519                 try self.replaceCommitted(insertion, committed, staged, candidate, control);
 520                 staged -= 1;
 521             } else {
 522                 try self.insertCommitted(insertion, committed, candidate, control);
 523                 committed += 1;
 524                 staged -= 1;
 525             }
 526             try self.checkFold(control);
 527         }
 528         self.committed_ref_count = committed;
 529     }
 530 
 531     fn replaceCommitted(
 532         self: *Scanner,
 533         insertion: u32,
 534         committed: u32,
 535         staged: u32,
 536         candidate: FrameRef,
 537         control: Control,
 538     ) error{Interrupted}!void {
 539         try self.checkFold(control);
 540         self.workspace.refs[insertion] = candidate;
 541         var index = committed;
 542         while (index + 1 < committed + staged) : (index += 1) {
 543             try self.checkFold(control);
 544             self.workspace.refs[index] = self.workspace.refs[index + 1];
 545         }
 546     }
 547 
 548     fn insertCommitted(
 549         self: *Scanner,
 550         insertion: u32,
 551         committed: u32,
 552         candidate: FrameRef,
 553         control: Control,
 554     ) error{Interrupted}!void {
 555         var index = committed;
 556         while (index > insertion) {
 557             try self.checkFold(control);
 558             self.workspace.refs[index] = self.workspace.refs[index - 1];
 559             index -= 1;
 560         }
 561         try self.checkFold(control);
 562         self.workspace.refs[insertion] = candidate;
 563     }
 564 
 565     fn checkFold(self: *Scanner, control: Control) error{Interrupted}!void {
 566         control.check() catch {
 567             self.state = .invalidated;
 568             return error.Interrupted;
 569         };
 570     }
 571 
 572     fn finishPhysicalFrame(self: *Scanner) void {
 573         std.debug.assert(self.covered_frames <= self.target_frames);
 574         if (self.covered_frames == self.target_frames) self.state = .ready;
 575         self.assertBounds();
 576     }
 577 
 578     fn invalidate(self: *Scanner, reason: MalformedFrame) FrameOutcome {
 579         self.state = .invalidated;
 580         return .{ .malformed_full_frame = reason };
 581     }
 582 
 583     fn storageBytes(
 584         frame_capacity: usize,
 585         scratch_bytes: usize,
 586         ref_bytes: usize,
 587     ) error{CapacityOverflow}!usize {
 588         const refs_bytes = std.math.mul(
 589             usize,
 590             frame_capacity,
 591             ref_bytes,
 592         ) catch return error.CapacityOverflow;
 593         return std.math.add(
 594             usize,
 595             scratch_bytes,
 596             refs_bytes,
 597         ) catch return error.CapacityOverflow;
 598     }
 599 
 600     fn assertBounds(self: *const Scanner) void {
 601         std.debug.assert(self.target_frames <= self.capacity.frames);
 602         std.debug.assert(self.covered_frames <= self.target_frames);
 603         std.debug.assert(self.committed_end_mark <= self.covered_frames);
 604         std.debug.assert(self.committed_ref_count <= self.covered_frames);
 605         std.debug.assert(self.staged_ref_count <= self.covered_frames);
 606         std.debug.assert(
 607             @as(u64, self.committed_ref_count) + self.staged_ref_count <=
 608                 self.capacity.frames,
 609         );
 610         std.debug.assert(
 611             @as(u64, self.committed_ref_count) + self.staged_ref_count <=
 612                 self.covered_frames,
 613         );
 614         std.debug.assert((self.staged_ref_count == 0) ==
 615             (self.staged_transaction_start == 0));
 616         if (self.state == .ready) {
 617             std.debug.assert(self.covered_frames == self.target_frames);
 618         }
 619         if (self.state == .scanning) {
 620             std.debug.assert(self.covered_frames < self.target_frames);
 621         }
 622     }
 623 };
 624 
 625 pub const FrameRequest = struct {
 626     frame: u32,
 627     offset: u64,
 628     bytes: *[frame_size]u8,
 629 };
 630 
 631 pub fn frameOffset(frame: u32) u64 {
 632     std.debug.assert(frame > 0);
 633     return @as(u64, header_size) +
 634         @as(u64, frame - 1) * @as(u64, frame_size);
 635 }
 636 
 637 pub const Writer = struct {
 638     pub const storage_alignment: usize = @alignOf(u8);
 639     pub const Storage = []align(storage_alignment) u8;
 640 
 641     pub const Limits = struct {
 642         header: Header,
 643         frames: usize,
 644     };
 645 
 646     pub const Capacity = struct {
 647         frames: usize,
 648         storage_bytes: usize,
 649 
 650         pub const DeriveError = error{CapacityOverflow};
 651 
 652         pub fn derive(limits: Limits) DeriveError!Capacity {
 653             const frame_bytes = std.math.mul(usize, limits.frames, frame_size) catch {
 654                 return error.CapacityOverflow;
 655             };
 656             const storage_bytes = std.math.add(usize, header_size, frame_bytes) catch {
 657                 return error.CapacityOverflow;
 658             };
 659             return .{ .frames = limits.frames, .storage_bytes = storage_bytes };
 660         }
 661     };
 662 
 663     pub const Exhaustion = error{WalFull};
 664     pub const InitError = Capacity.DeriveError || error{StorageTooShort};
 665     pub const LoadError = ReadError || Exhaustion;
 666     pub const ControlledLoadError = LoadError || error{Interrupted};
 667     pub const work_limits: alloc_phase.capacity.WorkLimits = .{
 668         .transition_steps_max = 1,
 669         .cleanup_steps_per_call_max = 0,
 670         .cleanup_calls_at_capacity_max = 0,
 671     };
 672 
 673     pub const claim: alloc_phase.capacity.Declaration = .{
 674         .source = .{
 675             .id = "sql.wal_writer",
 676             .kind = .phase_static,
 677             .limit_source = .caller,
 678             .storage = .{
 679                 .covered = &.{
 680                     .{
 681                         .id = "exact_header_and_committed_frame_byte_region",
 682                         .lifetime = .steady,
 683                         .detail = "mutable borrow of the caller-provisioned exact header and frame byte region",
 684                     },
 685                     .{
 686                         .id = "invisible_staged_frame_tail_transferred_into_the_co_0ed8d0c7aca0",
 687                         .lifetime = .transferred,
 688                         .detail = "invisible staged frame tail transferred into the committed prefix",
 689                     },
 690                 },
 691                 .excluded = &.{
 692                     "caller-owned storage outside the exact borrowed prefix",
 693                     "caller-owned page images and recovery source bytes",
 694                     "pager frame indexes, page indexes, checkpoint plans, files, and I/O runtime state",
 695                     "trace instrumentation and allocator implementation state",
 696                 },
 697             },
 698             .capacity = .{
 699                 .inputs = &.{
 700                     alloc_phase.capacity.bindInput(Limits, "frames", "frames"),
 701                 },
 702                 .type_selectors = &.{
 703                     alloc_phase.capacity.bindType([frame_size]u8, "walframebytes"),
 704                     alloc_phase.capacity.bindType([header_size]u8, "walheaderbytes"),
 705                 },
 706                 .nodes = &.{
 707                     .{ .input = 0 },
 708                     .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } },
 709                     .{ .constant = 1 },
 710                     .{ .scale = .{ .node = 2, .coefficient = .{ .size_of_concrete_type = 1 } } },
 711                     .{ .add = .{ .left = 1, .right = 3 } },
 712                 },
 713                 .assertions = &.{.{
 714                     .scope = .closure_total,
 715                     .measure = .retained,
 716                     .relation = .exact,
 717                     .expression = 4,
 718                 }},
 719             },
 720             .overload = .{
 721                 .kind = .reject_before_mutation,
 722                 .detail = "append staging and recovery load return WalFull before bytes length salt or checksum mutate",
 723             },
 724             .risks = .{
 725                 .transitive = .{
 726                     .status = .open,
 727                     .detail = "pager indexes and trace instrumentation are outside the writer-owned byte region",
 728                 },
 729                 .foreign = .{
 730                     .status = .open,
 731                     .detail = "file reads and writes that consume WAL bytes are owned by the file database",
 732                 },
 733             },
 734             .work = .{ .equation = "transition_steps <= transition_steps_max" },
 735             .obligations = &.{
 736                 .{ .key = "sql_wal_capacity", .role = .capacity_model },
 737                 .{ .key = "sql_wal_storage_rejection", .role = .initialization_failure },
 738                 .{ .key = "sql_wal_sealed", .role = .overload },
 739                 .{ .key = "sql_wal_work_bound", .role = .work_bound },
 740                 .{ .key = "sql_wal_semantics", .role = .custom },
 741                 .{ .key = "sql_wal_staged_transfer", .role = .custom },
 742                 .{ .key = "sql_wal_corruption", .role = .custom },
 743             },
 744         },
 745         .bindings = .{
 746             .owner = @This(),
 747             .seal = .{
 748                 .family = alloc_phase.capacity.selector(@This().activate),
 749                 .premise = .{
 750                     .class = .checked_semantic_fact,
 751                     .authority = .checker,
 752                 },
 753             },
 754             .teardown = .{
 755                 .family = alloc_phase.capacity.selector(@This().deinit),
 756                 .premise = .{
 757                     .class = .checked_semantic_fact,
 758                     .authority = .checker,
 759                 },
 760             },
 761         },
 762     };
 763 
 764     phase: alloc_phase.capacity.Phase,
 765     capacity: Capacity,
 766     storage: Storage,
 767     len: usize,
 768     salt: Salt,
 769     checksum: Checksum,
 770 
 771     pub const Position = struct {
 772         len: usize,
 773         checksum_first: u32,
 774         checksum_second: u32,
 775     };
 776 
 777     pub fn init(storage: Storage, limits: Limits) InitError!Writer {
 778         const capacity = try Capacity.derive(limits);
 779         if (storage.len < capacity.storage_bytes) return error.StorageTooShort;
 780         const borrowed = storage[0..capacity.storage_bytes];
 781         const checksum = encodeHeader(borrowed[0..header_size], limits.header);
 782         return .{
 783             .phase = .initialization,
 784             .capacity = capacity,
 785             .storage = borrowed,
 786             .len = header_size,
 787             .salt = limits.header.salt,
 788             .checksum = checksum,
 789         };
 790     }
 791 
 792     pub fn activate(self: *Writer) void {
 793         std.debug.assert(self.phase == .initialization);
 794         std.debug.assert(self.storage.len == self.capacity.storage_bytes);
 795         std.debug.assert(self.len == header_size);
 796         self.phase = .steady;
 797     }
 798 
 799     pub fn deinit(self: *Writer) Storage {
 800         std.debug.assert(self.phase != .teardown);
 801         std.debug.assert(self.storage.len == self.capacity.storage_bytes);
 802         self.phase = .teardown;
 803         const storage = self.storage;
 804         self.* = undefined;
 805         return storage;
 806     }
 807 
 808     pub fn bytes(self: *const Writer) []const u8 {
 809         std.debug.assert(self.phase == .steady);
 810         std.debug.assert(self.len <= self.storage.len);
 811         return self.storage[0..self.len];
 812     }
 813 
 814     pub fn frameCount(self: *const Writer) usize {
 815         std.debug.assert(self.phase == .steady);
 816         return (self.len - header_size) / frame_size;
 817     }
 818 
 819     pub fn frameCapacity(self: *const Writer) usize {
 820         std.debug.assert(self.phase == .steady);
 821         return self.capacity.frames;
 822     }
 823 
 824     pub fn byteCapacity(self: *const Writer) usize {
 825         std.debug.assert(self.phase == .steady);
 826         return self.capacity.storage_bytes;
 827     }
 828 
 829     pub fn remainingFrames(self: *const Writer) usize {
 830         std.debug.assert(self.phase == .steady);
 831         return self.capacity.frames - self.frameCount();
 832     }
 833 
 834     pub fn position(self: *const Writer) Position {
 835         std.debug.assert(self.phase == .steady);
 836         return .{
 837             .len = self.len,
 838             .checksum_first = self.checksum.first,
 839             .checksum_second = self.checksum.second,
 840         };
 841     }
 842 
 843     pub fn restore(self: *Writer, position_value: Position) void {
 844         std.debug.assert(self.phase == .steady);
 845         std.debug.assert(position_value.len >= header_size);
 846         std.debug.assert((position_value.len - header_size) % frame_size == 0);
 847         std.debug.assert(position_value.len <= self.len);
 848         self.len = position_value.len;
 849         self.checksum = .{
 850             .first = position_value.checksum_first,
 851             .second = position_value.checksum_second,
 852         };
 853     }
 854 
 855     pub fn load(self: *Writer, source: []const u8, committed_len: usize) LoadError!void {
 856         return self.loadControlled(source, committed_len, .{}) catch |err| switch (err) {
 857             error.Interrupted => unreachable,
 858             else => return @errorCast(err),
 859         };
 860     }
 861 
 862     pub fn loadControlled(
 863         self: *Writer,
 864         source: []const u8,
 865         committed_len: usize,
 866         control: Control,
 867     ) ControlledLoadError!void {
 868         std.debug.assert(self.phase == .steady);
 869         if (committed_len > source.len) return error.InvalidWal;
 870         if (committed_len > self.capacity.storage_bytes) return error.WalFull;
 871         var reader = try Reader.initControlled(source[0..committed_len], control);
 872         const frames_max = (committed_len - header_size) / frame_size;
 873         var frame_index: usize = 0;
 874         while (frame_index < frames_max) : (frame_index += 1) {
 875             if (try reader.nextControlled(control) == null) break;
 876         }
 877         if (reader.offset != committed_len) return error.InvalidWal;
 878         var copied: usize = 0;
 879         var chunks: usize = 0;
 880         const chunks_max = std.math.divCeil(
 881             usize,
 882             committed_len,
 883             controlled_copy_chunk_bytes,
 884         ) catch unreachable;
 885         while (copied < committed_len) : (chunks += 1) {
 886             std.debug.assert(chunks < chunks_max);
 887             try control.check();
 888             const end = copied + @min(
 889                 controlled_copy_chunk_bytes,
 890                 committed_len - copied,
 891             );
 892             std.mem.copyForwards(u8, self.storage[copied..end], source[copied..end]);
 893             copied = end;
 894         }
 895         try control.check();
 896         self.len = committed_len;
 897         self.salt = reader.salt;
 898         self.checksum = reader.checksum;
 899     }
 900 
 901     pub fn append(self: *Writer, page_id: u32, db_page_count: u32, image: *const [page.size]u8) Exhaustion!void {
 902         const phase = trace.scope("wal.append");
 903         defer phase.end();
 904         std.debug.assert(self.phase == .steady);
 905         if (self.remainingFrames() == 0) return error.WalFull;
 906         var frame_header = @as([frame_header_size]u8, @splat(0));
 907         writeU32(frame_header[0..4], page_id);
 908         writeU32(frame_header[4..8], db_page_count);
 909         writeU32(frame_header[8..12], self.salt.first);
 910         writeU32(frame_header[12..16], self.salt.second);
 911 
 912         var checksum = self.checksum;
 913         checksum.update(frame_header[0..8]);
 914         checksum.update(image[0..]);
 915         writeU32(frame_header[16..20], checksum.first);
 916         writeU32(frame_header[20..24], checksum.second);
 917 
 918         @memcpy(self.storage[self.len..][0..frame_header_size], frame_header[0..]);
 919         @memcpy(self.storage[self.len + frame_header_size ..][0..page.size], image[0..]);
 920         self.len += frame_size;
 921         self.checksum = checksum;
 922         trace.progress("wal.append.complete");
 923     }
 924 
 925     pub fn stagePage(self: *Writer, position_value: Position, index: usize, page_id: u32, image: *const [page.size]u8) Exhaustion!void {
 926         std.debug.assert(self.phase == .steady);
 927         std.debug.assert(position_value.len == self.len);
 928         const offset = try self.stagedFrameOffset(position_value, index);
 929         writeU32(self.storage[offset..][0..4], page_id);
 930         const staged = self.storage[offset + frame_header_size ..][0..page.size];
 931         if (staged != image) @memcpy(staged, image);
 932     }
 933 
 934     pub fn stagedPage(self: *const Writer, position_value: Position, index: usize) []const u8 {
 935         std.debug.assert(self.phase == .steady);
 936         std.debug.assert(position_value.len == self.len);
 937         const offset = self.stagedFrameOffset(position_value, index) catch unreachable;
 938         return self.storage[offset + frame_header_size ..][0..page.size];
 939     }
 940 
 941     /// Returns a staged image for editing in place. Staging the returned
 942     /// image again copies nothing, and the checksum covers the edits
 943     /// when the frame commits.
 944     pub fn stagedPageMut(self: *Writer, position_value: Position, index: usize) *[page.size]u8 {
 945         std.debug.assert(self.phase == .steady);
 946         std.debug.assert(position_value.len == self.len);
 947         const offset = self.stagedFrameOffset(position_value, index) catch unreachable;
 948         return self.storage[offset + frame_header_size ..][0..page.size];
 949     }
 950 
 951     pub fn stagedPageId(self: *const Writer, position_value: Position, index: usize) u32 {
 952         std.debug.assert(self.phase == .steady);
 953         const offset = self.stagedFrameOffset(position_value, index) catch unreachable;
 954         return readU32(self.storage[offset..][0..4]);
 955     }
 956 
 957     pub fn swapStagedFrames(self: *Writer, position_value: Position, left_index: usize, right_index: usize) void {
 958         std.debug.assert(self.phase == .steady);
 959         std.debug.assert(position_value.len == self.len);
 960         if (left_index == right_index) return;
 961         const left_offset = self.stagedFrameOffset(position_value, left_index) catch unreachable;
 962         const right_offset = self.stagedFrameOffset(position_value, right_index) catch unreachable;
 963         const left: *[frame_size]u8 = self.storage[left_offset..][0..frame_size];
 964         const right: *[frame_size]u8 = self.storage[right_offset..][0..frame_size];
 965         const temporary = left.*;
 966         left.* = right.*;
 967         right.* = temporary;
 968     }
 969 
 970     pub fn commitStagedFrame(self: *Writer, position_value: Position, index: usize, db_page_count: u32) void {
 971         std.debug.assert(self.phase == .steady);
 972         const offset = self.stagedFrameOffset(position_value, index) catch unreachable;
 973         std.debug.assert(self.len == offset);
 974         const frame_header = self.storage[offset..][0..frame_header_size];
 975         const image = self.storage[offset + frame_header_size ..][0..page.size];
 976         std.debug.assert(readU32(frame_header[0..4]) != 0);
 977         writeU32(frame_header[4..8], db_page_count);
 978         writeU32(frame_header[8..12], self.salt.first);
 979         writeU32(frame_header[12..16], self.salt.second);
 980 
 981         var checksum = self.checksum;
 982         checksum.update(frame_header[0..8]);
 983         checksum.update(image);
 984         writeU32(frame_header[16..20], checksum.first);
 985         writeU32(frame_header[20..24], checksum.second);
 986 
 987         self.len += frame_size;
 988         self.checksum = checksum;
 989     }
 990 
 991     pub fn rewriteTail(self: *Writer, checkpoint_mark: usize, header: Header) void {
 992         std.debug.assert(self.phase == .steady);
 993         std.debug.assert(checkpoint_mark <= self.frameCount());
 994         const source = self.bytes();
 995         var reader = Reader.init(source) catch @panic("invalid wal writer storage");
 996         self.writeHeader(header);
 997         while (reader.next() catch @panic("invalid wal writer storage")) |frame| {
 998             if (frame.index <= checkpoint_mark) continue;
 999             const image = frame.image[0..page.size].*;
1000             self.append(frame.page_id, frame.db_page_count, &image) catch unreachable;
1001         }
1002         if (reader.offset != source.len) @panic("invalid wal writer storage");
1003     }
1004 
1005     fn writeHeader(self: *Writer, header: Header) void {
1006         const checksum = encodeHeader(self.storage[0..header_size], header);
1007         self.len = header_size;
1008         self.salt = header.salt;
1009         self.checksum = checksum;
1010     }
1011 
1012     fn stagedFrameOffset(self: *const Writer, position_value: Position, index: usize) Exhaustion!usize {
1013         std.debug.assert(position_value.len >= header_size);
1014         std.debug.assert((position_value.len - header_size) % frame_size == 0);
1015         const staged_bytes = std.math.mul(usize, index, frame_size) catch return error.WalFull;
1016         const offset = std.math.add(usize, position_value.len, staged_bytes) catch return error.WalFull;
1017         if (offset > self.storage.len or frame_size > self.storage.len - offset) return error.WalFull;
1018         return offset;
1019     }
1020 };
1021 
1022 comptime {
1023     alloc_phase.capacity.requireProvisionedRejectingOwnerShape(Writer);
1024 }
1025 
1026 pub const Reader = struct {
1027     data: []const u8,
1028     salt: Salt,
1029     checksum: Checksum,
1030     offset: usize = header_size,
1031     index: usize = 0,
1032     stopped: bool = false,
1033 
1034     pub fn init(data: []const u8) ReadError!Reader {
1035         return initControlled(data, .{}) catch |err| switch (err) {
1036             error.Interrupted => unreachable,
1037             else => return @errorCast(err),
1038         };
1039     }
1040 
1041     pub fn initControlled(data: []const u8, control: Control) ControlledReadError!Reader {
1042         try control.check();
1043         if (data.len < header_size) return error.InvalidWal;
1044         if (readU32(data[0..4]) != magic) return error.InvalidWal;
1045         if (readU32(data[4..8]) != format_version) return error.InvalidWal;
1046         if (readU32(data[8..12]) != page.size) return error.UnsupportedPageSize;
1047         var checksum: Checksum = .{};
1048         checksum.update(data[0..24]);
1049         if (readU32(data[24..28]) != checksum.first) return error.InvalidChecksum;
1050         if (readU32(data[28..32]) != checksum.second) return error.InvalidChecksum;
1051         try control.check();
1052         return .{
1053             .data = data,
1054             .salt = .{
1055                 .first = readU32(data[16..20]),
1056                 .second = readU32(data[20..24]),
1057             },
1058             .checksum = checksum,
1059         };
1060     }
1061 
1062     pub fn next(self: *Reader) ReadError!?Frame {
1063         return self.nextControlled(.{}) catch |err| switch (err) {
1064             error.Interrupted => unreachable,
1065             else => return @errorCast(err),
1066         };
1067     }
1068 
1069     pub fn nextControlled(self: *Reader, control: Control) ControlledReadError!?Frame {
1070         const phase = trace.scope("wal.next");
1071         defer phase.end();
1072         try control.check();
1073         if (self.stopped) return null;
1074         if (self.offset + frame_size > self.data.len) {
1075             self.stopped = true;
1076             return null;
1077         }
1078 
1079         const frame_header = self.data[self.offset..][0..frame_header_size];
1080         const image = self.data[self.offset + frame_header_size ..][0..page.size];
1081         if (readU32(frame_header[8..12]) != self.salt.first or readU32(frame_header[12..16]) != self.salt.second) {
1082             self.stopped = true;
1083             return null;
1084         }
1085 
1086         var checksum = self.checksum;
1087         checksum.update(frame_header[0..8]);
1088         checksum.update(image);
1089         if (readU32(frame_header[16..20]) != checksum.first or readU32(frame_header[20..24]) != checksum.second) {
1090             self.stopped = true;
1091             return null;
1092         }
1093 
1094         try control.check();
1095         self.checksum = checksum;
1096         self.index += 1;
1097         self.offset += frame_size;
1098         return .{
1099             .index = self.index,
1100             .page_id = readU32(frame_header[0..4]),
1101             .db_page_count = readU32(frame_header[4..8]),
1102             .image = image,
1103         };
1104     }
1105 };
1106 
1107 pub fn endMark(data: []const u8) ReadError!usize {
1108     return endMarkControlled(data, .{}) catch |err| switch (err) {
1109         error.Interrupted => unreachable,
1110         else => return @errorCast(err),
1111     };
1112 }
1113 
1114 pub fn endMarkControlled(data: []const u8, control: Control) ControlledReadError!usize {
1115     const phase = trace.scope("wal.endmark");
1116     defer phase.end();
1117     var reader = try Reader.initControlled(data, control);
1118     var mark: usize = 0;
1119     const frames_max = (data.len - header_size) / frame_size;
1120     var frame_index: usize = 0;
1121     while (frame_index < frames_max) : (frame_index += 1) {
1122         const frame = (try reader.nextControlled(control)) orelse break;
1123         if (frame.committed()) mark = frame.index;
1124     }
1125     try control.check();
1126     return mark;
1127 }
1128 
1129 pub fn pageAt(data: []const u8, page_id: u32, max_frame: usize) ReadError!?[]const u8 {
1130     const phase = trace.scope("wal.page_at");
1131     defer phase.end();
1132     if (max_frame == 0) return null;
1133     var reader = try Reader.init(data);
1134     var latest: ?[]const u8 = null;
1135     while (try reader.next()) |frame| {
1136         if (frame.index > max_frame) break;
1137         if (frame.page_id == page_id) latest = frame.image;
1138     }
1139     return latest;
1140 }
1141 
1142 const magic: u32 = 0x74737177;
1143 const format_version: u32 = 1;
1144 
1145 const Checksum = struct {
1146     first: u32 = 0,
1147     second: u32 = 0,
1148 
1149     fn fromPair(words: [2]u32) Checksum {
1150         return .{ .first = words[0], .second = words[1] };
1151     }
1152 
1153     fn pair(self: Checksum) [2]u32 {
1154         return .{ self.first, self.second };
1155     }
1156 
1157     fn update(self: *Checksum, data: []const u8) void {
1158         std.debug.assert(data.len % 8 == 0);
1159         var index: usize = 0;
1160         while (index < data.len) : (index += 8) {
1161             self.updatePair(data[index..][0..8]);
1162         }
1163     }
1164 
1165     fn updateControlled(
1166         self: *Checksum,
1167         data: []const u8,
1168         control: Control,
1169     ) error{Interrupted}!void {
1170         std.debug.assert(data.len % 8 == 0);
1171         var index: usize = 0;
1172         while (index < data.len) : (index += 8) {
1173             try control.check();
1174             self.updatePair(data[index..][0..8]);
1175         }
1176     }
1177 
1178     fn updatePair(self: *Checksum, bytes: *const [8]u8) void {
1179         const left = readU32(bytes[0..4]);
1180         const right = readU32(bytes[4..8]);
1181         self.first +%= left +% self.second;
1182         self.second +%= right +% self.first;
1183     }
1184 };
1185 
1186 const DecodedHeader = struct {
1187     salt: Salt,
1188     checksum: Checksum,
1189 };
1190 
1191 fn decodeHeaderControlled(
1192     data: *const [header_size]u8,
1193     control: Control,
1194 ) (ReadError || error{Interrupted})!DecodedHeader {
1195     try control.check();
1196     if (readU32(data[0..4]) != magic) return error.InvalidWal;
1197     if (readU32(data[4..8]) != format_version) return error.InvalidWal;
1198     if (readU32(data[8..12]) != page.size) return error.UnsupportedPageSize;
1199     var checksum: Checksum = .{};
1200     try checksum.updateControlled(data[0..24], control);
1201     if (readU32(data[24..28]) != checksum.first) return error.InvalidChecksum;
1202     if (readU32(data[28..32]) != checksum.second) return error.InvalidChecksum;
1203     return .{
1204         .salt = .{
1205             .first = readU32(data[16..20]),
1206             .second = readU32(data[20..24]),
1207         },
1208         .checksum = checksum,
1209     };
1210 }
1211 
1212 fn disjoint(left: *const [frame_size]u8, right: *const [page.size]u8) bool {
1213     const left_start = @intFromPtr(left);
1214     const left_end = left_start + frame_size;
1215     const right_start = @intFromPtr(right);
1216     const right_end = right_start + page.size;
1217     return left_end <= right_start or right_end <= left_start;
1218 }
1219 
1220 fn readU32(bytes: []const u8) u32 {
1221     return std.mem.readInt(u32, bytes[0..4], .big);
1222 }
1223 
1224 fn writeU32(bytes: []u8, value: u32) void {
1225     std.mem.writeInt(u32, bytes[0..4], value, .big);
1226 }
1227 
1228 fn encodeHeader(bytes: *[header_size]u8, header: Header) Checksum {
1229     @memset(bytes, 0);
1230     writeU32(bytes[0..4], magic);
1231     writeU32(bytes[4..8], format_version);
1232     writeU32(bytes[8..12], page.size);
1233     writeU32(bytes[12..16], header.sequence);
1234     writeU32(bytes[16..20], header.salt.first);
1235     writeU32(bytes[20..24], header.salt.second);
1236     var checksum: Checksum = .{};
1237     checksum.update(bytes[0..24]);
1238     writeU32(bytes[24..28], checksum.first);
1239     writeU32(bytes[28..32], checksum.second);
1240     return checksum;
1241 }
1242 
1243 fn testingHeader() Header {
1244     return .{
1245         .sequence = 7,
1246         .salt = .{ .first = 0x1111_2222, .second = 0x3333_4444 },
1247     };
1248 }
1249 
1250 fn fillPage(bytes: *[page.size]u8, page_id: u8, value: u8) void {
1251     @memset(bytes, 0);
1252     bytes[0] = page_id;
1253     bytes[1] = value;
1254 }
1255 
1256 fn testingWriter(allocator: Allocator, frames: usize) !Writer {
1257     const limits: Writer.Limits = .{
1258         .header = testingHeader(),
1259         .frames = frames,
1260     };
1261     const capacity = try Writer.Capacity.derive(limits);
1262     const storage = try allocator.alloc(u8, capacity.storage_bytes);
1263     errdefer allocator.free(storage);
1264     var writer = try Writer.init(storage, limits);
1265     writer.activate();
1266     return writer;
1267 }
1268 
1269 fn testingScanner(
1270     scratch: *[frame_size]u8,
1271     refs: []FrameRef,
1272 ) !Scanner {
1273     return try Scanner.init(.{ .scratch = scratch, .refs = refs });
1274 }
1275 
1276 fn admittedExtent(data: []const u8, capacity: usize) !Extent.Admitted {
1277     return switch (try Extent.classify(data.len, capacity, .{})) {
1278         .admitted => |extent| extent,
1279         .partial_header, .over_capacity => error.TestUnexpectedResult,
1280     };
1281 }
1282 
1283 fn resetTestingScanner(
1284     scanner: *Scanner,
1285     data: []const u8,
1286     database_page_floor: u32,
1287 ) !void {
1288     const extent = try admittedExtent(data, scanner.workspace.refs.len);
1289     const header: *const [header_size]u8 = data[0..header_size];
1290     try scanner.reset(header, extent, database_page_floor, .{});
1291 }
1292 
1293 fn consumeTestingFrame(
1294     scanner: *Scanner,
1295     data: []const u8,
1296     control: Control,
1297 ) !Scanner.FrameOutcome {
1298     const request = (try scanner.nextFrame()) orelse return error.TestUnexpectedResult;
1299     const offset: usize = @intCast(request.offset);
1300     @memcpy(request.bytes, data[offset..][0..frame_size]);
1301     return try scanner.consumeFrame(control);
1302 }
1303 
1304 fn scanTestingData(
1305     scanner: *Scanner,
1306     data: []const u8,
1307     database_page_floor: u32,
1308 ) !Scanner.Outcome {
1309     try resetTestingScanner(scanner, data, database_page_floor);
1310     var consumed: u32 = 0;
1311     while (scanner.state == .scanning) {
1312         std.debug.assert(consumed < scanner.capacity.frames);
1313         _ = try consumeTestingFrame(scanner, data, .{});
1314         consumed += 1;
1315     }
1316     return try scanner.outcome();
1317 }
1318 
1319 fn loadSelectedFrame(scanner: *Scanner, data: []const u8, ref: FrameRef) !void {
1320     const request = try scanner.selectedFrame(ref, .{});
1321     const offset: usize = @intCast(request.offset);
1322     @memcpy(request.bytes, data[offset..][0..frame_size]);
1323 }
1324 
1325 const TestingInterrupt = struct {
1326     remaining: usize,
1327 
1328     fn control(self: *TestingInterrupt) Control {
1329         return .{ .context = self, .interrupted_fn = interrupted };
1330     }
1331 
1332     fn interrupted(context: ?*anyopaque) bool {
1333         const self: *TestingInterrupt = @ptrCast(@alignCast(context.?));
1334         if (self.remaining == 0) return true;
1335         self.remaining -= 1;
1336         return false;
1337     }
1338 };
1339 
1340 fn zeroFrameRef() FrameRef {
1341     return .{
1342         .page_id = 0,
1343         .frame = 0,
1344         .checksum_before = .{ 0, 0 },
1345         .checksum_after = .{ 0, 0 },
1346     };
1347 }
1348 
1349 fn refPagesEqual(refs: []const FrameRef, page_ids: []const u32) bool {
1350     if (refs.len != page_ids.len) return false;
1351     for (refs, page_ids) |ref, page_id| {
1352         if (ref.page_id != page_id) return false;
1353     }
1354     return true;
1355 }
1356 
1357 fn prepareFinalTestingFrame(scanner: *Scanner, data: []const u8) !void {
1358     try resetTestingScanner(scanner, data, 0);
1359     std.debug.assert(scanner.target_frames > 0);
1360     var consumed: u32 = 0;
1361     while (scanner.covered_frames + 1 < scanner.target_frames) {
1362         std.debug.assert(consumed < scanner.capacity.frames);
1363         _ = try consumeTestingFrame(scanner, data, .{});
1364         consumed += 1;
1365     }
1366     const request = (try scanner.nextFrame()) orelse return error.TestUnexpectedResult;
1367     const offset: usize = @intCast(request.offset);
1368     @memcpy(request.bytes, data[offset..][0..frame_size]);
1369 }
1370 
1371 fn expectScannerNotReady(scanner: *Scanner) !void {
1372     const ref = FrameRef{
1373         .page_id = 1,
1374         .frame = 1,
1375         .checksum_before = .{ 0, 0 },
1376         .checksum_after = .{ 0, 0 },
1377     };
1378     var destination: [page.size]u8 = @splat(0xa5);
1379     try std.testing.expectError(error.ScannerNotReady, scanner.outcome());
1380     try std.testing.expectError(error.ScannerNotReady, scanner.findCommitted(0, .{}));
1381     try std.testing.expectError(error.ScannerNotReady, scanner.selectedFrame(ref, .{}));
1382     try std.testing.expectError(
1383         error.ScannerNotReady,
1384         scanner.verifySelected(ref, &destination, .{}),
1385     );
1386     try std.testing.expectError(error.ScannerNotReady, scanner.nextFrame());
1387     try std.testing.expectError(error.ScannerNotReady, scanner.consumeFrame(.{}));
1388     try std.testing.expectError(
1389         error.ScannerNotReady,
1390         scanner.extend(.{ .complete_frames = 0, .partial_tail_bytes = 0 }, .{}),
1391     );
1392     const expected: [page.size]u8 = @splat(0xa5);
1393     try std.testing.expectEqualSlices(u8, &expected, &destination);
1394 }
1395 
1396 fn modelWriterCapacity(frames: usize) ?Writer.Capacity {
1397     const maximum_frames = (std.math.maxInt(usize) - header_size) / frame_size;
1398     if (frames > maximum_frames) return null;
1399     return .{
1400         .frames = frames,
1401         .storage_bytes = header_size + frames * frame_size,
1402     };
1403 }
1404 
1405 fn modelScannerCapacity(frames: usize) ?Scanner.Capacity {
1406     if (frames > std.math.maxInt(u32)) return null;
1407     const refs_bytes = std.math.mul(
1408         usize,
1409         frames,
1410         @sizeOf(FrameRef),
1411     ) catch return null;
1412     const storage_bytes = std.math.add(
1413         usize,
1414         frame_size,
1415         refs_bytes,
1416     ) catch return null;
1417     return .{
1418         .frames = @intCast(frames),
1419         .storage_bytes = storage_bytes,
1420     };
1421 }
1422 
1423 test "wal scanner capacity matches an independent typed byte model" {
1424     try std.testing.expectEqual(@as(usize, 24), @sizeOf(FrameRef));
1425     for (0..4_097) |frames| {
1426         try std.testing.expectEqual(
1427             modelScannerCapacity(frames).?,
1428             try Scanner.Capacity.derive(frames),
1429         );
1430     }
1431 
1432     const capacity = modelScannerCapacity(1_024).?;
1433     try std.testing.expectEqual(@as(u32, 1_024), capacity.frames);
1434     try std.testing.expectEqual(@as(usize, 28_696), capacity.storage_bytes);
1435     try std.testing.expectEqual(capacity, try Scanner.Capacity.derive(1_024));
1436 
1437     const maximum_frames: usize = std.math.maxInt(u32);
1438     if (modelScannerCapacity(maximum_frames)) |expected| {
1439         try std.testing.expectEqual(expected, try Scanner.Capacity.derive(maximum_frames));
1440     } else {
1441         try std.testing.expectError(
1442             error.CapacityOverflow,
1443             Scanner.Capacity.derive(maximum_frames),
1444         );
1445     }
1446     if (std.math.add(usize, maximum_frames, 1)) |over_u32| {
1447         try std.testing.expectError(
1448             error.CapacityOverflow,
1449             Scanner.Capacity.derive(over_u32),
1450         );
1451     } else |_| {}
1452     try std.testing.expectError(
1453         error.CapacityOverflow,
1454         Scanner.storageBytes(std.math.maxInt(usize), frame_size, @sizeOf(FrameRef)),
1455     );
1456     try std.testing.expectError(
1457         error.CapacityOverflow,
1458         Scanner.storageBytes(1, std.math.maxInt(usize), 1),
1459     );
1460 
1461     var scratch: [frame_size]u8 = undefined;
1462     var refs: [1_024]FrameRef = undefined;
1463     var scanner = try testingScanner(&scratch, &refs);
1464     try std.testing.expectEqual(&scratch, scanner.workspace.scratch);
1465     try std.testing.expectEqual(refs[0..].ptr, scanner.workspace.refs.ptr);
1466     const returned = scanner.deinit();
1467     try std.testing.expectEqual(&scratch, returned.scratch);
1468     try std.testing.expectEqual(refs[0..].ptr, returned.refs.ptr);
1469 }
1470 
1471 test "wal scanner extent distinguishes every incomplete boundary" {
1472     var header_bytes: u64 = 0;
1473     while (header_bytes < header_size) : (header_bytes += 1) {
1474         const extent = try Extent.classify(header_bytes, 2, .{});
1475         try std.testing.expectEqual(@as(u8, @intCast(header_bytes)), extent.partial_header);
1476     }
1477 
1478     var tail_bytes: u64 = 0;
1479     while (tail_bytes < frame_size) : (tail_bytes += 1) {
1480         const file_bytes = header_size + frame_size + tail_bytes;
1481         const extent = (try Extent.classify(file_bytes, 2, .{})).admitted;
1482         try std.testing.expectEqual(@as(u32, 1), extent.complete_frames);
1483         try std.testing.expectEqual(@as(u16, @intCast(tail_bytes)), extent.partial_tail_bytes);
1484     }
1485 }
1486 
1487 test "wal scanner rejects capacity plus one before frame payload" {
1488     const exact_bytes = header_size + 2 * frame_size;
1489     const exact = (try Extent.classify(exact_bytes, 2, .{})).admitted;
1490     try std.testing.expectEqual(@as(u32, 2), exact.complete_frames);
1491     try std.testing.expectEqual(@as(u16, 0), exact.partial_tail_bytes);
1492 
1493     const partial = (try Extent.classify(exact_bytes + 1, 2, .{})).over_capacity;
1494     try std.testing.expectEqual(@as(u64, 2), partial.complete_frames);
1495     try std.testing.expectEqual(@as(u16, 1), partial.partial_tail_bytes);
1496 
1497     const full = (try Extent.classify(exact_bytes + frame_size, 2, .{})).over_capacity;
1498     try std.testing.expectEqual(@as(u64, 3), full.complete_frames);
1499     try std.testing.expectEqual(@as(u16, 0), full.partial_tail_bytes);
1500 
1501     const maximum_admitted_bytes = header_size + 1_024 * frame_size;
1502     const maximum = (try Extent.classify(maximum_admitted_bytes, 1_024, .{})).admitted;
1503     try std.testing.expectEqual(@as(u32, 1_024), maximum.complete_frames);
1504     _ = (try Extent.classify(maximum_admitted_bytes + 1, 1_024, .{})).over_capacity;
1505 }
1506 
1507 test "wal scanner reports a partial uncommitted tail without consuming it" {
1508     var writer = try testingWriter(std.testing.allocator, 0);
1509     defer std.testing.allocator.free(writer.deinit());
1510     var scratch: [frame_size]u8 = undefined;
1511     var refs: [1]FrameRef = undefined;
1512     var scanner = try testingScanner(&scratch, &refs);
1513     const header: *const [header_size]u8 = writer.bytes()[0..header_size];
1514     try scanner.reset(header, .{
1515         .complete_frames = 0,
1516         .partial_tail_bytes = 73,
1517     }, 0, .{});
1518     try std.testing.expectEqual(
1519         @as(u16, 73),
1520         (try scanner.outcome()).partial_uncommitted_tail,
1521     );
1522     const progress = scanner.progress();
1523     try std.testing.expectEqual(@as(u32, 0), progress.covered_frames);
1524     try std.testing.expectEqualSlices(
1525         u8,
1526         writer.bytes()[0..header_size],
1527         &progress.header,
1528     );
1529 }
1530 
1531 test "wal scanner rejects every forged noncanonical partial tail" {
1532     var writer = try testingWriter(std.testing.allocator, 0);
1533     defer std.testing.allocator.free(writer.deinit());
1534     var scratch: [frame_size]u8 = undefined;
1535     var refs: [1]FrameRef = undefined;
1536     var scanner = try testingScanner(&scratch, &refs);
1537     const header: *const [header_size]u8 = writer.bytes()[0..header_size];
1538     try scanner.reset(header, .{
1539         .complete_frames = 0,
1540         .partial_tail_bytes = frame_size - 1,
1541     }, 0, .{});
1542     try std.testing.expectEqual(
1543         @as(u16, frame_size - 1),
1544         (try scanner.outcome()).partial_uncommitted_tail,
1545     );
1546 
1547     const before = scanner.progress();
1548     try std.testing.expectError(error.InvalidExtent, scanner.reset(
1549         header,
1550         .{ .complete_frames = 0, .partial_tail_bytes = frame_size },
1551         0,
1552         .{},
1553     ));
1554     try std.testing.expectEqualDeep(before, scanner.progress());
1555     try std.testing.expectError(error.InvalidExtent, scanner.extend(.{
1556         .complete_frames = 0,
1557         .partial_tail_bytes = std.math.maxInt(u16),
1558     }, .{}));
1559     try std.testing.expectEqualDeep(before, scanner.progress());
1560 }
1561 
1562 test "wal scanner makes rewind and capacity invalidation unreadable" {
1563     var writer = try testingWriter(std.testing.allocator, 0);
1564     defer std.testing.allocator.free(writer.deinit());
1565     var scratch: [frame_size]u8 = undefined;
1566     var refs: [1]FrameRef = undefined;
1567     var scanner = try testingScanner(&scratch, &refs);
1568     const header: *const [header_size]u8 = writer.bytes()[0..header_size];
1569     try scanner.reset(header, .{
1570         .complete_frames = 0,
1571         .partial_tail_bytes = 10,
1572     }, 0, .{});
1573     try std.testing.expectError(error.RewindRequired, scanner.extend(.{
1574         .complete_frames = 0,
1575         .partial_tail_bytes = 9,
1576     }, .{}));
1577     try expectScannerNotReady(&scanner);
1578 
1579     try scanner.reset(header, .{
1580         .complete_frames = 0,
1581         .partial_tail_bytes = 0,
1582     }, 0, .{});
1583     try std.testing.expectError(error.CapacityExceeded, scanner.extend(.{
1584         .complete_frames = 2,
1585         .partial_tail_bytes = 0,
1586     }, .{}));
1587     try expectScannerNotReady(&scanner);
1588 
1589     try scanner.reset(header, .{
1590         .complete_frames = 0,
1591         .partial_tail_bytes = 0,
1592     }, 0, .{});
1593     try std.testing.expectEqual(
1594         std.meta.Tag(Scanner.Outcome).complete,
1595         std.meta.activeTag(try scanner.outcome()),
1596     );
1597 }
1598 
1599 test "wal scanner reset interruption sweeps every header validation cut" {
1600     var writer = try testingWriter(std.testing.allocator, 0);
1601     defer std.testing.allocator.free(writer.deinit());
1602     const header: *const [header_size]u8 = writer.bytes()[0..header_size];
1603     var completed = false;
1604     for (0..16) |budget| {
1605         var scratch: [frame_size]u8 = undefined;
1606         var refs: [1]FrameRef = @splat(zeroFrameRef());
1607         var scanner = try testingScanner(&scratch, &refs);
1608         const before = scanner.progress();
1609         var interrupt = TestingInterrupt{ .remaining = budget };
1610         if (scanner.reset(
1611             header,
1612             .{ .complete_frames = 0, .partial_tail_bytes = 0 },
1613             0,
1614             interrupt.control(),
1615         )) |_| {
1616             try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1617             completed = true;
1618             break;
1619         } else |err| switch (err) {
1620             error.Interrupted => {
1621                 try std.testing.expectEqualDeep(before, scanner.progress());
1622                 try std.testing.expectEqualDeep(@as([1]FrameRef, @splat(zeroFrameRef())), refs);
1623             },
1624             else => return err,
1625         }
1626     }
1627     try std.testing.expect(completed);
1628 }
1629 
1630 test "wal scanner rejects a flip of every complete header byte" {
1631     var writer = try testingWriter(std.testing.allocator, 0);
1632     defer std.testing.allocator.free(writer.deinit());
1633     var scratch: [frame_size]u8 = undefined;
1634     var refs: [1]FrameRef = undefined;
1635     var scanner = try testingScanner(&scratch, &refs);
1636     const header: *[header_size]u8 = writer.storage[0..header_size];
1637 
1638     for (0..header_size) |index| {
1639         header[index] ^= 0xff;
1640         if (scanner.reset(
1641             header,
1642             .{ .complete_frames = 0, .partial_tail_bytes = 0 },
1643             0,
1644             .{},
1645         )) |_| {
1646             return error.TestUnexpectedResult;
1647         } else |err| switch (err) {
1648             error.InvalidChecksum,
1649             error.InvalidWal,
1650             error.UnsupportedPageSize,
1651             => {},
1652             error.CapacityExceeded,
1653             error.Interrupted,
1654             error.InvalidExtent,
1655             error.ScannerNotReady,
1656             => return err,
1657         }
1658         try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1659         header[index] ^= 0xff;
1660     }
1661 }
1662 
1663 test "wal scanner admits a completed prior partial frame" {
1664     var writer = try testingWriter(std.testing.allocator, 1);
1665     defer std.testing.allocator.free(writer.deinit());
1666     var image: [page.size]u8 = undefined;
1667     fillPage(&image, 1, 10);
1668     try writer.append(1, 1, &image);
1669     var scratch: [frame_size]u8 = undefined;
1670     var refs: [1]FrameRef = undefined;
1671     var scanner = try testingScanner(&scratch, &refs);
1672     const partial = writer.bytes()[0 .. header_size + 101];
1673     try resetTestingScanner(&scanner, partial, 0);
1674     try std.testing.expectEqual(
1675         @as(u16, 101),
1676         (try scanner.outcome()).partial_uncommitted_tail,
1677     );
1678 
1679     try scanner.extend(try admittedExtent(writer.bytes(), refs.len), .{});
1680     try std.testing.expectEqual(
1681         std.meta.Tag(Scanner.FrameOutcome).committed,
1682         std.meta.activeTag(try consumeTestingFrame(&scanner, writer.bytes(), .{})),
1683     );
1684     try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1685 }
1686 
1687 test "wal scanner terminal-invalidates page zero and malformed full frames" {
1688     var writer = try testingWriter(std.testing.allocator, 1);
1689     defer std.testing.allocator.free(writer.deinit());
1690     var image: [page.size]u8 = undefined;
1691     fillPage(&image, 1, 10);
1692     try writer.append(0, 1, &image);
1693 
1694     var scratch: [frame_size]u8 = undefined;
1695     var refs: [1]FrameRef = undefined;
1696     var scanner = try testingScanner(&scratch, &refs);
1697     try resetTestingScanner(&scanner, writer.bytes(), 0);
1698     const page_zero = try consumeTestingFrame(&scanner, writer.bytes(), .{});
1699     try std.testing.expectEqual(
1700         Scanner.MalformedFrame.invalid_page_id,
1701         page_zero.malformed_full_frame,
1702     );
1703     try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1704     try expectScannerNotReady(&scanner);
1705 
1706     writer.restore(.{
1707         .len = header_size,
1708         .checksum_first = readU32(writer.bytes()[24..28]),
1709         .checksum_second = readU32(writer.bytes()[28..32]),
1710     });
1711     try writer.append(1, 1, &image);
1712     _ = try scanTestingData(&scanner, writer.bytes(), 0);
1713     try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1714     writer.storage[header_size + 8] ^= 0xff;
1715     try resetTestingScanner(&scanner, writer.bytes(), 0);
1716     const salt = try consumeTestingFrame(&scanner, writer.bytes(), .{});
1717     try std.testing.expectEqual(
1718         Scanner.MalformedFrame.salt_mismatch,
1719         salt.malformed_full_frame,
1720     );
1721     try expectScannerNotReady(&scanner);
1722 
1723     writer.storage[header_size + 8] ^= 0xff;
1724     _ = try scanTestingData(&scanner, writer.bytes(), 0);
1725     writer.storage[header_size + frame_header_size + 9] ^= 0xff;
1726     try resetTestingScanner(&scanner, writer.bytes(), 0);
1727     const checksum = try consumeTestingFrame(&scanner, writer.bytes(), .{});
1728     try std.testing.expectEqual(
1729         Scanner.MalformedFrame.checksum_mismatch,
1730         checksum.malformed_full_frame,
1731     );
1732     try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1733     try expectScannerNotReady(&scanner);
1734     writer.storage[header_size + frame_header_size + 9] ^= 0xff;
1735     _ = try scanTestingData(&scanner, writer.bytes(), 0);
1736     try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1737 }
1738 
1739 fn scanSingleMarker(marker: u32, database_page_floor: u32) !Scanner.FrameOutcome {
1740     var writer = try testingWriter(std.testing.allocator, 1);
1741     defer std.testing.allocator.free(writer.deinit());
1742     var image: [page.size]u8 = undefined;
1743     fillPage(&image, 1, 10);
1744     try writer.append(1, marker, &image);
1745     var scratch: [frame_size]u8 = undefined;
1746     var refs: [1]FrameRef = undefined;
1747     var scanner = try testingScanner(&scratch, &refs);
1748     try resetTestingScanner(&scanner, writer.bytes(), database_page_floor);
1749     return try consumeTestingFrame(&scanner, writer.bytes(), .{});
1750 }
1751 
1752 test "wal scanner enforces the monotonic base page count floor" {
1753     const lower = try scanSingleMarker(9, 10);
1754     try std.testing.expectEqual(
1755         Scanner.MalformedFrame.database_page_count_regressed,
1756         lower.malformed_full_frame,
1757     );
1758     try std.testing.expectEqual(
1759         std.meta.Tag(Scanner.FrameOutcome).committed,
1760         std.meta.activeTag(try scanSingleMarker(10, 10)),
1761     );
1762     try std.testing.expectEqual(
1763         std.meta.Tag(Scanner.FrameOutcome).committed,
1764         std.meta.activeTag(try scanSingleMarker(12, 10)),
1765     );
1766 }
1767 
1768 test "wal scanner rejects a marker below any staged page" {
1769     var writer = try testingWriter(std.testing.allocator, 2);
1770     defer std.testing.allocator.free(writer.deinit());
1771     var image: [page.size]u8 = undefined;
1772     fillPage(&image, 11, 10);
1773     try writer.append(11, 0, &image);
1774     fillPage(&image, 1, 20);
1775     try writer.append(1, 10, &image);
1776 
1777     var scratch: [frame_size]u8 = undefined;
1778     var refs: [2]FrameRef = undefined;
1779     var scanner = try testingScanner(&scratch, &refs);
1780     try resetTestingScanner(&scanner, writer.bytes(), 0);
1781     try std.testing.expectEqual(
1782         std.meta.Tag(Scanner.FrameOutcome).staged,
1783         std.meta.activeTag(try consumeTestingFrame(&scanner, writer.bytes(), .{})),
1784     );
1785     const marker = try consumeTestingFrame(&scanner, writer.bytes(), .{});
1786     try std.testing.expectEqual(
1787         Scanner.MalformedFrame.page_outside_database,
1788         marker.malformed_full_frame,
1789     );
1790     try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1791     try expectScannerNotReady(&scanner);
1792 }
1793 
1794 test "wal scanner keeps an uncommitted high page outside authority" {
1795     var writer = try testingWriter(std.testing.allocator, 1);
1796     defer std.testing.allocator.free(writer.deinit());
1797     var image: [page.size]u8 = undefined;
1798     fillPage(&image, 100, 10);
1799     try writer.append(100, 0, &image);
1800 
1801     var scratch: [frame_size]u8 = undefined;
1802     var refs: [1]FrameRef = undefined;
1803     var scanner = try testingScanner(&scratch, &refs);
1804     _ = try scanTestingData(&scanner, writer.bytes(), 10);
1805     try std.testing.expect((try scanner.findCommitted(100, .{})) == null);
1806     try std.testing.expectEqual(@as(u32, 10), scanner.logical_database_page_count);
1807     try std.testing.expectEqual(@as(u32, 0), scanner.committed_ref_count);
1808     try std.testing.expectEqual(@as(u32, 1), scanner.staged_ref_count);
1809 }
1810 
1811 test "wal scanner folds duplicate pages with the greatest frame winning" {
1812     var writer = try testingWriter(std.testing.allocator, 5);
1813     defer std.testing.allocator.free(writer.deinit());
1814     var image: [page.size]u8 = undefined;
1815     fillPage(&image, 2, 20);
1816     try writer.append(2, 0, &image);
1817     fillPage(&image, 1, 10);
1818     try writer.append(1, 2, &image);
1819     fillPage(&image, 1, 11);
1820     try writer.append(1, 0, &image);
1821     fillPage(&image, 1, 12);
1822     try writer.append(1, 0, &image);
1823     fillPage(&image, 2, 22);
1824     try writer.append(2, 2, &image);
1825 
1826     var scratch: [frame_size]u8 = undefined;
1827     var refs: [5]FrameRef = undefined;
1828     var scanner = try testingScanner(&scratch, &refs);
1829     _ = try scanTestingData(&scanner, writer.bytes(), 0);
1830     const first = (try scanner.findCommitted(1, .{})).?;
1831     const second = (try scanner.findCommitted(2, .{})).?;
1832     try std.testing.expectEqual(@as(u32, 4), first.frame);
1833     try std.testing.expectEqual(@as(u32, 5), second.frame);
1834     try std.testing.expectEqual(@as(u32, 2), scanner.committed_ref_count);
1835     try std.testing.expectEqual(@as(u32, 5), scanner.committed_end_mark);
1836 
1837     var destination: [page.size]u8 = undefined;
1838     try loadSelectedFrame(&scanner, writer.bytes(), first);
1839     try scanner.verifySelected(first, &destination, .{});
1840     try std.testing.expectEqual(@as(u8, 12), destination[1]);
1841     try loadSelectedFrame(&scanner, writer.bytes(), second);
1842     try scanner.verifySelected(second, &destination, .{});
1843     try std.testing.expectEqual(@as(u8, 22), destination[1]);
1844 }
1845 
1846 test "wal scanner resumes from the physical checksum tail" {
1847     var writer = try testingWriter(std.testing.allocator, 4);
1848     defer std.testing.allocator.free(writer.deinit());
1849     var image: [page.size]u8 = undefined;
1850     fillPage(&image, 1, 10);
1851     try writer.append(1, 1, &image);
1852     fillPage(&image, 1, 15);
1853     try writer.append(1, 1, &image);
1854     fillPage(&image, 1, 20);
1855     try writer.append(1, 0, &image);
1856 
1857     var scratch: [frame_size]u8 = undefined;
1858     var refs: [4]FrameRef = undefined;
1859     var scanner = try testingScanner(&scratch, &refs);
1860     _ = try scanTestingData(&scanner, writer.bytes(), 0);
1861     try std.testing.expectEqual(
1862         @as(u32, 2),
1863         (try scanner.findCommitted(1, .{})).?.frame,
1864     );
1865 
1866     fillPage(&image, 1, 30);
1867     try writer.append(1, 1, &image);
1868     const extent = try admittedExtent(writer.bytes(), refs.len);
1869     try scanner.extend(extent, .{});
1870     _ = try consumeTestingFrame(&scanner, writer.bytes(), .{});
1871     const latest = (try scanner.findCommitted(1, .{})).?;
1872     try std.testing.expectEqual(@as(u32, 4), latest.frame);
1873     try std.testing.expectEqual(@as(u32, 4), scanner.covered_frames);
1874     try std.testing.expectEqual(@as(u32, 4), scanner.committed_end_mark);
1875 
1876     var destination: [page.size]u8 = undefined;
1877     try loadSelectedFrame(&scanner, writer.bytes(), latest);
1878     try scanner.verifySelected(latest, &destination, .{});
1879     try std.testing.expectEqual(@as(u8, 30), destination[1]);
1880 }
1881 
1882 const InsertSweep = struct {
1883     completed: bool = false,
1884     pre_mutation: bool = false,
1885     shifted_one: bool = false,
1886     shifted_two: bool = false,
1887     shifted_three: bool = false,
1888     inserted: bool = false,
1889 };
1890 
1891 fn exerciseInsertCut(data: []const u8, budget: usize) !InsertSweep {
1892     var scratch: [frame_size]u8 = undefined;
1893     var refs: [4]FrameRef = @splat(zeroFrameRef());
1894     var scanner = try testingScanner(&scratch, &refs);
1895     try prepareFinalTestingFrame(&scanner, data);
1896     const before = scanner.progress();
1897     const refs_before = refs;
1898     var interrupt = TestingInterrupt{ .remaining = budget };
1899     if (scanner.consumeFrame(interrupt.control())) |outcome| {
1900         try std.testing.expectEqual(
1901             std.meta.Tag(Scanner.FrameOutcome).committed,
1902             std.meta.activeTag(outcome),
1903         );
1904         try std.testing.expect(refPagesEqual(&refs, &.{ 1, 2, 4, 6 }));
1905         return .{ .completed = true };
1906     } else |err| switch (err) {
1907         error.Interrupted => {},
1908         else => return err,
1909     }
1910     if (scanner.state == .scanning) {
1911         try std.testing.expectEqualDeep(before, scanner.progress());
1912         try std.testing.expectEqualDeep(refs_before, refs);
1913         _ = try scanner.consumeFrame(.{});
1914         try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1915         return .{ .pre_mutation = true };
1916     }
1917     try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1918     const result = InsertSweep{
1919         .shifted_one = refPagesEqual(&refs, &.{ 2, 4, 6, 6 }),
1920         .shifted_two = refPagesEqual(&refs, &.{ 2, 4, 4, 6 }),
1921         .shifted_three = refPagesEqual(&refs, &.{ 2, 2, 4, 6 }),
1922         .inserted = refPagesEqual(&refs, &.{ 1, 2, 4, 6 }),
1923     };
1924     try expectScannerNotReady(&scanner);
1925     _ = try scanTestingData(&scanner, data, 0);
1926     try std.testing.expect((try scanner.findCommitted(1, .{})) != null);
1927     return result;
1928 }
1929 
1930 test "wal scanner interruption sweeps every insert fold cut" {
1931     var writer = try testingWriter(std.testing.allocator, 4);
1932     defer std.testing.allocator.free(writer.deinit());
1933     var image: [page.size]u8 = undefined;
1934     for ([_]u8{ 2, 4, 6 }) |page_id| {
1935         fillPage(&image, page_id, page_id);
1936         try writer.append(page_id, if (page_id == 6) 6 else 0, &image);
1937     }
1938     fillPage(&image, 1, 1);
1939     try writer.append(1, 6, &image);
1940 
1941     var observed = InsertSweep{};
1942     for (0..frame_size / 8 + 128) |budget| {
1943         const cut = try exerciseInsertCut(writer.bytes(), budget);
1944         observed.completed = observed.completed or cut.completed;
1945         observed.pre_mutation = observed.pre_mutation or cut.pre_mutation;
1946         observed.shifted_one = observed.shifted_one or cut.shifted_one;
1947         observed.shifted_two = observed.shifted_two or cut.shifted_two;
1948         observed.shifted_three = observed.shifted_three or cut.shifted_three;
1949         observed.inserted = observed.inserted or cut.inserted;
1950         if (cut.completed) break;
1951     }
1952     try std.testing.expect(observed.completed);
1953     try std.testing.expect(observed.pre_mutation);
1954     try std.testing.expect(observed.shifted_one);
1955     try std.testing.expect(observed.shifted_two);
1956     try std.testing.expect(observed.shifted_three);
1957     try std.testing.expect(observed.inserted);
1958 }
1959 
1960 const ReplaceSweep = struct {
1961     completed: bool = false,
1962     pre_mutation: bool = false,
1963     first_replacement: bool = false,
1964     tail_shift: bool = false,
1965     duplicate_replacement: bool = false,
1966     later_candidate: bool = false,
1967 };
1968 
1969 fn exerciseReplaceCut(data: []const u8, budget: usize) !ReplaceSweep {
1970     var scratch: [frame_size]u8 = undefined;
1971     var refs: [5]FrameRef = @splat(zeroFrameRef());
1972     var scanner = try testingScanner(&scratch, &refs);
1973     try prepareFinalTestingFrame(&scanner, data);
1974     const before = scanner.progress();
1975     const refs_before = refs;
1976     var interrupt = TestingInterrupt{ .remaining = budget };
1977     if (scanner.consumeFrame(interrupt.control())) |outcome| {
1978         try std.testing.expectEqual(
1979             std.meta.Tag(Scanner.FrameOutcome).committed,
1980             std.meta.activeTag(outcome),
1981         );
1982         try std.testing.expectEqual(@as(u32, 4), refs[0].frame);
1983         try std.testing.expectEqual(@as(u32, 5), refs[1].frame);
1984         return .{ .completed = true };
1985     } else |err| switch (err) {
1986         error.Interrupted => {},
1987         else => return err,
1988     }
1989     if (scanner.state == .scanning) {
1990         try std.testing.expectEqualDeep(before, scanner.progress());
1991         try std.testing.expectEqualDeep(refs_before, refs);
1992         _ = try scanner.consumeFrame(.{});
1993         try std.testing.expectEqual(Scanner.State.ready, scanner.state);
1994         return .{ .pre_mutation = true };
1995     }
1996     try std.testing.expectEqual(Scanner.State.invalidated, scanner.state);
1997     const result = ReplaceSweep{
1998         .first_replacement = refs[0].frame == 3,
1999         .tail_shift = refs[2].frame == 4 and refs[3].frame == 5,
2000         .duplicate_replacement = refs[0].frame == 4,
2001         .later_candidate = refs[1].frame == 5,
2002     };
2003     try expectScannerNotReady(&scanner);
2004     _ = try scanTestingData(&scanner, data, 0);
2005     try std.testing.expectEqual(@as(u32, 4), (try scanner.findCommitted(1, .{})).?.frame);
2006     try std.testing.expectEqual(@as(u32, 5), (try scanner.findCommitted(2, .{})).?.frame);
2007     return result;
2008 }
2009 
2010 test "wal scanner interruption sweeps replacement and duplicate fold cuts" {
2011     var writer = try testingWriter(std.testing.allocator, 5);
2012     defer std.testing.allocator.free(writer.deinit());
2013     var image: [page.size]u8 = undefined;
2014     const pages = [_]u8{ 1, 2, 1, 1, 2 };
2015     for (pages, 0..) |page_id, index| {
2016         fillPage(&image, page_id, @intCast(index + 1));
2017         const marker: u32 = if (index == 1 or index == 4) 2 else 0;
2018         try writer.append(page_id, marker, &image);
2019     }
2020 
2021     var observed = ReplaceSweep{};
2022     for (0..frame_size / 8 + 128) |budget| {
2023         const cut = try exerciseReplaceCut(writer.bytes(), budget);
2024         observed.completed = observed.completed or cut.completed;
2025         observed.pre_mutation = observed.pre_mutation or cut.pre_mutation;
2026         observed.first_replacement = observed.first_replacement or cut.first_replacement;
2027         observed.tail_shift = observed.tail_shift or cut.tail_shift;
2028         observed.duplicate_replacement = observed.duplicate_replacement or
2029             cut.duplicate_replacement;
2030         observed.later_candidate = observed.later_candidate or cut.later_candidate;
2031         if (cut.completed) break;
2032     }
2033     try std.testing.expect(observed.completed);
2034     try std.testing.expect(observed.pre_mutation);
2035     try std.testing.expect(observed.first_replacement);
2036     try std.testing.expect(observed.tail_shift);
2037     try std.testing.expect(observed.duplicate_replacement);
2038     try std.testing.expect(observed.later_candidate);
2039 }
2040 
2041 fn expectSelectedMutationRejected(
2042     scanner: *Scanner,
2043     data: []const u8,
2044     ref: FrameRef,
2045     mutation_index: usize,
2046 ) !void {
2047     try loadSelectedFrame(scanner, data, ref);
2048     scanner.workspace.scratch[mutation_index] ^= 0xff;
2049     var destination: [page.size]u8 = @splat(0xa5);
2050     const before = scanner.progress();
2051     try std.testing.expectError(
2052         error.SelectedFrameMismatch,
2053         scanner.verifySelected(ref, &destination, .{}),
2054     );
2055     try std.testing.expectEqualDeep(before, scanner.progress());
2056     const expected: [page.size]u8 = @splat(0xa5);
2057     try std.testing.expectEqualSlices(u8, &expected, &destination);
2058 }
2059 
2060 fn exerciseVerifyCut(
2061     scanner: *Scanner,
2062     data: []const u8,
2063     ref: FrameRef,
2064     expected: *const [page.size]u8,
2065     budget: usize,
2066 ) !bool {
2067     try loadSelectedFrame(scanner, data, ref);
2068     var destination: [page.size]u8 = @splat(0xa5);
2069     const before = scanner.progress();
2070     const ref_before = (try scanner.findCommitted(ref.page_id, .{})).?;
2071     var interrupt = TestingInterrupt{ .remaining = budget };
2072     if (scanner.verifySelected(ref, &destination, interrupt.control())) |_| {
2073         try std.testing.expectEqualSlices(u8, expected, &destination);
2074         return true;
2075     } else |err| switch (err) {
2076         error.Interrupted => {},
2077         else => return err,
2078     }
2079     try std.testing.expectEqualDeep(before, scanner.progress());
2080     try std.testing.expectEqualDeep(
2081         ref_before,
2082         (try scanner.findCommitted(ref.page_id, .{})).?,
2083     );
2084     const sentinel: [page.size]u8 = @splat(0xa5);
2085     try std.testing.expectEqualSlices(u8, &sentinel, &destination);
2086     return false;
2087 }
2088 
2089 test "wal scanner selected verification changes destination only after proof" {
2090     var writer = try testingWriter(std.testing.allocator, 1);
2091     defer std.testing.allocator.free(writer.deinit());
2092     var image: [page.size]u8 = undefined;
2093     fillPage(&image, 1, 10);
2094     try writer.append(1, 1, &image);
2095     var scratch: [frame_size]u8 = undefined;
2096     var refs: [1]FrameRef = undefined;
2097     var scanner = try testingScanner(&scratch, &refs);
2098     _ = try scanTestingData(&scanner, writer.bytes(), 0);
2099     const ref = (try scanner.findCommitted(1, .{})).?;
2100 
2101     try expectSelectedMutationRejected(&scanner, writer.bytes(), ref, 0);
2102     try expectSelectedMutationRejected(&scanner, writer.bytes(), ref, 8);
2103     try expectSelectedMutationRejected(&scanner, writer.bytes(), ref, 16);
2104     try expectSelectedMutationRejected(
2105         &scanner,
2106         writer.bytes(),
2107         ref,
2108         frame_header_size + 9,
2109     );
2110 
2111     var completed = false;
2112     for (0..frame_size / 8 + 128) |budget| {
2113         if (try exerciseVerifyCut(&scanner, writer.bytes(), ref, &image, budget)) {
2114             completed = true;
2115             break;
2116         }
2117     }
2118     try std.testing.expect(completed);
2119 }
2120 
2121 test "wal scanner rolling checksum does not claim colliding byte identity" {
2122     var writer = try testingWriter(std.testing.allocator, 1);
2123     defer std.testing.allocator.free(writer.deinit());
2124     var image: [page.size]u8 = undefined;
2125     fillPage(&image, 1, 10);
2126     try writer.append(1, 1, &image);
2127     var scratch: [frame_size]u8 = undefined;
2128     var refs: [1]FrameRef = undefined;
2129     var scanner = try testingScanner(&scratch, &refs);
2130     _ = try scanTestingData(&scanner, writer.bytes(), 0);
2131     const ref = (try scanner.findCommitted(1, .{})).?;
2132     try loadSelectedFrame(&scanner, writer.bytes(), ref);
2133 
2134     const payload = scanner.workspace.scratch[frame_header_size..];
2135     writeU32(payload[0..4], readU32(payload[0..4]) +% 1);
2136     writeU32(payload[8..12], readU32(payload[8..12]) -% 2);
2137     writeU32(payload[12..16], readU32(payload[12..16]) -% 1);
2138     var destination: [page.size]u8 = undefined;
2139     const before = scanner.progress();
2140     try scanner.verifySelected(ref, &destination, .{});
2141     try std.testing.expectEqualDeep(before, scanner.progress());
2142     try std.testing.expectEqualSlices(u8, payload, &destination);
2143     try std.testing.expect(!std.mem.eql(u8, &image, &destination));
2144 }
2145 
2146 test "wal writer capacity matches an independent additive model" {
2147     comptime {
2148         @stardustClaim(
2149             @import("alloc_phase").capacity.witness(Writer, "sql_wal_capacity"),
2150             null,
2151             null,
2152             null,
2153             null,
2154             null,
2155             null,
2156         );
2157     }
2158 
2159     for (0..4097) |frames| {
2160         try std.testing.expectEqual(
2161             modelWriterCapacity(frames).?,
2162             try Writer.Capacity.derive(.{ .header = testingHeader(), .frames = frames }),
2163         );
2164     }
2165 
2166     const maximum_frames = (std.math.maxInt(usize) - header_size) / frame_size;
2167     try std.testing.expectEqual(
2168         modelWriterCapacity(maximum_frames).?,
2169         try Writer.Capacity.derive(.{ .header = testingHeader(), .frames = maximum_frames }),
2170     );
2171     const overflow_frames = maximum_frames + 1;
2172     try std.testing.expect(modelWriterCapacity(overflow_frames) == null);
2173     try std.testing.expectError(
2174         error.CapacityOverflow,
2175         Writer.Capacity.derive(.{ .header = testingHeader(), .frames = overflow_frames }),
2176     );
2177     var empty_storage: [0]u8 = .{};
2178     try std.testing.expectError(
2179         error.CapacityOverflow,
2180         Writer.init(&empty_storage, .{ .header = testingHeader(), .frames = overflow_frames }),
2181     );
2182 }
2183 
2184 test "wal writer rejects short storage and releases its exact borrow" {
2185     comptime {
2186         @stardustClaim(
2187             @import("alloc_phase").capacity.witness(Writer, "sql_wal_storage_rejection"),
2188             null,
2189             null,
2190             null,
2191             null,
2192             null,
2193             null,
2194         );
2195         @stardustClaim(
2196             @import("alloc_phase").capacity.witness(Writer, "sql_wal_work_bound"),
2197             null,
2198             null,
2199             null,
2200             null,
2201             null,
2202             null,
2203         );
2204     }
2205 
2206     const limits: Writer.Limits = .{ .header = testingHeader(), .frames = 3 };
2207     const capacity = try Writer.Capacity.derive(limits);
2208     const storage = try std.testing.allocator.alloc(u8, capacity.storage_bytes);
2209     defer std.testing.allocator.free(storage);
2210     try std.testing.expectError(error.StorageTooShort, Writer.init(storage[0 .. storage.len - 1], limits));
2211 
2212     var writer = try Writer.init(storage, limits);
2213     try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, writer.phase);
2214     writer.activate();
2215     try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, writer.phase);
2216     const released = writer.deinit();
2217     try std.testing.expectEqual(storage.ptr, released.ptr);
2218     try std.testing.expectEqual(storage.len, released.len);
2219 }
2220 
2221 test "wal writer is sealed before maximum append recovery and rewrite" {
2222     comptime {
2223         @stardustClaim(
2224             @import("alloc_phase").capacity.witness(Writer, "sql_wal_sealed"),
2225             null,
2226             null,
2227             null,
2228             null,
2229             null,
2230             null,
2231         );
2232     }
2233 
2234     var source = try testingWriter(std.testing.allocator, 4);
2235     defer std.testing.allocator.free(source.deinit());
2236     var image: [page.size]u8 = undefined;
2237     for (0..4) |index| {
2238         fillPage(&image, @intCast(index + 1), @intCast(10 + index));
2239         try source.append(@intCast(index + 1), if (index == 0) 4 else 0, &image);
2240     }
2241 
2242     var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
2243     const limits: Writer.Limits = .{ .header = testingHeader(), .frames = 3 };
2244     const storage_capacity = try Writer.Capacity.derive(limits);
2245     const storage = phase_allocator.initializationAllocator().alloc(u8, storage_capacity.storage_bytes) catch |err| {
2246         phase_allocator.abortInitialization();
2247         phase_allocator.deinit();
2248         return err;
2249     };
2250     var writer = Writer.init(storage, limits) catch |err| {
2251         phase_allocator.abortInitialization();
2252         phase_allocator.deinit();
2253         return err;
2254     };
2255     defer {
2256         if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
2257         if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
2258         if (writer.phase != .teardown) phase_allocator.teardownAllocator().free(writer.deinit());
2259         phase_allocator.deinit();
2260     }
2261 
2262     const storage_pointer = writer.storage.ptr;
2263     const capacity = writer.capacity;
2264     phase_allocator.seal();
2265     writer.activate();
2266 
2267     const committed_len = header_size + 3 * frame_size;
2268     try writer.load(source.bytes(), committed_len);
2269     try std.testing.expectEqual(@as(usize, 3), writer.frameCount());
2270     writer.rewriteTail(1, .{
2271         .sequence = 8,
2272         .salt = .{ .first = 0x5555_6666, .second = 0x7777_8888 },
2273     });
2274     try std.testing.expectEqual(@as(usize, 2), writer.frameCount());
2275 
2276     fillPage(&image, 9, 99);
2277     const staged_position = writer.position();
2278     try writer.stagePage(staged_position, 0, 9, &image);
2279     writer.commitStagedFrame(staged_position, 0, 9);
2280     try std.testing.expectEqual(@as(usize, 3), writer.frameCount());
2281     const position_before_overload = writer.position();
2282     var bytes_before_overload: [header_size + 3 * frame_size]u8 = undefined;
2283     @memcpy(bytes_before_overload[0..], writer.bytes());
2284 
2285     try std.testing.expectError(error.WalFull, writer.append(10, 10, &image));
2286     try std.testing.expectError(error.WalFull, writer.load(source.bytes(), source.bytes().len));
2287     try std.testing.expectEqual(position_before_overload, writer.position());
2288     try std.testing.expectEqual(storage_pointer, writer.storage.ptr);
2289     try std.testing.expectEqual(capacity, writer.capacity);
2290     try std.testing.expectEqualSlices(u8, bytes_before_overload[0..], writer.bytes());
2291     try std.testing.expectEqual(@as(u64, 0), phase_allocator.violations().total());
2292 }
2293 
2294 test "wal header validates and has no committed frames" {
2295     var writer = try testingWriter(std.testing.allocator, 0);
2296     defer std.testing.allocator.free(writer.deinit());
2297 
2298     var reader = try Reader.init(writer.bytes());
2299     try std.testing.expect(try reader.next() == null);
2300     try std.testing.expectEqual(@as(usize, 0), try endMark(writer.bytes()));
2301 }
2302 
2303 test "wal commit marker makes preceding frames visible" {
2304     comptime {
2305         @stardustClaim(
2306             @import("alloc_phase").capacity.witness(Writer, "sql_wal_semantics"),
2307             null,
2308             null,
2309             null,
2310             null,
2311             null,
2312             null,
2313         );
2314     }
2315 
2316     var writer = try testingWriter(std.testing.allocator, 2);
2317     defer std.testing.allocator.free(writer.deinit());
2318 
2319     var first: [page.size]u8 = undefined;
2320     var second: [page.size]u8 = undefined;
2321     fillPage(&first, 1, 10);
2322     fillPage(&second, 2, 20);
2323     try writer.append(1, 0, &first);
2324     try writer.append(2, 2, &second);
2325 
2326     const mark = try endMark(writer.bytes());
2327     try std.testing.expectEqual(@as(usize, 2), mark);
2328     const recovered_first = (try pageAt(writer.bytes(), 1, mark)).?;
2329     const recovered_second = (try pageAt(writer.bytes(), 2, mark)).?;
2330     try std.testing.expectEqual(@as(u8, 10), recovered_first[1]);
2331     try std.testing.expectEqual(@as(u8, 20), recovered_second[1]);
2332 }
2333 
2334 test "wal staged frames transfer in place with byte exact output" {
2335     comptime {
2336         @stardustClaim(
2337             @import("alloc_phase").capacity.witness(Writer, "sql_wal_staged_transfer"),
2338             null,
2339             null,
2340             null,
2341             null,
2342             null,
2343             null,
2344         );
2345     }
2346 
2347     var staged = try testingWriter(std.testing.allocator, 2);
2348     defer std.testing.allocator.free(staged.deinit());
2349     var appended = try testingWriter(std.testing.allocator, 2);
2350     defer std.testing.allocator.free(appended.deinit());
2351 
2352     var first: [page.size]u8 = undefined;
2353     var second: [page.size]u8 = undefined;
2354     fillPage(&first, 1, 10);
2355     fillPage(&second, 2, 20);
2356     const position = staged.position();
2357     try staged.stagePage(position, 0, 2, &second);
2358     try staged.stagePage(position, 1, 1, &first);
2359     try std.testing.expectEqual(@as(u32, 2), staged.stagedPageId(position, 0));
2360     try std.testing.expectEqual(@as(u8, 20), staged.stagedPage(position, 0)[1]);
2361     staged.swapStagedFrames(position, 0, 1);
2362     try std.testing.expectEqual(@as(u32, 1), staged.stagedPageId(position, 0));
2363     try std.testing.expectEqual(@as(u32, 2), staged.stagedPageId(position, 1));
2364     staged.commitStagedFrame(position, 0, 0);
2365     staged.commitStagedFrame(position, 1, 2);
2366 
2367     try appended.append(1, 0, &first);
2368     try appended.append(2, 2, &second);
2369     try std.testing.expectEqualSlices(u8, appended.bytes(), staged.bytes());
2370 
2371     const full_position = staged.position();
2372     try std.testing.expectError(error.WalFull, staged.stagePage(full_position, 0, 3, &first));
2373     try std.testing.expectEqual(full_position, staged.position());
2374 }
2375 
2376 test "wal ignores uncommitted tail frames" {
2377     var writer = try testingWriter(std.testing.allocator, 2);
2378     defer std.testing.allocator.free(writer.deinit());
2379 
2380     var first: [page.size]u8 = undefined;
2381     var second: [page.size]u8 = undefined;
2382     fillPage(&first, 1, 10);
2383     fillPage(&second, 1, 99);
2384     try writer.append(1, 1, &first);
2385     try writer.append(1, 0, &second);
2386 
2387     const mark = try endMark(writer.bytes());
2388     try std.testing.expectEqual(@as(usize, 1), mark);
2389     const recovered = (try pageAt(writer.bytes(), 1, mark)).?;
2390     try std.testing.expectEqual(@as(u8, 10), recovered[1]);
2391 }
2392 
2393 test "wal writer restores a prior append position" {
2394     var writer = try testingWriter(std.testing.allocator, 1);
2395     defer std.testing.allocator.free(writer.deinit());
2396 
2397     var first: [page.size]u8 = undefined;
2398     var second: [page.size]u8 = undefined;
2399     fillPage(&first, 1, 10);
2400     fillPage(&second, 1, 20);
2401 
2402     const position = writer.position();
2403     try writer.append(1, 1, &first);
2404     writer.restore(position);
2405     try std.testing.expectEqual(@as(usize, 0), writer.frameCount());
2406     try writer.append(1, 1, &second);
2407 
2408     const mark = try endMark(writer.bytes());
2409     try std.testing.expectEqual(@as(usize, 1), mark);
2410     const recovered = (try pageAt(writer.bytes(), 1, mark)).?;
2411     try std.testing.expectEqual(@as(u8, 20), recovered[1]);
2412 }
2413 
2414 test "wal recovery copies the committed prefix into fixed storage" {
2415     var writer = try testingWriter(std.testing.allocator, 2);
2416     defer std.testing.allocator.free(writer.deinit());
2417 
2418     var first: [page.size]u8 = undefined;
2419     var second: [page.size]u8 = undefined;
2420     fillPage(&first, 1, 10);
2421     fillPage(&second, 2, 20);
2422     try writer.append(1, 1, &first);
2423     try writer.append(2, 0, &second);
2424 
2425     const committed_len = header_size + frame_size;
2426     var recovered = try testingWriter(std.testing.allocator, 2);
2427     defer std.testing.allocator.free(recovered.deinit());
2428     const recovered_storage = recovered.storage.ptr;
2429     try recovered.load(writer.bytes(), committed_len);
2430 
2431     try std.testing.expectEqual(recovered_storage, recovered.bytes().ptr);
2432     try std.testing.expect(writer.bytes().ptr != recovered.bytes().ptr);
2433     try std.testing.expectEqual(committed_len, recovered.bytes().len);
2434     try std.testing.expectEqual(@as(usize, 1), recovered.frameCount());
2435     try recovered.append(2, 2, &second);
2436     try std.testing.expectEqual(@as(usize, 2), try endMark(recovered.bytes()));
2437 }
2438 
2439 test "wal recovery rejects an invalid prefix before mutation" {
2440     var writer = try testingWriter(std.testing.allocator, 1);
2441     defer std.testing.allocator.free(writer.deinit());
2442     const position = writer.position();
2443     try std.testing.expectError(error.InvalidWal, writer.load("invalid", "invalid".len));
2444     try std.testing.expectEqual(position, writer.position());
2445 }
2446 
2447 test "wal stops recovery before a corrupted frame" {
2448     var writer = try testingWriter(std.testing.allocator, 2);
2449     defer std.testing.allocator.free(writer.deinit());
2450 
2451     var first: [page.size]u8 = undefined;
2452     var second: [page.size]u8 = undefined;
2453     fillPage(&first, 1, 10);
2454     fillPage(&second, 2, 20);
2455     try writer.append(1, 1, &first);
2456     try writer.append(2, 2, &second);
2457     writer.storage[header_size + frame_size + frame_header_size + 9] ^= 0xff;
2458 
2459     const mark = try endMark(writer.bytes());
2460     try std.testing.expectEqual(@as(usize, 1), mark);
2461     try std.testing.expect(try pageAt(writer.bytes(), 2, mark) == null);
2462 }
2463 
2464 test "wal stops recovery at a torn frame" {
2465     var writer = try testingWriter(std.testing.allocator, 2);
2466     defer std.testing.allocator.free(writer.deinit());
2467 
2468     var first: [page.size]u8 = undefined;
2469     var second: [page.size]u8 = undefined;
2470     fillPage(&first, 1, 10);
2471     fillPage(&second, 2, 20);
2472     try writer.append(1, 1, &first);
2473     try writer.append(2, 2, &second);
2474     writer.len -= 17;
2475 
2476     const mark = try endMark(writer.bytes());
2477     try std.testing.expectEqual(@as(usize, 1), mark);
2478 }
2479 
2480 fn sweepWriter() !Writer {
2481     var writer = try testingWriter(std.testing.allocator, 3);
2482     errdefer std.testing.allocator.free(writer.deinit());
2483 
2484     var image: [page.size]u8 = undefined;
2485     fillPage(&image, 1, 10);
2486     try writer.append(1, 0, &image);
2487     fillPage(&image, 2, 20);
2488     try writer.append(2, 2, &image);
2489     fillPage(&image, 1, 30);
2490     try writer.append(1, 0, &image);
2491     return writer;
2492 }
2493 
2494 fn sweepMarkForReadableFrames(readable: usize) usize {
2495     return if (readable >= 2) 2 else 0;
2496 }
2497 
2498 fn expectSweepState(data: []const u8, mark: usize) !void {
2499     switch (mark) {
2500         0 => {
2501             try std.testing.expect(try pageAt(data, 1, mark) == null);
2502             try std.testing.expect(try pageAt(data, 2, mark) == null);
2503         },
2504         2 => {
2505             try std.testing.expectEqual(@as(u8, 10), (try pageAt(data, 1, mark)).?[1]);
2506             try std.testing.expectEqual(@as(u8, 20), (try pageAt(data, 2, mark)).?[1]);
2507         },
2508         else => return error.TestUnexpectedResult,
2509     }
2510 }
2511 
2512 test "wal recovery survives truncation at every byte offset" {
2513     var writer = try sweepWriter();
2514     defer std.testing.allocator.free(writer.deinit());
2515     const data = writer.bytes();
2516 
2517     var length: usize = 0;
2518     while (length <= data.len) : (length += 1) {
2519         const prefix = data[0..length];
2520         if (length < header_size) {
2521             try std.testing.expectError(error.InvalidWal, endMark(prefix));
2522             continue;
2523         }
2524         const readable = (length - header_size) / frame_size;
2525         const mark = try endMark(prefix);
2526         try std.testing.expectEqual(sweepMarkForReadableFrames(readable), mark);
2527         try expectSweepState(prefix, mark);
2528     }
2529 }
2530 
2531 test "wal recovery stops before a flip of any byte" {
2532     comptime {
2533         @stardustClaim(
2534             @import("alloc_phase").capacity.witness(Writer, "sql_wal_corruption"),
2535             null,
2536             null,
2537             null,
2538             null,
2539             null,
2540             null,
2541         );
2542     }
2543 
2544     var writer = try sweepWriter();
2545     defer std.testing.allocator.free(writer.deinit());
2546     const data = writer.storage[0..writer.len];
2547 
2548     var index: usize = 0;
2549     while (index < data.len) : (index += 1) {
2550         data[index] ^= 0xff;
2551         defer data[index] ^= 0xff;
2552 
2553         if (index < header_size) {
2554             if (endMark(data)) |_| {
2555                 return error.TestUnexpectedResult;
2556             } else |err| switch (err) {
2557                 error.InvalidWal, error.InvalidChecksum, error.UnsupportedPageSize => {},
2558             }
2559             continue;
2560         }
2561         const flipped_frame = (index - header_size) / frame_size;
2562         const mark = try endMark(data);
2563         try std.testing.expectEqual(sweepMarkForReadableFrames(flipped_frame), mark);
2564         try expectSweepState(data, mark);
2565     }
2566 }
2567 
2568 test "wal recovery loads exactly the frame-aligned valid prefixes" {
2569     var writer = try sweepWriter();
2570     defer std.testing.allocator.free(writer.deinit());
2571     const data = writer.bytes();
2572     var recovered = try testingWriter(std.testing.allocator, 3);
2573     defer std.testing.allocator.free(recovered.deinit());
2574 
2575     var length: usize = 0;
2576     while (length <= data.len) : (length += 1) {
2577         const aligned = length >= header_size and (length - header_size) % frame_size == 0;
2578         if (!aligned) {
2579             try std.testing.expectError(error.InvalidWal, recovered.load(data, length));
2580             continue;
2581         }
2582         try recovered.load(data, length);
2583         try std.testing.expectEqual((length - header_size) / frame_size, recovered.frameCount());
2584     }
2585 }