tiny.sql.history.segment.manifest

1const std = @import("std");2const sql = @import("../../root.zig");3const segment = @import("root.zig");45const Hash = sql.Hash;6const Sha256 = std.crypto.hash.sha2.Sha256;78pub const root_bytes: usize = 128;9pub const page_bytes: usize = 4096;10pub const page_header_bytes: usize = 96;11pub const entry_bytes: usize = 208;12pub const page_trailer_bytes: usize = Sha256.digest_length;13pub const entries_per_page: usize =14    (page_bytes - page_header_bytes - page_trailer_bytes) / entry_bytes;1516pub const PageRange = struct {17    pub const Count = u64;18    pub const unit_precision_segments: Count = 1;19    pub const maximum_segments_per_second: Count =20        segment.SegmentRange.maximum_units_per_second;21    pub const maximum_seconds_per_year: Count =22        segment.SegmentRange.maximum_seconds_per_year;23    pub const service_lifetime_years: Count =24        segment.SegmentRange.service_lifetime_years;25    pub const service_lifetime_seconds: Count =26        segment.SegmentRange.service_lifetime_seconds;27    pub const segments_per_page: Count = entries_per_page;28    pub const maximum_increment_pages: Count = 1;29    pub const budget_pages: Count = total(30        maximum_segments_per_second,31        service_lifetime_seconds,32    ).?;3334    pub fn total(segments_per_second: Count, lifetime_seconds: Count) ?Count {35        const segment_count = @as(u128, unit_precision_segments) *36            @as(u128, segments_per_second) * @as(u128, lifetime_seconds);37        const page_count = if (segment_count == 0) 0 else (segment_count - 1) / @as(u128, segments_per_page) + 1;38        return std.math.cast(Count, page_count);39    }4041    pub fn advance(current: Count, increment: Count) ?Count {42        if (!containsCount(current) or increment > maximum_increment_pages) {43            return null;44        }45        const next = std.math.add(Count, current, increment) catch return null;46        return if (next <= budget_pages) next else null;47    }4849    pub fn containsCount(count: Count) bool {50        return count > 0 and count <= budget_pages;51    }52};5354comptime {55    std.debug.assert(PageRange.budget_pages > 0);56    std.debug.assert(57        PageRange.budget_pages ==58            (segment.SegmentRange.budget_segments - 1) /59                PageRange.segments_per_page + 1,60    );61    std.debug.assert(62        @as(u128, PageRange.budget_pages) +63            @as(u128, PageRange.maximum_increment_pages) <=64            std.math.maxInt(PageRange.Count),65    );66}6768const root_magic = "TINYMNRT".*;69const page_magic = "TINYMPAG".*;70const format_version: u32 = 1;7172pub const Error = segment.Error || error{73    InvalidManifest,74    InvalidLifecycleTransition,75    ManifestChanged,76};7778pub const Tier = enum(u8) {79    sealed = 1,80    cold = 2,81};8283pub const Entry = struct {84    segment: Hash,85    previous_segment: Hash,86    stream_identity: Hash,87    checkpoint_root: Hash,88    cold_receipt: Hash = @splat(0),89    epoch: u64,90    ordinal: segment.SegmentRange.Count,91    first_event: segment.EventRange.Count,92    event_count: segment.EventRange.Count,93    encoded_bytes: u64,94    tier: Tier = .sealed,9596    pub fn fromVerified(value: segment.View) Entry {97        return .{98            .segment = value.digest,99            .previous_segment = value.spec.previous_segment,100            .stream_identity = value.spec.stream_identity,101            .checkpoint_root = value.spec.checkpoint_root,102            .epoch = value.spec.epoch,103            .ordinal = value.spec.ordinal,104            .first_event = value.spec.first_event,105            .event_count = value.spec.event_count,106            .encoded_bytes = value.bytes.len,107        };108    }109110    pub fn localReleaseAllowed(self: Entry) bool {111        return self.tier == .cold and !isZero(self.cold_receipt);112    }113};114115pub const Root = struct {116    digest: Hash,117    tail_page: Hash,118    last_segment: Hash,119    segment_count: segment.SegmentRange.Count,120    page_count: PageRange.Count,121    tail_entries: u32,122123    pub fn demand(self: Root) Error!ReplayDemand {124        return .{125            .manifest_pages = self.page_count,126            .segments = self.segment_count,127            .io_operations = std.math.add(128                u64,129                self.page_count,130                self.segment_count,131            ) catch return error.InvalidManifest,132        };133    }134};135136pub const ReplayDemand = struct {137    manifest_pages: u64,138    segments: u64,139    sealed_segments: u64 = 0,140    cold_segments: u64 = 0,141    event_count: u64 = 0,142    segment_bytes: u64 = 0,143    max_segment_bytes: u64 = 0,144    cold_fetch_count: u64 = 0,145    cold_bytes: u64 = 0,146    io_operations: u64,147};148149pub const Page = struct {150    bytes: []const u8,151    digest: Hash,152    previous_page: Hash,153    first_ordinal: segment.SegmentRange.Count,154    count: u32,155156    pub fn entry(self: Page, index: usize) Error!Entry {157        if (index >= self.count) return error.InvalidManifest;158        const start = page_header_bytes + index * entry_bytes;159        return try readEntry(self.bytes[start..][0..entry_bytes]);160    }161162    pub fn latest(self: Page) Error!Entry {163        if (self.count == 0) return error.InvalidManifest;164        return try self.entry(self.count - 1);165    }166};167168pub const Publication = struct {169    root: Root,170    page: Page,171};172173pub const Reader = struct {174    expected_page: Hash,175    expected_last_segment: Hash,176    pages_remaining: PageRange.Count,177    segments_remaining: segment.SegmentRange.Count,178    first: bool = true,179    tail_entries: u32,180    replay_demand: ReplayDemand,181182    pub fn init(root: Root) Reader {183        std.debug.assert(root.page_count != 0);184        std.debug.assert(root.segment_count != 0);185        return .{186            .expected_page = root.tail_page,187            .expected_last_segment = root.last_segment,188            .pages_remaining = root.page_count,189            .segments_remaining = root.segment_count,190            .tail_entries = root.tail_entries,191            .replay_demand = .{192                .manifest_pages = root.page_count,193                .segments = 0,194                .io_operations = root.page_count,195            },196        };197    }198199    pub fn accept(self: *Reader, bytes: []const u8) Error!Page {200        if (self.pages_remaining == 0) return error.InvalidManifest;201        const page = try verifyPage(bytes);202        if (!same(page.digest, self.expected_page)) return error.ManifestChanged;203        if (self.first and page.count != self.tail_entries) {204            return error.InvalidManifest;205        }206        if (page.count > self.segments_remaining) return error.InvalidManifest;207        const latest = try page.latest();208        if (!same(latest.segment, self.expected_last_segment)) {209            return error.InvalidManifest;210        }211        const ordinal_count = segment.SegmentRange.advance(latest.ordinal) orelse212            return error.InvalidManifest;213        if (ordinal_count != self.segments_remaining) {214            return error.InvalidManifest;215        }216        const oldest = try page.entry(0);217        var next_demand = self.replay_demand;218        for (0..page.count) |index| {219            try addDemand(&next_demand, try page.entry(index));220        }221        self.segments_remaining -= page.count;222        self.pages_remaining -= 1;223        self.expected_page = page.previous_page;224        self.expected_last_segment = oldest.previous_segment;225        self.first = false;226        self.replay_demand = next_demand;227        if ((self.pages_remaining == 0) != isZero(self.expected_page)) {228            return error.InvalidManifest;229        }230        if (self.pages_remaining == 0 and self.segments_remaining != 0) {231            return error.InvalidManifest;232        }233        if (self.pages_remaining == 0 and234            !isZero(self.expected_last_segment))235        {236            return error.InvalidManifest;237        }238        return page;239    }240241    pub fn done(self: Reader) bool {242        return self.pages_remaining == 0 and self.segments_remaining == 0;243    }244245    pub fn demand(self: Reader) Error!ReplayDemand {246        if (!self.done()) return error.InvalidManifest;247        return self.replay_demand;248    }249};250251pub fn append(252    current_root_bytes: ?[]const u8,253    current_tail_bytes: ?[]const u8,254    entry_value: Entry,255    page_out: *[page_bytes]u8,256    root_out: *[root_bytes]u8,257) Error!Publication {258    if (entry_value.tier != .sealed or !isZero(entry_value.cold_receipt)) {259        return error.InvalidLifecycleTransition;260    }261    if (!entryWithinRanges(entry_value)) return error.InvalidManifest;262    const current = try validateCurrent(current_root_bytes, current_tail_bytes);263    if (current) |state| {264        if (entry_value.ordinal != state.root.segment_count or265            !same(entry_value.previous_segment, state.root.last_segment))266        {267            return error.InvalidManifest;268        }269        const page_count = PageRange.advance(270            state.root.page_count,271            @intFromBool(state.page.count == entries_per_page),272        ) orelse return error.InvalidManifest;273        const segment_count = segment.SegmentRange.advance(274            state.root.segment_count,275        ) orelse return error.InvalidManifest;276        if (state.page.count < entries_per_page) {277            copyPageAndAppend(state.page, entry_value, page_out);278        } else {279            writePage(page_out, state.page.digest, entry_value);280        }281        const page = try verifyPage(page_out);282        writeRoot(root_out, .{283            .tail_page = page.digest,284            .last_segment = entry_value.segment,285            .segment_count = segment_count,286            .page_count = page_count,287            .tail_entries = page.count,288        });289        return .{ .root = try verifyRoot(root_out), .page = page };290    }291    if (entry_value.ordinal != 0 or !isZero(entry_value.previous_segment)) {292        return error.InvalidManifest;293    }294    writePage(page_out, @splat(0), entry_value);295    const page = try verifyPage(page_out);296    writeRoot(root_out, .{297        .tail_page = page.digest,298        .last_segment = entry_value.segment,299        .segment_count = 1,300        .page_count = 1,301        .tail_entries = 1,302    });303    return .{ .root = try verifyRoot(root_out), .page = page };304}305306pub fn promoteTailCold(307    current_root_bytes: []const u8,308    current_tail_bytes: []const u8,309    copied_segment: []const u8,310    page_out: *[page_bytes]u8,311    root_out: *[root_bytes]u8,312) Error!Publication {313    const state = (try validateCurrent(314        current_root_bytes,315        current_tail_bytes,316    )) orelse return error.InvalidManifest;317    var latest = try state.page.latest();318    if (latest.tier != .sealed or !isZero(latest.cold_receipt)) {319        return error.InvalidLifecycleTransition;320    }321    const verified = try segment.verify(copied_segment);322    if (!same(verified.digest, latest.segment) or323        copied_segment.len != latest.encoded_bytes)324    {325        return error.InvalidLifecycleTransition;326    }327    latest.tier = .cold;328    latest.cold_receipt = coldReceipt(latest.segment, copied_segment.len);329    page_out.* = state.page.bytes[0..page_bytes].*;330    const latest_index: usize = state.page.count - 1;331    const start = page_header_bytes + latest_index * entry_bytes;332    writeEntry(page_out[start..][0..entry_bytes], latest);333    finishPage(page_out, state.page.count);334    const page = try verifyPage(page_out);335    writeRoot(root_out, .{336        .tail_page = page.digest,337        .last_segment = state.root.last_segment,338        .segment_count = state.root.segment_count,339        .page_count = state.root.page_count,340        .tail_entries = state.root.tail_entries,341    });342    return .{ .root = try verifyRoot(root_out), .page = page };343}344345pub fn verifyRoot(bytes: []const u8) Error!Root {346    if (bytes.len != root_bytes or !std.mem.eql(u8, bytes[0..8], &root_magic)) {347        return error.InvalidManifest;348    }349    if (std.mem.readInt(u32, bytes[8..12], .little) != format_version) {350        return error.InvalidManifest;351    }352    const root_checksum = digest(bytes[0 .. root_bytes - Sha256.digest_length]);353    if (!std.mem.eql(u8, &root_checksum, bytes[root_bytes - Sha256.digest_length ..])) {354        return error.InvalidManifest;355    }356    const root_digest = digest(bytes);357    const segment_count = std.mem.readInt(u64, bytes[16..24], .little);358    const page_count = std.mem.readInt(u64, bytes[24..32], .little);359    const tail_entries = std.mem.readInt(u32, bytes[12..16], .little);360    if (!segment.SegmentRange.containsCount(segment_count) or361        !PageRange.containsCount(page_count) or tail_entries == 0 or362        tail_entries > entries_per_page or page_count > segment_count)363    {364        return error.InvalidManifest;365    }366    const page_width: u64 = entries_per_page;367    const expected_page_count = (segment_count - 1) / page_width + 1;368    const expected_tail_entries = (segment_count - 1) % page_width + 1;369    if (page_count != expected_page_count or370        @as(u64, tail_entries) != expected_tail_entries)371    {372        return error.InvalidManifest;373    }374    return .{375        .digest = root_digest,376        .tail_page = bytes[32..64].*,377        .last_segment = bytes[64..96].*,378        .segment_count = segment_count,379        .page_count = page_count,380        .tail_entries = tail_entries,381    };382}383384pub fn verifyPage(bytes: []const u8) Error!Page {385    if (bytes.len != page_bytes or !std.mem.eql(u8, bytes[0..8], &page_magic)) {386        return error.InvalidManifest;387    }388    if (std.mem.readInt(u32, bytes[8..12], .little) != format_version) {389        return error.InvalidManifest;390    }391    const count = std.mem.readInt(u32, bytes[12..16], .little);392    if (count == 0 or count > entries_per_page) return error.InvalidManifest;393    const used_end = page_header_bytes + count * entry_bytes;394    if (!allZero(bytes[used_end .. page_bytes - page_trailer_bytes])) {395        return error.InvalidManifest;396    }397    if (!std.mem.eql(u8, bytes[64..96], &digest(bytes[page_header_bytes..used_end]))) {398        return error.InvalidManifest;399    }400    const page_checksum = digest(bytes[0 .. page_bytes - page_trailer_bytes]);401    if (!std.mem.eql(u8, &page_checksum, bytes[page_bytes - page_trailer_bytes ..])) {402        return error.InvalidManifest;403    }404    const page_digest = digest(bytes);405    const page = Page{406        .bytes = bytes,407        .digest = page_digest,408        .previous_page = bytes[32..64].*,409        .first_ordinal = std.mem.readInt(u64, bytes[16..24], .little),410        .count = count,411    };412    try verifyEntries(page);413    return page;414}415416const Current = struct {417    root: Root,418    page: Page,419};420421const RootFields = struct {422    tail_page: Hash,423    last_segment: Hash,424    segment_count: segment.SegmentRange.Count,425    page_count: PageRange.Count,426    tail_entries: u32,427};428429fn validateCurrent(430    root_value: ?[]const u8,431    page_value: ?[]const u8,432) Error!?Current {433    if ((root_value == null) != (page_value == null)) return error.InvalidManifest;434    const root_bytes_value = root_value orelse return null;435    const page_bytes_value = page_value.?;436    const root = try verifyRoot(root_bytes_value);437    const page = try verifyPage(page_bytes_value);438    if (!same(root.tail_page, page.digest) or root.tail_entries != page.count) {439        return error.ManifestChanged;440    }441    const latest = try page.latest();442    const segment_count = segment.SegmentRange.advance(latest.ordinal) orelse443        return error.InvalidManifest;444    if (!same(root.last_segment, latest.segment) or445        root.segment_count != segment_count)446    {447        return error.InvalidManifest;448    }449    return .{ .root = root, .page = page };450}451452fn verifyEntries(page: Page) Error!void {453    var previous: ?Entry = null;454    for (0..page.count) |index| {455        const entry_value = try page.entry(index);456        if (!entryWithinRanges(entry_value)) return error.InvalidManifest;457        const expected_ordinal = std.math.add(458            segment.SegmentRange.Count,459            page.first_ordinal,460            @intCast(index),461        ) catch return error.InvalidManifest;462        if (entry_value.ordinal != expected_ordinal) {463            return error.InvalidManifest;464        }465        if (entry_value.event_count == 0 or entry_value.encoded_bytes == 0 or466            entry_value.encoded_bytes > segment.encoded_bytes_max or467            isZero(entry_value.segment) or isZero(entry_value.stream_identity))468        {469            return error.InvalidManifest;470        }471        if (previous) |prior| {472            if (!same(entry_value.previous_segment, prior.segment)) {473                return error.InvalidManifest;474            }475        }476        switch (entry_value.tier) {477            .sealed => if (!isZero(entry_value.cold_receipt)) {478                return error.InvalidManifest;479            },480            .cold => if (isZero(entry_value.cold_receipt)) {481                return error.InvalidManifest;482            },483        }484        previous = entry_value;485    }486}487488fn addDemand(demand: *ReplayDemand, entry_value: Entry) Error!void {489    demand.segments = std.math.add(u64, demand.segments, 1) catch490        return error.InvalidManifest;491    demand.event_count = std.math.add(492        u64,493        demand.event_count,494        entry_value.event_count,495    ) catch return error.InvalidManifest;496    demand.segment_bytes = std.math.add(497        u64,498        demand.segment_bytes,499        entry_value.encoded_bytes,500    ) catch return error.InvalidManifest;501    demand.max_segment_bytes = @max(502        demand.max_segment_bytes,503        entry_value.encoded_bytes,504    );505    demand.io_operations = std.math.add(u64, demand.io_operations, 1) catch506        return error.InvalidManifest;507    switch (entry_value.tier) {508        .sealed => demand.sealed_segments = std.math.add(509            u64,510            demand.sealed_segments,511            1,512        ) catch return error.InvalidManifest,513        .cold => {514            demand.cold_segments = std.math.add(515                u64,516                demand.cold_segments,517                1,518            ) catch return error.InvalidManifest;519            demand.cold_fetch_count = std.math.add(520                u64,521                demand.cold_fetch_count,522                1,523            ) catch return error.InvalidManifest;524            demand.cold_bytes = std.math.add(525                u64,526                demand.cold_bytes,527                entry_value.encoded_bytes,528            ) catch return error.InvalidManifest;529        },530    }531}532533fn writeRoot(out: *[root_bytes]u8, fields: RootFields) void {534    std.debug.assert(segment.SegmentRange.containsCount(fields.segment_count));535    std.debug.assert(PageRange.containsCount(fields.page_count));536    std.debug.assert(fields.tail_entries != 0);537    @memset(out, 0);538    @memcpy(out[0..8], &root_magic);539    std.mem.writeInt(u32, out[8..12], format_version, .little);540    std.mem.writeInt(u32, out[12..16], fields.tail_entries, .little);541    std.mem.writeInt(u64, out[16..24], fields.segment_count, .little);542    std.mem.writeInt(u64, out[24..32], fields.page_count, .little);543    @memcpy(out[32..64], &fields.tail_page);544    @memcpy(out[64..96], &fields.last_segment);545    const root_checksum = digest(out[0 .. root_bytes - Sha256.digest_length]);546    @memcpy(out[root_bytes - Sha256.digest_length ..], &root_checksum);547}548549fn writePage(out: *[page_bytes]u8, previous_page: Hash, entry_value: Entry) void {550    std.debug.assert(entryWithinRanges(entry_value));551    @memset(out, 0);552    @memcpy(out[0..8], &page_magic);553    std.mem.writeInt(u32, out[8..12], format_version, .little);554    std.mem.writeInt(u32, out[12..16], 1, .little);555    std.mem.writeInt(u64, out[16..24], entry_value.ordinal, .little);556    @memcpy(out[32..64], &previous_page);557    writeEntry(out[page_header_bytes..][0..entry_bytes], entry_value);558    finishPage(out, 1);559}560561fn copyPageAndAppend(page: Page, entry_value: Entry, out: *[page_bytes]u8) void {562    std.debug.assert(page.count < entries_per_page);563    out.* = page.bytes[0..page_bytes].*;564    const index: usize = page.count;565    const start = page_header_bytes + index * entry_bytes;566    writeEntry(out[start..][0..entry_bytes], entry_value);567    const count = page.count + 1;568    std.mem.writeInt(u32, out[12..16], count, .little);569    finishPage(out, count);570}571572fn finishPage(out: *[page_bytes]u8, count: u32) void {573    const used_end = page_header_bytes + count * entry_bytes;574    @memset(out[used_end .. page_bytes - page_trailer_bytes], 0);575    const entries_digest = digest(out[page_header_bytes..used_end]);576    @memcpy(out[64..96], &entries_digest);577    const page_checksum = digest(out[0 .. page_bytes - page_trailer_bytes]);578    @memcpy(out[page_bytes - page_trailer_bytes ..], &page_checksum);579}580581fn writeEntry(out: []u8, value: Entry) void {582    std.debug.assert(out.len == entry_bytes);583    std.debug.assert(entryWithinRanges(value));584    @memset(out, 0);585    @memcpy(out[0..32], &value.segment);586    @memcpy(out[32..64], &value.previous_segment);587    @memcpy(out[64..96], &value.stream_identity);588    @memcpy(out[96..128], &value.checkpoint_root);589    @memcpy(out[128..160], &value.cold_receipt);590    std.mem.writeInt(u64, out[160..168], value.epoch, .little);591    std.mem.writeInt(u64, out[168..176], value.ordinal, .little);592    std.mem.writeInt(u64, out[176..184], value.first_event, .little);593    std.mem.writeInt(u64, out[184..192], value.event_count, .little);594    std.mem.writeInt(u64, out[192..200], value.encoded_bytes, .little);595    out[200] = @backingInt(value.tier);596}597598fn readEntry(bytes: []const u8) Error!Entry {599    if (bytes.len != entry_bytes or !allZero(bytes[201..208])) {600        return error.InvalidManifest;601    }602    return .{603        .segment = bytes[0..32].*,604        .previous_segment = bytes[32..64].*,605        .stream_identity = bytes[64..96].*,606        .checkpoint_root = bytes[96..128].*,607        .cold_receipt = bytes[128..160].*,608        .epoch = std.mem.readInt(u64, bytes[160..168], .little),609        .ordinal = std.mem.readInt(u64, bytes[168..176], .little),610        .first_event = std.mem.readInt(u64, bytes[176..184], .little),611        .event_count = std.mem.readInt(u64, bytes[184..192], .little),612        .encoded_bytes = std.mem.readInt(u64, bytes[192..200], .little),613        .tier = std.enums.fromInt(Tier, bytes[200]) orelse614            return error.InvalidManifest,615    };616}617618fn entryWithinRanges(value: Entry) bool {619    return segment.SegmentRange.containsOrdinal(value.ordinal) and620        segment.EventRange.advance(value.first_event, value.event_count) != null;621}622623fn coldReceipt(segment_digest: Hash, bytes: usize) Hash {624    var hasher = Sha256.init(.{});625    hasher.update("tiny.sql.history.cold-receipt/v1");626    hasher.update(&segment_digest);627    var frame: [8]u8 = undefined;628    std.mem.writeInt(u64, &frame, bytes, .little);629    hasher.update(&frame);630    var value: Hash = undefined;631    hasher.final(&value);632    return value;633}634635fn digest(bytes: []const u8) Hash {636    var value: Hash = undefined;637    Sha256.hash(bytes, &value, .{});638    return value;639}640641fn same(left: Hash, right: Hash) bool {642    return std.mem.eql(u8, &left, &right);643}644645fn isZero(value: Hash) bool {646    return same(value, @splat(0));647}648649fn allZero(bytes: []const u8) bool {650    for (bytes) |byte| if (byte != 0) return false;651    return true;652}

1pub const manifest = @import("manifest.zig");