tiny.sql.pager
Defined in tiny.sql.
API (33)
Actions
Public operations.
PreparedCheckpoint.checkpointPagePreparedCheckpoint.checkpointPageCountPreparedCheckpoint.deinitPreparedCheckpoint.resultWalIndex.Capacity.deriveWalIndex.activateWalIndex.deinitWalIndex.initWalIndex.reserve
Types and contracts
Public types and contracts.
CapacityCheckpointCheckpointOptionsCheckpointPageViewCheckpointPlanCheckpointReadersErrorInitOptionsMarkedImage: A committed page image and the check mark stored with it.PagerPreparedCheckpointSnapshotStorageViewWalIndexWalIndex.CapacityWalIndex.Capacity.DeriveErrorWalIndex.ExhaustionWalIndex.InitErrorWalIndex.LimitsWalIndex.Storage
Values and defaults
Public values and defaults.
Source
Source: lib/sql/src/pager.zig
zig
const std = @import("std");const alloc_phase = @import("alloc_phase");const page = @import("page.zig");const trace = @import("trace.zig");const wal = @import("wal.zig");const Allocator = std.mem.Allocator;pub const Error = Allocator.Error || wal.Error || error{ CapacityOverflow, CheckpointPlanCapacityExceeded, CheckpointPlanInUse, DurableCheckpointReaders, GenerationOverflow, PagerTooLarge, StaleCheckpoint, StorageTooShort,};pub const View = struct { base_generation: u64, end_mark: usize,};pub const CheckpointReaders = union(enum) { none, oldest: View,};pub const CheckpointOptions = struct { readers: CheckpointReaders = .none, restart_header: ?wal.Header = null,};pub const Checkpoint = struct { end_mark: usize, pages: usize, base_generation: u64, restarted: bool,};pub const Capacity = struct { base_pages: usize = 0, wal_frames: usize = 0, wal_pages: ?usize = null,};pub const InitOptions = struct { header: wal.Header, wal_frames: usize,};const BaseImage = struct { id: u32, generation: u64, bytes: [page.size]u8, /// The check mark that `markedPageAt` hands to readers. It starts clear /// and clears again when the bytes change in place. checked: bool = false,};const WalImage = struct { page_id: u32, frame: usize, offset: usize, previous: ?usize, /// The check mark that `markedPageAt` hands to readers. Frame bytes /// never change while the record exists, since restore and checkpoint /// restart drop the records of the frames they rewrite. checked: bool = false,};/// A committed page image and the check mark stored with it.pub const MarkedImage = struct { bytes: *const [page.size]u8, /// Null for an image in the unindexed log tail, which has no record. checked: ?*bool,};const ImageLocation = union(enum) { wal: usize, tail: *const [page.size]u8, base: usize,};const WalPage = struct { page_id: u32, frame_index: usize,};fn FixedList(comptime T: type) type { return struct { items: []T, buffer: []T, capacity: usize, fn initBuffer(buffer: []T) @This() { return .{ .items = buffer[0..0], .buffer = buffer, .capacity = buffer.len, }; } fn appendAssumeCapacity(self: *@This(), value: T) void { std.debug.assert(self.items.len < self.capacity); self.buffer[self.items.len] = value; self.items = self.buffer[0 .. self.items.len + 1]; } fn insertAssumeCapacity(self: *@This(), index: usize, value: T) void { std.debug.assert(index <= self.items.len); std.debug.assert(self.items.len < self.capacity); const next_len = self.items.len + 1; std.mem.copyBackwards( T, self.buffer[index + 1 .. next_len], self.buffer[index..self.items.len], ); self.buffer[index] = value; self.items = self.buffer[0..next_len]; } fn shrinkRetainingCapacity(self: *@This(), len: usize) void { std.debug.assert(len <= self.items.len); self.items = self.buffer[0..len]; } fn clearRetainingCapacity(self: *@This()) void { self.items = self.buffer[0..0]; } };}pub const WalIndex = struct { pub const storage_alignment: usize = @max(@alignOf(WalImage), @alignOf(WalPage)); pub const Storage = []align(storage_alignment) u8; pub const Limits = struct { frames: usize, }; pub const Capacity = struct { frames: usize, frame_bytes: usize, page_offset: usize, page_bytes: usize, storage_bytes: usize, pub const DeriveError = error{CapacityOverflow}; pub fn derive(limits: Limits) DeriveError!@This() { const frame_bytes = std.math.mul( usize, limits.frames, @sizeOf(WalImage), ) catch return error.CapacityOverflow; const page_offset = try alignForward(frame_bytes, @alignOf(WalPage)); const page_bytes = std.math.mul( usize, limits.frames, @sizeOf(WalPage), ) catch return error.CapacityOverflow; return .{ .frames = limits.frames, .frame_bytes = frame_bytes, .page_offset = page_offset, .page_bytes = page_bytes, .storage_bytes = std.math.add( usize, page_offset, page_bytes, ) catch return error.CapacityOverflow, }; } }; pub const InitError = WalIndex.Capacity.DeriveError || error{StorageTooShort}; pub const Exhaustion = error{WalFull}; pub const work_limits: alloc_phase.capacity.WorkLimits = .{ .transition_steps_max = std.math.maxInt(usize), .cleanup_steps_per_call_max = 0, .cleanup_calls_at_capacity_max = 0, }; pub const claim: alloc_phase.capacity.Declaration = .{ .source = .{ .id = "sql.wal_index", .kind = .phase_static, .limit_source = .caller, .storage = .{ .covered = &.{ .{ .id = "wal_frame_metadata", .lifetime = .steady, .detail = "exclusive mutable loan for one metadata descriptor per retained WAL frame", }, .{ .id = "wal_page_lookup", .lifetime = .steady, .detail = "exclusive mutable loan for at most one latest-frame lookup per distinct WAL page", }, }, .excluded = &.{ "WAL header and frame bytes owned by sql.wal_writer", "pager base images, base lookup index, and reader snapshots", "checkpoint plans, files, I/O state, and trace instrumentation", }, }, .capacity = .{ .inputs = &.{ alloc_phase.capacity.bindInput(Limits, "frames", "frames"), }, .type_selectors = &.{ alloc_phase.capacity.bindType(WalImage, "walimage"), alloc_phase.capacity.bindType(WalPage, "walpage"), }, .nodes = &.{ .{ .input = 0 }, .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } }, .{ .alignment = .{ .node = 1, .alignment = .{ .concrete_type = 1 } } }, .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 1 } } }, .{ .add = .{ .left = 2, .right = 3 } }, }, .assertions = &.{.{ .scope = .closure_total, .measure = .retained, .relation = .exact, .expression = 4, }}, }, .overload = .{ .kind = .reject_before_mutation, .detail = "checked capacity and short caller storage reject before activation; the parent WAL limit rejects max plus one before either index changes", }, .risks = .{ .transitive = .{ .status = .witnessed, .detail = "append, staged commit, recovery load, restore, and WAL restart use only the activated fixed frame and page regions", }, .foreign = .{ .status = .excluded, .detail = "the index stores offsets and frame positions while file and operating-system effects remain in the file database", }, }, .work = .{ .equation = "single operations visit at most frames descriptors; staged insertion and rebuild visit at most frames multiplied by frames descriptors", }, .obligations = &.{ .{ .key = "sql_wal_index_capacity", .role = .capacity_model }, .{ .key = "sql_wal_index_storage_rejection", .role = .initialization_failure }, .{ .key = "sql_wal_index_sealed_overload", .role = .overload }, .{ .key = "sql_wal_index_work_bound", .role = .work_bound }, .{ .key = "sql_wal_index_sealed_transitive_risk", .role = .transitive_risk }, }, }, .bindings = .{ .owner = @This(), .seal = .{ .family = alloc_phase.capacity.selector(@This().activate), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, .teardown = .{ .family = alloc_phase.capacity.selector(@This().deinit), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, }, }; phase: alloc_phase.capacity.Phase, capacity: WalIndex.Capacity, storage: WalIndex.Storage, frames: FixedList(WalImage), pages: FixedList(WalPage), pub fn init(storage: WalIndex.Storage, limits: Limits) InitError!WalIndex { const capacity = try WalIndex.Capacity.derive(limits); if (storage.len < capacity.storage_bytes) return error.StorageTooShort; const borrowed = storage[0..capacity.storage_bytes]; return .{ .phase = .initialization, .capacity = capacity, .storage = borrowed, .frames = .initBuffer(typedRegion( WalImage, borrowed, 0, capacity.frames, )), .pages = .initBuffer(typedRegion( WalPage, borrowed, capacity.page_offset, capacity.frames, )), }; } pub fn activate(self: *WalIndex) void { std.debug.assert(self.phase == .initialization); self.assertStorage(); self.phase = .steady; } pub fn reserve( self: *const WalIndex, additional_frames: usize, additional_pages: usize, ) Exhaustion!void { std.debug.assert(self.phase == .steady); if (additional_frames > self.frames.capacity - self.frames.items.len or additional_pages > self.pages.capacity - self.pages.items.len) { return error.WalFull; } } pub fn deinit(self: *WalIndex) WalIndex.Storage { std.debug.assert(self.phase != .teardown); self.assertStorage(); self.phase = .teardown; const storage = self.storage; self.* = undefined; return storage; } fn assertStorage(self: *const WalIndex) void { std.debug.assert(self.storage.len == self.capacity.storage_bytes); std.debug.assert(self.frames.capacity == self.capacity.frames); std.debug.assert(self.pages.capacity == self.capacity.frames); std.debug.assert(self.frames.items.len <= self.frames.capacity); std.debug.assert(self.pages.items.len <= self.pages.capacity); std.debug.assert(self.pages.items.len <= self.frames.items.len); } fn typedRegion( comptime T: type, storage: WalIndex.Storage, offset: usize, count: usize, ) []T { const byte_count = count * @sizeOf(T); const bytes: []align(@alignOf(T)) u8 = @alignCast( storage[offset..][0..byte_count], ); return std.mem.bytesAsSlice(T, bytes); } fn alignForward(value: usize, alignment: usize) error{CapacityOverflow}!usize { std.debug.assert(std.math.isPowerOfTwo(alignment)); const mask = alignment - 1; const padded = std.math.add(usize, value, mask) catch return error.CapacityOverflow; return padded & ~mask; }};comptime { alloc_phase.capacity.requireProvisionedRejectingOwnerShape(WalIndex);}const CheckpointPage = struct { page_id: u32, wal_offset: usize,};pub const CheckpointPageView = struct { page_id: u32, bytes: []const u8,};pub const CheckpointPlan = struct { pub const storage_alignment: usize = @alignOf(CheckpointPage); pub const Storage = []align(storage_alignment) u8; pub const Limits = struct { pages: usize, }; pub const Capacity = struct { pages: usize, storage_bytes: usize, pub const DeriveError = error{CapacityOverflow}; pub fn derive(limits: Limits) DeriveError!@This() { return .{ .pages = limits.pages, .storage_bytes = std.math.mul(usize, limits.pages, @sizeOf(CheckpointPage)) catch return error.CapacityOverflow, }; } }; pub const InitError = CheckpointPlan.Capacity.DeriveError || error{StorageTooShort}; pub const Exhaustion = error{ CheckpointPlanCapacityExceeded, CheckpointPlanInUse, }; pub const work_limits: alloc_phase.capacity.WorkLimits = .{ .transition_steps_max = 1, .cleanup_steps_per_call_max = 0, .cleanup_calls_at_capacity_max = 0, }; pub const claim: alloc_phase.capacity.Declaration = .{ .source = .{ .id = "sql.checkpoint_plan", .kind = .phase_static, .limit_source = .caller, .storage = .{ .covered = &.{ .{ .id = "reusable_maximum_latest_committed_wal_page_descript_860da2340194", .lifetime = .steady, .detail = "exclusive mutable loan for the reusable maximum latest committed WAL page descriptors for one prepared or synchronous checkpoint", }, }, .excluded = &.{ "pager base images, WAL page bytes, frame and page indexes, and reader snapshots", "file handles, base-file writes, WAL rewrites, and operating-system cache state", "prepared checkpoint metadata, caller state, and trace instrumentation", }, }, .capacity = .{ .inputs = &.{ alloc_phase.capacity.bindInput(Limits, "pages", "pages"), }, .type_selectors = &.{ alloc_phase.capacity.bindType(CheckpointPage, "checkpointpage"), }, .nodes = &.{ .{ .input = 0 }, .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } }, }, .assertions = &.{.{ .scope = .closure_total, .measure = .retained, .relation = .exact, .expression = 1, }}, }, .overload = .{ .kind = .reject_before_mutation, .detail = "checked descriptor arithmetic and short caller storage reject before activation; begin rejects max plus one and concurrent use before clearing the reusable prefix", }, .risks = .{ .transitive = .{ .status = .witnessed, .detail = "WAL page selection and descriptor page views remain allocation-free across repeated prepared and durable checkpoints after plan initialization", }, .foreign = .{ .status = .excluded, .detail = "base-file writes consume borrowed immutable WAL page bytes and WAL rewrite effects occur outside the plan-owned descriptor claim", }, }, .work = .{ .equation = "transition_steps <= transition_steps_max" }, .obligations = &.{ .{ .key = "sql_checkpoint_plan_capacity", .role = .capacity_model }, .{ .key = "sql_checkpoint_plan_storage_rejection", .role = .initialization_failure }, .{ .key = "sql_checkpoint_plan_sealed_overload", .role = .overload }, .{ .key = "sql_checkpoint_plan_work_bound", .role = .work_bound }, .{ .key = "sql_checkpoint_plan_sealed_transitive_risk", .role = .transitive_risk }, .{ .key = "sql_checkpoint_plan_semantics_transitive_risk", .role = .transitive_risk }, .{ .key = "sql_checkpoint_plan_semantics_foreign_risk", .role = .foreign_risk }, }, }, .bindings = .{ .owner = @This(), .seal = .{ .family = alloc_phase.capacity.selector(@This().activate), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, .teardown = .{ .family = alloc_phase.capacity.selector(@This().deinit), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, }, }; phase: alloc_phase.capacity.Phase, capacity: CheckpointPlan.Capacity, storage: CheckpointPlan.Storage, filled: usize = 0, admitted: usize = 0, in_use: bool = false, pub fn init(storage: CheckpointPlan.Storage, limits: Limits) InitError!CheckpointPlan { const capacity = try CheckpointPlan.Capacity.derive(limits); if (storage.len < capacity.storage_bytes) return error.StorageTooShort; return .{ .phase = .initialization, .capacity = capacity, .storage = storage[0..capacity.storage_bytes], }; } pub fn begin(self: *CheckpointPlan, required_pages: usize) error{ CheckpointPlanCapacityExceeded, CheckpointPlanInUse, }!void { std.debug.assert(self.phase == .steady); if (self.in_use) return error.CheckpointPlanInUse; if (required_pages > self.capacity.pages) return error.CheckpointPlanCapacityExceeded; self.filled = 0; self.admitted = required_pages; self.in_use = true; } pub fn append(self: *CheckpointPlan, checkpoint_page: CheckpointPage) error{CheckpointPlanCapacityExceeded}!void { std.debug.assert(self.phase == .steady); std.debug.assert(self.in_use); if (self.filled >= self.admitted) return error.CheckpointPlanCapacityExceeded; self.pageStorage()[self.filled] = checkpoint_page; self.filled += 1; } pub fn activate(self: *CheckpointPlan) void { std.debug.assert(self.phase == .initialization); std.debug.assert(self.filled == 0); std.debug.assert(!self.in_use); self.phase = .steady; } pub fn pages(self: *const CheckpointPlan) []const CheckpointPage { std.debug.assert(self.phase == .steady); std.debug.assert(self.in_use); std.debug.assert(self.filled == self.admitted); return self.pageStorageConst()[0..self.filled]; } pub fn release(self: *CheckpointPlan) void { std.debug.assert(self.phase == .steady); std.debug.assert(self.in_use); self.filled = 0; self.admitted = 0; self.in_use = false; } pub fn deinit(self: *CheckpointPlan) CheckpointPlan.Storage { std.debug.assert(self.phase != .teardown); std.debug.assert(!self.in_use); std.debug.assert(self.storage.len == self.capacity.storage_bytes); self.phase = .teardown; const storage = self.storage; self.* = undefined; return storage; } fn pageStorage(self: *CheckpointPlan) []CheckpointPage { return std.mem.bytesAsSlice(CheckpointPage, self.storage); } fn pageStorageConst(self: *const CheckpointPlan) []const CheckpointPage { return std.mem.bytesAsSlice(CheckpointPage, self.storage); }};comptime { alloc_phase.capacity.requireProvisionedRejectingOwnerShape(CheckpointPlan);}const PagerWorkspace = struct { journal: wal.Writer.Storage, checkpoint: CheckpointPlan.Storage, checkpoint_once: CheckpointPlan.Storage, wal_index: WalIndex.Storage, pub const Capacity = struct { journal: wal.Writer.Capacity, checkpoint: CheckpointPlan.Capacity, checkpoint_once: CheckpointPlan.Capacity, wal_index: WalIndex.Capacity, pub const DeriveError = wal.Writer.Capacity.DeriveError || CheckpointPlan.Capacity.DeriveError || WalIndex.Capacity.DeriveError; pub fn derive(options: InitOptions) DeriveError!@This() { const checkpoint = try CheckpointPlan.Capacity.derive(.{ .pages = options.wal_frames }); return .{ .journal = try wal.Writer.Capacity.derive(.{ .header = options.header, .frames = options.wal_frames, }), .checkpoint = checkpoint, .checkpoint_once = checkpoint, .wal_index = try WalIndex.Capacity.derive(.{ .frames = options.wal_frames, }), }; } }; pub const AllocateError = Allocator.Error || PagerWorkspace.Capacity.DeriveError; pub fn init( journal: wal.Writer.Storage, checkpoint: CheckpointPlan.Storage, checkpoint_once: CheckpointPlan.Storage, wal_index: WalIndex.Storage, ) PagerWorkspace { return .{ .journal = journal, .checkpoint = checkpoint, .checkpoint_once = checkpoint_once, .wal_index = wal_index, }; } pub fn allocate(allocator: Allocator, options: InitOptions) AllocateError!PagerWorkspace { const capacity = try PagerWorkspace.Capacity.derive(options); const journal = try allocator.alloc(u8, capacity.journal.storage_bytes); errdefer allocator.free(journal); const checkpoint = if (capacity.checkpoint.storage_bytes == 0) @as(CheckpointPlan.Storage, &.{}) else try allocator.alignedAlloc( u8, .fromByteUnits(CheckpointPlan.storage_alignment), capacity.checkpoint.storage_bytes, ); errdefer if (checkpoint.len != 0) allocator.free(checkpoint); const checkpoint_once = if (capacity.checkpoint_once.storage_bytes == 0) @as(CheckpointPlan.Storage, &.{}) else try allocator.alignedAlloc( u8, .fromByteUnits(CheckpointPlan.storage_alignment), capacity.checkpoint_once.storage_bytes, ); errdefer if (checkpoint_once.len != 0) allocator.free(checkpoint_once); const wal_index = if (capacity.wal_index.storage_bytes == 0) @as(WalIndex.Storage, &.{}) else try allocator.alignedAlloc( u8, .fromByteUnits(WalIndex.storage_alignment), capacity.wal_index.storage_bytes, ); return init(journal, checkpoint, checkpoint_once, wal_index); } pub fn deallocate(self: *PagerWorkspace, allocator: Allocator) void { if (self.wal_index.len != 0) allocator.free(self.wal_index); if (self.checkpoint_once.len != 0) allocator.free(self.checkpoint_once); if (self.checkpoint.len != 0) allocator.free(self.checkpoint); allocator.free(self.journal); self.* = undefined; }};const CheckpointState = struct { position: Pager.Position, base_generation: u64, base_images: usize, wal_pages: usize,};pub const PreparedCheckpoint = struct { pager: *const Pager, plan: *CheckpointPlan, checkpoint: Checkpoint, retain_generation: u64, restart_header: ?wal.Header, has_readers: bool, serial: u64, state: CheckpointState, pub fn result(self: PreparedCheckpoint) Checkpoint { return self.checkpoint; } pub fn checkpointPageCount(self: PreparedCheckpoint) usize { return self.plan.pages().len; } pub fn checkpointPage(self: PreparedCheckpoint, index: usize) CheckpointPageView { std.debug.assert(self.plan.phase == .steady); std.debug.assert(std.meta.eql(self.pager.position(), self.state.position)); const checkpoint_page = self.plan.pages()[index]; const bytes = self.pager.walBytes(); std.debug.assert(checkpoint_page.wal_offset <= bytes.len); std.debug.assert(page.size <= bytes.len - checkpoint_page.wal_offset); return .{ .page_id = checkpoint_page.page_id, .bytes = bytes[checkpoint_page.wal_offset..][0..page.size], }; } pub fn deinit(self: *PreparedCheckpoint) void { self.plan.release(); self.* = undefined; }};pub const Storage = struct { base_images: usize, base_capacity: usize, wal_frames: usize, wal_frame_capacity: usize, wal_pages: usize, wal_page_capacity: usize,};pub const Pager = struct { pub const Workspace = PagerWorkspace; allocator: Allocator, base: std.ArrayList(BaseImage) = .empty, base_index: std.AutoHashMapUnmanaged(u32, usize) = .empty, wal_index: WalIndex, journal: wal.Writer, checkpoint_plan: CheckpointPlan, checkpoint_once_plan: CheckpointPlan, base_generation: u64 = 0, base_page_count: u32 = 0, database_page_count: u32 = 0, end_mark: usize = 0, checkpoint_serial: u64 = 0, pub const Position = struct { journal: wal.Writer.Position, frames_len: usize, database_page_count: u32, end_mark: usize, }; pub const RestoreEpoch = struct { base_generation: u64, base_page_count: u32, checkpoint_serial: u64, }; pub fn init(allocator: Allocator, workspace: *Workspace, options: InitOptions) Error!Pager { const journal_limits: wal.Writer.Limits = .{ .header = options.header, .frames = options.wal_frames, }; const workspace_capacity = try Workspace.Capacity.derive(options); if (workspace.journal.len < workspace_capacity.journal.storage_bytes or workspace.checkpoint.len < workspace_capacity.checkpoint.storage_bytes or workspace.checkpoint_once.len < workspace_capacity.checkpoint_once.storage_bytes or workspace.wal_index.len < workspace_capacity.wal_index.storage_bytes) { return error.StorageTooShort; } var journal = wal.Writer.init(workspace.journal, journal_limits) catch |err| switch (err) { error.CapacityOverflow, error.StorageTooShort => unreachable, }; var checkpoint_plan = CheckpointPlan.init(workspace.checkpoint, .{ .pages = options.wal_frames, }) catch |err| switch (err) { error.CapacityOverflow, error.StorageTooShort => unreachable, }; var checkpoint_once_plan = CheckpointPlan.init(workspace.checkpoint_once, .{ .pages = options.wal_frames, }) catch |err| switch (err) { error.CapacityOverflow, error.StorageTooShort => unreachable, }; var wal_index = WalIndex.init(workspace.wal_index, .{ .frames = options.wal_frames, }) catch |err| switch (err) { error.CapacityOverflow, error.StorageTooShort => unreachable, }; journal.activate(); checkpoint_plan.activate(); checkpoint_once_plan.activate(); wal_index.activate(); workspace.* = undefined; return .{ .allocator = allocator, .wal_index = wal_index, .journal = journal, .checkpoint_plan = checkpoint_plan, .checkpoint_once_plan = checkpoint_once_plan, }; } pub fn deinit(self: *Pager) Workspace { self.base.deinit(self.allocator); self.base_index.deinit(self.allocator); const workspace: Workspace = .{ .journal = self.journal.deinit(), .checkpoint = self.checkpoint_plan.deinit(), .checkpoint_once = self.checkpoint_once_plan.deinit(), .wal_index = self.wal_index.deinit(), }; self.* = undefined; return workspace; } pub fn reserve(self: *Pager, capacity: Capacity) Error!void { const phase = trace.scope("pager.reserve"); defer phase.end(); if (capacity.wal_frames > self.journal.remainingFrames()) return error.WalFull; const wal_pages = capacity.wal_pages orelse capacity.wal_frames; try self.wal_index.reserve(capacity.wal_frames, wal_pages); try self.base.ensureTotalCapacityPrecise( self.allocator, try additionalCapacity(self.base.items.len, capacity.base_pages), ); try self.base_index.ensureUnusedCapacity(self.allocator, try hashMapSize(capacity.base_pages)); } pub fn replaceWal(self: *Pager, bytes: []const u8, committed_len: usize) Error!void { return self.replaceWalControlled(bytes, committed_len, .{}) catch |err| switch (err) { error.Interrupted => unreachable, else => return @errorCast(err), }; } pub fn replaceWalControlled( self: *Pager, bytes: []const u8, committed_len: usize, control: wal.Control, ) (Error || error{Interrupted})!void { const phase = trace.scope("pager.replace_wal"); defer phase.end(); if (committed_len < wal.header_size or (committed_len - wal.header_size) % wal.frame_size != 0) return error.InvalidWal; const frames_count = (committed_len - wal.header_size) / wal.frame_size; if (frames_count > self.journal.frameCapacity()) return error.WalFull; std.debug.assert(frames_count <= self.wal_index.frames.capacity); std.debug.assert(frames_count <= self.wal_index.pages.capacity); try self.journal.loadControlled(bytes, committed_len, control); try self.rebuildWalFramesFromJournalControlled(control); trace.progress("pager.replace_wal.complete"); } pub fn installBase(self: *Pager, page_id: u32, image: *const [page.size]u8) Error!void { const phase = trace.scope("pager.install_base"); defer phase.end(); const generation = try self.nextBaseGeneration(); try self.installBaseAtGeneration(page_id, image, generation); trace.progress("pager.install_base.complete"); } pub fn installBaseAtGeneration(self: *Pager, page_id: u32, image: *const [page.size]u8, generation: u64) Error!void { const phase = trace.scope("pager.install_base.generation"); defer phase.end(); if (self.base_index.get(page_id)) |index| { const cached = &self.base.items[index]; if (cached.id == page_id and cached.generation == generation) { cached.bytes = image.*; cached.checked = false; self.base_generation = @max(self.base_generation, generation); self.base_page_count = @max(self.base_page_count, page_id); self.database_page_count = @max(self.database_page_count, page_id); trace.progress("pager.install_base.generation.complete"); return; } } try self.base.ensureUnusedCapacity(self.allocator, 1); try self.base_index.ensureUnusedCapacity(self.allocator, 1); const index = self.base.items.len; self.base.appendAssumeCapacity(.{ .id = page_id, .generation = generation, .bytes = image.*, }); if (self.base_index.get(page_id)) |existing| { if (self.base.items[existing].generation <= generation) self.base_index.putAssumeCapacity(page_id, index); } else { self.base_index.putAssumeCapacity(page_id, index); } self.base_generation = @max(self.base_generation, generation); self.base_page_count = @max(self.base_page_count, page_id); self.database_page_count = @max(self.database_page_count, page_id); trace.progress("pager.install_base.generation.complete"); } pub fn appendWal(self: *Pager, page_id: u32, db_page_count: u32, image: *const [page.size]u8) Error!void { const phase = trace.scope("pager.append_wal"); defer phase.end(); if (self.journal.remainingFrames() == 0) return error.WalFull; const page_index = self.lowerBoundWalPage(page_id); const existing = page_index < self.wal_index.pages.items.len and self.wal_index.pages.items[page_index].page_id == page_id; try self.wal_index.reserve(1, @intFromBool(!existing)); try self.journal.append(page_id, db_page_count, image); self.indexAppendedWalFrame(page_id, db_page_count, page_index, existing); trace.progress("pager.append_wal.complete"); } pub fn walStagingCapacity(self: *const Pager, position_value: Position) usize { std.debug.assert(std.meta.eql(self.position(), position_value)); return self.journal.remainingFrames(); } pub fn stageWalPage(self: *Pager, position_value: Position, index: usize, page_id: u32, image: *const [page.size]u8) Error!void { std.debug.assert(std.meta.eql(self.position(), position_value)); try self.journal.stagePage(position_value.journal, index, page_id, image); } pub fn stagedWalPage(self: *const Pager, position_value: Position, index: usize) []const u8 { std.debug.assert(std.meta.eql(self.position(), position_value)); return self.journal.stagedPage(position_value.journal, index); } pub fn stagedWalPageMut(self: *Pager, position_value: Position, index: usize) *[page.size]u8 { std.debug.assert(std.meta.eql(self.position(), position_value)); return self.journal.stagedPageMut(position_value.journal, index); } pub fn stagedWalPageId(self: *const Pager, position_value: Position, index: usize) u32 { return self.journal.stagedPageId(position_value.journal, index); } pub fn swapStagedWalFrames(self: *Pager, position_value: Position, left_index: usize, right_index: usize) void { std.debug.assert(std.meta.eql(self.position(), position_value)); self.journal.swapStagedFrames(position_value.journal, left_index, right_index); } pub fn commitStagedWal(self: *Pager, position_value: Position, count: usize, database_page_count: u32) Error!void { const phase = trace.scope("pager.commit_staged_wal"); defer phase.end(); std.debug.assert(std.meta.eql(self.position(), position_value)); if (count > self.walStagingCapacity(position_value)) return error.WalFull; var new_pages: usize = 0; var previous_page_id: u32 = 0; for (0..count) |index| { const page_id = self.stagedWalPageId(position_value, index); std.debug.assert(page_id != 0); if (index != 0) std.debug.assert(page_id > previous_page_id); previous_page_id = page_id; const page_index = self.lowerBoundWalPage(page_id); if (page_index == self.wal_index.pages.items.len or self.wal_index.pages.items[page_index].page_id != page_id) { new_pages += 1; } } try self.wal_index.reserve(count, new_pages); for (0..count) |index| { const page_id = self.stagedWalPageId(position_value, index); const page_index = self.lowerBoundWalPage(page_id); const existing = page_index < self.wal_index.pages.items.len and self.wal_index.pages.items[page_index].page_id == page_id; self.journal.commitStagedFrame(position_value.journal, index, if (index + 1 == count) database_page_count else 0); self.indexAppendedWalFrame(page_id, if (index + 1 == count) database_page_count else 0, page_index, existing); } trace.progress("pager.commit_staged_wal.complete"); } pub fn beginRead(self: *const Pager) Error!Snapshot { const phase = trace.scope("pager.begin_read"); defer phase.end(); return .{ .pager = self, .view = try self.currentView() }; } pub fn checkpoint(self: *Pager, options: CheckpointOptions) Error!Checkpoint { const phase = trace.scope("pager.checkpoint"); defer phase.end(); var prepared = try self.prepareCheckpointWithPlan( &self.checkpoint_once_plan, options, ); defer prepared.deinit(); return try self.commitCheckpoint(prepared); } pub fn prepareCheckpoint(self: *Pager, options: CheckpointOptions) Error!PreparedCheckpoint { return try self.prepareCheckpointWithPlan(&self.checkpoint_plan, options); } fn prepareCheckpointWithPlan( self: *Pager, plan: *CheckpointPlan, options: CheckpointOptions, ) Error!PreparedCheckpoint { const phase = trace.scope("pager.checkpoint.prepare"); defer phase.end(); const current = try self.currentView(); const has_readers = switch (options.readers) { .none => false, .oldest => true, }; const target_mark = switch (options.readers) { .none => current.end_mark, .oldest => |oldest| @min(oldest.end_mark, current.end_mark), }; const can_rewrite_wal = options.restart_header != null and !has_readers; if (can_rewrite_wal) { const retained_frames = self.frameCount() - target_mark; std.debug.assert(retained_frames <= self.wal_index.frames.capacity); std.debug.assert(retained_frames <= self.wal_index.pages.capacity); } const bytes = self.walBytes(); try plan.begin(self.checkpointPageCount(target_mark, bytes.len)); errdefer plan.release(); const serial = std.math.add(u64, self.checkpoint_serial, 1) catch return error.GenerationOverflow; _ = std.math.add(u64, serial, 1) catch return error.GenerationOverflow; self.checkpoint_serial = serial; try self.collectCheckpointPages(target_mark, bytes.len, plan); const pages = plan.pages().len; var generation = self.base_generation; if (pages > 0) generation = try self.nextBaseGeneration(); const retain_generation = switch (options.readers) { .none => generation, .oldest => |oldest| @min(oldest.base_generation, generation), }; return .{ .pager = self, .plan = plan, .checkpoint = .{ .end_mark = target_mark, .pages = pages, .base_generation = generation, .restarted = can_rewrite_wal, }, .retain_generation = retain_generation, .restart_header = options.restart_header, .has_readers = has_readers, .serial = serial, .state = .{ .position = self.position(), .base_generation = self.base_generation, .base_images = self.base.items.len, .wal_pages = self.wal_index.pages.items.len, }, }; } pub fn commitCheckpoint(self: *Pager, prepared: PreparedCheckpoint) Error!Checkpoint { const phase = trace.scope("pager.checkpoint.commit_memory"); defer phase.end(); if (!self.preparedCheckpointCurrent(prepared)) return error.StaleCheckpoint; const checkpoint_value = prepared.checkpoint; if (checkpoint_value.pages > 0) { try self.installCheckpointPages(prepared, checkpoint_value.base_generation); self.base_generation = checkpoint_value.base_generation; } _ = try self.compactBaseHistory(prepared.retain_generation); self.finishCheckpointFrames(prepared); self.checkpoint_serial = prepared.serial + 1; trace.progress("pager.checkpoint.complete"); return checkpoint_value; } pub fn commitDurableCheckpoint(self: *Pager, prepared: PreparedCheckpoint) Error!Checkpoint { const phase = trace.scope("pager.checkpoint.commit_durable"); defer phase.end(); if (!self.preparedCheckpointCurrent(prepared)) return error.StaleCheckpoint; if (prepared.has_readers) return error.DurableCheckpointReaders; const checkpoint_value = prepared.checkpoint; if (checkpoint_value.pages > 0) { self.base_generation = checkpoint_value.base_generation; for (prepared.plan.pages()) |checkpoint_page| { self.base_page_count = @max(self.base_page_count, checkpoint_page.page_id); self.database_page_count = @max(self.database_page_count, checkpoint_page.page_id); } } self.releaseDurableBase(); self.finishCheckpointFrames(prepared); self.checkpoint_serial = prepared.serial + 1; trace.progress("pager.checkpoint.complete"); return checkpoint_value; } pub fn storage(self: *const Pager) Storage { return .{ .base_images = self.base.items.len, .base_capacity = self.base.capacity, .wal_frames = self.wal_index.frames.items.len, .wal_frame_capacity = self.wal_index.frames.capacity, .wal_pages = self.wal_index.pages.items.len, .wal_page_capacity = self.wal_index.pages.capacity, }; } pub fn releaseDurableBase(self: *Pager) void { self.base.deinit(self.allocator); self.base_index.deinit(self.allocator); self.base = .empty; self.base_index = .empty; } pub fn currentView(self: *const Pager) Error!View { return .{ .base_generation = self.base_generation, .end_mark = self.end_mark, }; } pub fn pageAt(self: *const Pager, page_id: u32, view: View) Error!?[]const u8 { const location = try self.locateImage(page_id, view) orelse return null; return switch (location) { .wal => |index| self.walImageBytes(index), .tail => |bytes| bytes, .base => |index| self.base.items[index].bytes[0..], }; } /// Returns the image `pageAt` returns, with the check mark stored with /// it. The pager only clears a mark, when it stores new bytes under it. /// A reader sets it after the image passes the reader's checks, so later /// readers can skip them. pub fn markedPageAt(self: *Pager, page_id: u32, view: View) Error!?MarkedImage { const location = try self.locateImage(page_id, view) orelse return null; return switch (location) { .wal => |index| .{ .bytes = self.walImageBytes(index), .checked = &self.wal_index.frames.items[index].checked, }, .tail => |bytes| .{ .bytes = bytes, .checked = null }, .base => |index| .{ .bytes = &self.base.items[index].bytes, .checked = &self.base.items[index].checked, }, }; } fn locateImage(self: *const Pager, page_id: u32, view: View) Error!?ImageLocation { const phase = trace.scope("pager.page_at"); defer phase.end(); if (self.indexedWalImage(page_id, view.end_mark)) |index| return .{ .wal = index }; if (view.end_mark > self.frameCount()) { if (try wal.pageAt(self.walBytes(), page_id, view.end_mark)) |bytes| { return .{ .tail = bytes[0..page.size] }; } } if (self.baseVisibleIndex(page_id, view.base_generation)) |index| return .{ .base = index }; return null; } fn walImageBytes(self: *const Pager, index: usize) *const [page.size]u8 { return self.walBytes()[self.wal_index.frames.items[index].offset..][0..page.size]; } pub fn frameCount(self: *const Pager) usize { return self.journal.frameCount(); } pub fn position(self: *const Pager) Position { return .{ .journal = self.journal.position(), .frames_len = self.wal_index.frames.items.len, .database_page_count = self.database_page_count, .end_mark = self.end_mark, }; } pub fn restoreEpoch(self: *const Pager) RestoreEpoch { return .{ .base_generation = self.base_generation, .base_page_count = self.base_page_count, .checkpoint_serial = self.checkpoint_serial, }; } pub fn canRestore( self: *const Pager, position_value: Position, epoch: RestoreEpoch, ) bool { if (!std.meta.eql(self.restoreEpoch(), epoch)) return false; if (self.wal_index.frames.items.len < position_value.frames_len) return false; if (self.journal.position().len < position_value.journal.len) return false; return true; } pub fn restore(self: *Pager, position_value: Position) void { self.journal.restore(position_value.journal); self.wal_index.frames.shrinkRetainingCapacity(position_value.frames_len); self.database_page_count = position_value.database_page_count; self.end_mark = position_value.end_mark; self.rebuildWalPages(); } pub fn walBytes(self: *const Pager) []const u8 { return self.journal.bytes(); } pub fn walCapacityBytes(self: *const Pager) usize { return self.journal.byteCapacity(); } pub fn baseGeneration(self: *const Pager) u64 { return self.base_generation; } pub fn setBasePageCount(self: *Pager, count: u32) void { self.base_page_count = @max(self.base_page_count, count); self.database_page_count = @max(self.database_page_count, count); if (count > 0 and self.base_generation == 0) self.base_generation = 1; } pub fn basePageCount(self: *const Pager) u32 { return self.base_page_count; } pub fn databasePageCount(self: *const Pager) u32 { return self.database_page_count; } fn lowerBoundWalPage(self: *const Pager, page_id: u32) usize { var low: usize = 0; var high = self.wal_index.pages.items.len; while (low < high) { const mid = low + (high - low) / 2; if (self.wal_index.pages.items[mid].page_id < page_id) { low = mid + 1; } else { high = mid; } } return low; } fn indexAppendedWalFrame(self: *Pager, page_id: u32, db_page_count: u32, page_index: usize, existing: bool) void { const frame = self.journal.frameCount(); if (db_page_count != 0) self.end_mark = frame; self.database_page_count = @max(self.database_page_count, @max(page_id, db_page_count)); const frame_index = self.wal_index.frames.items.len; self.wal_index.frames.appendAssumeCapacity(.{ .page_id = page_id, .frame = frame, .offset = self.walBytes().len - page.size, .previous = if (existing) self.wal_index.pages.items[page_index].frame_index else null, }); if (existing) { self.wal_index.pages.items[page_index].frame_index = frame_index; } else { self.wal_index.pages.insertAssumeCapacity(page_index, .{ .page_id = page_id, .frame_index = frame_index, }); } } fn lowerBoundBase(self: *const Pager, page_id: u32, generation: u64) usize { var low: usize = 0; var high = self.base.items.len; while (low < high) { const mid = low + (high - low) / 2; const image = self.base.items[mid]; if (image.id < page_id or (image.id == page_id and image.generation < generation)) { low = mid + 1; } else { high = mid; } } return low; } fn baseVisibleIndex(self: *const Pager, page_id: u32, generation: u64) ?usize { if (self.base_index.get(page_id)) |index| { const image = &self.base.items[index]; if (image.id == page_id and image.generation <= generation) return index; } var best_index: ?usize = null; var best_generation: u64 = 0; for (self.base.items, 0..) |*image, index| { if (image.id == page_id and image.generation <= generation and (best_index == null or image.generation > best_generation)) { best_index = index; best_generation = image.generation; } } return best_index; } fn upperBoundBase(self: *const Pager, page_id: u32, generation: u64) usize { var low: usize = 0; var high = self.base.items.len; while (low < high) { const mid = low + (high - low) / 2; const image = self.base.items[mid]; if (image.id < page_id or (image.id == page_id and image.generation <= generation)) { low = mid + 1; } else { high = mid; } } return low; } fn indexedWalImage(self: *const Pager, page_id: u32, max_frame: usize) ?usize { const page_index = self.lowerBoundWalPage(page_id); if (page_index == self.wal_index.pages.items.len or self.wal_index.pages.items[page_index].page_id != page_id) { return null; } const wal_len = self.walBytes().len; var frame_index: ?usize = self.wal_index.pages.items[page_index].frame_index; while (frame_index) |index| { const image = self.wal_index.frames.items[index]; if (image.frame <= max_frame and image.offset + page.size <= wal_len) return index; frame_index = image.previous; } return null; } fn latestCheckpointFrame(self: *const Pager, wal_page: WalPage, target_mark: usize, wal_len: usize) ?WalImage { var frame_index: ?usize = wal_page.frame_index; while (frame_index) |index| { const image = self.wal_index.frames.items[index]; if (image.frame <= target_mark and image.offset + page.size <= wal_len) return image; frame_index = image.previous; } return null; } fn checkpointPageCount(self: *const Pager, target_mark: usize, wal_len: usize) usize { var count: usize = 0; for (self.wal_index.pages.items) |wal_page| { if (self.latestCheckpointFrame(wal_page, target_mark, wal_len) != null) count += 1; } return count; } fn collectCheckpointPages( self: *const Pager, target_mark: usize, wal_len: usize, plan: *CheckpointPlan, ) error{CheckpointPlanCapacityExceeded}!void { for (self.wal_index.pages.items) |wal_page| { const image = self.latestCheckpointFrame(wal_page, target_mark, wal_len) orelse continue; try plan.append(.{ .page_id = image.page_id, .wal_offset = image.offset, }); } } fn installCheckpointPages(self: *Pager, prepared: PreparedCheckpoint, generation: u64) Error!void { const checkpoint_page_count = prepared.checkpointPageCount(); try self.base.ensureUnusedCapacity(self.allocator, checkpoint_page_count); try self.base_index.ensureUnusedCapacity(self.allocator, try hashMapSize(checkpoint_page_count)); for (0..checkpoint_page_count) |page_index| { const checkpoint_page = prepared.checkpointPage(page_index); const index = self.base.items.len; self.base.appendAssumeCapacity(.{ .id = checkpoint_page.page_id, .generation = generation, .bytes = checkpoint_page.bytes[0..page.size].*, }); if (self.base_index.get(checkpoint_page.page_id)) |existing| { if (self.base.items[existing].generation <= generation) self.base_index.putAssumeCapacity(checkpoint_page.page_id, index); } else { self.base_index.putAssumeCapacity(checkpoint_page.page_id, index); } self.base_page_count = @max(self.base_page_count, checkpoint_page.page_id); self.database_page_count = @max(self.database_page_count, checkpoint_page.page_id); } } fn preparedCheckpointCurrent(self: *const Pager, prepared: PreparedCheckpoint) bool { if (prepared.pager != self) return false; if (self.checkpoint_serial != prepared.serial) return false; if (!std.meta.eql(self.position(), prepared.state.position)) return false; if (self.base_generation != prepared.state.base_generation) return false; if (self.base.items.len != prepared.state.base_images) return false; if (self.wal_index.pages.items.len != prepared.state.wal_pages) return false; return true; } fn finishCheckpointFrames(self: *Pager, prepared: PreparedCheckpoint) void { const checkpoint_value = prepared.checkpoint; if (!prepared.has_readers and !checkpoint_value.restarted) _ = self.compactWalFrames(checkpoint_value.end_mark); if (checkpoint_value.restarted) { self.journal.rewriteTail(checkpoint_value.end_mark, prepared.restart_header.?); self.rebuildWalFramesFromJournal(); } } fn compactBaseHistory(self: *Pager, retain_generation: u64) Error!usize { const phase = trace.scope("pager.base_history.compact"); defer phase.end(); var retained_floor: std.AutoHashMapUnmanaged(u32, usize) = .empty; defer retained_floor.deinit(self.allocator); try retained_floor.ensureTotalCapacity(self.allocator, try hashMapSize(self.base.items.len)); for (self.base.items, 0..) |*image, index| { if (image.generation > retain_generation) continue; if (retained_floor.get(image.id)) |existing| { if (self.base.items[existing].generation < image.generation) retained_floor.putAssumeCapacity(image.id, index); } else { retained_floor.putAssumeCapacity(image.id, index); } } var write_index: usize = 0; for (self.base.items, 0..) |image, index| { const keep = image.generation > retain_generation or (retained_floor.get(image.id) orelse std.math.maxInt(usize)) == index; if (keep) { self.base.items[write_index] = image; write_index += 1; } } const removed = self.base.items.len - write_index; self.base.shrinkRetainingCapacity(write_index); try self.rebuildBaseIndex(); if (removed > 0) trace.progress("pager.base_history.compact.complete"); return removed; } fn compactWalFrames(self: *Pager, checkpoint_mark: usize) usize { const phase = trace.scope("pager.wal_frames.compact"); defer phase.end(); var write_index: usize = 0; for (self.wal_index.frames.items) |image| { if (image.frame > checkpoint_mark) { self.wal_index.frames.items[write_index] = image; write_index += 1; } } const removed = self.wal_index.frames.items.len - write_index; self.wal_index.frames.shrinkRetainingCapacity(write_index); if (removed > 0) { self.rebuildWalPages(); trace.progress("pager.wal_frames.compact.complete"); } return removed; } fn rebuildWalFramesFromJournal(self: *Pager) void { self.rebuildWalFramesFromJournalControlled(.{}) catch unreachable; } fn rebuildWalFramesFromJournalControlled( self: *Pager, control: wal.Control, ) error{Interrupted}!void { self.wal_index.frames.clearRetainingCapacity(); self.wal_index.pages.clearRetainingCapacity(); self.database_page_count = self.base_page_count; self.end_mark = 0; var reader = wal.Reader.initControlled(self.walBytes(), control) catch |err| switch (err) { error.Interrupted => return error.Interrupted, else => unreachable, }; const frames_max = self.journal.frameCount(); var frame_index: usize = 0; while (frame_index < frames_max) : (frame_index += 1) { const frame = (reader.nextControlled(control) catch |err| switch (err) { error.Interrupted => return error.Interrupted, else => unreachable, }) orelse unreachable; self.wal_index.frames.appendAssumeCapacity(.{ .page_id = frame.page_id, .frame = frame.index, .offset = wal.header_size + (frame.index - 1) * wal.frame_size + wal.frame_header_size, .previous = null, }); self.database_page_count = @max(self.database_page_count, @max(frame.page_id, frame.db_page_count)); if (frame.committed()) self.end_mark = frame.index; } try self.rebuildWalPagesControlled(control); } fn rebuildWalPages(self: *Pager) void { self.rebuildWalPagesControlled(.{}) catch unreachable; } fn rebuildWalPagesControlled( self: *Pager, control: wal.Control, ) error{Interrupted}!void { self.wal_index.pages.clearRetainingCapacity(); for (self.wal_index.frames.items, 0..) |*image, index| { try control.check(); const page_index = self.lowerBoundWalPage(image.page_id); if (page_index < self.wal_index.pages.items.len and self.wal_index.pages.items[page_index].page_id == image.page_id) { image.previous = self.wal_index.pages.items[page_index].frame_index; self.wal_index.pages.items[page_index].frame_index = index; } else { image.previous = null; self.wal_index.pages.insertAssumeCapacity(page_index, .{ .page_id = image.page_id, .frame_index = index, }); } } try control.check(); } fn rebuildBaseIndex(self: *Pager) Error!void { try self.base_index.ensureTotalCapacity(self.allocator, try hashMapSize(self.base.items.len)); self.base_index.clearRetainingCapacity(); for (self.base.items, 0..) |*image, index| { if (self.base_index.get(image.id)) |existing| { if (self.base.items[existing].generation <= image.generation) self.base_index.putAssumeCapacity(image.id, index); } else { self.base_index.putAssumeCapacity(image.id, index); } } } fn nextBaseGeneration(self: *const Pager) Error!u64 { if (self.base_generation == std.math.maxInt(u64)) return error.GenerationOverflow; return self.base_generation + 1; }};fn hashMapSize(count: usize) Error!u32 { if (count > std.math.maxInt(u32)) return error.PagerTooLarge; return @intCast(count);}fn additionalCapacity(current: usize, additional: usize) Error!usize { return std.math.add(usize, current, additional) catch error.PagerTooLarge;}pub const Snapshot = struct { pager: *const Pager, view: View, pub fn get(self: Snapshot, page_id: u32) Error!?[]const u8 { const phase = trace.scope("pager.snapshot.get"); defer phase.end(); return self.pager.pageAt(page_id, self.view); }};fn testingHeader() wal.Header { return .{ .sequence = 31, .salt = .{ .first = 0x5151_7171, .second = 0x9191_b1b1 }, };}fn restartHeader() wal.Header { return .{ .sequence = 32, .salt = .{ .first = 0xc1c1_d1d1, .second = 0xe1e1_f1f1 }, };}fn fillImage(image: *[page.size]u8, page_id: u32, value: u8) void { @memset(image, 0); image[0] = @intCast(page_id); image[1] = value;}fn testingPager(wal_frames: usize) !Pager { const options: InitOptions = .{ .header = testingHeader(), .wal_frames = wal_frames, }; var workspace = try Pager.Workspace.allocate(std.testing.allocator, options); return Pager.init(std.testing.allocator, &workspace, options) catch |err| { workspace.deallocate(std.testing.allocator); return err; };}fn deinitTestingPager(pager: *Pager) void { var workspace = pager.deinit(); workspace.deallocate(std.testing.allocator);}test "pager workspace rejects short regions before transfer and returns exact loans" { const wal_frames = 2; const options: InitOptions = .{ .header = testingHeader(), .wal_frames = wal_frames, }; const capacity = try Pager.Workspace.Capacity.derive(options); const wal_index_capacity = comptime WalIndex.Capacity.derive(.{ .frames = wal_frames, }) catch unreachable; var journal_storage: [wal.header_size + wal_frames * wal.frame_size]u8 align(wal.Writer.storage_alignment) = undefined; var checkpoint_storage: [wal_frames * @sizeOf(CheckpointPage)]u8 align(CheckpointPlan.storage_alignment) = undefined; var checkpoint_once_storage: [wal_frames * @sizeOf(CheckpointPage)]u8 align(CheckpointPlan.storage_alignment) = undefined; var wal_index_storage: [wal_index_capacity.storage_bytes]u8 align(WalIndex.storage_alignment) = undefined; var workspace = Pager.Workspace.init( &journal_storage, &checkpoint_storage, &checkpoint_once_storage, &wal_index_storage, ); try std.testing.expectEqual(journal_storage.len, capacity.journal.storage_bytes); try std.testing.expectEqual(checkpoint_storage.len, capacity.checkpoint.storage_bytes); try std.testing.expectEqual(checkpoint_once_storage.len, capacity.checkpoint_once.storage_bytes); try std.testing.expectEqual(wal_index_storage.len, capacity.wal_index.storage_bytes); @memset(workspace.journal, 0xa5); var short_journal_workspace = Pager.Workspace.init( workspace.journal[0 .. capacity.journal.storage_bytes - 1], workspace.checkpoint, workspace.checkpoint_once, workspace.wal_index, ); try std.testing.expectError(error.StorageTooShort, Pager.init( std.testing.allocator, &short_journal_workspace, options, )); try std.testing.expectEqual( journal_storage[0..].ptr, short_journal_workspace.journal.ptr, ); var short_checkpoint_workspace = Pager.Workspace.init( workspace.journal, workspace.checkpoint[0 .. capacity.checkpoint.storage_bytes - 1], workspace.checkpoint_once, workspace.wal_index, ); try std.testing.expectError(error.StorageTooShort, Pager.init( std.testing.allocator, &short_checkpoint_workspace, options, )); try std.testing.expectEqual( checkpoint_storage[0..].ptr, short_checkpoint_workspace.checkpoint.ptr, ); var short_wal_index_workspace = Pager.Workspace.init( workspace.journal, workspace.checkpoint, workspace.checkpoint_once, workspace.wal_index[0 .. capacity.wal_index.storage_bytes - 1], ); try std.testing.expectError(error.StorageTooShort, Pager.init( std.testing.allocator, &short_wal_index_workspace, options, )); try std.testing.expectEqual( wal_index_storage[0..].ptr, short_wal_index_workspace.wal_index.ptr, ); for (workspace.journal) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte); const journal_pointer = workspace.journal.ptr; const checkpoint_pointer = workspace.checkpoint.ptr; const checkpoint_once_pointer = workspace.checkpoint_once.ptr; const wal_index_pointer = workspace.wal_index.ptr; var pager = try Pager.init(std.testing.allocator, &workspace, options); workspace = pager.deinit(); try std.testing.expectEqual(journal_pointer, workspace.journal.ptr); try std.testing.expectEqual(checkpoint_pointer, workspace.checkpoint.ptr); try std.testing.expectEqual(checkpoint_once_pointer, workspace.checkpoint_once.ptr); try std.testing.expectEqual(wal_index_pointer, workspace.wal_index.ptr);}test "pager snapshot reads base page without wal frames" { var pager = try testingPager(0); defer deinitTestingPager(&pager); var base: [page.size]u8 = undefined; fillImage(&base, 1, 10); try pager.installBase(1, &base); const snapshot = try pager.beginRead(); try std.testing.expectEqual(@as(u64, 1), snapshot.view.base_generation); try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark); const image = (try snapshot.get(1)).?; try std.testing.expectEqual(@as(u8, 10), image[1]);}test "pager reserve keeps base wal and checkpoint behavior intact" { var pager = try testingPager(2); defer deinitTestingPager(&pager); try pager.reserve(.{ .base_pages = 2, .wal_frames = 2 }); var base: [page.size]u8 = undefined; var wal_image: [page.size]u8 = undefined; fillImage(&base, 1, 10); fillImage(&wal_image, 1, 20); try pager.installBase(1, &base); try pager.appendWal(1, 1, &wal_image); const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() }); const snapshot = try pager.beginRead(); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);}test "pager reserve uses sealed wal frame and page index capacities" { var pager = try testingPager(32); defer deinitTestingPager(&pager); try pager.reserve(.{ .wal_frames = 32, .wal_pages = 2 }); try std.testing.expectEqual(@as(usize, 32), pager.wal_index.frames.capacity); try std.testing.expectEqual(@as(usize, 32), pager.wal_index.pages.capacity);}test "pager prepared checkpoint preserves state and rejects stale commit" { var pager = try testingPager(3); defer deinitTestingPager(&pager); try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 }); var committed_first: [page.size]u8 = undefined; var committed_second: [page.size]u8 = undefined; var tail: [page.size]u8 = undefined; fillImage(&committed_first, 2, 20); fillImage(&committed_second, 1, 25); fillImage(&tail, 3, 30); try pager.appendWal(2, 0, &committed_first); try pager.appendWal(1, 2, &committed_second); const before = pager.storage(); var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() }); defer prepared.deinit(); try std.testing.expectEqual(@as(usize, 2), prepared.result().pages); try std.testing.expectEqual(@as(u32, 1), prepared.checkpointPage(0).page_id); try std.testing.expectEqual(@as(u32, 2), prepared.checkpointPage(1).page_id); try std.testing.expectEqual(@as(usize, 2), pager.frameCount()); try std.testing.expectEqual(before.base_images, pager.storage().base_images); try std.testing.expectEqual(@as(u64, 0), pager.baseGeneration()); var peer = try testingPager(3); defer deinitTestingPager(&peer); try peer.reserve(.{ .wal_frames = 3, .wal_pages = 3 }); try peer.appendWal(2, 0, &committed_first); try peer.appendWal(1, 2, &committed_second); var peer_prepared = try peer.prepareCheckpoint(.{ .restart_header = restartHeader() }); defer peer_prepared.deinit(); try std.testing.expectError(error.StaleCheckpoint, peer.commitCheckpoint(prepared)); try pager.appendWal(3, 0, &tail); try std.testing.expectError(error.StaleCheckpoint, pager.commitCheckpoint(prepared)); const checkpoint_value = try pager.checkpoint(.{ .restart_header = restartHeader() }); try std.testing.expect(checkpoint_value.restarted); try std.testing.expectEqual(@as(usize, 2), checkpoint_value.pages);}test "pager durable checkpoint drops memory base and leaves file fallback" { var pager = try testingPager(1); defer deinitTestingPager(&pager); try pager.reserve(.{ .base_pages = 1, .wal_frames = 1, .wal_pages = 1 }); var base: [page.size]u8 = undefined; var committed: [page.size]u8 = undefined; fillImage(&base, 1, 10); fillImage(&committed, 1, 20); try pager.installBase(1, &base); try pager.appendWal(1, 1, &committed); var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() }); defer prepared.deinit(); const checkpoint_value = try pager.commitDurableCheckpoint(prepared); try std.testing.expect(checkpoint_value.restarted); try std.testing.expectEqual(@as(u64, 2), checkpoint_value.base_generation); try std.testing.expectEqual(@as(usize, 0), pager.storage().base_images); try std.testing.expectEqual(@as(usize, 0), pager.storage().base_capacity); try std.testing.expect(try pager.pageAt(1, .{ .base_generation = checkpoint_value.base_generation, .end_mark = 0 }) == null);}test "pager concurrent preparation rejects without invalidating prepared page views" { var pager = try testingPager(2); defer deinitTestingPager(&pager); try pager.reserve(.{ .wal_frames = 2, .wal_pages = 2 }); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 20); fillImage(&second, 2, 30); try pager.appendWal(1, 1, &first); const oldest = try pager.currentView(); try pager.appendWal(2, 2, &second); var prepared = try pager.prepareCheckpoint(.{ .readers = .{ .oldest = oldest } }); defer prepared.deinit(); try std.testing.expectEqual(@as(usize, 1), prepared.checkpointPageCount()); try std.testing.expectError( error.CheckpointPlanInUse, pager.prepareCheckpoint(.{ .restart_header = restartHeader() }), ); try std.testing.expectEqual(@as(usize, 1), prepared.checkpointPageCount()); try std.testing.expectEqual(@as(u32, 1), prepared.checkpointPage(0).page_id); _ = try pager.commitCheckpoint(prepared);}test "pager durable checkpoint rejects active readers" { var pager = try testingPager(1); defer deinitTestingPager(&pager); try pager.reserve(.{ .base_pages = 1, .wal_frames = 1, .wal_pages = 1 }); var base: [page.size]u8 = undefined; var committed: [page.size]u8 = undefined; fillImage(&base, 1, 10); fillImage(&committed, 1, 20); try pager.installBase(1, &base); const oldest = try pager.beginRead(); try pager.appendWal(1, 1, &committed); var prepared = try pager.prepareCheckpoint(.{ .readers = .{ .oldest = oldest.view } }); defer prepared.deinit(); try std.testing.expectError(error.DurableCheckpointReaders, pager.commitDurableCheckpoint(prepared)); _ = try pager.commitCheckpoint(prepared); try std.testing.expectEqual(@as(u8, 10), (try oldest.get(1)).?[1]);}fn modelWalIndexCapacity(limits: WalIndex.Limits) ?WalIndex.Capacity { const frame_bytes = @as(u128, limits.frames) * @sizeOf(WalImage); const page_alignment = @as(u128, @alignOf(WalPage)); const page_offset = (frame_bytes + page_alignment - 1) & ~(page_alignment - 1); const page_bytes = @as(u128, limits.frames) * @sizeOf(WalPage); const storage_bytes = page_offset + page_bytes; const maximum = std.math.maxInt(usize); if (frame_bytes > maximum or page_offset > maximum or page_bytes > maximum or storage_bytes > maximum) { return null; } return .{ .frames = limits.frames, .frame_bytes = @intCast(frame_bytes), .page_offset = @intCast(page_offset), .page_bytes = @intCast(page_bytes), .storage_bytes = @intCast(storage_bytes), };}test "wal index capacity matches independent aligned typed regions" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_capacity"), null, null, null, null, null, null, ); } for (0..4097) |frames| { const limits: WalIndex.Limits = .{ .frames = frames }; try std.testing.expectEqual( modelWalIndexCapacity(limits).?, try WalIndex.Capacity.derive(limits), ); } const overflow: WalIndex.Limits = .{ .frames = std.math.maxInt(usize) }; try std.testing.expect(modelWalIndexCapacity(overflow) == null); try std.testing.expectError(error.CapacityOverflow, WalIndex.Capacity.derive(overflow));}test "wal index rejects short storage and releases its exact borrow" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_storage_rejection"), null, null, null, null, null, null, ); @stardustClaim( @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_work_bound"), null, null, null, null, null, null, ); } const limits: WalIndex.Limits = .{ .frames = 3 }; const capacity = try WalIndex.Capacity.derive(limits); const storage = try std.testing.allocator.alignedAlloc( u8, .fromByteUnits(WalIndex.storage_alignment), capacity.storage_bytes, ); defer std.testing.allocator.free(storage); try std.testing.expectError( error.StorageTooShort, WalIndex.init(storage[0 .. storage.len - 1], limits), ); var index = try WalIndex.init(storage, limits); try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, index.phase); try std.testing.expectEqual(@as(usize, 3), index.frames.capacity); try std.testing.expectEqual(@as(usize, 3), index.pages.capacity); index.activate(); try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, index.phase); const released = index.deinit(); try std.testing.expectEqual(storage.ptr, released.ptr); try std.testing.expectEqual(storage.len, released.len);}test "wal index stays sealed through transitive operations and overload" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_sealed_overload"), null, null, null, null, null, null, ); @stardustClaim( @import("alloc_phase").capacity.witness(WalIndex, "sql_wal_index_sealed_transitive_risk"), null, null, null, null, null, null, ); } const allocator = std.testing.allocator; var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(allocator); const initialization_allocator = phase_allocator.initializationAllocator(); const options: InitOptions = .{ .header = testingHeader(), .wal_frames = 3, }; var maybe_workspace: ?Pager.Workspace = try Pager.Workspace.allocate( initialization_allocator, options, ); var maybe_pager: ?Pager = null; defer { if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization(); if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown(); const teardown_allocator = phase_allocator.teardownAllocator(); if (maybe_pager) |*pager| { var workspace = pager.deinit(); workspace.deallocate(teardown_allocator); } else if (maybe_workspace) |*workspace| { workspace.deallocate(teardown_allocator); } phase_allocator.deinit(); } maybe_pager = try Pager.init(initialization_allocator, &maybe_workspace.?, options); maybe_workspace = null; const pager = &maybe_pager.?; const index_pointer = @intFromPtr(pager.wal_index.storage.ptr); phase_allocator.seal(); try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 }); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; var third: [page.size]u8 = undefined; var rejected: [page.size]u8 = undefined; fillImage(&first, 1, 10); fillImage(&second, 2, 20); fillImage(&third, 3, 30); fillImage(&rejected, 4, 40); try pager.appendWal(1, 1, &first); const after_first = pager.position(); try pager.appendWal(2, 0, &second); try pager.appendWal(3, 3, &third); const before = pager.position(); try std.testing.expectError(error.WalFull, pager.appendWal(4, 4, &rejected)); try std.testing.expectEqual(before, pager.position()); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len); pager.restore(after_first); const staged_position = pager.position(); try pager.stageWalPage(staged_position, 0, 2, &second); try pager.stageWalPage(staged_position, 1, 3, &third); try pager.commitStagedWal(staged_position, 2, 3); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len); var recovered_wal: [wal.header_size + 3 * wal.frame_size]u8 = undefined; @memcpy(&recovered_wal, pager.walBytes()); try pager.replaceWal(&recovered_wal, recovered_wal.len); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len); var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() }); defer prepared.deinit(); const checkpoint = try pager.commitDurableCheckpoint(prepared); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 0), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 0), pager.wal_index.pages.items.len); try std.testing.expectEqual(index_pointer, @intFromPtr(pager.wal_index.storage.ptr)); try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());}fn modelCheckpointPlanBytes(limits: CheckpointPlan.Limits) ?usize { if (limits.pages > std.math.maxInt(usize) / @sizeOf(CheckpointPage)) return null; return limits.pages * @sizeOf(CheckpointPage);}test "checkpoint plan capacity matches an independent typed storage model" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_capacity"), null, null, null, null, null, null, ); } for (0..64) |pages| { const limits: CheckpointPlan.Limits = .{ .pages = pages }; const capacity = try CheckpointPlan.Capacity.derive(limits); try std.testing.expectEqual(pages, capacity.pages); try std.testing.expectEqual(modelCheckpointPlanBytes(limits).?, capacity.storage_bytes); } const overflow: CheckpointPlan.Limits = .{ .pages = std.math.maxInt(usize) }; try std.testing.expect(modelCheckpointPlanBytes(overflow) == null); try std.testing.expectError(error.CapacityOverflow, CheckpointPlan.Capacity.derive(overflow));}test "checkpoint plan rejects short storage and releases its exact borrow" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_storage_rejection"), null, null, null, null, null, null, ); @stardustClaim( @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_work_bound"), null, null, null, null, null, null, ); } const limits: CheckpointPlan.Limits = .{ .pages = 3 }; const capacity = try CheckpointPlan.Capacity.derive(limits); const storage = try std.testing.allocator.alignedAlloc( u8, .fromByteUnits(CheckpointPlan.storage_alignment), capacity.storage_bytes, ); defer std.testing.allocator.free(storage); try std.testing.expectError( error.StorageTooShort, CheckpointPlan.init(storage[0 .. storage.len - 1], limits), ); var plan = try CheckpointPlan.init(storage, limits); try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, plan.phase); try std.testing.expectEqual(capacity.storage_bytes, plan.storage.len); plan.activate(); try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, plan.phase); const released = plan.deinit(); try std.testing.expectEqual(storage.ptr, released.ptr); try std.testing.expectEqual(storage.len, released.len);}test "checkpoint plan stays sealed through exact page selection and overload" { var pager = try testingPager(3); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var first_latest: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 10); fillImage(&first_latest, 1, 11); fillImage(&second, 2, 20); try pager.appendWal(1, 0, &first); try pager.appendWal(1, 0, &first_latest); try pager.appendWal(2, 2, &second); var prepared = try pager.prepareCheckpoint(.{}); defer prepared.deinit(); try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, prepared.plan.phase); try std.testing.expectEqual(@as(usize, 3), prepared.plan.capacity.pages); try std.testing.expectEqual(@as(usize, 2), prepared.checkpointPageCount()); try std.testing.expectEqual(@as(u8, 11), prepared.checkpointPage(0).bytes[1]); try std.testing.expectEqual(@as(u8, 20), prepared.checkpointPage(1).bytes[1]); const bounded_capacity = try CheckpointPlan.Capacity.derive(.{ .pages = 1 }); const bounded_storage = try std.testing.allocator.alignedAlloc( u8, .fromByteUnits(CheckpointPlan.storage_alignment), bounded_capacity.storage_bytes, ); defer std.testing.allocator.free(bounded_storage); var bounded = try CheckpointPlan.init(bounded_storage, .{ .pages = 1 }); defer _ = bounded.deinit(); bounded.activate(); try std.testing.expectError(error.CheckpointPlanCapacityExceeded, bounded.begin(2)); try bounded.begin(1); try std.testing.expectError(error.CheckpointPlanInUse, bounded.begin(1)); try bounded.append(.{ .page_id = 1, .wal_offset = wal.header_size + wal.frame_header_size }); try std.testing.expectError(error.CheckpointPlanCapacityExceeded, bounded.append(.{ .page_id = 2, .wal_offset = wal.header_size + wal.frame_size + wal.frame_header_size })); try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, bounded.phase); try std.testing.expectEqual(@as(usize, 1), bounded.filled); try std.testing.expectEqual(@as(usize, 1), bounded.pages().len); bounded.release();}test "pager reuses sealed checkpoint plans across durable checkpoints" { comptime { @stardustClaim( @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_sealed_overload"), null, null, null, null, null, null, ); } comptime { @stardustClaim( @import("alloc_phase").capacity.witness(CheckpointPlan, "sql_checkpoint_plan_sealed_transitive_risk"), null, null, null, null, null, null, ); } const allocator = std.testing.allocator; var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(allocator); const initialization_allocator = phase_allocator.initializationAllocator(); const options: InitOptions = .{ .header = testingHeader(), .wal_frames = 3, }; var maybe_workspace: ?Pager.Workspace = try Pager.Workspace.allocate(initialization_allocator, options); var maybe_pager: ?Pager = null; defer { if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization(); if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown(); const teardown_allocator = phase_allocator.teardownAllocator(); if (maybe_pager) |*pager| { var workspace = pager.deinit(); workspace.deallocate(teardown_allocator); } else if (maybe_workspace) |*workspace| { workspace.deallocate(teardown_allocator); } phase_allocator.deinit(); } maybe_pager = try Pager.init(initialization_allocator, &maybe_workspace.?, options); maybe_workspace = null; const pager = &maybe_pager.?; try pager.reserve(.{ .wal_frames = 3, .wal_pages = 3 }); const plan_pointer = @intFromPtr(pager.checkpoint_plan.storage.ptr); const once_pointer = @intFromPtr(pager.checkpoint_once_plan.storage.ptr); phase_allocator.seal(); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 10); fillImage(&second, 2, 20); try pager.appendWal(1, 0, &first); try pager.appendWal(2, 2, &second); var prepared = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() }); _ = try pager.commitDurableCheckpoint(prepared); prepared.deinit(); try pager.appendWal(1, 1, &second); var repeated = try pager.prepareCheckpoint(.{ .restart_header = restartHeader() }); _ = try pager.commitDurableCheckpoint(repeated); repeated.deinit(); try std.testing.expectEqual(plan_pointer, @intFromPtr(pager.checkpoint_plan.storage.ptr)); try std.testing.expectEqual(once_pointer, @intFromPtr(pager.checkpoint_once_plan.storage.ptr)); try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());}test "pager base lookup finds visible generation by page order" { var pager = try testingPager(0); defer deinitTestingPager(&pager); var second_old: [page.size]u8 = undefined; var first: [page.size]u8 = undefined; var second_new: [page.size]u8 = undefined; fillImage(&second_old, 2, 20); fillImage(&first, 1, 10); fillImage(&second_new, 2, 22); try pager.installBase(2, &second_old); try pager.installBase(1, &first); try pager.installBase(2, &second_new); try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(2, .{ .base_generation = 1, .end_mark = 0 })).?[1]); try std.testing.expect(try pager.pageAt(1, .{ .base_generation = 1, .end_mark = 0 }) == null); try std.testing.expectEqual(@as(u8, 10), (try pager.pageAt(1, .{ .base_generation = 2, .end_mark = 0 })).?[1]); try std.testing.expectEqual(@as(u8, 22), (try pager.pageAt(2, .{ .base_generation = std.math.maxInt(u64), .end_mark = 0 })).?[1]);}test "pager snapshots preserve base generations" { var pager = try testingPager(0); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 10); fillImage(&second, 1, 20); try pager.installBase(1, &first); const before = try pager.beginRead(); try pager.installBase(1, &second); const after = try pager.beginRead(); try std.testing.expectEqual(@as(u8, 10), (try before.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 20), (try after.get(1)).?[1]);}test "pager end mark freezes wal page view" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var base: [page.size]u8 = undefined; var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&base, 1, 10); fillImage(&first, 1, 20); fillImage(&second, 1, 30); try pager.installBase(1, &base); try pager.appendWal(1, 1, &first); const before = try pager.beginRead(); try pager.appendWal(1, 1, &second); const after = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), before.view.end_mark); try std.testing.expectEqual(@as(usize, 2), after.view.end_mark); try std.testing.expectEqual(@as(u8, 20), (try before.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 30), (try after.get(1)).?[1]);}test "pager restore rebuilds wal page index" { var pager = try testingPager(4); defer deinitTestingPager(&pager); var second: [page.size]u8 = undefined; var first: [page.size]u8 = undefined; var second_newer: [page.size]u8 = undefined; var seventh: [page.size]u8 = undefined; fillImage(&second, 2, 20); fillImage(&first, 1, 10); fillImage(&second_newer, 2, 22); fillImage(&seventh, 7, 77); try pager.appendWal(2, 2, &second); const position_value = pager.position(); try pager.appendWal(1, 2, &first); try pager.appendWal(2, 2, &second_newer); try pager.appendWal(7, 7, &seventh); const before = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 3), pager.wal_index.pages.items.len); try std.testing.expectEqual(@as(u8, 10), (try before.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 22), (try before.get(2)).?[1]); try std.testing.expectEqual(@as(u8, 77), (try before.get(7)).?[1]); try std.testing.expectEqual(@as(u32, 7), pager.databasePageCount()); pager.restore(position_value); const after = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), pager.frameCount()); try std.testing.expectEqual(@as(usize, 1), after.view.end_mark); try std.testing.expectEqual(@as(usize, 1), pager.wal_index.pages.items.len); try std.testing.expectEqual(@as(u32, 2), pager.wal_index.pages.items[0].page_id); try std.testing.expectEqual(@as(u32, 2), pager.databasePageCount()); try std.testing.expect(try after.get(1) == null); try std.testing.expectEqual(@as(u8, 20), (try after.get(2)).?[1]);}test "pager check marks clear when a stored image changes" { var pager = try testingPager(4); defer deinitTestingPager(&pager); var base: [page.size]u8 = undefined; var replaced: [page.size]u8 = undefined; var logged: [page.size]u8 = undefined; var superseding: [page.size]u8 = undefined; var relogged: [page.size]u8 = undefined; fillImage(&base, 1, 10); fillImage(&replaced, 1, 11); fillImage(&logged, 2, 20); fillImage(&superseding, 2, 21); fillImage(&relogged, 2, 22); try pager.installBase(1, &base); const based = (try pager.markedPageAt(1, try pager.currentView())).?; try std.testing.expect(!based.checked.?.*); based.checked.?.* = true; try std.testing.expect((try pager.markedPageAt(1, try pager.currentView())).?.checked.?.*); try pager.installBaseAtGeneration(1, &replaced, pager.baseGeneration()); const rebased = (try pager.markedPageAt(1, try pager.currentView())).?; try std.testing.expectEqual(@as(u8, 11), rebased.bytes[1]); try std.testing.expect(!rebased.checked.?.*); const before = pager.position(); try pager.appendWal(2, 2, &logged); const appended = (try pager.markedPageAt(2, try pager.currentView())).?; try std.testing.expect(!appended.checked.?.*); appended.checked.?.* = true; try std.testing.expect((try pager.markedPageAt(2, try pager.currentView())).?.checked.?.*); try pager.appendWal(2, 2, &superseding); const superseded = (try pager.markedPageAt(2, try pager.currentView())).?; try std.testing.expectEqual(@as(u8, 21), superseded.bytes[1]); try std.testing.expect(!superseded.checked.?.*); superseded.checked.?.* = true; pager.restore(before); try pager.appendWal(2, 2, &relogged); const reappended = (try pager.markedPageAt(2, try pager.currentView())).?; try std.testing.expectEqual(@as(u8, 22), reappended.bytes[1]); try std.testing.expect(!reappended.checked.?.*);}test "pager ignores uncommitted wal tail" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var committed: [page.size]u8 = undefined; var uncommitted: [page.size]u8 = undefined; fillImage(&committed, 1, 20); fillImage(&uncommitted, 9, 99); try pager.appendWal(1, 1, &committed); try pager.appendWal(9, 0, &uncommitted); const snapshot = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), snapshot.view.end_mark); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]); try std.testing.expectEqual(@as(u32, 9), pager.databasePageCount());}test "pager falls back to base pages not present in wal" { var pager = try testingPager(1); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; var wal_image: [page.size]u8 = undefined; fillImage(&first, 1, 10); fillImage(&second, 2, 20); fillImage(&wal_image, 1, 30); try pager.installBase(1, &first); try pager.installBase(2, &second); try pager.appendWal(1, 2, &wal_image); const snapshot = try pager.beginRead(); try std.testing.expectEqual(@as(u8, 30), (try snapshot.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(2)).?[1]); try std.testing.expect(try snapshot.get(3) == null);}test "pager snapshot survives frames appended after its end mark" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 20); fillImage(&second, 1, 30); try pager.appendWal(1, 1, &first); const snapshot = try pager.beginRead(); try pager.appendWal(1, 1, &second); try std.testing.expectEqual(@as(usize, 1), snapshot.view.end_mark); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]); try std.testing.expectEqual(@as(usize, 2), (try pager.beginRead()).view.end_mark);}test "pager checkpoint applies committed wal frames to a base generation" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 20); fillImage(&second, 2, 30); try pager.appendWal(1, 0, &first); try pager.appendWal(2, 2, &second); const checkpoint = try pager.checkpoint(.{}); const base_view = View{ .base_generation = checkpoint.base_generation, .end_mark = 0 }; try std.testing.expectEqual(@as(usize, 2), checkpoint.end_mark); try std.testing.expectEqual(@as(usize, 2), checkpoint.pages); try std.testing.expectEqual(@as(u64, 1), checkpoint.base_generation); try std.testing.expect(!checkpoint.restarted); try std.testing.expectEqual(@as(usize, 2), pager.frameCount()); try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(1, base_view)).?[1]); try std.testing.expectEqual(@as(u8, 30), (try pager.pageAt(2, base_view)).?[1]);}test "pager checkpoint prunes obsolete base images without readers" { var pager = try testingPager(0); defer deinitTestingPager(&pager); var first_old: [page.size]u8 = undefined; var second_old: [page.size]u8 = undefined; var first_new: [page.size]u8 = undefined; var second_new: [page.size]u8 = undefined; fillImage(&first_old, 1, 10); fillImage(&second_old, 2, 20); fillImage(&first_new, 1, 11); fillImage(&second_new, 2, 22); try pager.installBase(1, &first_old); try pager.installBase(2, &second_old); try pager.installBase(1, &first_new); try pager.installBase(2, &second_new); const checkpoint = try pager.checkpoint(.{}); const current = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 0), checkpoint.pages); try std.testing.expectEqual(@as(usize, 2), pager.base.items.len); try std.testing.expectEqual(@as(u64, 4), pager.baseGeneration()); try std.testing.expect(try pager.pageAt(1, .{ .base_generation = 1, .end_mark = 0 }) == null); try std.testing.expectEqual(@as(u8, 11), (try current.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 22), (try current.get(2)).?[1]);}test "pager checkpoint retains base floor for oldest reader" { var pager = try testingPager(0); defer deinitTestingPager(&pager); var first_old: [page.size]u8 = undefined; var second_old: [page.size]u8 = undefined; var first_floor: [page.size]u8 = undefined; var first_new: [page.size]u8 = undefined; var second_new: [page.size]u8 = undefined; fillImage(&first_old, 1, 10); fillImage(&second_old, 2, 20); fillImage(&first_floor, 1, 11); fillImage(&first_new, 1, 12); fillImage(&second_new, 2, 22); try pager.installBase(1, &first_old); try pager.installBase(2, &second_old); try pager.installBase(1, &first_floor); const oldest = try pager.beginRead(); try pager.installBase(1, &first_new); try pager.installBase(2, &second_new); const checkpoint = try pager.checkpoint(.{ .readers = .{ .oldest = oldest.view } }); const current = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 0), checkpoint.pages); try std.testing.expectEqual(@as(usize, 4), pager.base.items.len); try std.testing.expectEqual(@as(u8, 11), (try oldest.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 20), (try oldest.get(2)).?[1]); try std.testing.expectEqual(@as(u8, 12), (try current.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 22), (try current.get(2)).?[1]);}test "pager checkpoint compacts wal-applied base images without readers" { var pager = try testingPager(1); defer deinitTestingPager(&pager); var base_old: [page.size]u8 = undefined; var base_new: [page.size]u8 = undefined; var committed: [page.size]u8 = undefined; fillImage(&base_old, 1, 10); fillImage(&base_new, 1, 20); fillImage(&committed, 1, 30); try pager.installBase(1, &base_old); try pager.installBase(1, &base_new); try pager.appendWal(1, 1, &committed); const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() }); const current = try pager.beginRead(); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expectEqual(@as(usize, 1), pager.base.items.len); try std.testing.expectEqual(@as(u8, 30), (try current.get(1)).?[1]);}test "pager checkpoint prunes applied wal frame index without readers" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 20); fillImage(&second, 2, 30); try pager.appendWal(1, 0, &first); try pager.appendWal(2, 2, &second); const checkpoint = try pager.checkpoint(.{}); const snapshot = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 2), checkpoint.end_mark); try std.testing.expectEqual(@as(usize, 2), checkpoint.pages); try std.testing.expectEqual(@as(usize, 2), pager.frameCount()); try std.testing.expectEqual(@as(usize, 0), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 0), pager.wal_index.pages.items.len); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 30), (try snapshot.get(2)).?[1]);}test "pager checkpoint carries uncommitted tail index across wal rewrite" { var pager = try testingPager(3); defer deinitTestingPager(&pager); var committed: [page.size]u8 = undefined; var tail: [page.size]u8 = undefined; var marker: [page.size]u8 = undefined; fillImage(&committed, 1, 20); fillImage(&tail, 2, 40); fillImage(&marker, 3, 60); try pager.appendWal(1, 1, &committed); try pager.appendWal(2, 0, &tail); const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() }); const before_commit = try pager.beginRead(); try pager.appendWal(3, 3, &marker); const after_commit = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 2), pager.frameCount()); try std.testing.expectEqual(@as(usize, 2), pager.wal_index.frames.items.len); try std.testing.expectEqual(@as(usize, 2), pager.wal_index.pages.items.len); try std.testing.expectEqual(@as(u8, 20), (try before_commit.get(1)).?[1]); try std.testing.expect(try before_commit.get(2) == null); try std.testing.expectEqual(@as(u8, 20), (try after_commit.get(1)).?[1]); try std.testing.expectEqual(@as(u8, 40), (try after_commit.get(2)).?[1]); try std.testing.expectEqual(@as(u8, 60), (try after_commit.get(3)).?[1]);}test "pager checkpoint restarts wal when no reader is active" { var pager = try testingPager(1); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; fillImage(&first, 1, 20); try pager.appendWal(1, 1, &first); const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() }); const snapshot = try pager.beginRead(); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expectEqual(@as(usize, 0), pager.frameCount()); try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]);}test "pager checkpoint honors the oldest reader end mark" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var first: [page.size]u8 = undefined; var second: [page.size]u8 = undefined; fillImage(&first, 1, 20); fillImage(&second, 2, 30); try pager.appendWal(1, 1, &first); const oldest = try pager.beginRead(); try pager.appendWal(2, 2, &second); const checkpoint = try pager.checkpoint(.{ .readers = .{ .oldest = oldest.view }, .restart_header = restartHeader(), }); const base_view = View{ .base_generation = checkpoint.base_generation, .end_mark = 0 }; const current = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expect(!checkpoint.restarted); try std.testing.expectEqual(@as(usize, 2), pager.frameCount()); try std.testing.expectEqual(@as(u8, 20), (try pager.pageAt(1, base_view)).?[1]); try std.testing.expect(try pager.pageAt(2, base_view) == null); try std.testing.expectEqual(@as(u8, 30), (try current.get(2)).?[1]);}test "pager checkpoint rewrites wal to an uncommitted tail" { var pager = try testingPager(2); defer deinitTestingPager(&pager); var committed: [page.size]u8 = undefined; var uncommitted: [page.size]u8 = undefined; fillImage(&committed, 1, 20); fillImage(&uncommitted, 2, 99); try pager.appendWal(1, 1, &committed); try pager.appendWal(2, 0, &uncommitted); const checkpoint = try pager.checkpoint(.{ .restart_header = restartHeader() }); const snapshot = try pager.beginRead(); try std.testing.expectEqual(@as(usize, 1), checkpoint.end_mark); try std.testing.expectEqual(@as(usize, 1), checkpoint.pages); try std.testing.expect(checkpoint.restarted); try std.testing.expectEqual(@as(usize, 1), pager.frameCount()); try std.testing.expectEqual(@as(usize, 1), pager.wal_index.frames.items.len); try std.testing.expectEqual(wal.header_size + wal.frame_size, pager.walBytes().len); try std.testing.expectEqual(@as(usize, 0), snapshot.view.end_mark); try std.testing.expectEqual(@as(u8, 20), (try snapshot.get(1)).?[1]); try std.testing.expect(try snapshot.get(2) == null);}Source: lib/sql/src/root.zig:21
zig
pub const pager = @import("pager.zig");Audit
| Definitions | 24 |
|---|---|
| Public names | 24 |
| Members | 25 |
| Version | 26.7.0 |
| Revision | daab053ee433 |