tiny.sql.history.segment

1pub const Block = struct {2    first_event: EventRange.Count,3    event_count: u32,4    raw_bytes: u32,5    bytes: []const u8,6};

1pub const Error = error{2    CapacityOverflow,3    CapacityExceeded,4    EmptySegment,5    InvalidSegment,6    SegmentBytesExceeded,7    StorageTooShort,8    TooManyBlocks,9    TooManyClosureDigests,10};

1pub const Iterator = struct {2    view: *const View,3    cursor: usize,4    closures_remaining: u32,5    blocks_remaining: u32,6    next_event: EventRange.Count,78    pub fn next(self: *Iterator) Error!?Record {9        if (self.closures_remaining != 0) {10            const end = self.cursor + closure_record_bytes;11            if (end > self.view.bytes.len - trailer_bytes) {12                return error.InvalidSegment;13            }14            const value = try readClosure(self.view.bytes[self.cursor..end]);15            self.cursor = end;16            self.closures_remaining -= 1;17            return .{ .closure = value };18        }19        if (self.blocks_remaining == 0) {20            if (self.cursor != self.view.bytes.len - trailer_bytes) {21                return error.InvalidSegment;22            }23            return null;24        }25        const block = try readBlock(self.view.bytes, &self.cursor);26        if (block.first_event != self.next_event) return error.InvalidSegment;27        self.next_event = EventRange.advance(28            block.first_event,29            block.event_count,30        ) orelse return error.InvalidSegment;31        self.blocks_remaining -= 1;32        return .{ .block = block };33    }34};

1pub const Record = union(enum) {2    closure: Hash,3    block: Block,4};

1pub const ReplayDemand = struct {2    segments: u64 = 1,3    blocks: u64,4    events: u64,5    raw_event_bytes: u64 = 0,6    encoded_block_bytes: u64,7    max_record_bytes: u64 = closure_record_bytes,8    checkpoint_roots: u64,9    closure_digests: u64,10    segment_bytes: u64,11    io_operations: u64 = 1,12};

1pub const Sealed = struct {2    bytes: []const u8,3    digest: Hash,4    spec: Spec,5};

