tiny.sql.WalWriter

1pub const Writer = struct {2    pub const storage_alignment: usize = @alignOf(u8);3    pub const Storage = []align(storage_alignment) u8;45    pub const Limits = struct {6        header: Header,7        frames: usize,8    };910    pub const Capacity = struct {11        frames: usize,12        storage_bytes: usize,1314        pub const DeriveError = error{CapacityOverflow};1516        pub fn derive(limits: Limits) DeriveError!Capacity {17            const frame_bytes = std.math.mul(usize, limits.frames, frame_size) catch {18                return error.CapacityOverflow;19            };20            const storage_bytes = std.math.add(usize, header_size, frame_bytes) catch {21                return error.CapacityOverflow;22            };23            return .{ .frames = limits.frames, .storage_bytes = storage_bytes };24        }25    };2627    pub const Exhaustion = error{WalFull};28    pub const InitError = Capacity.DeriveError || error{StorageTooShort};29    pub const LoadError = ReadError || Exhaustion;30    pub const ControlledLoadError = LoadError || error{Interrupted};31    pub const work_limits: alloc_phase.capacity.WorkLimits = .{32        .transition_steps_max = 1,33        .cleanup_steps_per_call_max = 0,34        .cleanup_calls_at_capacity_max = 0,35    };3637    pub const claim: alloc_phase.capacity.Declaration = .{38        .source = .{39            .id = "sql.wal_writer",40            .kind = .phase_static,41            .limit_source = .caller,42            .storage = .{43                .covered = &.{44                    .{45                        .id = "exact_header_and_committed_frame_byte_region",46                        .lifetime = .steady,47                        .detail = "mutable borrow of the caller-provisioned exact header and frame byte region",48                    },49                    .{50                        .id = "invisible_staged_frame_tail_transferred_into_the_co_0ed8d0c7aca0",51                        .lifetime = .transferred,52                        .detail = "invisible staged frame tail transferred into the committed prefix",53                    },54                },55                .excluded = &.{56                    "caller-owned storage outside the exact borrowed prefix",57                    "caller-owned page images and recovery source bytes",58                    "pager frame indexes, page indexes, checkpoint plans, files, and I/O runtime state",59                    "trace instrumentation and allocator implementation state",60                },61            },62            .capacity = .{63                .inputs = &.{64                    alloc_phase.capacity.bindInput(Limits, "frames", "frames"),65                },66                .type_selectors = &.{67                    alloc_phase.capacity.bindType([frame_size]u8, "walframebytes"),68                    alloc_phase.capacity.bindType([header_size]u8, "walheaderbytes"),69                },70                .nodes = &.{71                    .{ .input = 0 },72                    .{ .scale = .{ .node = 0, .coefficient = .{ .size_of_concrete_type = 0 } } },73                    .{ .constant = 1 },74                    .{ .scale = .{ .node = 2, .coefficient = .{ .size_of_concrete_type = 1 } } },75                    .{ .add = .{ .left = 1, .right = 3 } },76                },77                .assertions = &.{.{78                    .scope = .closure_total,79                    .measure = .retained,80                    .relation = .exact,81                    .expression = 4,82                }},83            },84            .overload = .{85                .kind = .reject_before_mutation,86                .detail = "append staging and recovery load return WalFull before bytes length salt or checksum mutate",87            },88            .risks = .{89                .transitive = .{90                    .status = .open,91                    .detail = "pager indexes and trace instrumentation are outside the writer-owned byte region",92                },93                .foreign = .{94                    .status = .open,95                    .detail = "file reads and writes that consume WAL bytes are owned by the file database",96                },97            },98            .work = .{ .equation = "transition_steps <= transition_steps_max" },99            .obligations = &.{100                .{ .key = "sql_wal_capacity", .role = .capacity_model },101                .{ .key = "sql_wal_storage_rejection", .role = .initialization_failure },102                .{ .key = "sql_wal_sealed", .role = .overload },103                .{ .key = "sql_wal_work_bound", .role = .work_bound },104                .{ .key = "sql_wal_semantics", .role = .custom },105                .{ .key = "sql_wal_staged_transfer", .role = .custom },106                .{ .key = "sql_wal_corruption", .role = .custom },107            },108        },109        .bindings = .{110            .owner = @This(),111            .seal = .{112                .family = alloc_phase.capacity.selector(@This().activate),113                .premise = .{114                    .class = .checked_semantic_fact,115                    .authority = .checker,116                },117            },118            .teardown = .{119                .family = alloc_phase.capacity.selector(@This().deinit),120                .premise = .{121                    .class = .checked_semantic_fact,122                    .authority = .checker,123                },124            },125        },126    };127128    phase: alloc_phase.capacity.Phase,129    capacity: Capacity,130    storage: Storage,131    len: usize,132    salt: Salt,133    checksum: Checksum,134135    pub const Position = struct {136        len: usize,137        checksum_first: u32,138        checksum_second: u32,139    };140141    pub fn init(storage: Storage, limits: Limits) InitError!Writer {142        const capacity = try Capacity.derive(limits);143        if (storage.len < capacity.storage_bytes) return error.StorageTooShort;144        const borrowed = storage[0..capacity.storage_bytes];145        const checksum = encodeHeader(borrowed[0..header_size], limits.header);146        return .{147            .phase = .initialization,148            .capacity = capacity,149            .storage = borrowed,150            .len = header_size,151            .salt = limits.header.salt,152            .checksum = checksum,153        };154    }155156    pub fn activate(self: *Writer) void {157        std.debug.assert(self.phase == .initialization);158        std.debug.assert(self.storage.len == self.capacity.storage_bytes);159        std.debug.assert(self.len == header_size);160        self.phase = .steady;161    }162163    pub fn deinit(self: *Writer) Storage {164        std.debug.assert(self.phase != .teardown);165        std.debug.assert(self.storage.len == self.capacity.storage_bytes);166        self.phase = .teardown;167        const storage = self.storage;168        self.* = undefined;169        return storage;170    }171172    pub fn bytes(self: *const Writer) []const u8 {173        std.debug.assert(self.phase == .steady);174        std.debug.assert(self.len <= self.storage.len);175        return self.storage[0..self.len];176    }177178    pub fn frameCount(self: *const Writer) usize {179        std.debug.assert(self.phase == .steady);180        return (self.len - header_size) / frame_size;181    }182183    pub fn frameCapacity(self: *const Writer) usize {184        std.debug.assert(self.phase == .steady);185        return self.capacity.frames;186    }187188    pub fn byteCapacity(self: *const Writer) usize {189        std.debug.assert(self.phase == .steady);190        return self.capacity.storage_bytes;191    }192193    pub fn remainingFrames(self: *const Writer) usize {194        std.debug.assert(self.phase == .steady);195        return self.capacity.frames - self.frameCount();196    }197198    pub fn position(self: *const Writer) Position {199        std.debug.assert(self.phase == .steady);200        return .{201            .len = self.len,202            .checksum_first = self.checksum.first,203            .checksum_second = self.checksum.second,204        };205    }206207    pub fn restore(self: *Writer, position_value: Position) void {208        std.debug.assert(self.phase == .steady);209        std.debug.assert(position_value.len >= header_size);210        std.debug.assert((position_value.len - header_size) % frame_size == 0);211        std.debug.assert(position_value.len <= self.len);212        self.len = position_value.len;213        self.checksum = .{214            .first = position_value.checksum_first,215            .second = position_value.checksum_second,216        };217    }218219    pub fn load(self: *Writer, source: []const u8, committed_len: usize) LoadError!void {220        return self.loadControlled(source, committed_len, .{}) catch |err| switch (err) {221            error.Interrupted => unreachable,222            else => return @errorCast(err),223        };224    }225226    pub fn loadControlled(227        self: *Writer,228        source: []const u8,229        committed_len: usize,230        control: Control,231    ) ControlledLoadError!void {232        std.debug.assert(self.phase == .steady);233        if (committed_len > source.len) return error.InvalidWal;234        if (committed_len > self.capacity.storage_bytes) return error.WalFull;235        var reader = try Reader.initControlled(source[0..committed_len], control);236        const frames_max = (committed_len - header_size) / frame_size;237        var frame_index: usize = 0;238        while (frame_index < frames_max) : (frame_index += 1) {239            if (try reader.nextControlled(control) == null) break;240        }241        if (reader.offset != committed_len) return error.InvalidWal;242        var copied: usize = 0;243        var chunks: usize = 0;244        const chunks_max = std.math.divCeil(245            usize,246            committed_len,247            controlled_copy_chunk_bytes,248        ) catch unreachable;249        while (copied < committed_len) : (chunks += 1) {250            std.debug.assert(chunks < chunks_max);251            try control.check();252            const end = copied + @min(253                controlled_copy_chunk_bytes,254                committed_len - copied,255            );256            std.mem.copyForwards(u8, self.storage[copied..end], source[copied..end]);257            copied = end;258        }259        try control.check();260        self.len = committed_len;261        self.salt = reader.salt;262        self.checksum = reader.checksum;263    }264265    pub fn append(self: *Writer, page_id: u32, db_page_count: u32, image: *const [page.size]u8) Exhaustion!void {266        const phase = trace.scope("wal.append");267        defer phase.end();268        std.debug.assert(self.phase == .steady);269        if (self.remainingFrames() == 0) return error.WalFull;270        var frame_header = @as([frame_header_size]u8, @splat(0));271        writeU32(frame_header[0..4], page_id);272        writeU32(frame_header[4..8], db_page_count);273        writeU32(frame_header[8..12], self.salt.first);274        writeU32(frame_header[12..16], self.salt.second);275276        var checksum = self.checksum;277        checksum.update(frame_header[0..8]);278        checksum.update(image[0..]);279        writeU32(frame_header[16..20], checksum.first);280        writeU32(frame_header[20..24], checksum.second);281282        @memcpy(self.storage[self.len..][0..frame_header_size], frame_header[0..]);283        @memcpy(self.storage[self.len + frame_header_size ..][0..page.size], image[0..]);284        self.len += frame_size;285        self.checksum = checksum;286        trace.progress("wal.append.complete");287    }288289    pub fn stagePage(self: *Writer, position_value: Position, index: usize, page_id: u32, image: *const [page.size]u8) Exhaustion!void {290        std.debug.assert(self.phase == .steady);291        std.debug.assert(position_value.len == self.len);292        const offset = try self.stagedFrameOffset(position_value, index);293        writeU32(self.storage[offset..][0..4], page_id);294        const staged = self.storage[offset + frame_header_size ..][0..page.size];295        if (staged != image) @memcpy(staged, image);296    }297298    pub fn stagedPage(self: *const Writer, position_value: Position, index: usize) []const u8 {299        std.debug.assert(self.phase == .steady);300        std.debug.assert(position_value.len == self.len);301        const offset = self.stagedFrameOffset(position_value, index) catch unreachable;302        return self.storage[offset + frame_header_size ..][0..page.size];303    }304305    /// Returns a staged image for editing in place. Staging the returned306    /// image again copies nothing, and the checksum covers the edits307    /// when the frame commits.308    pub fn stagedPageMut(self: *Writer, position_value: Position, index: usize) *[page.size]u8 {309        std.debug.assert(self.phase == .steady);310        std.debug.assert(position_value.len == self.len);311        const offset = self.stagedFrameOffset(position_value, index) catch unreachable;312        return self.storage[offset + frame_header_size ..][0..page.size];313    }314315    pub fn stagedPageId(self: *const Writer, position_value: Position, index: usize) u32 {316        std.debug.assert(self.phase == .steady);317        const offset = self.stagedFrameOffset(position_value, index) catch unreachable;318        return readU32(self.storage[offset..][0..4]);319    }320321    pub fn swapStagedFrames(self: *Writer, position_value: Position, left_index: usize, right_index: usize) void {322        std.debug.assert(self.phase == .steady);323        std.debug.assert(position_value.len == self.len);324        if (left_index == right_index) return;325        const left_offset = self.stagedFrameOffset(position_value, left_index) catch unreachable;326        const right_offset = self.stagedFrameOffset(position_value, right_index) catch unreachable;327        const left: *[frame_size]u8 = self.storage[left_offset..][0..frame_size];328        const right: *[frame_size]u8 = self.storage[right_offset..][0..frame_size];329        const temporary = left.*;330        left.* = right.*;331        right.* = temporary;332    }333334    pub fn commitStagedFrame(self: *Writer, position_value: Position, index: usize, db_page_count: u32) void {335        std.debug.assert(self.phase == .steady);336        const offset = self.stagedFrameOffset(position_value, index) catch unreachable;337        std.debug.assert(self.len == offset);338        const frame_header = self.storage[offset..][0..frame_header_size];339        const image = self.storage[offset + frame_header_size ..][0..page.size];340        std.debug.assert(readU32(frame_header[0..4]) != 0);341        writeU32(frame_header[4..8], db_page_count);342        writeU32(frame_header[8..12], self.salt.first);343        writeU32(frame_header[12..16], self.salt.second);344345        var checksum = self.checksum;346        checksum.update(frame_header[0..8]);347        checksum.update(image);348        writeU32(frame_header[16..20], checksum.first);349        writeU32(frame_header[20..24], checksum.second);350351        self.len += frame_size;352        self.checksum = checksum;353    }354355    pub fn rewriteTail(self: *Writer, checkpoint_mark: usize, header: Header) void {356        std.debug.assert(self.phase == .steady);357        std.debug.assert(checkpoint_mark <= self.frameCount());358        const source = self.bytes();359        var reader = Reader.init(source) catch @panic("invalid wal writer storage");360        self.writeHeader(header);361        while (reader.next() catch @panic("invalid wal writer storage")) |frame| {362            if (frame.index <= checkpoint_mark) continue;363            const image = frame.image[0..page.size].*;364            self.append(frame.page_id, frame.db_page_count, &image) catch unreachable;365        }366        if (reader.offset != source.len) @panic("invalid wal writer storage");367    }368369    fn writeHeader(self: *Writer, header: Header) void {370        const checksum = encodeHeader(self.storage[0..header_size], header);371        self.len = header_size;372        self.salt = header.salt;373        self.checksum = checksum;374    }375376    fn stagedFrameOffset(self: *const Writer, position_value: Position, index: usize) Exhaustion!usize {377        std.debug.assert(position_value.len >= header_size);378        std.debug.assert((position_value.len - header_size) % frame_size == 0);379        const staged_bytes = std.math.mul(usize, index, frame_size) catch return error.WalFull;380        const offset = std.math.add(usize, position_value.len, staged_bytes) catch return error.WalFull;381        if (offset > self.storage.len or frame_size > self.storage.len - offset) return error.WalFull;382        return offset;383    }384};

1pub const WalWriter = wal.Writer;