lib/sql/src/pager.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 const wal = @import("wal.zig");
6
7 const Allocator = std.mem.Allocator;
8
9 pub const Error = Allocator.Error || wal.Error || error{
10 CapacityOverflow,
11 CheckpointPlanCapacityExceeded,
12 CheckpointPlanInUse,
13 DurableCheckpointReaders,
14 GenerationOverflow,
15 PagerTooLarge,
16 StaleCheckpoint,
17 StorageTooShort,
18 };
19
20 pub const View = struct {
21 base_generation: u64,
22 end_mark: usize,
23 };
24
25 pub const CheckpointReaders = union(enum) {
26 none,
27 oldest: View,
28 };
29
30 pub const CheckpointOptions = struct {
31 readers: CheckpointReaders = .none,
32 restart_header: ?wal.Header = null,
33 };
34
35 pub const Checkpoint = struct {
36 end_mark: usize,
37 pages: usize,
38 base_generation: u64,
39 restarted: bool,
40 };
41
42 pub const Capacity = struct {
43 base_pages: usize = 0,
44 wal_frames: usize = 0,
45 wal_pages: ?usize = null,
46 };
47
48 pub const InitOptions = struct {
49 header: wal.Header,
50 wal_frames: usize,
51 };
52
53 const BaseImage = struct {
54 id: u32,
55 generation: u64,
56 bytes: [page.size]u8,
57 /// The check mark that `markedPageAt` hands to readers. It starts clear
58 /// and clears again when the bytes change in place.
59 checked: bool = false,
60 };
61
62 const WalImage = struct {
63 page_id: u32,
64 frame: usize,
65 offset: usize,
66 previous: ?usize,
67 /// The check mark that `markedPageAt` hands to readers. Frame bytes
68 /// never change while the record exists, since restore and checkpoint
69 /// restart drop the records of the frames they rewrite.
70 checked: bool = false,
71 };
72
73 /// A committed page image and the check mark stored with it.
74 pub const MarkedImage = struct {
75 bytes: *const [page.size]u8,
76 /// Null for an image in the unindexed log tail, which has no record.
77 checked: ?*bool,
78 };
79
80 const ImageLocation = union(enum) {
81 wal: usize,
82 tail: *const [page.size]u8,
83 base: usize,
84 };
85
86 const WalPage = struct {
87 page_id: u32,
88 frame_index: usize,
89 };
90
91 fn FixedList(comptime T: type) type {
92 return struct {
93 items: []T,
94 buffer: []T,
95 capacity: usize,
96
97 fn initBuffer(buffer: []T) @This() {
98 return .{
99 .items = buffer[0..0],
100 .buffer = buffer,
101 .capacity = buffer.len,
102 };
103 }
104
105 fn appendAssumeCapacity(self: *@This(), value: T) void {
106 std.debug.assert(self.items.len < self.capacity);
107 self.buffer[self.items.len] = value;
108 self.items = self.buffer[0 .. self.items.len + 1];
109 }
110
111 fn insertAssumeCapacity(self: *@This(), index: usize, value: T) void {
112 std.debug.assert(index <= self.items.len);
113 std.debug.assert(self.items.len < self.capacity);
114 const next_len = self.items.len + 1;
115 std.mem.copyBackwards(
116 T,
117 self.buffer[index + 1 .. next_len],
118 self.buffer[index..self.items.len],
119 );
120 self.buffer[index] = value;
121 self.items = self.buffer[0..next_len];
122 }
123
124 fn shrinkRetainingCapacity(self: *@This(), len: usize) void {
125 std.debug.assert(len <= self.items.len);
126 self.items = self.buffer[0..len];
127 }
128
129 fn clearRetainingCapacity(self: *@This()) void {
130 self.items = self.buffer[0..0];
131 }
132 };
133 }
134
135 pub const WalIndex = struct {
136 pub const storage_alignment: usize = @max(@alignOf(WalImage), @alignOf(WalPage));
137 pub const Storage = []align(storage_alignment) u8;
138
139 pub const Limits = struct {
140 frames: usize,
141 };
142
143 pub const Capacity = struct {
144 frames: usize,
145 frame_bytes: usize,
146 page_offset: usize,
147 page_bytes: usize,
148 storage_bytes: usize,
149
150 pub const DeriveError = error{CapacityOverflow};
151
152 pub fn derive(limits: Limits) DeriveError!@This() {
153 const frame_bytes = std.math.mul(
154 usize,
155 limits.frames,
156 @sizeOf(WalImage),
157 ) catch return error.CapacityOverflow;
158 const page_offset = try alignForward(frame_bytes, @alignOf(WalPage));
159 const page_bytes = std.math.mul(
160 usize,
161 limits.frames,
162 @sizeOf(WalPage),
163 ) catch return error.CapacityOverflow;
164 return .{
165 .frames = limits.frames,
166 .frame_bytes = frame_bytes,
167 .page_offset = page_offset,
168 .page_bytes = page_bytes,
169 .storage_bytes = std.math.add(
170 usize,
171 page_offset,
172 page_bytes,
173 ) catch return error.CapacityOverflow,
174 };
175 }
176 };
177
178 pub const InitError = WalIndex.Capacity.DeriveError || error{StorageTooShort};
179 pub const Exhaustion = error{WalFull};
180 pub const work_limits: alloc_phase.capacity.WorkLimits = .{
181 .transition_steps_max = std.math.maxInt(usize),
182 .cleanup_steps_per_call_max = 0,
183 .cleanup_calls_at_capacity_max = 0,
184 };
185
186 pub const claim: alloc_phase.capacity.Declaration = .{
187 .source = .{
188 .id = "sql.wal_index",
189 .kind = .phase_static,
190 .limit_source = .caller,
191 .storage = .{
192 .covered = &.{
193 .{
194 .id = "wal_frame_metadata",
195 .lifetime = .steady,
196 .detail = "exclusive mutable loan for one metadata descriptor per retained WAL frame",
197 },
198 .{
199 .id = "wal_page_lookup",
200 .lifetime = .steady,
201 .detail = "exclusive mutable loan for at most one latest-frame lookup per distinct WAL page",
202 },
203 },
204 .excluded = &.{
205 "WAL header and frame bytes owned by sql.wal_writer",
206 "pager base images, base lookup index, and reader snapshots",
207 "checkpoint plans, files, I/O state, and trace instrumentation",
208 },
209 },
210 .capacity = .{
211 .inputs = &.{
212 alloc_phase.capacity.bindInput(Limits, "frames", "frames"),
213 },
214 .type_selectors = &.{
215 alloc_phase.capacity.bindType(WalImage, "walimage"),
216 alloc_phase.capacity.bindType(WalPage, "walpage"),
217 },
218 .nodes = &.{
219 .{ .input = 0 },
220 .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } },
221 .{ .alignment = .{ .node = 1, .alignment = .{ .concrete_type = 1 } } },
222 .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 1 } } },
223 .{ .add = .{ .left = 2, .right = 3 } },
224 },
225 .assertions = &.{.{
226 .scope = .closure_total,
227 .measure = .retained,
228 .relation = .exact,
229 .expression = 4,
230 }},
231 },
232 .overload = .{
233 .kind = .reject_before_mutation,
234 .detail = "checked capacity and short caller storage reject before activation; the parent WAL limit rejects max plus one before either index changes",
235 },
236 .risks = .{
237 .transitive = .{
238 .status = .witnessed,
239 .detail = "append, staged commit, recovery load, restore, and WAL restart use only the activated fixed frame and page regions",
240 },
241 .foreign = .{
242 .status = .excluded,
243 .detail = "the index stores offsets and frame positions while file and operating-system effects remain in the file database",
244 },
245 },
246 .work = .{
247 .equation = "single operations visit at most frames descriptors; staged insertion and rebuild visit at most frames multiplied by frames descriptors",
248 },
249 .obligations = &.{
250 .{ .key = "sql_wal_index_capacity", .role = .capacity_model },
251 .{ .key = "sql_wal_index_storage_rejection", .role = .initialization_failure },
252 .{ .key = "sql_wal_index_sealed_overload", .role = .overload },
253 .{ .key = "sql_wal_index_work_bound", .role = .work_bound },
254 .{ .key = "sql_wal_index_sealed_transitive_risk", .role = .transitive_risk },
255 },
256 },
257 .bindings = .{
258 .owner = @This(),
259 .seal = .{
260 .family = alloc_phase.capacity.selector(@This().activate),
261 .premise = .{
262 .class = .checked_semantic_fact,
263 .authority = .checker,
264 },
265 },
266 .teardown = .{
267 .family = alloc_phase.capacity.selector(@This().deinit),
268 .premise = .{
269 .class = .checked_semantic_fact,
270 .authority = .checker,
271 },
272 },
273 },
274 };
275
276 phase: alloc_phase.capacity.Phase,
277 capacity: WalIndex.Capacity,
278 storage: WalIndex.Storage,
279 frames: FixedList(WalImage),
280 pages: FixedList(WalPage),
281
282 pub fn init(storage: WalIndex.Storage, limits: Limits) InitError!WalIndex {
283 const capacity = try WalIndex.Capacity.derive(limits);
284 if (storage.len < capacity.storage_bytes) return error.StorageTooShort;
285 const borrowed = storage[0..capacity.storage_bytes];
286 return .{
287 .phase = .initialization,
288 .capacity = capacity,
289 .storage = borrowed,
290 .frames = .initBuffer(typedRegion(
291 WalImage,
292 borrowed,
293 0,
294 capacity.frames,
295 )),
296 .pages = .initBuffer(typedRegion(
297 WalPage,
298 borrowed,
299 capacity.page_offset,
300 capacity.frames,
301 )),
302 };
303 }
304
305 pub fn activate(self: *WalIndex) void {
306 std.debug.assert(self.phase == .initialization);
307 self.assertStorage();
308 self.phase = .steady;
309 }
310
311 pub fn reserve(
312 self: *const WalIndex,
313 additional_frames: usize,
314 additional_pages: usize,
315 ) Exhaustion!void {
316 std.debug.assert(self.phase == .steady);
317 if (additional_frames > self.frames.capacity - self.frames.items.len or
318 additional_pages > self.pages.capacity - self.pages.items.len)
319 {
320 return error.WalFull;
321 }
322 }
323
324 pub fn deinit(self: *WalIndex) WalIndex.Storage {
325 std.debug.assert(self.phase != .teardown);
326 self.assertStorage();
327 self.phase = .teardown;
328 const storage = self.storage;
329 self.* = undefined;
330 return storage;
331 }
332
333 fn assertStorage(self: *const WalIndex) void {
334 std.debug.assert(self.storage.len == self.capacity.storage_bytes);
335 std.debug.assert(self.frames.capacity == self.capacity.frames);
336 std.debug.assert(self.pages.capacity == self.capacity.frames);
337 std.debug.assert(self.frames.items.len <= self.frames.capacity);
338 std.debug.assert(self.pages.items.len <= self.pages.capacity);
339 std.debug.assert(self.pages.items.len <= self.frames.items.len);
340 }
341
342 fn typedRegion(
343 comptime T: type,
344 storage: WalIndex.Storage,
345 offset: usize,
346 count: usize,
347 ) []T {
348 const byte_count = count * @sizeOf(T);
349 const bytes: []align(@alignOf(T)) u8 = @alignCast(
350 storage[offset..][0..byte_count],
351 );
352 return std.mem.bytesAsSlice(T, bytes);
353 }
354
355 fn alignForward(value: usize, alignment: usize) error{CapacityOverflow}!usize {
356 std.debug.assert(std.math.isPowerOfTwo(alignment));
357 const mask = alignment - 1;
358 const padded = std.math.add(usize, value, mask) catch
359 return error.CapacityOverflow;
360 return padded & ~mask;
361 }
362 };
363
364 comptime {
365 alloc_phase.capacity.requireProvisionedRejectingOwnerShape(WalIndex);
366 }
367
368 const CheckpointPage = struct {
369 page_id: u32,
370 wal_offset: usize,
371 };
372
373 pub const CheckpointPageView = struct {
374 page_id: u32,
375 bytes: []const u8,
376 };
377
378 pub const CheckpointPlan = struct {
379 pub const storage_alignment: usize = @alignOf(CheckpointPage);
380 pub const Storage = []align(storage_alignment) u8;
381
382 pub const Limits = struct {
383 pages: usize,
384 };
385
386 pub const Capacity = struct {
387 pages: usize,
388 storage_bytes: usize,
389
390 pub const DeriveError = error{CapacityOverflow};
391
392 pub fn derive(limits: Limits) DeriveError!@This() {
393 return .{
394 .pages = limits.pages,
395 .storage_bytes = std.math.mul(usize, limits.pages, @sizeOf(CheckpointPage)) catch return error.CapacityOverflow,
396 };
397 }
398 };
399
400 pub const InitError = CheckpointPlan.Capacity.DeriveError || error{StorageTooShort};
401 pub const Exhaustion = error{
402 CheckpointPlanCapacityExceeded,
403 CheckpointPlanInUse,
404 };
405 pub const work_limits: alloc_phase.capacity.WorkLimits = .{
406 .transition_steps_max = 1,
407 .cleanup_steps_per_call_max = 0,
408 .cleanup_calls_at_capacity_max = 0,
409 };
410
411 pub const claim: alloc_phase.capacity.Declaration = .{
412 .source = .{
413 .id = "sql.checkpoint_plan",
414 .kind = .phase_static,
415 .limit_source = .caller,
416 .storage = .{
417 .covered = &.{
418 .{
419 .id = "reusable_maximum_latest_committed_wal_page_descript_860da2340194",
420 .lifetime = .steady,
421 .detail = "exclusive mutable loan for the reusable maximum latest committed WAL page descriptors for one prepared or synchronous checkpoint",
422 },
423 },
424 .excluded = &.{
425 "pager base images, WAL page bytes, frame and page indexes, and reader snapshots",
426 "file handles, base-file writes, WAL rewrites, and operating-system cache state",
427 "prepared checkpoint metadata, caller state, and trace instrumentation",
428 },
429 },
430 .capacity = .{
431 .inputs = &.{
432 alloc_phase.capacity.bindInput(Limits, "pages", "pages"),
433 },
434 .type_selectors = &.{
435 alloc_phase.capacity.bindType(CheckpointPage, "checkpointpage"),
436 },
437 .nodes = &.{
438 .{ .input = 0 },
439 .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } },
440 },
441 .assertions = &.{.{
442 .scope = .closure_total,
443 .measure = .retained,
444 .relation = .exact,
445 .expression = 1,
446 }},
447 },
448 .overload = .{
449 .kind = .reject_before_mutation,
450 .detail = "checked descriptor arithmetic and short caller storage reject before activation; begin rejects max plus one and concurrent use before clearing the reusable prefix",
451 },
452 .risks = .{
453 .transitive = .{
454 .status = .witnessed,
455 .detail = "WAL page selection and descriptor page views remain allocation-free across repeated prepared and durable checkpoints after plan initialization",
456 },
457 .foreign = .{
458 .status = .excluded,
459 .detail = "base-file writes consume borrowed immutable WAL page bytes and WAL rewrite effects occur outside the plan-owned descriptor claim",
460 },
461 },
462 .work = .{ .equation = "transition_steps <= transition_steps_max" },
463 .obligations = &.{
464 .{ .key = "sql_checkpoint_plan_capacity", .role = .capacity_model },
465 .{ .key = "sql_checkpoint_plan_storage_rejection", .role = .initialization_failure },
466 .{ .key = "sql_checkpoint_plan_sealed_overload", .role = .overload },
467 .{ .key = "sql_checkpoint_plan_work_bound", .role = .work_bound },
468 .{ .key = "sql_checkpoint_plan_sealed_transitive_risk", .role = .transitive_risk },
469 .{ .key = "sql_checkpoint_plan_semantics_transitive_risk", .role = .transitive_risk },
470 .{ .key = "sql_checkpoint_plan_semantics_foreign_risk", .role = .foreign_risk },
471 },
472 },
473 .bindings = .{
474 .owner = @This(),
475 .seal = .{
476 .family = alloc_phase.capacity.selector(@This().activate),
477 .premise = .{
478 .class = .checked_semantic_fact,
479 .authority = .checker,
480 },
481 },
482 .teardown = .{
483 .family = alloc_phase.capacity.selector(@This().deinit),
484 .premise = .{
485 .class = .checked_semantic_fact,
486 .authority = .checker,
487 },
488 },
489 },
490 };
491
492 phase: alloc_phase.capacity.Phase,
493 capacity: CheckpointPlan.Capacity,
494 storage: CheckpointPlan.Storage,
495 filled: usize = 0,
496 admitted: usize = 0,
497 in_use: bool = false,
498
499 pub fn init(storage: CheckpointPlan.Storage, limits: Limits) InitError!CheckpointPlan {
500 const capacity = try CheckpointPlan.Capacity.derive(limits);
501 if (storage.len < capacity.storage_bytes) return error.StorageTooShort;
502 return .{
503 .phase = .initialization,
504 .capacity = capacity,
505 .storage = storage[0..capacity.storage_bytes],
506 };
507 }
508
509 pub fn begin(self: *CheckpointPlan, required_pages: usize) error{
510 CheckpointPlanCapacityExceeded,
511 CheckpointPlanInUse,
512 }!void {
513 std.debug.assert(self.phase == .steady);
514 if (self.in_use) return error.CheckpointPlanInUse;
515 if (required_pages > self.capacity.pages) return error.CheckpointPlanCapacityExceeded;
516 self.filled = 0;
517 self.admitted = required_pages;
518 self.in_use = true;
519 }
520
521 pub fn append(self: *CheckpointPlan, checkpoint_page: CheckpointPage) error{CheckpointPlanCapacityExceeded}!void {
522 std.debug.assert(self.phase == .steady);
523 std.debug.assert(self.in_use);
524 if (self.filled >= self.admitted) return error.CheckpointPlanCapacityExceeded;
525 self.pageStorage()[self.filled] = checkpoint_page;
526 self.filled += 1;
527 }
528
529 pub fn activate(self: *CheckpointPlan) void {
530 std.debug.assert(self.phase == .initialization);
531 std.debug.assert(self.filled == 0);
532 std.debug.assert(!self.in_use);
533 self.phase = .steady;
534 }
535
536 pub fn pages(self: *const CheckpointPlan) []const CheckpointPage {
537 std.debug.assert(self.phase == .steady);
538 std.debug.assert(self.in_use);
539 std.debug.assert(self.filled == self.admitted);
540 return self.pageStorageConst()[0..self.filled];
541 }
542
543 pub fn release(self: *CheckpointPlan) void {
544 std.debug.assert(self.phase == .steady);
545 std.debug.assert(self.in_use);
546 self.filled = 0;
547 self.admitted = 0;
548 self.in_use = false;
549 }
550
551 pub fn deinit(self: *CheckpointPlan) CheckpointPlan.Storage {
552 std.debug.assert(self.phase != .teardown);
553 std.debug.assert(!self.in_use);
554 std.debug.assert(self.storage.len == self.capacity.storage_bytes);
555 self.phase = .teardown;
556 const storage = self.storage;
557 self.* = undefined;
558 return storage;
559 }
560
561 fn pageStorage(self: *CheckpointPlan) []CheckpointPage {
562 return std.mem.bytesAsSlice(CheckpointPage, self.storage);
563 }
564
565 fn pageStorageConst(self: *const CheckpointPlan) []const CheckpointPage {
566 return std.mem.bytesAsSlice(CheckpointPage, self.storage);
567 }
568 };
569
570 comptime {
571 alloc_phase.capacity.requireProvisionedRejectingOwnerShape(CheckpointPlan);
572 }
573
574 const PagerWorkspace = struct {
575 journal: wal.Writer.Storage,
576 checkpoint: CheckpointPlan.Storage,
577 checkpoint_once: CheckpointPlan.Storage,
578 wal_index: WalIndex.Storage,
579
580 pub const Capacity = struct {
581 journal: wal.Writer.Capacity,
582 checkpoint: CheckpointPlan.Capacity,
583 checkpoint_once: CheckpointPlan.Capacity,
584 wal_index: WalIndex.Capacity,
585
586 pub const DeriveError = wal.Writer.Capacity.DeriveError ||
587 CheckpointPlan.Capacity.DeriveError || WalIndex.Capacity.DeriveError;
588
589 pub fn derive(options: InitOptions) DeriveError!@This() {
590 const checkpoint = try CheckpointPlan.Capacity.derive(.{ .pages = options.wal_frames });
591 return .{
592 .journal = try wal.Writer.Capacity.derive(.{
593 .header = options.header,
594 .frames = options.wal_frames,
595 }),
596 .checkpoint = checkpoint,
597 .checkpoint_once = checkpoint,
598 .wal_index = try WalIndex.Capacity.derive(.{
599 .frames = options.wal_frames,
600 }),
601 };
602 }
603 };
604
605 pub const AllocateError = Allocator.Error || PagerWorkspace.Capacity.DeriveError;
606
607 pub fn init(
608 journal: wal.Writer.Storage,
609 checkpoint: CheckpointPlan.Storage,
610 checkpoint_once: CheckpointPlan.Storage,
611 wal_index: WalIndex.Storage,
612 ) PagerWorkspace {
613 return .{
614 .journal = journal,
615 .checkpoint = checkpoint,
616 .checkpoint_once = checkpoint_once,
617 .wal_index = wal_index,
618 };
619 }
620
621 pub fn allocate(allocator: Allocator, options: InitOptions) AllocateError!PagerWorkspace {
622 const capacity = try PagerWorkspace.Capacity.derive(options);
623 const journal = try allocator.alloc(u8, capacity.journal.storage_bytes);
624 errdefer allocator.free(journal);
625 const checkpoint = if (capacity.checkpoint.storage_bytes == 0)
626 @as(CheckpointPlan.Storage, &.{})
627 else
628 try allocator.alignedAlloc(
629 u8,
630 .fromByteUnits(CheckpointPlan.storage_alignment),
631 capacity.checkpoint.storage_bytes,
632 );
633 errdefer if (checkpoint.len != 0) allocator.free(checkpoint);
634 const checkpoint_once = if (capacity.checkpoint_once.storage_bytes == 0)
635 @as(CheckpointPlan.Storage, &.{})
636 else
637 try allocator.alignedAlloc(
638 u8,
639 .fromByteUnits(CheckpointPlan.storage_alignment),
640 capacity.checkpoint_once.storage_bytes,
641 );
642 errdefer if (checkpoint_once.len != 0) allocator.free(checkpoint_once);
643 const wal_index = if (capacity.wal_index.storage_bytes == 0)
644 @as(WalIndex.Storage, &.{})
645 else
646 try allocator.alignedAlloc(
647 u8,
648 .fromByteUnits(WalIndex.storage_alignment),
649 capacity.wal_index.storage_bytes,
650 );
651 return init(journal, checkpoint, checkpoint_once, wal_index);
652 }
653
654 pub fn deallocate(self: *PagerWorkspace, allocator: Allocator) void {
655 if (self.wal_index.len != 0) allocator.free(self.wal_index);
656 if (self.checkpoint_once.len != 0) allocator.free(self.checkpoint_once);
657 if (self.checkpoint.len != 0) allocator.free(self.checkpoint);
658 allocator.free(self.journal);
659 self.* = undefined;
660 }
661 };
662
663 const CheckpointState = struct {
664 position: Pager.Position,
665 base_generation: u64,
666 base_images: usize,
667 wal_pages: usize,
668 };
669
670 pub const PreparedCheckpoint = struct {
671 pager: *const Pager,
672 plan: *CheckpointPlan,
673 checkpoint: Checkpoint,
674 retain_generation: u64,
675 restart_header: ?wal.Header,
676 has_readers: bool,
677 serial: u64,
678 state: CheckpointState,
679
680 pub fn result(self: PreparedCheckpoint) Checkpoint {
681 return self.checkpoint;
682 }
683
684 pub fn checkpointPageCount(self: PreparedCheckpoint) usize {
685 return self.plan.pages().len;
686 }
687
688 pub fn checkpointPage(self: PreparedCheckpoint, index: usize) CheckpointPageView {
689 std.debug.assert(self.plan.phase == .steady);
690 std.debug.assert(std.meta.eql(self.pager.position(), self.state.position));
691 const checkpoint_page = self.plan.pages()[index];
692 const bytes = self.pager.walBytes();
693 std.debug.assert(checkpoint_page.wal_offset <= bytes.len);
694 std.debug.assert(page.size <= bytes.len - checkpoint_page.wal_offset);
695 return .{
696 .page_id = checkpoint_page.page_id,
697 .bytes = bytes[checkpoint_page.wal_offset..][0..page.size],
698 };
699 }
700
701 pub fn deinit(self: *PreparedCheckpoint) void {
702 self.plan.release();
703 self.* = undefined;
704 }
705 };
706
707 pub const Storage = struct {
708 base_images: usize,
709 base_capacity: usize,
710 wal_frames: usize,
711 wal_frame_capacity: usize,
712 wal_pages: usize,
713 wal_page_capacity: usize,
714 };
715
716 pub const Pager = struct {
717 pub const Workspace = PagerWorkspace;
718
719 allocator: Allocator,
720 base: std.ArrayList(BaseImage) = .empty,
721 base_index: std.AutoHashMapUnmanaged(u32, usize) = .empty,
722 wal_index: WalIndex,
723 journal: wal.Writer,
724 checkpoint_plan: CheckpointPlan,
725 checkpoint_once_plan: CheckpointPlan,
726 base_generation: u64 = 0,
727 base_page_count: u32 = 0,
728 database_page_count: u32 = 0,
729 end_mark: usize = 0,
730 checkpoint_serial: u64 = 0,
731
732 pub const Position = struct {
733 journal: wal.Writer.Position,
734 frames_len: usize,
735 database_page_count: u32,
736 end_mark: usize,
737 };
738
739 pub const RestoreEpoch = struct {
740 base_generation: u64,
741 base_page_count: u32,
742 checkpoint_serial: u64,
743 };
744
745 pub fn init(allocator: Allocator, workspace: *Workspace, options: InitOptions) Error!Pager {
746 const journal_limits: wal.Writer.Limits = .{
747 .header = options.header,
748 .frames = options.wal_frames,
749 };
750 const workspace_capacity = try Workspace.Capacity.derive(options);
751 if (workspace.journal.len < workspace_capacity.journal.storage_bytes or
752 workspace.checkpoint.len < workspace_capacity.checkpoint.storage_bytes or
753 workspace.checkpoint_once.len < workspace_capacity.checkpoint_once.storage_bytes or
754 workspace.wal_index.len < workspace_capacity.wal_index.storage_bytes)
755 {
756 return error.StorageTooShort;
757 }
758 var journal = wal.Writer.init(workspace.journal, journal_limits) catch |err| switch (err) {
759 error.CapacityOverflow, error.StorageTooShort => unreachable,
760 };
761 var checkpoint_plan = CheckpointPlan.init(workspace.checkpoint, .{
762 .pages = options.wal_frames,
763 }) catch |err| switch (err) {
764 error.CapacityOverflow, error.StorageTooShort => unreachable,
765 };
766 var checkpoint_once_plan = CheckpointPlan.init(workspace.checkpoint_once, .{
767 .pages = options.wal_frames,
768 }) catch |err| switch (err) {
769 error.CapacityOverflow, error.StorageTooShort => unreachable,
770 };
771 var wal_index = WalIndex.init(workspace.wal_index, .{
772 .frames = options.wal_frames,
773 }) catch |err| switch (err) {
774 error.CapacityOverflow, error.StorageTooShort => unreachable,
775 };
776 journal.activate();
777 checkpoint_plan.activate();
778 checkpoint_once_plan.activate();
779 wal_index.activate();
780 workspace.* = undefined;
781 return .{
782 .allocator = allocator,
783 .wal_index = wal_index,
784 .journal = journal,
785 .checkpoint_plan = checkpoint_plan,
786 .checkpoint_once_plan = checkpoint_once_plan,
787 };
788 }
789
790 pub fn deinit(self: *Pager) Workspace {
791 self.base.deinit(self.allocator);
792 self.base_index.deinit(self.allocator);
793 const workspace: Workspace = .{
794 .journal = self.journal.deinit(),
795 .checkpoint = self.checkpoint_plan.deinit(),
796 .checkpoint_once = self.checkpoint_once_plan.deinit(),
797 .wal_index = self.wal_index.deinit(),
798 };
799 self.* = undefined;
800 return workspace;
801 }
802
803 pub fn reserve(self: *Pager, capacity: Capacity) Error!void {
804 const phase = trace.scope("pager.reserve");
805 defer phase.end();
806 if (capacity.wal_frames > self.journal.remainingFrames()) return error.WalFull;
807 const wal_pages = capacity.wal_pages orelse capacity.wal_frames;
808 try self.wal_index.reserve(capacity.wal_frames, wal_pages);
809 try self.base.ensureTotalCapacityPrecise(
810 self.allocator,
811 try additionalCapacity(self.base.items.len, capacity.base_pages),
812 );
813 try self.base_index.ensureUnusedCapacity(self.allocator, try hashMapSize(capacity.base_pages));
814 }
815
816 pub fn replaceWal(self: *Pager, bytes: []const u8, committed_len: usize) Error!void {
817 return self.replaceWalControlled(bytes, committed_len, .{}) catch |err| switch (err) {
818 error.Interrupted => unreachable,
819 else => return @errorCast(err),
820 };
821 }
822
823 pub fn replaceWalControlled(
824 self: *Pager,
825 bytes: []const u8,
826 committed_len: usize,
827 control: wal.Control,
828 ) (Error || error{Interrupted})!void {
829 const phase = trace.scope("pager.replace_wal");
830 defer phase.end();
831 if (committed_len < wal.header_size or (committed_len - wal.header_size) % wal.frame_size != 0) return error.InvalidWal;
832 const frames_count = (committed_len - wal.header_size) / wal.frame_size;
833 if (frames_count > self.journal.frameCapacity()) return error.WalFull;
834 std.debug.assert(frames_count <= self.wal_index.frames.capacity);
835 std.debug.assert(frames_count <= self.wal_index.pages.capacity);
836 try self.journal.loadControlled(bytes, committed_len, control);
837 try self.rebuildWalFramesFromJournalControlled(control);
838 trace.progress("pager.replace_wal.complete");
839 }
840
841 pub fn installBase(self: *Pager, page_id: u32, image: *const [page.size]u8) Error!void {
842 const phase = trace.scope("pager.install_base");
843 defer phase.end();
844 const generation = try self.nextBaseGeneration();
845 try self.installBaseAtGeneration(page_id, image, generation);
846 trace.progress("pager.install_base.complete");
847 }
848
849 pub fn installBaseAtGeneration(self: *Pager, page_id: u32, image: *const [page.size]u8, generation: u64) Error!void {
850 const phase = trace.scope("pager.install_base.generation");
851 defer phase.end();
852 if (self.base_index.get(page_id)) |index| {
853 const cached = &self.base.items[index];
854 if (cached.id == page_id and cached.generation == generation) {
855 cached.bytes = image.*;
856 cached.checked = false;
857 self.base_generation = @max(self.base_generation, generation);
858 self.base_page_count = @max(self.base_page_count, page_id);
859 self.database_page_count = @max(self.database_page_count, page_id);
860 trace.progress("pager.install_base.generation.complete");
861 return;
862 }
863 }
864 try self.base.ensureUnusedCapacity(self.allocator, 1);
865 try self.base_index.ensureUnusedCapacity(self.allocator, 1);
866 const index = self.base.items.len;
867 self.base.appendAssumeCapacity(.{
868 .id = page_id,
869 .generation = generation,
870 .bytes = image.*,
871 });
872 if (self.base_index.get(page_id)) |existing| {
873 if (self.base.items[existing].generation <= generation) self.base_index.putAssumeCapacity(page_id, index);
874 } else {
875 self.base_index.putAssumeCapacity(page_id, index);
876 }
877 self.base_generation = @max(self.base_generation, generation);
878 self.base_page_count = @max(self.base_page_count, page_id);
879 self.database_page_count = @max(self.database_page_count, page_id);
880 trace.progress("pager.install_base.generation.complete");
881 }
882
883 pub fn appendWal(self: *Pager, page_id: u32, db_page_count: u32, image: *const [page.size]u8) Error!void {
884 const phase = trace.scope("pager.append_wal");
885 defer phase.end();
886 if (self.journal.remainingFrames() == 0) return error.WalFull;
887 const page_index = self.lowerBoundWalPage(page_id);
888 const existing = page_index < self.wal_index.pages.items.len and
889 self.wal_index.pages.items[page_index].page_id == page_id;
890 try self.wal_index.reserve(1, @intFromBool(!existing));
891 try self.journal.append(page_id, db_page_count, image);
892 self.indexAppendedWalFrame(page_id, db_page_count, page_index, existing);
893 trace.progress("pager.append_wal.complete");
894 }
895
896 pub fn walStagingCapacity(self: *const Pager, position_value: Position) usize {
897 std.debug.assert(std.meta.eql(self.position(), position_value));
898 return self.journal.remainingFrames();
899 }
900
901 pub fn stageWalPage(self: *Pager, position_value: Position, index: usize, page_id: u32, image: *const [page.size]u8) Error!void {
902 std.debug.assert(std.meta.eql(self.position(), position_value));
903 try self.journal.stagePage(position_value.journal, index, page_id, image);
904 }
905
906 pub fn stagedWalPage(self: *const Pager, position_value: Position, index: usize) []const u8 {
907 std.debug.assert(std.meta.eql(self.position(), position_value));
908 return self.journal.stagedPage(position_value.journal, index);
909 }
910
911 pub fn stagedWalPageMut(self: *Pager, position_value: Position, index: usize) *[page.size]u8 {
912 std.debug.assert(std.meta.eql(self.position(), position_value));
913 return self.journal.stagedPageMut(position_value.journal, index);
914 }
915
916 pub fn stagedWalPageId(self: *const Pager, position_value: Position, index: usize) u32 {
917 return self.journal.stagedPageId(position_value.journal, index);
918 }
919
920 pub fn swapStagedWalFrames(self: *Pager, position_value: Position, left_index: usize, right_index: usize) void {
921 std.debug.assert(std.meta.eql(self.position(), position_value));
922 self.journal.swapStagedFrames(position_value.journal, left_index, right_index);
923 }
924
925 pub fn commitStagedWal(self: *Pager, position_value: Position, count: usize, database_page_count: u32) Error!void {
926 const phase = trace.scope("pager.commit_staged_wal");
927 defer phase.end();
928 std.debug.assert(std.meta.eql(self.position(), position_value));
929 if (count > self.walStagingCapacity(position_value)) return error.WalFull;
930
931 var new_pages: usize = 0;
932 var previous_page_id: u32 = 0;
933 for (0..count) |index| {
934 const page_id = self.stagedWalPageId(position_value, index);
935 std.debug.assert(page_id != 0);
936 if (index != 0) std.debug.assert(page_id > previous_page_id);
937 previous_page_id = page_id;
938 const page_index = self.lowerBoundWalPage(page_id);
939 if (page_index == self.wal_index.pages.items.len or
940 self.wal_index.pages.items[page_index].page_id != page_id)
941 {
942 new_pages += 1;
943 }
944 }
945
946 try self.wal_index.reserve(count, new_pages);
947 for (0..count) |index| {
948 const page_id = self.stagedWalPageId(position_value, index);
949 const page_index = self.lowerBoundWalPage(page_id);
950 const existing = page_index < self.wal_index.pages.items.len and
951 self.wal_index.pages.items[page_index].page_id == page_id;
952 self.journal.commitStagedFrame(position_value.journal, index, if (index + 1 == count) database_page_count else 0);
953 self.indexAppendedWalFrame(page_id, if (index + 1 == count) database_page_count else 0, page_index, existing);
954 }
955 trace.progress("pager.commit_staged_wal.complete");
956 }
957
958 pub fn beginRead(self: *const Pager) Error!Snapshot {
959 const phase = trace.scope("pager.begin_read");
960 defer phase.end();
961 return .{ .pager = self, .view = try self.currentView() };
962 }
963
964 pub fn checkpoint(self: *Pager, options: CheckpointOptions) Error!Checkpoint {
965 const phase = trace.scope("pager.checkpoint");
966 defer phase.end();
967
968 var prepared = try self.prepareCheckpointWithPlan(
969 &self.checkpoint_once_plan,
970 options,
971 );
972 defer prepared.deinit();
973 return try self.commitCheckpoint(prepared);
974 }
975
976 pub fn prepareCheckpoint(self: *Pager, options: CheckpointOptions) Error!PreparedCheckpoint {
977 return try self.prepareCheckpointWithPlan(&self.checkpoint_plan, options);
978 }
979
980 fn prepareCheckpointWithPlan(
981 self: *Pager,
982 plan: *CheckpointPlan,
983 options: CheckpointOptions,
984 ) Error!PreparedCheckpoint {
985 const phase = trace.scope("pager.checkpoint.prepare");
986 defer phase.end();
987
988 const current = try self.currentView();
989 const has_readers = switch (options.readers) {
990 .none => false,
991 .oldest => true,
992 };
993 const target_mark = switch (options.readers) {
994 .none => current.end_mark,
995 .oldest => |oldest| @min(oldest.end_mark, current.end_mark),
996 };
997 const can_rewrite_wal = options.restart_header != null and !has_readers;
998 if (can_rewrite_wal) {
999 const retained_frames = self.frameCount() - target_mark;
1000 std.debug.assert(retained_frames <= self.wal_index.frames.capacity);
1001 std.debug.assert(retained_frames <= self.wal_index.pages.capacity);
1002 }
1003
1004 const bytes = self.walBytes();
1005 try plan.begin(self.checkpointPageCount(target_mark, bytes.len));
1006 errdefer plan.release();
1007 const serial = std.math.add(u64, self.checkpoint_serial, 1) catch return error.GenerationOverflow;
1008 _ = std.math.add(u64, serial, 1) catch return error.GenerationOverflow;
1009 self.checkpoint_serial = serial;
1010 try self.collectCheckpointPages(target_mark, bytes.len, plan);
1011 const pages = plan.pages().len;
1012
1013 var generation = self.base_generation;
1014 if (pages > 0) generation = try self.nextBaseGeneration();
1015 const retain_generation = switch (options.readers) {
1016 .none => generation,
1017 .oldest => |oldest| @min(oldest.base_generation, generation),
1018 };
1019 return .{
1020 .pager = self,
1021 .plan = plan,
1022 .checkpoint = .{
1023 .end_mark = target_mark,
1024 .pages = pages,
1025 .base_generation = generation,
1026 .restarted = can_rewrite_wal,
1027 },
1028 .retain_generation = retain_generation,
1029 .restart_header = options.restart_header,
1030 .has_readers = has_readers,
1031 .serial = serial,
1032 .state = .{
1033 .position = self.position(),
1034 .base_generation = self.base_generation,
1035 .base_images = self.base.items.len,
1036 .wal_pages = self.wal_index.pages.items.len,
1037 },
1038 };
1039 }
1040
1041 pub fn commitCheckpoint(self: *Pager, prepared: PreparedCheckpoint) Error!Checkpoint {
1042 const phase = trace.scope("pager.checkpoint.commit_memory");
1043 defer phase.end();
1044 if (!self.preparedCheckpointCurrent(prepared)) return error.StaleCheckpoint;
1045
1046 const checkpoint_value = prepared.checkpoint;
1047 if (checkpoint_value.pages > 0) {
1048 try self.installCheckpointPages(prepared, checkpoint_value.base_generation);
1049 self.base_generation = checkpoint_value.base_generation;
1050 }
1051 _ = try self.compactBaseHistory(prepared.retain_generation);
1052 self.finishCheckpointFrames(prepared);
1053 self.checkpoint_serial = prepared.serial + 1;
1054
1055 trace.progress("pager.checkpoint.complete");
1056 return checkpoint_value;
1057 }
1058
1059 pub fn commitDurableCheckpoint(self: *Pager, prepared: PreparedCheckpoint) Error!Checkpoint {
1060 const phase = trace.scope("pager.checkpoint.commit_durable");
1061 defer phase.end();
1062 if (!self.preparedCheckpointCurrent(prepared)) return error.StaleCheckpoint;
1063 if (prepared.has_readers) return error.DurableCheckpointReaders;
1064
1065 const checkpoint_value = prepared.checkpoint;
1066 if (checkpoint_value.pages > 0) {
1067 self.base_generation = checkpoint_value.base_generation;
1068 for (prepared.plan.pages()) |checkpoint_page| {
1069 self.base_page_count = @max(self.base_page_count, checkpoint_page.page_id);
1070 self.database_page_count = @max(self.database_page_count, checkpoint_page.page_id);
1071 }
1072 }
1073 self.releaseDurableBase();
1074 self.finishCheckpointFrames(prepared);
1075 self.checkpoint_serial = prepared.serial + 1;
1076
1077 trace.progress("pager.checkpoint.complete");
1078 return checkpoint_value;
1079 }
1080
1081 pub fn storage(self: *const Pager) Storage {
1082 return .{
1083 .base_images = self.base.items.len,
1084 .base_capacity = self.base.capacity,
1085 .wal_frames = self.wal_index.frames.items.len,
1086 .wal_frame_capacity = self.wal_index.frames.capacity,
1087 .wal_pages = self.wal_index.pages.items.len,
1088 .wal_page_capacity = self.wal_index.pages.capacity,
1089 };
1090 }
1091
1092 pub fn releaseDurableBase(self: *Pager) void {
1093 self.base.deinit(self.allocator);
1094 self.base_index.deinit(self.allocator);
1095 self.base = .empty;
1096 self.base_index = .empty;
1097 }
1098
1099 pub fn currentView(self: *const Pager) Error!View {
1100 return .{
1101 .base_generation = self.base_generation,
1102 .end_mark = self.end_mark,
1103 };
1104 }
1105
1106 pub fn pageAt(self: *const Pager, page_id: u32, view: View) Error!?[]const u8 {
1107 const location = try self.locateImage(page_id, view) orelse return null;
1108 return switch (location) {
1109 .wal => |index| self.walImageBytes(index),
1110 .tail => |bytes| bytes,
1111 .base => |index| self.base.items[index].bytes[0..],
1112 };
1113 }
1114
1115 /// Returns the image `pageAt` returns, with the check mark stored with
1116 /// it. The pager only clears a mark, when it stores new bytes under it.
1117 /// A reader sets it after the image passes the reader's checks, so later
1118 /// readers can skip them.
1119 pub fn markedPageAt(self: *Pager, page_id: u32, view: View) Error!?MarkedImage {
1120 const location = try self.locateImage(page_id, view) orelse return null;
1121 return switch (location) {
1122 .wal => |index| .{
1123 .bytes = self.walImageBytes(index),
1124 .checked = &self.wal_index.frames.items[index].checked,
1125 },
1126 .tail => |bytes| .{ .bytes = bytes, .checked = null },
1127 .base => |index| .{
1128 .bytes = &self.base.items[index].bytes,
1129 .checked = &self.base.items[index].checked,
1130 },
1131 };
1132 }
1133
1134 fn locateImage(self: *const Pager, page_id: u32, view: View) Error!?ImageLocation {
1135 const phase = trace.scope("pager.page_at");
1136 defer phase.end();
1137 if (self.indexedWalImage(page_id, view.end_mark)) |index| return .{ .wal = index };
1138 if (view.end_mark > self.frameCount()) {
1139 if (try wal.pageAt(self.walBytes(), page_id, view.end_mark)) |bytes| {
1140 return .{ .tail = bytes[0..page.size] };
1141 }
1142 }
1143 if (self.baseVisibleIndex(page_id, view.base_generation)) |index| return .{ .base = index };
1144 return null;
1145 }
1146
1147 fn walImageBytes(self: *const Pager, index: usize) *const [page.size]u8 {
1148 return self.walBytes()[self.wal_index.frames.items[index].offset..][0..page.size];
1149 }
1150
1151 pub fn frameCount(self: *const Pager) usize {
1152 return self.journal.frameCount();
1153 }
1154
1155 pub fn position(self: *const Pager) Position {
1156 return .{
1157 .journal = self.journal.position(),
1158 .frames_len = self.wal_index.frames.items.len,
1159 .database_page_count = self.database_page_count,
1160 .end_mark = self.end_mark,
1161 };
1162 }
1163
1164 pub fn restoreEpoch(self: *const Pager) RestoreEpoch {
1165 return .{
1166 .base_generation = self.base_generation,
1167 .base_page_count = self.base_page_count,
1168 .checkpoint_serial = self.checkpoint_serial,
1169 };
1170 }
1171
1172 pub fn canRestore(
1173 self: *const Pager,
1174 position_value: Position,
1175 epoch: RestoreEpoch,
1176 ) bool {
1177 if (!std.meta.eql(self.restoreEpoch(), epoch)) return false;
1178 if (self.wal_index.frames.items.len < position_value.frames_len) return false;
1179 if (self.journal.position().len < position_value.journal.len) return false;
1180 return true;
1181 }
1182
1183 pub fn restore(self: *Pager, position_value: Position) void {
1184 self.journal.restore(position_value.journal);
1185 self.wal_index.frames.shrinkRetainingCapacity(position_value.frames_len);
1186 self.database_page_count = position_value.database_page_count;
1187 self.end_mark = position_value.end_mark;
1188 self.rebuildWalPages();
1189 }
1190
1191 pub fn walBytes(self: *const Pager) []const u8 {
1192 return self.journal.bytes();
1193 }
1194
1195 pub fn walCapacityBytes(self: *const Pager) usize {
1196 return self.journal.byteCapacity();
1197 }
1198
1199 pub fn baseGeneration(self: *const Pager) u64 {
1200 return self.base_generation;
1201 }
1202
1203 pub fn setBasePageCount(self: *Pager, count: u32) void {
1204 self.base_page_count = @max(self.base_page_count, count);
1205 self.database_page_count = @max(self.database_page_count, count);
1206 if (count > 0 and self.base_generation == 0) self.base_generation = 1;
1207 }
1208
1209 pub fn basePageCount(self: *const Pager) u32 {
1210 return self.base_page_count;
1211 }
1212
1213 pub fn databasePageCount(self: *const Pager) u32 {
1214 return self.database_page_count;
1215 }
1216
1217 fn lowerBoundWalPage(self: *const Pager, page_id: u32) usize {
1218 var low: usize = 0;
1219 var high = self.wal_index.pages.items.len;
1220 while (low < high) {
1221 const mid = low + (high - low) / 2;
1222 if (self.wal_index.pages.items[mid].page_id < page_id) {
1223 low = mid + 1;
1224 } else {
1225 high = mid;
1226 }
1227 }
1228 return low;
1229 }
1230
1231 fn indexAppendedWalFrame(self: *Pager, page_id: u32, db_page_count: u32, page_index: usize, existing: bool) void {
1232 const frame = self.journal.frameCount();
1233 if (db_page_count != 0) self.end_mark = frame;
1234 self.database_page_count = @max(self.database_page_count, @max(page_id, db_page_count));
1235 const frame_index = self.wal_index.frames.items.len;
1236 self.wal_index.frames.appendAssumeCapacity(.{
1237 .page_id = page_id,
1238 .frame = frame,
1239 .offset = self.walBytes().len - page.size,
1240 .previous = if (existing) self.wal_index.pages.items[page_index].frame_index else null,
1241 });
1242 if (existing) {
1243 self.wal_index.pages.items[page_index].frame_index = frame_index;
1244 } else {
1245 self.wal_index.pages.insertAssumeCapacity(page_index, .{
1246 .page_id = page_id,
1247 .frame_index = frame_index,
1248 });
1249 }
1250 }
1251
1252 fn lowerBoundBase(self: *const Pager, page_id: u32, generation: u64) usize {
1253 var low: usize = 0;
1254 var high = self.base.items.len;
1255 while (low < high) {
1256 const mid = low + (high - low) / 2;
1257 const image = self.base.items[mid];
1258 if (image.id < page_id or (image.id == page_id and image.generation < generation)) {
1259 low = mid + 1;
1260 } else {
1261 high = mid;
1262 }
1263 }
1264 return low;
1265 }
1266
1267 fn baseVisibleIndex(self: *const Pager, page_id: u32, generation: u64) ?usize {
1268 if (self.base_index.get(page_id)) |index| {
1269 const image = &self.base.items[index];
1270 if (image.id == page_id and image.generation <= generation) return index;
1271 }
1272
1273 var best_index: ?usize = null;
1274 var best_generation: u64 = 0;
1275 for (self.base.items, 0..) |*image, index| {
1276 if (image.id == page_id and image.generation <= generation and (best_index == null or image.generation > best_generation)) {
1277 best_index = index;
1278 best_generation = image.generation;
1279 }
1280 }
1281 return best_index;
1282 }
1283
1284 fn upperBoundBase(self: *const Pager, page_id: u32, generation: u64) usize {
1285 var low: usize = 0;
1286 var high = self.base.items.len;
1287 while (low < high) {
1288 const mid = low + (high - low) / 2;
1289 const image = self.base.items[mid];
1290 if (image.id < page_id or (image.id == page_id and image.generation <= generation)) {
1291 low = mid + 1;
1292 } else {
1293 high = mid;
1294 }
1295 }
1296 return low;
1297 }
1298
1299 fn indexedWalImage(self: *const Pager, page_id: u32, max_frame: usize) ?usize {
1300 const page_index = self.lowerBoundWalPage(page_id);
1301 if (page_index == self.wal_index.pages.items.len or
1302 self.wal_index.pages.items[page_index].page_id != page_id)
1303 {
1304 return null;
1305 }
1306
1307 const wal_len = self.walBytes().len;
1308 var frame_index: ?usize = self.wal_index.pages.items[page_index].frame_index;
1309 while (frame_index) |index| {
1310 const image = self.wal_index.frames.items[index];
1311 if (image.frame <= max_frame and image.offset + page.size <= wal_len) return index;
1312 frame_index = image.previous;
1313 }
1314 return null;
1315 }
1316
1317 fn latestCheckpointFrame(self: *const Pager, wal_page: WalPage, target_mark: usize, wal_len: usize) ?WalImage {
1318 var frame_index: ?usize = wal_page.frame_index;
1319 while (frame_index) |index| {
1320 const image = self.wal_index.frames.items[index];
1321 if (image.frame <= target_mark and image.offset + page.size <= wal_len) return image;
1322 frame_index = image.previous;
1323 }
1324 return null;
1325 }
1326
1327 fn checkpointPageCount(self: *const Pager, target_mark: usize, wal_len: usize) usize {
1328 var count: usize = 0;
1329 for (self.wal_index.pages.items) |wal_page| {
1330 if (self.latestCheckpointFrame(wal_page, target_mark, wal_len) != null) count += 1;
1331 }
1332 return count;
1333 }
1334
1335 fn collectCheckpointPages(
1336 self: *const Pager,
1337 target_mark: usize,
1338 wal_len: usize,
1339 plan: *CheckpointPlan,
1340 ) error{CheckpointPlanCapacityExceeded}!void {
1341 for (self.wal_index.pages.items) |wal_page| {
1342 const image = self.latestCheckpointFrame(wal_page, target_mark, wal_len) orelse continue;
1343 try plan.append(.{
1344 .page_id = image.page_id,
1345 .wal_offset = image.offset,
1346 });
1347 }
1348 }
1349
1350 fn installCheckpointPages(self: *Pager, prepared: PreparedCheckpoint, generation: u64) Error!void {
1351 const checkpoint_page_count = prepared.checkpointPageCount();
1352 try self.base.ensureUnusedCapacity(self.allocator, checkpoint_page_count);
1353 try self.base_index.ensureUnusedCapacity(self.allocator, try hashMapSize(checkpoint_page_count));
1354 for (0..checkpoint_page_count) |page_index| {
1355 const checkpoint_page = prepared.checkpointPage(page_index);
1356 const index = self.base.items.len;
1357 self.base.appendAssumeCapacity(.{
1358 .id = checkpoint_page.page_id,
1359 .generation = generation,
1360 .bytes = checkpoint_page.bytes[0..page.size].*,
1361 });
1362 if (self.base_index.get(checkpoint_page.page_id)) |existing| {
1363 if (self.base.items[existing].generation <= generation) self.base_index.putAssumeCapacity(checkpoint_page.page_id, index);
1364 } else {
1365 self.base_index.putAssumeCapacity(checkpoint_page.page_id, index);
1366 }
1367 self.base_page_count = @max(self.base_page_count, checkpoint_page.page_id);
1368 self.database_page_count = @max(self.database_page_count, checkpoint_page.page_id);
1369 }
1370 }
1371
1372 fn preparedCheckpointCurrent(self: *const Pager, prepared: PreparedCheckpoint) bool {
1373 if (prepared.pager != self) return false;
1374 if (self.checkpoint_serial != prepared.serial) return false;
1375 if (!std.meta.eql(self.position(), prepared.state.position)) return false;
1376 if (self.base_generation != prepared.state.base_generation) return false;
1377 if (self.base.items.len != prepared.state.base_images) return false;
1378 if (self.wal_index.pages.items.len != prepared.state.wal_pages) return false;
1379 return true;
1380 }
1381
1382 fn finishCheckpointFrames(self: *Pager, prepared: PreparedCheckpoint) void {
1383 const checkpoint_value = prepared.checkpoint;
1384 if (!prepared.has_readers and !checkpoint_value.restarted) _ = self.compactWalFrames(checkpoint_value.end_mark);
1385 if (checkpoint_value.restarted) {
1386 self.journal.rewriteTail(checkpoint_value.end_mark, prepared.restart_header.?);
1387 self.rebuildWalFramesFromJournal();
1388 }
1389 }
1390
1391 fn compactBaseHistory(self: *Pager, retain_generation: u64) Error!usize {
1392 const phase = trace.scope("pager.base_history.compact");
1393 defer phase.end();
1394 var retained_floor: std.AutoHashMapUnmanaged(u32, usize) = .empty;
1395 defer retained_floor.deinit(self.allocator);
1396 try retained_floor.ensureTotalCapacity(self.allocator, try hashMapSize(self.base.items.len));
1397 for (self.base.items, 0..) |*image, index| {
1398 if (image.generation > retain_generation) continue;
1399 if (retained_floor.get(image.id)) |existing| {
1400 if (self.base.items[existing].generation < image.generation) retained_floor.putAssumeCapacity(image.id, index);
1401 } else {
1402 retained_floor.putAssumeCapacity(image.id, index);
1403 }
1404 }
1405
1406 var write_index: usize = 0;
1407 for (self.base.items, 0..) |image, index| {
1408 const keep = image.generation > retain_generation or (retained_floor.get(image.id) orelse std.math.maxInt(usize)) == index;
1409 if (keep) {
1410 self.base.items[write_index] = image;
1411 write_index += 1;
1412 }
1413 }
1414 const removed = self.base.items.len - write_index;
1415 self.base.shrinkRetainingCapacity(write_index);
1416 try self.rebuildBaseIndex();
1417 if (removed > 0) trace.progress("pager.base_history.compact.complete");
1418 return removed;
1419 }
1420
1421 fn compactWalFrames(self: *Pager, checkpoint_mark: usize) usize {
1422 const phase = trace.scope("pager.wal_frames.compact");
1423 defer phase.end();
1424 var write_index: usize = 0;
1425 for (self.wal_index.frames.items) |image| {
1426 if (image.frame > checkpoint_mark) {
1427 self.wal_index.frames.items[write_index] = image;
1428 write_index += 1;
1429 }
1430 }
1431 const removed = self.wal_index.frames.items.len - write_index;
1432 self.wal_index.frames.shrinkRetainingCapacity(write_index);
1433 if (removed > 0) {
1434 self.rebuildWalPages();
1435 trace.progress("pager.wal_frames.compact.complete");
1436 }
1437 return removed;
1438 }
1439
1440 fn rebuildWalFramesFromJournal(self: *Pager) void {
1441 self.rebuildWalFramesFromJournalControlled(.{}) catch unreachable;
1442 }
1443
1444 fn rebuildWalFramesFromJournalControlled(
1445 self: *Pager,
1446 control: wal.Control,
1447 ) error{Interrupted}!void {
1448 self.wal_index.frames.clearRetainingCapacity();
1449 self.wal_index.pages.clearRetainingCapacity();
1450 self.database_page_count = self.base_page_count;
1451 self.end_mark = 0;
1452 var reader = wal.Reader.initControlled(self.walBytes(), control) catch |err| switch (err) {
1453 error.Interrupted => return error.Interrupted,
1454 else => unreachable,
1455 };
1456 const frames_max = self.journal.frameCount();
1457 var frame_index: usize = 0;
1458 while (frame_index < frames_max) : (frame_index += 1) {
1459 const frame = (reader.nextControlled(control) catch |err| switch (err) {
1460 error.Interrupted => return error.Interrupted,
1461 else => unreachable,
1462 }) orelse unreachable;
1463 self.wal_index.frames.appendAssumeCapacity(.{
1464 .page_id = frame.page_id,
1465 .frame = frame.index,
1466 .offset = wal.header_size + (frame.index - 1) * wal.frame_size + wal.frame_header_size,
1467 .previous = null,
1468 });
1469 self.database_page_count = @max(self.database_page_count, @max(frame.page_id, frame.db_page_count));
1470 if (frame.committed()) self.end_mark = frame.index;
1471 }
1472 try self.rebuildWalPagesControlled(control);
1473 }
1474
1475 fn rebuildWalPages(self: *Pager) void {
1476 self.rebuildWalPagesControlled(.{}) catch unreachable;
1477 }
1478
1479 fn rebuildWalPagesControlled(
1480 self: *Pager,
1481 control: wal.Control,
1482 ) error{Interrupted}!void {
1483 self.wal_index.pages.clearRetainingCapacity();
1484 for (self.wal_index.frames.items, 0..) |*image, index| {
1485 try control.check();
1486 const page_index = self.lowerBoundWalPage(image.page_id);
1487 if (page_index < self.wal_index.pages.items.len and
1488 self.wal_index.pages.items[page_index].page_id == image.page_id)
1489 {
1490 image.previous = self.wal_index.pages.items[page_index].frame_index;
1491 self.wal_index.pages.items[page_index].frame_index = index;
1492 } else {
1493 image.previous = null;
1494 self.wal_index.pages.insertAssumeCapacity(page_index, .{
1495 .page_id = image.page_id,
1496 .frame_index = index,
1497 });
1498 }
1499 }
1500 try control.check();
1501 }
1502
1503 fn rebuildBaseIndex(self: *Pager) Error!void {
1504 try self.base_index.ensureTotalCapacity(self.allocator, try hashMapSize(self.base.items.len));
1505 self.base_index.clearRetainingCapacity();
1506 for (self.base.items, 0..) |*image, index| {
1507 if (self.base_index.get(image.id)) |existing| {
1508 if (self.base.items[existing].generation <= image.generation) self.base_index.putAssumeCapacity(image.id, index);
1509 } else {
1510 self.base_index.putAssumeCapacity(image.id, index);
1511 }
1512 }
1513 }
1514
1515 fn nextBaseGeneration(self: *const Pager) Error!u64 {
1516 if (self.base_generation == std.math.maxInt(u64)) return error.GenerationOverflow;
1517 return self.base_generation + 1;
1518 }
1519 };
1520
1521 fn hashMapSize(count: usize) Error!u32 {
1522 if (count > std.math.maxInt(u32)) return error.PagerTooLarge;
1523 return @intCast(count);
1524 }
1525
1526 fn additionalCapacity(current: usize, additional: usize) Error!usize {
1527 return std.math.add(usize, current, additional) catch error.PagerTooLarge;
1528 }
1529
1530 pub const Snapshot = struct {
1531 pager: *const Pager,
1532 view: View,
1533
1534 pub fn get(self: Snapshot, page_id: u32) Error!?[]const u8 {
1535 const phase = trace.scope("pager.snapshot.get");
1536 defer phase.end();
1537 return self.pager.pageAt(page_id, self.view);
1538 }
1539 };
1540
1541 fn testingHeader() wal.Header {
1542 return .{
1543 .sequence = 31,
1544 .salt = .{ .first = 0x5151_7171, .second = 0x9191_b1b1 },
1545 };
1546 }
1547
1548 fn restartHeader() wal.Header {
1549 return .{
1550 .sequence = 32,
1551 .salt = .{ .first = 0xc1c1_d1d1, .second = 0xe1e1_f1f1 },
1552 };
1553 }
1554
1555 fn fillImage(image: *[page.size]u8, page_id: u32, value: u8) void {
1556 @memset(image, 0);
1557 image[0] = @intCast(page_id);
1558 image[1] = value;
1559 }
1560
1561 fn testingPager(wal_frames: usize) !Pager {
1562 const options: InitOptions = .{
1563 .header = testingHeader(),
1564 .wal_frames = wal_frames,
1565 };
1566 var workspace = try Pager.Workspace.allocate(std.testing.allocator, options);
1567 return Pager.init(std.testing.allocator, &workspace, options) catch |err| {
1568 workspace.deallocate(std.testing.allocator);
1569 return err;
1570 };
1571 }
1572
1573 fn deinitTestingPager(pager: *Pager) void {
1574 var workspace = pager.deinit();
1575 workspace.deallocate(std.testing.allocator);
1576 }
1577
1578 test "pager workspace rejects short regions before transfer and returns exact loans" {
1579 const wal_frames = 2;
1580 const options: InitOptions = .{
1581 .header = testingHeader(),
1582 .wal_frames = wal_frames,
1583 };
1584 const capacity = try Pager.Workspace.Capacity.derive(options);
1585 const wal_index_capacity = comptime WalIndex.Capacity.derive(.{
1586 .frames = wal_frames,
1587 }) catch unreachable;
1588 var journal_storage: [wal.header_size + wal_frames * wal.frame_size]u8 align(wal.Writer.storage_alignment) = undefined;
1589 var checkpoint_storage: [wal_frames * @sizeOf(CheckpointPage)]u8 align(CheckpointPlan.storage_alignment) = undefined;
1590 var checkpoint_once_storage: [wal_frames * @sizeOf(CheckpointPage)]u8 align(CheckpointPlan.storage_alignment) = undefined;
1591 var wal_index_storage: [wal_index_capacity.storage_bytes]u8 align(WalIndex.storage_alignment) = undefined;
1592 var workspace = Pager.Workspace.init(
1593 &journal_storage,
1594 &checkpoint_storage,
1595 &checkpoint_once_storage,
1596 &wal_index_storage,
1597 );
1598 try std.testing.expectEqual(journal_storage.len, capacity.journal.storage_bytes);
1599 try std.testing.expectEqual(checkpoint_storage.len, capacity.checkpoint.storage_bytes);
1600 try std.testing.expectEqual(checkpoint_once_storage.len, capacity.checkpoint_once.storage_bytes);
1601 try std.testing.expectEqual(wal_index_storage.len, capacity.wal_index.storage_bytes);
1602 @memset(workspace.journal, 0xa5);
1603
1604 var short_journal_workspace = Pager.Workspace.init(
1605 workspace.journal[0 .. capacity.journal.storage_bytes - 1],
1606 workspace.checkpoint,
1607 workspace.checkpoint_once,
1608 workspace.wal_index,
1609 );
1610 try std.testing.expectError(error.StorageTooShort, Pager.init(
1611 std.testing.allocator,
1612 &short_journal_workspace,
1613 options,
1614 ));
1615 try std.testing.expectEqual(
1616 journal_storage[0..].ptr,
1617 short_journal_workspace.journal.ptr,
1618 );
1619 var short_checkpoint_workspace = Pager.Workspace.init(
1620 workspace.journal,
1621 workspace.checkpoint[0 .. capacity.checkpoint.storage_bytes - 1],
1622 workspace.checkpoint_once,
1623 workspace.wal_index,
1624 );
1625 try std.testing.expectError(error.StorageTooShort, Pager.init(
1626 std.testing.allocator,
1627 &short_checkpoint_workspace,
1628 options,
1629 ));
1630 try std.testing.expectEqual(
1631 checkpoint_storage[0..].ptr,
1632 short_checkpoint_workspace.checkpoint.ptr,
1633 );
1634 var short_wal_index_workspace = Pager.Workspace.init(
1635 workspace.journal,
1636 workspace.checkpoint,
1637 workspace.checkpoint_once,
1638 workspace.wal_index[0 .. capacity.wal_index.storage_bytes - 1],
1639 );
1640 try std.testing.expectError(error.StorageTooShort, Pager.init(
1641 std.testing.allocator,
1642 &short_wal_index_workspace,
1643 options,
1644 ));
1645 try std.testing.expectEqual(
1646 wal_index_storage[0..].ptr,
1647 short_wal_index_workspace.wal_index.ptr,
1648 );
1649 for (workspace.journal) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte);
1650
1651 const journal_pointer = workspace.journal.ptr;
1652 const checkpoint_pointer = workspace.checkpoint.ptr;
1653 const checkpoint_once_pointer = workspace.checkpoint_once.ptr;
1654 const wal_index_pointer = workspace.wal_index.ptr;
1655 var pager = try Pager.init(std.testing.allocator, &workspace, options);
1656 workspace = pager.deinit();
1657 try std.testing.expectEqual(journal_pointer, workspace.journal.ptr);
1658 try std.testing.expectEqual(checkpoint_pointer, workspace.checkpoint.ptr);
1659 try std.testing.expectEqual(checkpoint_once_pointer, workspace.checkpoint_once.ptr);
1660 try std.testing.expectEqual(wal_index_pointer, workspace.wal_index.ptr);
1661 }
1662
1663 test "pager snapshot reads base page without wal frames" {
1664 var pager = try testingPager(0);
1665 defer deinitTestingPager(&pager);
1666
1667 var base: [page.size]u8 = undefined;
1668 fillImage(&base, 1, 10);
1669 try pager.installBase(1, &base);
1670
1671 const snapshot = try pager.beginRead();
1672 try std.testing.expectEqual(@as(u64, 1), snapshot.view.base_generation);
1673 try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark);
1674 const image = (try snapshot.get(1)).?;
1675 try std.testing.expectEqual(@as(u8, 10), image[1]);
1676 }
1677
1678 test "pager reserve keeps base wal and checkpoint behavior intact" {
1679 var pager = try testingPager(2);
1680 defer deinitTestingPager(&pager);
1681 try pager.reserve(.{ .base_pages = 2, .wal_frames = 2 });
1682
1683 var base: [page.size]u8 = undefined;
1684 var wal_image: [page.size]u8 = undefined;
1685 fillImage(&base, 1, 10);
1686 fillImage(&wal_image, 1, 20);
1687
1688 try pager.installBase(1, &base);
1689 try pager.appendWal(1, 1, &wal_image);
1690 const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() });
1691 const snapshot = try pager.beginRead();
1692
1693 try std.testing.expect(checkpoint.restarted);
1694 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
1695 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
1696 }
1697
1698 test "pager reserve uses sealed wal frame and page index capacities" {
1699 var pager = try testingPager(32);
1700 defer deinitTestingPager(&pager);
1701 try pager.reserve(.{ .wal_frames = 32, .wal_pages = 2 });
1702
1703 try std.testing.expectEqual(@as(usize, 32), pager.wal_index.frames.capacity);
1704 try std.testing.expectEqual(@as(usize, 32), pager.wal_index.pages.capacity);
1705 }
1706
1707 test "pager prepared checkpoint preserves state and rejects stale commit" {
1708 var pager = try testingPager(3);
1709 defer deinitTestingPager(&pager);
1710 try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 });
1711
1712 var committed_first: [page.size]u8 = undefined;
1713 var committed_second: [page.size]u8 = undefined;
1714 var tail: [page.size]u8 = undefined;
1715 fillImage(&committed_first, 2, 20);
1716 fillImage(&committed_second, 1, 25);
1717 fillImage(&tail, 3, 30);
1718 try pager.appendWal(2, 0, &committed_first);
1719 try pager.appendWal(1, 2, &committed_second);
1720
1721 const before = pager.storage();
1722 var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() });
1723 defer prepared.deinit();
1724 try std.testing.expectEqual(@as(usize, 2), prepared.result().pages);
1725 try std.testing.expectEqual(@as(u32, 1), prepared.checkpointPage(0).page_id);
1726 try std.testing.expectEqual(@as(u32, 2), prepared.checkpointPage(1).page_id);
1727 try std.testing.expectEqual(@as(usize, 2), pager.frameCount());
1728 try std.testing.expectEqual(before.base_images, pager.storage().base_images);
1729 try std.testing.expectEqual(@as(u64, 0), pager.baseGeneration());
1730
1731 var peer = try testingPager(3);
1732 defer deinitTestingPager(&peer);
1733 try peer.reserve(.{ .wal_frames = 3, .wal_pages = 3 });
1734 try peer.appendWal(2, 0, &committed_first);
1735 try peer.appendWal(1, 2, &committed_second);
1736 var peer_prepared = try peer.prepareCheckpoint(.{ .restart_header = restartHeader() });
1737 defer peer_prepared.deinit();
1738 try std.testing.expectError(error.StaleCheckpoint, peer.commitCheckpoint(prepared));
1739
1740 try pager.appendWal(3, 0, &tail);
1741 try std.testing.expectError(error.StaleCheckpoint, pager.commitCheckpoint(prepared));
1742 const checkpoint_value = try pager.checkpoint(.{ .restart_header = restartHeader() });
1743 try std.testing.expect(checkpoint_value.restarted);
1744 try std.testing.expectEqual(@as(usize, 2), checkpoint_value.pages);
1745 }
1746
1747 test "pager durable checkpoint drops memory base and leaves file fallback" {
1748 var pager = try testingPager(1);
1749 defer deinitTestingPager(&pager);
1750 try pager.reserve(.{ .base_pages = 1, .wal_frames = 1, .wal_pages = 1 });
1751
1752 var base: [page.size]u8 = undefined;
1753 var committed: [page.size]u8 = undefined;
1754 fillImage(&base, 1, 10);
1755 fillImage(&committed, 1, 20);
1756 try pager.installBase(1, &base);
1757 try pager.appendWal(1, 1, &committed);
1758 var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() });
1759 defer prepared.deinit();
1760 const checkpoint_value = try pager.commitDurableCheckpoint(prepared);
1761
1762 try std.testing.expect(checkpoint_value.restarted);
1763 try std.testing.expectEqual(@as(u64, 2), checkpoint_value.base_generation);
1764 try std.testing.expectEqual(@as(usize, 0), pager.storage().base_images);
1765 try std.testing.expectEqual(@as(usize, 0), pager.storage().base_capacity);
1766 try std.testing.expect(try pager.pageAt(1, .{ .base_generation = checkpoint_value.base_generation, .end_mark = 0 }) == null);
1767 }
1768
1769 test "pager concurrent preparation rejects without invalidating prepared page views" {
1770 var pager = try testingPager(2);
1771 defer deinitTestingPager(&pager);
1772 try pager.reserve(.{ .wal_frames = 2, .wal_pages = 2 });
1773
1774 var first: [page.size]u8 = undefined;
1775 var second: [page.size]u8 = undefined;
1776 fillImage(&first, 1, 20);
1777 fillImage(&second, 2, 30);
1778 try pager.appendWal(1, 1, &first);
1779 const oldest = try pager.currentView();
1780 try pager.appendWal(2, 2, &second);
1781 var prepared = try pager.prepareCheckpoint(.{ .readers = .{ .oldest = oldest } });
1782 defer prepared.deinit();
1783 try std.testing.expectEqual(@as(usize, 1), prepared.checkpointPageCount());
1784
1785 try std.testing.expectError(
1786 error.CheckpointPlanInUse,
1787 pager.prepareCheckpoint(.{ .restart_header = restartHeader() }),
1788 );
1789 try std.testing.expectEqual(@as(usize, 1), prepared.checkpointPageCount());
1790 try std.testing.expectEqual(@as(u32, 1), prepared.checkpointPage(0).page_id);
1791 _ = try pager.commitCheckpoint(prepared);
1792 }
1793
1794 test "pager durable checkpoint rejects active readers" {
1795 var pager = try testingPager(1);
1796 defer deinitTestingPager(&pager);
1797 try pager.reserve(.{ .base_pages = 1, .wal_frames = 1, .wal_pages = 1 });
1798
1799 var base: [page.size]u8 = undefined;
1800 var committed: [page.size]u8 = undefined;
1801 fillImage(&base, 1, 10);
1802 fillImage(&committed, 1, 20);
1803 try pager.installBase(1, &base);
1804 const oldest = try pager.beginRead();
1805 try pager.appendWal(1, 1, &committed);
1806 var prepared = try pager.prepareCheckpoint(.{ .readers = .{ .oldest = oldest.view } });
1807 defer prepared.deinit();
1808 try std.testing.expectError(error.DurableCheckpointReaders, pager.commitDurableCheckpoint(prepared));
1809 _ = try pager.commitCheckpoint(prepared);
1810 try std.testing.expectEqual(@as(u8, 10), (try oldest.get(1)).?[1]);
1811 }
1812
1813 fn modelWalIndexCapacity(limits: WalIndex.Limits) ?WalIndex.Capacity {
1814 const frame_bytes = @as(u128, limits.frames) * @sizeOf(WalImage);
1815 const page_alignment = @as(u128, @alignOf(WalPage));
1816 const page_offset = (frame_bytes + page_alignment - 1) & ~(page_alignment - 1);
1817 const page_bytes = @as(u128, limits.frames) * @sizeOf(WalPage);
1818 const storage_bytes = page_offset + page_bytes;
1819 const maximum = std.math.maxInt(usize);
1820 if (frame_bytes > maximum or
1821 page_offset > maximum or
1822 page_bytes > maximum or
1823 storage_bytes > maximum)
1824 {
1825 return null;
1826 }
1827 return .{
1828 .frames = limits.frames,
1829 .frame_bytes = @intCast(frame_bytes),
1830 .page_offset = @intCast(page_offset),
1831 .page_bytes = @intCast(page_bytes),
1832 .storage_bytes = @intCast(storage_bytes),
1833 };
1834 }
1835
1836 test "wal index capacity matches independent aligned typed regions" {
1837 comptime {
1838 @stardustClaim(
1839 @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_capacity"),
1840 null,
1841 null,
1842 null,
1843 null,
1844 null,
1845 null,
1846 );
1847 }
1848
1849 for (0..4097) |frames| {
1850 const limits: WalIndex.Limits = .{ .frames = frames };
1851 try std.testing.expectEqual(
1852 modelWalIndexCapacity(limits).?,
1853 try WalIndex.Capacity.derive(limits),
1854 );
1855 }
1856 const overflow: WalIndex.Limits = .{ .frames = std.math.maxInt(usize) };
1857 try std.testing.expect(modelWalIndexCapacity(overflow) == null);
1858 try std.testing.expectError(error.CapacityOverflow, WalIndex.Capacity.derive(overflow));
1859 }
1860
1861 test "wal index rejects short storage and releases its exact borrow" {
1862 comptime {
1863 @stardustClaim(
1864 @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_storage_rejection"),
1865 null,
1866 null,
1867 null,
1868 null,
1869 null,
1870 null,
1871 );
1872 @stardustClaim(
1873 @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_work_bound"),
1874 null,
1875 null,
1876 null,
1877 null,
1878 null,
1879 null,
1880 );
1881 }
1882
1883 const limits: WalIndex.Limits = .{ .frames = 3 };
1884 const capacity = try WalIndex.Capacity.derive(limits);
1885 const storage = try std.testing.allocator.alignedAlloc(
1886 u8,
1887 .fromByteUnits(WalIndex.storage_alignment),
1888 capacity.storage_bytes,
1889 );
1890 defer std.testing.allocator.free(storage);
1891 try std.testing.expectError(
1892 error.StorageTooShort,
1893 WalIndex.init(storage[0 .. storage.len - 1], limits),
1894 );
1895
1896 var index = try WalIndex.init(storage, limits);
1897 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, index.phase);
1898 try std.testing.expectEqual(@as(usize, 3), index.frames.capacity);
1899 try std.testing.expectEqual(@as(usize, 3), index.pages.capacity);
1900 index.activate();
1901 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, index.phase);
1902 const released = index.deinit();
1903 try std.testing.expectEqual(storage.ptr, released.ptr);
1904 try std.testing.expectEqual(storage.len, released.len);
1905 }
1906
1907 test "wal index stays sealed through transitive operations and overload" {
1908 comptime {
1909 @stardustClaim(
1910 @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_sealed_overload"),
1911 null,
1912 null,
1913 null,
1914 null,
1915 null,
1916 null,
1917 );
1918 @stardustClaim(
1919 @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_sealed_transitive_risk"),
1920 null,
1921 null,
1922 null,
1923 null,
1924 null,
1925 null,
1926 );
1927 }
1928
1929 const allocator = std.testing.allocator;
1930 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(allocator);
1931 const initialization_allocator = phase_allocator.initializationAllocator();
1932 const options: InitOptions = .{
1933 .header = testingHeader(),
1934 .wal_frames = 3,
1935 };
1936 var maybe_workspace: ?Pager.Workspace = try Pager.Workspace.allocate(
1937 initialization_allocator,
1938 options,
1939 );
1940 var maybe_pager: ?Pager = null;
1941 defer {
1942 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
1943 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
1944 const teardown_allocator = phase_allocator.teardownAllocator();
1945 if (maybe_pager) |*pager| {
1946 var workspace = pager.deinit();
1947 workspace.deallocate(teardown_allocator);
1948 } else if (maybe_workspace) |*workspace| {
1949 workspace.deallocate(teardown_allocator);
1950 }
1951 phase_allocator.deinit();
1952 }
1953 maybe_pager = try Pager.init(initialization_allocator, &maybe_workspace.?, options);
1954 maybe_workspace = null;
1955 const pager = &maybe_pager.?;
1956 const index_pointer = @intFromPtr(pager.wal_index.storage.ptr);
1957 phase_allocator.seal();
1958 try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 });
1959
1960 var first: [page.size]u8 = undefined;
1961 var second: [page.size]u8 = undefined;
1962 var third: [page.size]u8 = undefined;
1963 var rejected: [page.size]u8 = undefined;
1964 fillImage(&first, 1, 10);
1965 fillImage(&second, 2, 20);
1966 fillImage(&third, 3, 30);
1967 fillImage(&rejected, 4, 40);
1968 try pager.appendWal(1, 1, &first);
1969 const after_first = pager.position();
1970 try pager.appendWal(2, 0, &second);
1971 try pager.appendWal(3, 3, &third);
1972 const before = pager.position();
1973 try std.testing.expectError(error.WalFull, pager.appendWal(4, 4, &rejected));
1974 try std.testing.expectEqual(before, pager.position());
1975 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len);
1976 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len);
1977
1978 pager.restore(after_first);
1979 const staged_position = pager.position();
1980 try pager.stageWalPage(staged_position, 0, 2, &second);
1981 try pager.stageWalPage(staged_position, 1, 3, &third);
1982 try pager.commitStagedWal(staged_position, 2, 3);
1983 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len);
1984 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len);
1985
1986 var recovered_wal: [wal.header_size + 3 * wal.frame_size]u8 = undefined;
1987 @memcpy(&recovered_wal, pager.walBytes());
1988 try pager.replaceWal(&recovered_wal, recovered_wal.len);
1989 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len);
1990 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len);
1991
1992 var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() });
1993 defer prepared.deinit();
1994 const checkpoint = try pager.commitDurableCheckpoint(prepared);
1995 try std.testing.expect(checkpoint.restarted);
1996 try std.testing.expectEqual(@as(usize, 0), pager.wal_index.frames.items.len);
1997 try std.testing.expectEqual(@as(usize, 0), pager.wal_index.pages.items.len);
1998 try std.testing.expectEqual(index_pointer, @intFromPtr(pager.wal_index.storage.ptr));
1999 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
2000 }
2001
2002 fn modelCheckpointPlanBytes(limits: CheckpointPlan.Limits) ?usize {
2003 if (limits.pages > std.math.maxInt(usize) / @sizeOf(CheckpointPage)) return null;
2004 return limits.pages * @sizeOf(CheckpointPage);
2005 }
2006
2007 test "checkpoint plan capacity matches an independent typed storage model" {
2008 comptime {
2009 @stardustClaim(
2010 @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_capacity"),
2011 null,
2012 null,
2013 null,
2014 null,
2015 null,
2016 null,
2017 );
2018 }
2019
2020 for (0..64) |pages| {
2021 const limits: CheckpointPlan.Limits = .{ .pages = pages };
2022 const capacity = try CheckpointPlan.Capacity.derive(limits);
2023 try std.testing.expectEqual(pages, capacity.pages);
2024 try std.testing.expectEqual(modelCheckpointPlanBytes(limits).?, capacity.storage_bytes);
2025 }
2026 const overflow: CheckpointPlan.Limits = .{ .pages = std.math.maxInt(usize) };
2027 try std.testing.expect(modelCheckpointPlanBytes(overflow) == null);
2028 try std.testing.expectError(error.CapacityOverflow, CheckpointPlan.Capacity.derive(overflow));
2029 }
2030
2031 test "checkpoint plan rejects short storage and releases its exact borrow" {
2032 comptime {
2033 @stardustClaim(
2034 @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_storage_rejection"),
2035 null,
2036 null,
2037 null,
2038 null,
2039 null,
2040 null,
2041 );
2042 @stardustClaim(
2043 @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_work_bound"),
2044 null,
2045 null,
2046 null,
2047 null,
2048 null,
2049 null,
2050 );
2051 }
2052
2053 const limits: CheckpointPlan.Limits = .{ .pages = 3 };
2054 const capacity = try CheckpointPlan.Capacity.derive(limits);
2055 const storage = try std.testing.allocator.alignedAlloc(
2056 u8,
2057 .fromByteUnits(CheckpointPlan.storage_alignment),
2058 capacity.storage_bytes,
2059 );
2060 defer std.testing.allocator.free(storage);
2061 try std.testing.expectError(
2062 error.StorageTooShort,
2063 CheckpointPlan.init(storage[0 .. storage.len - 1], limits),
2064 );
2065
2066 var plan = try CheckpointPlan.init(storage, limits);
2067 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, plan.phase);
2068 try std.testing.expectEqual(capacity.storage_bytes, plan.storage.len);
2069 plan.activate();
2070 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, plan.phase);
2071 const released = plan.deinit();
2072 try std.testing.expectEqual(storage.ptr, released.ptr);
2073 try std.testing.expectEqual(storage.len, released.len);
2074 }
2075
2076 test "checkpoint plan stays sealed through exact page selection and overload" {
2077 var pager = try testingPager(3);
2078 defer deinitTestingPager(&pager);
2079
2080 var first: [page.size]u8 = undefined;
2081 var first_latest: [page.size]u8 = undefined;
2082 var second: [page.size]u8 = undefined;
2083 fillImage(&first, 1, 10);
2084 fillImage(&first_latest, 1, 11);
2085 fillImage(&second, 2, 20);
2086 try pager.appendWal(1, 0, &first);
2087 try pager.appendWal(1, 0, &first_latest);
2088 try pager.appendWal(2, 2, &second);
2089
2090 var prepared = try pager.prepareCheckpoint(.{});
2091 defer prepared.deinit();
2092 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, prepared.plan.phase);
2093 try std.testing.expectEqual(@as(usize, 3), prepared.plan.capacity.pages);
2094 try std.testing.expectEqual(@as(usize, 2), prepared.checkpointPageCount());
2095 try std.testing.expectEqual(@as(u8, 11), prepared.checkpointPage(0).bytes[1]);
2096 try std.testing.expectEqual(@as(u8, 20), prepared.checkpointPage(1).bytes[1]);
2097
2098 const bounded_capacity = try CheckpointPlan.Capacity.derive(.{ .pages = 1 });
2099 const bounded_storage = try std.testing.allocator.alignedAlloc(
2100 u8,
2101 .fromByteUnits(CheckpointPlan.storage_alignment),
2102 bounded_capacity.storage_bytes,
2103 );
2104 defer std.testing.allocator.free(bounded_storage);
2105 var bounded = try CheckpointPlan.init(bounded_storage, .{ .pages = 1 });
2106 defer _ = bounded.deinit();
2107 bounded.activate();
2108 try std.testing.expectError(error.CheckpointPlanCapacityExceeded, bounded.begin(2));
2109 try bounded.begin(1);
2110 try std.testing.expectError(error.CheckpointPlanInUse, bounded.begin(1));
2111 try bounded.append(.{ .page_id = 1, .wal_offset = wal.header_size + wal.frame_header_size });
2112 try std.testing.expectError(error.CheckpointPlanCapacityExceeded, bounded.append(.{ .page_id = 2, .wal_offset = wal.header_size + wal.frame_size + wal.frame_header_size }));
2113 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, bounded.phase);
2114 try std.testing.expectEqual(@as(usize, 1), bounded.filled);
2115 try std.testing.expectEqual(@as(usize, 1), bounded.pages().len);
2116 bounded.release();
2117 }
2118
2119 test "pager reuses sealed checkpoint plans across durable checkpoints" {
2120 comptime {
2121 @stardustClaim(
2122 @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_sealed_overload"),
2123 null,
2124 null,
2125 null,
2126 null,
2127 null,
2128 null,
2129 );
2130 }
2131 comptime {
2132 @stardustClaim(
2133 @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_sealed_transitive_risk"),
2134 null,
2135 null,
2136 null,
2137 null,
2138 null,
2139 null,
2140 );
2141 }
2142
2143 const allocator = std.testing.allocator;
2144 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(allocator);
2145 const initialization_allocator = phase_allocator.initializationAllocator();
2146 const options: InitOptions = .{
2147 .header = testingHeader(),
2148 .wal_frames = 3,
2149 };
2150 var maybe_workspace: ?Pager.Workspace = try Pager.Workspace.allocate(initialization_allocator, options);
2151 var maybe_pager: ?Pager = null;
2152 defer {
2153 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
2154 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
2155 const teardown_allocator = phase_allocator.teardownAllocator();
2156 if (maybe_pager) |*pager| {
2157 var workspace = pager.deinit();
2158 workspace.deallocate(teardown_allocator);
2159 } else if (maybe_workspace) |*workspace| {
2160 workspace.deallocate(teardown_allocator);
2161 }
2162 phase_allocator.deinit();
2163 }
2164 maybe_pager = try Pager.init(initialization_allocator, &maybe_workspace.?, options);
2165 maybe_workspace = null;
2166 const pager = &maybe_pager.?;
2167 try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 });
2168 const plan_pointer = @intFromPtr(pager.checkpoint_plan.storage.ptr);
2169 const once_pointer = @intFromPtr(pager.checkpoint_once_plan.storage.ptr);
2170 phase_allocator.seal();
2171
2172 var first: [page.size]u8 = undefined;
2173 var second: [page.size]u8 = undefined;
2174 fillImage(&first, 1, 10);
2175 fillImage(&second, 2, 20);
2176 try pager.appendWal(1, 0, &first);
2177 try pager.appendWal(2, 2, &second);
2178 var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() });
2179 _ = try pager.commitDurableCheckpoint(prepared);
2180 prepared.deinit();
2181
2182 try pager.appendWal(1, 1, &second);
2183 var repeated = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() });
2184 _ = try pager.commitDurableCheckpoint(repeated);
2185 repeated.deinit();
2186
2187 try std.testing.expectEqual(plan_pointer, @intFromPtr(pager.checkpoint_plan.storage.ptr));
2188 try std.testing.expectEqual(once_pointer, @intFromPtr(pager.checkpoint_once_plan.storage.ptr));
2189 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
2190 }
2191
2192 test "pager base lookup finds visible generation by page order" {
2193 var pager = try testingPager(0);
2194 defer deinitTestingPager(&pager);
2195
2196 var second_old: [page.size]u8 = undefined;
2197 var first: [page.size]u8 = undefined;
2198 var second_new: [page.size]u8 = undefined;
2199 fillImage(&second_old, 2, 20);
2200 fillImage(&first, 1, 10);
2201 fillImage(&second_new, 2, 22);
2202
2203 try pager.installBase(2, &second_old);
2204 try pager.installBase(1, &first);
2205 try pager.installBase(2, &second_new);
2206
2207 try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(2, .{ .base_generation = 1, .end_mark = 0 })).?[1]);
2208 try std.testing.expect(try pager.pageAt(1, .{ .base_generation = 1, .end_mark = 0 }) == null);
2209 try std.testing.expectEqual(@as(u8, 10), (try pager.pageAt(1, .{ .base_generation = 2, .end_mark = 0 })).?[1]);
2210 try std.testing.expectEqual(@as(u8, 22), (try pager.pageAt(2, .{ .base_generation = std.math.maxInt(u64), .end_mark = 0 })).?[1]);
2211 }
2212
2213 test "pager snapshots preserve base generations" {
2214 var pager = try testingPager(0);
2215 defer deinitTestingPager(&pager);
2216
2217 var first: [page.size]u8 = undefined;
2218 var second: [page.size]u8 = undefined;
2219 fillImage(&first, 1, 10);
2220 fillImage(&second, 1, 20);
2221
2222 try pager.installBase(1, &first);
2223 const before = try pager.beginRead();
2224 try pager.installBase(1, &second);
2225 const after = try pager.beginRead();
2226
2227 try std.testing.expectEqual(@as(u8, 10), (try before.get(1)).?[1]);
2228 try std.testing.expectEqual(@as(u8, 20), (try after.get(1)).?[1]);
2229 }
2230
2231 test "pager end mark freezes wal page view" {
2232 var pager = try testingPager(2);
2233 defer deinitTestingPager(&pager);
2234
2235 var base: [page.size]u8 = undefined;
2236 var first: [page.size]u8 = undefined;
2237 var second: [page.size]u8 = undefined;
2238 fillImage(&base, 1, 10);
2239 fillImage(&first, 1, 20);
2240 fillImage(&second, 1, 30);
2241
2242 try pager.installBase(1, &base);
2243 try pager.appendWal(1, 1, &first);
2244 const before = try pager.beginRead();
2245 try pager.appendWal(1, 1, &second);
2246 const after = try pager.beginRead();
2247
2248 try std.testing.expectEqual(@as(usize, 1), before.view.end_mark);
2249 try std.testing.expectEqual(@as(usize, 2), after.view.end_mark);
2250 try std.testing.expectEqual(@as(u8, 20), (try before.get(1)).?[1]);
2251 try std.testing.expectEqual(@as(u8, 30), (try after.get(1)).?[1]);
2252 }
2253
2254 test "pager restore rebuilds wal page index" {
2255 var pager = try testingPager(4);
2256 defer deinitTestingPager(&pager);
2257
2258 var second: [page.size]u8 = undefined;
2259 var first: [page.size]u8 = undefined;
2260 var second_newer: [page.size]u8 = undefined;
2261 var seventh: [page.size]u8 = undefined;
2262 fillImage(&second, 2, 20);
2263 fillImage(&first, 1, 10);
2264 fillImage(&second_newer, 2, 22);
2265 fillImage(&seventh, 7, 77);
2266
2267 try pager.appendWal(2, 2, &second);
2268 const position_value = pager.position();
2269 try pager.appendWal(1, 2, &first);
2270 try pager.appendWal(2, 2, &second_newer);
2271 try pager.appendWal(7, 7, &seventh);
2272 const before = try pager.beginRead();
2273
2274 try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len);
2275 try std.testing.expectEqual(@as(u8, 10), (try before.get(1)).?[1]);
2276 try std.testing.expectEqual(@as(u8, 22), (try before.get(2)).?[1]);
2277 try std.testing.expectEqual(@as(u8, 77), (try before.get(7)).?[1]);
2278 try std.testing.expectEqual(@as(u32, 7), pager.databasePageCount());
2279
2280 pager.restore(position_value);
2281 const after = try pager.beginRead();
2282
2283 try std.testing.expectEqual(@as(usize, 1), pager.frameCount());
2284 try std.testing.expectEqual(@as(usize, 1), after.view.end_mark);
2285 try std.testing.expectEqual(@as(usize, 1), pager.wal_index.pages.items.len);
2286 try std.testing.expectEqual(@as(u32, 2), pager.wal_index.pages.items[0].page_id);
2287 try std.testing.expectEqual(@as(u32, 2), pager.databasePageCount());
2288 try std.testing.expect(try after.get(1) == null);
2289 try std.testing.expectEqual(@as(u8, 20), (try after.get(2)).?[1]);
2290 }
2291
2292 test "pager check marks clear when a stored image changes" {
2293 var pager = try testingPager(4);
2294 defer deinitTestingPager(&pager);
2295
2296 var base: [page.size]u8 = undefined;
2297 var replaced: [page.size]u8 = undefined;
2298 var logged: [page.size]u8 = undefined;
2299 var superseding: [page.size]u8 = undefined;
2300 var relogged: [page.size]u8 = undefined;
2301 fillImage(&base, 1, 10);
2302 fillImage(&replaced, 1, 11);
2303 fillImage(&logged, 2, 20);
2304 fillImage(&superseding, 2, 21);
2305 fillImage(&relogged, 2, 22);
2306
2307 try pager.installBase(1, &base);
2308 const based = (try pager.markedPageAt(1, try pager.currentView())).?;
2309 try std.testing.expect(!based.checked.?.*);
2310 based.checked.?.* = true;
2311 try std.testing.expect((try pager.markedPageAt(1, try pager.currentView())).?.checked.?.*);
2312 try pager.installBaseAtGeneration(1, &replaced, pager.baseGeneration());
2313 const rebased = (try pager.markedPageAt(1, try pager.currentView())).?;
2314 try std.testing.expectEqual(@as(u8, 11), rebased.bytes[1]);
2315 try std.testing.expect(!rebased.checked.?.*);
2316
2317 const before = pager.position();
2318 try pager.appendWal(2, 2, &logged);
2319 const appended = (try pager.markedPageAt(2, try pager.currentView())).?;
2320 try std.testing.expect(!appended.checked.?.*);
2321 appended.checked.?.* = true;
2322 try std.testing.expect((try pager.markedPageAt(2, try pager.currentView())).?.checked.?.*);
2323 try pager.appendWal(2, 2, &superseding);
2324 const superseded = (try pager.markedPageAt(2, try pager.currentView())).?;
2325 try std.testing.expectEqual(@as(u8, 21), superseded.bytes[1]);
2326 try std.testing.expect(!superseded.checked.?.*);
2327 superseded.checked.?.* = true;
2328 pager.restore(before);
2329 try pager.appendWal(2, 2, &relogged);
2330 const reappended = (try pager.markedPageAt(2, try pager.currentView())).?;
2331 try std.testing.expectEqual(@as(u8, 22), reappended.bytes[1]);
2332 try std.testing.expect(!reappended.checked.?.*);
2333 }
2334
2335 test "pager ignores uncommitted wal tail" {
2336 var pager = try testingPager(2);
2337 defer deinitTestingPager(&pager);
2338
2339 var committed: [page.size]u8 = undefined;
2340 var uncommitted: [page.size]u8 = undefined;
2341 fillImage(&committed, 1, 20);
2342 fillImage(&uncommitted, 9, 99);
2343
2344 try pager.appendWal(1, 1, &committed);
2345 try pager.appendWal(9, 0, &uncommitted);
2346 const snapshot = try pager.beginRead();
2347
2348 try std.testing.expectEqual(@as(usize, 1), snapshot.view.end_mark);
2349 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
2350 try std.testing.expectEqual(@as(u32, 9), pager.databasePageCount());
2351 }
2352
2353 test "pager falls back to base pages not present in wal" {
2354 var pager = try testingPager(1);
2355 defer deinitTestingPager(&pager);
2356
2357 var first: [page.size]u8 = undefined;
2358 var second: [page.size]u8 = undefined;
2359 var wal_image: [page.size]u8 = undefined;
2360 fillImage(&first, 1, 10);
2361 fillImage(&second, 2, 20);
2362 fillImage(&wal_image, 1, 30);
2363
2364 try pager.installBase(1, &first);
2365 try pager.installBase(2, &second);
2366 try pager.appendWal(1, 2, &wal_image);
2367 const snapshot = try pager.beginRead();
2368
2369 try std.testing.expectEqual(@as(u8, 30), (try snapshot.get(1)).?[1]);
2370 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(2)).?[1]);
2371 try std.testing.expect(try snapshot.get(3) == null);
2372 }
2373
2374 test "pager snapshot survives frames appended after its end mark" {
2375 var pager = try testingPager(2);
2376 defer deinitTestingPager(&pager);
2377
2378 var first: [page.size]u8 = undefined;
2379 var second: [page.size]u8 = undefined;
2380 fillImage(&first, 1, 20);
2381 fillImage(&second, 1, 30);
2382
2383 try pager.appendWal(1, 1, &first);
2384 const snapshot = try pager.beginRead();
2385 try pager.appendWal(1, 1, &second);
2386
2387 try std.testing.expectEqual(@as(usize, 1), snapshot.view.end_mark);
2388 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
2389 try std.testing.expectEqual(@as(usize, 2), (try pager.beginRead()).view.end_mark);
2390 }
2391
2392 test "pager checkpoint applies committed wal frames to a base generation" {
2393 var pager = try testingPager(2);
2394 defer deinitTestingPager(&pager);
2395
2396 var first: [page.size]u8 = undefined;
2397 var second: [page.size]u8 = undefined;
2398 fillImage(&first, 1, 20);
2399 fillImage(&second, 2, 30);
2400
2401 try pager.appendWal(1, 0, &first);
2402 try pager.appendWal(2, 2, &second);
2403 const checkpoint = try pager.checkpoint(.{});
2404 const base_view = View{ .base_generation = checkpoint.base_generation, .end_mark = 0 };
2405
2406 try std.testing.expectEqual(@as(usize, 2), checkpoint.end_mark);
2407 try std.testing.expectEqual(@as(usize, 2), checkpoint.pages);
2408 try std.testing.expectEqual(@as(u64, 1), checkpoint.base_generation);
2409 try std.testing.expect(!checkpoint.restarted);
2410 try std.testing.expectEqual(@as(usize, 2), pager.frameCount());
2411 try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(1, base_view)).?[1]);
2412 try std.testing.expectEqual(@as(u8, 30), (try pager.pageAt(2, base_view)).?[1]);
2413 }
2414
2415 test "pager checkpoint prunes obsolete base images without readers" {
2416 var pager = try testingPager(0);
2417 defer deinitTestingPager(&pager);
2418
2419 var first_old: [page.size]u8 = undefined;
2420 var second_old: [page.size]u8 = undefined;
2421 var first_new: [page.size]u8 = undefined;
2422 var second_new: [page.size]u8 = undefined;
2423 fillImage(&first_old, 1, 10);
2424 fillImage(&second_old, 2, 20);
2425 fillImage(&first_new, 1, 11);
2426 fillImage(&second_new, 2, 22);
2427
2428 try pager.installBase(1, &first_old);
2429 try pager.installBase(2, &second_old);
2430 try pager.installBase(1, &first_new);
2431 try pager.installBase(2, &second_new);
2432 const checkpoint = try pager.checkpoint(.{});
2433 const current = try pager.beginRead();
2434
2435 try std.testing.expectEqual(@as(usize, 0), checkpoint.pages);
2436 try std.testing.expectEqual(@as(usize, 2), pager.base.items.len);
2437 try std.testing.expectEqual(@as(u64, 4), pager.baseGeneration());
2438 try std.testing.expect(try pager.pageAt(1, .{ .base_generation = 1, .end_mark = 0 }) == null);
2439 try std.testing.expectEqual(@as(u8, 11), (try current.get(1)).?[1]);
2440 try std.testing.expectEqual(@as(u8, 22), (try current.get(2)).?[1]);
2441 }
2442
2443 test "pager checkpoint retains base floor for oldest reader" {
2444 var pager = try testingPager(0);
2445 defer deinitTestingPager(&pager);
2446
2447 var first_old: [page.size]u8 = undefined;
2448 var second_old: [page.size]u8 = undefined;
2449 var first_floor: [page.size]u8 = undefined;
2450 var first_new: [page.size]u8 = undefined;
2451 var second_new: [page.size]u8 = undefined;
2452 fillImage(&first_old, 1, 10);
2453 fillImage(&second_old, 2, 20);
2454 fillImage(&first_floor, 1, 11);
2455 fillImage(&first_new, 1, 12);
2456 fillImage(&second_new, 2, 22);
2457
2458 try pager.installBase(1, &first_old);
2459 try pager.installBase(2, &second_old);
2460 try pager.installBase(1, &first_floor);
2461 const oldest = try pager.beginRead();
2462 try pager.installBase(1, &first_new);
2463 try pager.installBase(2, &second_new);
2464 const checkpoint = try pager.checkpoint(.{ .readers = .{ .oldest = oldest.view } });
2465 const current = try pager.beginRead();
2466
2467 try std.testing.expectEqual(@as(usize, 0), checkpoint.pages);
2468 try std.testing.expectEqual(@as(usize, 4), pager.base.items.len);
2469 try std.testing.expectEqual(@as(u8, 11), (try oldest.get(1)).?[1]);
2470 try std.testing.expectEqual(@as(u8, 20), (try oldest.get(2)).?[1]);
2471 try std.testing.expectEqual(@as(u8, 12), (try current.get(1)).?[1]);
2472 try std.testing.expectEqual(@as(u8, 22), (try current.get(2)).?[1]);
2473 }
2474
2475 test "pager checkpoint compacts wal-applied base images without readers" {
2476 var pager = try testingPager(1);
2477 defer deinitTestingPager(&pager);
2478
2479 var base_old: [page.size]u8 = undefined;
2480 var base_new: [page.size]u8 = undefined;
2481 var committed: [page.size]u8 = undefined;
2482 fillImage(&base_old, 1, 10);
2483 fillImage(&base_new, 1, 20);
2484 fillImage(&committed, 1, 30);
2485
2486 try pager.installBase(1, &base_old);
2487 try pager.installBase(1, &base_new);
2488 try pager.appendWal(1, 1, &committed);
2489 const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() });
2490 const current = try pager.beginRead();
2491
2492 try std.testing.expect(checkpoint.restarted);
2493 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
2494 try std.testing.expectEqual(@as(usize, 1), pager.base.items.len);
2495 try std.testing.expectEqual(@as(u8, 30), (try current.get(1)).?[1]);
2496 }
2497
2498 test "pager checkpoint prunes applied wal frame index without readers" {
2499 var pager = try testingPager(2);
2500 defer deinitTestingPager(&pager);
2501
2502 var first: [page.size]u8 = undefined;
2503 var second: [page.size]u8 = undefined;
2504 fillImage(&first, 1, 20);
2505 fillImage(&second, 2, 30);
2506
2507 try pager.appendWal(1, 0, &first);
2508 try pager.appendWal(2, 2, &second);
2509 const checkpoint = try pager.checkpoint(.{});
2510 const snapshot = try pager.beginRead();
2511
2512 try std.testing.expectEqual(@as(usize, 2), checkpoint.end_mark);
2513 try std.testing.expectEqual(@as(usize, 2), checkpoint.pages);
2514 try std.testing.expectEqual(@as(usize, 2), pager.frameCount());
2515 try std.testing.expectEqual(@as(usize, 0), pager.wal_index.frames.items.len);
2516 try std.testing.expectEqual(@as(usize, 0), pager.wal_index.pages.items.len);
2517 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
2518 try std.testing.expectEqual(@as(u8, 30), (try snapshot.get(2)).?[1]);
2519 }
2520
2521 test "pager checkpoint carries uncommitted tail index across wal rewrite" {
2522 var pager = try testingPager(3);
2523 defer deinitTestingPager(&pager);
2524
2525 var committed: [page.size]u8 = undefined;
2526 var tail: [page.size]u8 = undefined;
2527 var marker: [page.size]u8 = undefined;
2528 fillImage(&committed, 1, 20);
2529 fillImage(&tail, 2, 40);
2530 fillImage(&marker, 3, 60);
2531
2532 try pager.appendWal(1, 1, &committed);
2533 try pager.appendWal(2, 0, &tail);
2534 const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() });
2535 const before_commit = try pager.beginRead();
2536 try pager.appendWal(3, 3, &marker);
2537 const after_commit = try pager.beginRead();
2538
2539 try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark);
2540 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
2541 try std.testing.expect(checkpoint.restarted);
2542 try std.testing.expectEqual(@as(usize, 2), pager.frameCount());
2543 try std.testing.expectEqual(@as(usize, 2), pager.wal_index.frames.items.len);
2544 try std.testing.expectEqual(@as(usize, 2), pager.wal_index.pages.items.len);
2545 try std.testing.expectEqual(@as(u8, 20), (try before_commit.get(1)).?[1]);
2546 try std.testing.expect(try before_commit.get(2) == null);
2547 try std.testing.expectEqual(@as(u8, 20), (try after_commit.get(1)).?[1]);
2548 try std.testing.expectEqual(@as(u8, 40), (try after_commit.get(2)).?[1]);
2549 try std.testing.expectEqual(@as(u8, 60), (try after_commit.get(3)).?[1]);
2550 }
2551
2552 test "pager checkpoint restarts wal when no reader is active" {
2553 var pager = try testingPager(1);
2554 defer deinitTestingPager(&pager);
2555
2556 var first: [page.size]u8 = undefined;
2557 fillImage(&first, 1, 20);
2558
2559 try pager.appendWal(1, 1, &first);
2560 const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() });
2561 const snapshot = try pager.beginRead();
2562
2563 try std.testing.expect(checkpoint.restarted);
2564 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
2565 try std.testing.expectEqual(@as(usize, 0), pager.frameCount());
2566 try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark);
2567 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
2568 }
2569
2570 test "pager checkpoint honors the oldest reader end mark" {
2571 var pager = try testingPager(2);
2572 defer deinitTestingPager(&pager);
2573
2574 var first: [page.size]u8 = undefined;
2575 var second: [page.size]u8 = undefined;
2576 fillImage(&first, 1, 20);
2577 fillImage(&second, 2, 30);
2578
2579 try pager.appendWal(1, 1, &first);
2580 const oldest = try pager.beginRead();
2581 try pager.appendWal(2, 2, &second);
2582 const checkpoint = try pager.checkpoint(.{
2583 .readers = .{ .oldest = oldest.view },
2584 .restart_header = restartHeader(),
2585 });
2586 const base_view = View{ .base_generation = checkpoint.base_generation, .end_mark = 0 };
2587 const current = try pager.beginRead();
2588
2589 try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark);
2590 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
2591 try std.testing.expect(!checkpoint.restarted);
2592 try std.testing.expectEqual(@as(usize, 2), pager.frameCount());
2593 try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(1, base_view)).?[1]);
2594 try std.testing.expect(try pager.pageAt(2, base_view) == null);
2595 try std.testing.expectEqual(@as(u8, 30), (try current.get(2)).?[1]);
2596 }
2597
2598 test "pager checkpoint rewrites wal to an uncommitted tail" {
2599 var pager = try testingPager(2);
2600 defer deinitTestingPager(&pager);
2601
2602 var committed: [page.size]u8 = undefined;
2603 var uncommitted: [page.size]u8 = undefined;
2604 fillImage(&committed, 1, 20);
2605 fillImage(&uncommitted, 2, 99);
2606
2607 try pager.appendWal(1, 1, &committed);
2608 try pager.appendWal(2, 0, &uncommitted);
2609 const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() });
2610 const snapshot = try pager.beginRead();
2611
2612 try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark);
2613 try std.testing.expectEqual(@as(usize, 1), checkpoint.pages);
2614 try std.testing.expect(checkpoint.restarted);
2615 try std.testing.expectEqual(@as(usize, 1), pager.frameCount());
2616 try std.testing.expectEqual(@as(usize, 1), pager.wal_index.frames.items.len);
2617 try std.testing.expectEqual(wal.header_size + wal.frame_size, pager.walBytes().len);
2618 try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark);
2619 try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);
2620 try std.testing.expect(try snapshot.get(2) == null);
2621 }