1pub const Sealer = struct {2    pub const storage_alignment: usize = 8;3    pub const Storage = []align(storage_alignment) u8;4    pub const Limits = struct {5        blocks: usize,6        block_bytes: usize,7        closure_digests: usize,8    };910    pub const Capacity = struct {11        blocks: usize,12        block_bytes: usize,13        closure_digests: usize,14        storage_bytes: usize,1516        pub const DeriveError = error{17            CapacityOverflow,18            CapacityExceeded,19            EmptySegment,20        };2122        pub fn derive(limits: Limits) DeriveError!Capacity {23            if (limits.blocks == 0 or limits.block_bytes == 0) {24                return error.EmptySegment;25            }26            if (limits.blocks > blocks_max or27                limits.closure_digests > closure_digests_max or28                limits.block_bytes > block_bytes_max)29            {30                return error.CapacityExceeded;31            }32            const block_headers = try multiply(limits.blocks, block_header_bytes);33            const closures = try multiply(34                limits.closure_digests,35                closure_record_bytes,36            );37            var storage_bytes = try add(header_bytes, block_headers);38            storage_bytes = try add(storage_bytes, closures);39            storage_bytes = try add(storage_bytes, limits.block_bytes);40            storage_bytes = try add(storage_bytes, trailer_bytes);41            return .{42                .blocks = limits.blocks,43                .block_bytes = limits.block_bytes,44                .closure_digests = limits.closure_digests,45                .storage_bytes = storage_bytes,46            };47        }48    };4950    pub const Exhaustion = error{51        InvalidSegment,52        SegmentBytesExceeded,53        TooManyBlocks,54        TooManyClosureDigests,55    };56    pub const InitError = Capacity.DeriveError || error{StorageTooShort};57    pub const work_limits: alloc_phase.capacity.WorkLimits = .{58        .transition_steps_max = blocks_max + closure_digests_max,59        .cleanup_steps_per_call_max = 0,60        .cleanup_calls_at_capacity_max = 0,61    };6263    pub const claim: alloc_phase.capacity.Declaration = .{64        .source = .{65            .id = "sql.history_segment_sealer",66            .kind = .phase_static,67            .limit_source = .caller,68            .storage = .{69                .covered = &.{.{70                    .id = "reserved_immutable_segment_output",71                    .lifetime = .transferred,72                    .detail = "exact header record payload and trailer output",73                }},74                .excluded = &.{75                    "caller-owned event blocks and content closure digests",76                    "filesystem writes manifest pages and archive transport",77                },78            },79            .capacity = .{80                .inputs = &.{81                    alloc_phase.capacity.bindInput(Limits, "blocks", "blocks"),82                    alloc_phase.capacity.bindInput(83                        Limits,84                        "block_bytes",85                        "block_bytes",86                    ),87                    alloc_phase.capacity.bindInput(88                        Limits,89                        "closure_digests",90                        "closure_digests",91                    ),92                },93                .type_selectors = &.{},94                .nodes = &.{95                    .{ .input = 0 },96                    .{ .input = 1 },97                    .{ .input = 2 },98                    .{ .constant = 64 },99                    .{ .product = .{ .left = 0, .right = 3 } },100                    .{ .constant = 40 },101                    .{ .product = .{ .left = 2, .right = 5 } },102                    .{ .constant = 256 },103                    .{ .add = .{ .left = 7, .right = 4 } },104                    .{ .add = .{ .left = 8, .right = 6 } },105                    .{ .add = .{ .left = 9, .right = 1 } },106                    .{ .constant = 32 },107                    .{ .add = .{ .left = 10, .right = 11 } },108                },109                .assertions = &.{.{110                    .scope = .closure_total,111                    .measure = .retained,112                    .relation = .exact,113                    .expression = 12,114                }},115            },116            .overload = .{117                .kind = .reject_before_mutation,118                .detail = "record count and byte exhaustion reject before output changes",119            },120            .risks = .{121                .transitive = .{122                    .status = .witnessed,123                    .detail = "sealing hashes and writes only the caller reservation",124                },125                .foreign = .{126                    .status = .excluded,127                    .detail = "durable file and archive publication are outside this owner",128                },129            },130            .work = .{131                .equation = "record writes <= blocks + closure_digests",132            },133            .obligations = &.{134                .{135                    .key = "sql_history_segment_capacity",136                    .role = .capacity_model,137                },138                .{139                    .key = "sql_history_segment_overload",140                    .role = .overload,141                },142                .{143                    .key = "sql_history_segment_verification",144                    .role = .transitive_risk,145                },146                .{147                    .key = "sql_history_segment_work_bound",148                    .role = .work_bound,149                },150            },151        },152        .bindings = .{153            .owner = @This(),154            .seal = .{155                .family = alloc_phase.capacity.selector(@This().activate),156                .premise = .{157                    .class = .checked_semantic_fact,158                    .authority = .checker,159                },160            },161            .teardown = .{162                .family = alloc_phase.capacity.selector(@This().deinit),163                .premise = .{164                    .class = .checked_semantic_fact,165                    .authority = .checker,166                },167            },168        },169    };170171    phase: alloc_phase.capacity.Phase,172    capacity: Capacity,173    storage: Storage,174    cursor: usize = header_bytes,175    blocks: usize = 0,176    block_bytes: usize = 0,177    closure_digests: usize = 0,178    first_event: ?EventRange.Count = null,179    next_event: EventRange.Count = 0,180    blocks_started: bool = false,181182    pub fn init(storage: Storage, limits: Limits) InitError!Sealer {183        const capacity = try Capacity.derive(limits);184        if (storage.len < capacity.storage_bytes) return error.StorageTooShort;185        return .{186            .phase = .initialization,187            .capacity = capacity,188            .storage = storage[0..capacity.storage_bytes],189        };190    }191192    pub fn activate(self: *Sealer) void {193        std.debug.assert(self.phase == .initialization);194        std.debug.assert(self.storage.len == self.capacity.storage_bytes);195        self.phase = .steady;196    }197198    pub fn addClosure(self: *Sealer, value: Hash) Exhaustion!void {199        self.assertSteady();200        if (self.blocks_started) return error.InvalidSegment;201        if (self.closure_digests == self.capacity.closure_digests) {202            return error.TooManyClosureDigests;203        }204        const end = self.cursor + closure_record_bytes;205        writeClosure(self.storage[self.cursor..end], value);206        self.cursor = end;207        self.closure_digests += 1;208    }209210    pub fn addBlock(self: *Sealer, block: Block) Exhaustion!void {211        self.assertSteady();212        if (block.event_count == 0 or block.raw_bytes == 0 or block.bytes.len == 0) {213            return error.InvalidSegment;214        }215        if (self.blocks == self.capacity.blocks) return error.TooManyBlocks;216        if (block.bytes.len > self.capacity.block_bytes - self.block_bytes) {217            return error.SegmentBytesExceeded;218        }219        const next_event = EventRange.advance(220            block.first_event,221            block.event_count,222        ) orelse return error.InvalidSegment;223        if (self.first_event) |_| {224            if (block.first_event != self.next_event) return error.InvalidSegment;225        } else self.first_event = block.first_event;226        const end = self.cursor + block_header_bytes + block.bytes.len;227        writeBlock(self.storage[self.cursor..end], block);228        self.cursor = end;229        self.blocks += 1;230        self.block_bytes += block.bytes.len;231        self.next_event = next_event;232        self.blocks_started = true;233    }234235    pub fn finish(self: *Sealer, spec: Spec) Exhaustion!Sealed {236        self.assertSteady();237        if (self.blocks != self.capacity.blocks or238            self.block_bytes != self.capacity.block_bytes or239            self.closure_digests != self.capacity.closure_digests)240        {241            return error.InvalidSegment;242        }243        if (!SegmentRange.containsOrdinal(spec.ordinal)) {244            return error.InvalidSegment;245        }246        const first_event = self.first_event orelse return error.InvalidSegment;247        if (spec.first_event != first_event) return error.InvalidSegment;248        const next_event = EventRange.advance(249            spec.first_event,250            spec.event_count,251        ) orelse return error.InvalidSegment;252        if (next_event != self.next_event) {253            return error.InvalidSegment;254        }255        std.debug.assert(self.cursor + trailer_bytes == self.storage.len);256        var records_digest: Hash = undefined;257        Sha256.hash(self.storage[header_bytes..self.cursor], &records_digest, .{});258        writeHeader(self.storage[0..header_bytes], spec, self.capacity, records_digest);259        const trailer_digest = digest(self.storage[0..self.cursor]);260        @memcpy(self.storage[self.cursor..], &trailer_digest);261        const segment_digest = digest(self.storage);262        const view = verify(self.storage) catch return error.InvalidSegment;263        std.debug.assert(std.mem.eql(u8, &view.digest, &segment_digest));264        return .{ .bytes = self.storage, .digest = segment_digest, .spec = spec };265    }266267    pub fn deinit(self: *Sealer) Storage {268        std.debug.assert(self.phase == .steady);269        const storage = self.storage;270        self.phase = .teardown;271        self.* = undefined;272        return storage;273    }274275    fn assertSteady(self: *const Sealer) void {276        std.debug.assert(self.phase == .steady);277        std.debug.assert(self.storage.len == self.capacity.storage_bytes);278        std.debug.assert(self.blocks <= self.capacity.blocks);279        std.debug.assert(self.block_bytes <= self.capacity.block_bytes);280        std.debug.assert(self.closure_digests <= self.capacity.closure_digests);281    }282};

1pub const Spec = struct {2    epoch: u64,3    ordinal: SegmentRange.Count,4    first_event: EventRange.Count,5    event_count: EventRange.Count,6    previous_segment: Hash,7    event_chain: Hash,8    checkpoint_root: Hash,9    schema_identity: Hash,10    stream_identity: Hash,11};

1pub const View = struct {2    bytes: []const u8,3    digest: Hash,4    spec: Spec,5    block_count: u32,6    closure_count: u32,7    block_bytes: u64,89    pub fn iterator(self: *const View) Iterator {10        return .{11            .view = self,12            .cursor = header_bytes,13            .closures_remaining = self.closure_count,14            .blocks_remaining = self.block_count,15            .next_event = self.spec.first_event,16        };17    }1819    pub fn demand(self: *const View) Error!ReplayDemand {20        var demand_value = ReplayDemand{21            .blocks = self.block_count,22            .events = self.spec.event_count,23            .encoded_block_bytes = self.block_bytes,24            .closure_digests = self.closure_count,25            .checkpoint_roots = @intFromBool(!allZero(&self.spec.checkpoint_root)),26            .segment_bytes = @intCast(self.bytes.len),27        };28        var iterator_value = self.iterator();29        while (try iterator_value.next()) |record| switch (record) {30            .closure => {},31            .block => |block| {32                demand_value.raw_event_bytes = std.math.add(33                    u64,34                    demand_value.raw_event_bytes,35                    block.raw_bytes,36                ) catch return error.InvalidSegment;37                demand_value.max_record_bytes = @max(38                    demand_value.max_record_bytes,39                    @as(u64, block_header_bytes) +40                        @as(u64, @intCast(block.bytes.len)),41                );42            },43        };44        return demand_value;45    }46};

1pub const block_bytes_max: usize = 64 * 1024 * 1024;

1pub const block_header_bytes: usize = 64;

1pub const blocks_max: usize = 64;

1pub const closure_digests_max: usize = 64 * 1024;

1pub const closure_record_bytes: usize = 40;

1pub const encoded_bytes_max: usize = header_bytes +2    blocks_max * block_header_bytes +3    closure_digests_max * closure_record_bytes +4    block_bytes_max + trailer_bytes;

1pub const header_bytes: usize = 256;

1pub fn verify(bytes: []const u8) Error!View {2    if (bytes.len < header_bytes + block_header_bytes + trailer_bytes) {3        return error.InvalidSegment;4    }5    const decoded = try readHeader(bytes[0..header_bytes]);6    const expected_len = try encodedBytes(7        decoded.block_count,8        decoded.block_bytes,9        decoded.closure_count,10    );11    if (bytes.len != expected_len) return error.InvalidSegment;12    const records_end = bytes.len - trailer_bytes;13    if (!std.mem.eql(14        u8,15        &decoded.records_digest,16        &digest(bytes[header_bytes..records_end]),17    )) return error.InvalidSegment;18    const trailer_digest = digest(bytes[0..records_end]);19    if (!std.mem.eql(u8, &trailer_digest, bytes[records_end..])) {20        return error.InvalidSegment;21    }22    const segment_digest = digest(bytes);23    var view = View{24        .bytes = bytes,25        .digest = segment_digest,26        .spec = decoded.spec,27        .block_count = decoded.block_count,28        .closure_count = decoded.closure_count,29        .block_bytes = decoded.block_bytes,30    };31    try verifyRecords(&view);32    return view;33}

1pub const trailer_bytes: usize = Sha256.digest_length;

1pub const segment = @import("segment/root.zig");

1const format = @import("format.zig");23pub const manifest = @import("manifest.zig");4pub const Error = format.Error;5pub const SegmentRange = format.SegmentRange;6pub const EventRange = format.EventRange;7pub const Spec = format.Spec;8pub const Block = format.Block;9pub const Sealed = format.Sealed;10pub const Sealer = format.Sealer;11pub const View = format.View;12pub const ReplayDemand = format.ReplayDemand;13pub const Record = format.Record;14pub const Iterator = format.Iterator;15pub const header_bytes = format.header_bytes;16pub const block_header_bytes = format.block_header_bytes;17pub const closure_record_bytes = format.closure_record_bytes;18pub const trailer_bytes = format.trailer_bytes;19pub const blocks_max = format.blocks_max;20pub const closure_digests_max = format.closure_digests_max;21pub const block_bytes_max = format.block_bytes_max;22pub const encoded_bytes_max = format.encoded_bytes_max;23pub const verify = format.verify;