tiny.sql.history.refs

1const std = @import("std");2const simd = @import("simd");3const sql = @import("../root.zig");4const record_mod = @import("record.zig");5const validate_mod = @import("validate.zig");67const Bytes = simd.ScalableTag(u8);8const version = sql.version;910const Allocator = std.mem.Allocator;1112pub const Error = Allocator.Error;13pub const ReadError = Error || std.Io.Dir.ReadFileAllocError;1415pub const suffix = ".refs";16pub const magic: u32 = 0x7473_7266;17pub const format_version: u32 = 2;18pub const max_snapshot_bytes: usize = 4 * 1024 * 1024;19pub const max_entries: u32 = 65_536;20pub const max_name_bytes: u32 = 4_096;21pub const path_bytes_max: usize = 1_024;22pub const repair_metadata_record_bytes_max: usize = max_snapshot_bytes;2324const repair_stream_bytes: usize = 64 * 1024;25const repair_dependencies_max: usize = repair_metadata_record_bytes_max /26    version.hash_bytes;27const repair_conflict_bytes_max: usize = max_snapshot_bytes;28const repair_limits = validate_mod.Limits{29    .metadata_bytes_max = repair_metadata_record_bytes_max,30    .dependencies_max = repair_dependencies_max,31    .conflict_bytes_max = repair_conflict_bytes_max,32};3334const header_size: usize = 4 + 4 + 8 + 4;35const entry_fixed_size: usize = 4 + 3 * version.hash_bytes;36const digest_size: usize = version.hash_bytes;37const write_suffix = ".next";38const write_path_bytes_max = path_bytes_max + write_suffix.len;3940pub const Entry = struct {41    name: []const u8,42    head: version.Hash,43    root: version.Hash,44    conflicts: version.Hash,45};4647pub const Snapshot = struct {48    allocator: Allocator,49    covered_length: u64,50    entries: []Entry,51    names: []u8,5253    pub fn deinit(self: *Snapshot) void {54        self.allocator.free(self.entries);55        self.allocator.free(self.names);56        self.* = undefined;57    }5859    pub fn find(self: *const Snapshot, name: []const u8) ?Entry {60        std.debug.assert(self.entries.len <= max_entries);61        for (self.entries) |entry| {62            if (std.mem.eql(u8, entry.name, name)) return entry;63        }64        return null;65    }66};6768pub fn pathFor(buffer: *[path_bytes_max]u8, history_path: []const u8) ?[]const u8 {69    if (history_path.len + suffix.len > buffer.len) return null;70    @memcpy(buffer[0..history_path.len], history_path);71    @memcpy(buffer[history_path.len..][0..suffix.len], suffix);72    return buffer[0 .. history_path.len + suffix.len];73}7475pub fn load(76    allocator: Allocator,77    io: std.Io,78    dir: std.Io.Dir,79    history_path: []const u8,80) Error!?Snapshot {81    return loadExisting(allocator, io, dir, history_path) catch |err| switch (err) {82        error.OutOfMemory => return error.OutOfMemory,83        else => return null,84    };85}8687pub fn loadExisting(88    allocator: Allocator,89    io: std.Io,90    dir: std.Io.Dir,91    history_path: []const u8,92) ReadError!?Snapshot {93    var path_buffer: [path_bytes_max]u8 = undefined;94    const path = pathFor(&path_buffer, history_path) orelse return null;95    const bytes = dir.readFileAlloc(io, path, allocator, .limited(max_snapshot_bytes)) catch |err| switch (err) {96        error.StreamTooLong => return null,97        else => return err,98    };99    defer allocator.free(bytes);100    return try decode(allocator, bytes);101}102103pub fn store(104    allocator: Allocator,105    io: std.Io,106    dir: std.Io.Dir,107    history_path: []const u8,108    covered_length: u64,109    entries: []const Entry,110) (Error || std.Io.File.OpenError || std.Io.File.WritePositionalError ||111    std.Io.Dir.RenameError)!void {112    std.debug.assert(entries.len <= max_entries);113    var path_buffer: [path_bytes_max]u8 = undefined;114    const path = pathFor(&path_buffer, history_path) orelse return;115    var write_buffer: [write_path_bytes_max]u8 = undefined;116    @memcpy(write_buffer[0..path.len], path);117    @memcpy(write_buffer[path.len..][0..write_suffix.len], write_suffix);118    const write_path = write_buffer[0 .. path.len + write_suffix.len];119120    const bytes = try encode(allocator, covered_length, entries);121    defer allocator.free(bytes);122123    {124        var file = try dir.createFile(io, write_path, .{ .read = true, .truncate = true });125        defer file.close(io);126        try file.writePositionalAll(io, bytes, 0);127    }128    try dir.rename(write_path, dir, path, io);129}130131fn encode(allocator: Allocator, covered_length: u64, entries: []const Entry) Error![]u8 {132    std.debug.assert(entries.len <= max_entries);133    var total: usize = header_size;134    for (entries) |entry| {135        std.debug.assert(entry.name.len <= max_name_bytes);136        total += entry_fixed_size + entry.name.len;137    }138    total += digest_size;139    std.debug.assert(total <= max_snapshot_bytes);140141    const bytes = try allocator.alloc(u8, total);142    errdefer allocator.free(bytes);143    var cursor: usize = 0;144    writeU32(bytes, &cursor, magic);145    writeU32(bytes, &cursor, format_version);146    writeU64(bytes, &cursor, covered_length);147    writeU32(bytes, &cursor, @intCast(entries.len));148    for (entries) |entry| {149        writeU32(bytes, &cursor, @intCast(entry.name.len));150        @memcpy(bytes[cursor..][0..entry.name.len], entry.name);151        cursor += entry.name.len;152        @memcpy(bytes[cursor..][0..version.hash_bytes], entry.head[0..]);153        cursor += version.hash_bytes;154        @memcpy(bytes[cursor..][0..version.hash_bytes], entry.root[0..]);155        cursor += version.hash_bytes;156        @memcpy(bytes[cursor..][0..version.hash_bytes], entry.conflicts[0..]);157        cursor += version.hash_bytes;158    }159    std.debug.assert(cursor + digest_size == total);160    var digest: version.Hash = undefined;161    std.crypto.hash.sha2.Sha256.hash(bytes[0..cursor], &digest, .{});162    @memcpy(bytes[cursor..][0..digest_size], digest[0..]);163    return bytes;164}165166fn decode(allocator: Allocator, bytes: []const u8) Error!?Snapshot {167    if (bytes.len < header_size + digest_size) return null;168    if (bytes.len > max_snapshot_bytes) return null;169    const body = bytes[0 .. bytes.len - digest_size];170    var digest: version.Hash = undefined;171    std.crypto.hash.sha2.Sha256.hash(body, &digest, .{});172    if (!std.mem.eql(u8, digest[0..], bytes[body.len..])) return null;173174    var cursor: usize = 0;175    if (readU32(body, &cursor) != magic) return null;176    if (readU32(body, &cursor) != format_version) return null;177    const covered_length = readU64(body, &cursor);178    const entry_count = readU32(body, &cursor);179    if (entry_count > max_entries) return null;180181    var name_bytes: usize = 0;182    {183        var scan = cursor;184        var index: u32 = 0;185        while (index < entry_count) : (index += 1) {186            if (body.len - scan < 4) return null;187            var peek = scan;188            const name_len = readU32(body, &peek);189            if (name_len == 0 or name_len > max_name_bytes) return null;190            if (body.len - peek < name_len + 3 * version.hash_bytes) return null;191            name_bytes += name_len;192            scan = peek + name_len + 3 * version.hash_bytes;193        }194        if (scan != body.len) return null;195    }196197    const entries = try allocator.alloc(Entry, entry_count);198    errdefer allocator.free(entries);199    const names = try allocator.alloc(u8, name_bytes);200    errdefer allocator.free(names);201202    var name_cursor: usize = 0;203    var index: u32 = 0;204    while (index < entry_count) : (index += 1) {205        const name_len = readU32(body, &cursor);206        const name = names[name_cursor..][0..name_len];207        @memcpy(name, body[cursor..][0..name_len]);208        cursor += name_len;209        name_cursor += name_len;210        var head: version.Hash = undefined;211        @memcpy(head[0..], body[cursor..][0..version.hash_bytes]);212        cursor += version.hash_bytes;213        var root: version.Hash = undefined;214        @memcpy(root[0..], body[cursor..][0..version.hash_bytes]);215        cursor += version.hash_bytes;216        var conflicts: version.Hash = undefined;217        @memcpy(conflicts[0..], body[cursor..][0..version.hash_bytes]);218        cursor += version.hash_bytes;219        entries[index] = .{220            .name = name,221            .head = head,222            .root = root,223            .conflicts = conflicts,224        };225    }226    std.debug.assert(cursor == body.len);227    std.debug.assert(name_cursor == names.len);228229    return .{230        .allocator = allocator,231        .covered_length = covered_length,232        .entries = entries,233        .names = names,234    };235}236237fn writeU32(bytes: []u8, cursor: *usize, value: u32) void {238    std.mem.writeInt(u32, bytes[cursor.*..][0..4], value, .big);239    cursor.* += 4;240}241242fn writeU64(bytes: []u8, cursor: *usize, value: u64) void {243    std.mem.writeInt(u64, bytes[cursor.*..][0..8], value, .big);244    cursor.* += 8;245}246247fn readU32(bytes: []const u8, cursor: *usize) u32 {248    const value = std.mem.readInt(u32, bytes[cursor.*..][0..4], .big);249    cursor.* += 4;250    return value;251}252253fn readU64(bytes: []const u8, cursor: *usize) u64 {254    const value = std.mem.readInt(u64, bytes[cursor.*..][0..8], .big);255    cursor.* += 8;256    return value;257}258259const RepairStage = enum {260    current,261    head,262    root,263};264265const RepairEntry = struct {266    name: []u8,267    head: version.Hash,268    root: version.Hash,269    conflicts: version.Hash,270    stage: RepairStage,271};272273const FastForwardDecision = enum {274    pending,275    baseline,276    target,277};278279const FastForward = struct {280    id: version.Hash,281    name: []u8,282    expected: version.Hash,283    target: version.Hash,284    decision: FastForwardDecision = .pending,285};286287const RepairState = struct {288    allocator: Allocator,289    entries: std.ArrayList(RepairEntry) = .empty,290    encoded_bytes: usize = header_size + digest_size,291    fast_forward: ?FastForward = null,292293    fn init(allocator: Allocator, snapshot: *const Snapshot) !RepairState {294        var state = RepairState{ .allocator = allocator };295        errdefer state.deinit();296        try state.entries.ensureTotalCapacity(allocator, snapshot.entries.len);297        for (snapshot.entries) |entry| {298            const name = try allocator.dupe(u8, entry.name);299            state.entries.appendAssumeCapacity(.{300                .name = name,301                .head = entry.head,302                .root = entry.root,303                .conflicts = entry.conflicts,304                .stage = .current,305            });306            state.encoded_bytes += entry_fixed_size + name.len;307        }308        std.debug.assert(state.entries.items.len <= max_entries);309        std.debug.assert(state.encoded_bytes <= max_snapshot_bytes);310        return state;311    }312313    fn deinit(self: *RepairState) void {314        if (self.fast_forward) |pending| self.allocator.free(pending.name);315        for (self.entries.items) |entry| self.allocator.free(entry.name);316        self.entries.deinit(self.allocator);317        self.* = undefined;318    }319320    fn find(self: *RepairState, name: []const u8) ?usize {321        std.debug.assert(self.entries.items.len <= max_entries);322        for (self.entries.items, 0..) |entry, index| {323            if (std.mem.eql(u8, entry.name, name)) return index;324        }325        return null;326    }327328    fn setRef(self: *RepairState, name: []const u8, head: version.Hash) !void {329        if (name.len == 0 or name.len > max_name_bytes) return error.InvalidHistory;330        if (self.find(name)) |index| {331            if (version.same(self.entries.items[index].head, head)) return;332            const metadata = self.metadataForHead(head);333            self.entries.items[index].head = head;334            if (metadata) |entry| {335                self.entries.items[index].root = entry.root;336                self.entries.items[index].conflicts = entry.conflicts;337                self.entries.items[index].stage = .current;338            } else {339                self.entries.items[index].stage = .head;340            }341            return;342        }343        if (self.entries.items.len >= max_entries) return error.StreamTooLong;344        const next_bytes = std.math.add(345            usize,346            self.encoded_bytes,347            entry_fixed_size + name.len,348        ) catch return error.StreamTooLong;349        if (next_bytes > max_snapshot_bytes) return error.StreamTooLong;350        const owned_name = try self.allocator.dupe(u8, name);351        errdefer self.allocator.free(owned_name);352        try self.entries.append(self.allocator, .{353            .name = owned_name,354            .head = head,355            .root = undefined,356            .conflicts = undefined,357            .stage = .head,358        });359        self.encoded_bytes = next_bytes;360    }361362    fn metadataForHead(self: *const RepairState, head: version.Hash) ?RepairEntry {363        std.debug.assert(self.entries.items.len <= max_entries);364        for (self.entries.items) |entry| {365            if (entry.stage == .current and version.same(entry.head, head)) return entry;366        }367        return null;368    }369370    fn deleteRef(self: *RepairState, name: []const u8) void {371        const index = self.find(name) orelse return;372        const removed = self.entries.orderedRemove(index);373        self.encoded_bytes -= entry_fixed_size + removed.name.len;374        self.allocator.free(removed.name);375    }376377    fn finish(self: *RepairState, covered_length: u64) !Snapshot {378        if (self.fast_forward != null) return error.InvalidHistory;379        for (self.entries.items) |entry| {380            if (entry.stage != .current) return error.InvalidHistory;381        }382        const entries = try self.allocator.alloc(Entry, self.entries.items.len);383        errdefer self.allocator.free(entries);384        var names_bytes: usize = 0;385        for (self.entries.items) |entry| names_bytes += entry.name.len;386        const names = try self.allocator.alloc(u8, names_bytes);387        errdefer self.allocator.free(names);388        var cursor: usize = 0;389        for (self.entries.items, entries) |source, *target| {390            const name = names[cursor..][0..source.name.len];391            @memcpy(name, source.name);392            cursor += name.len;393            target.* = .{394                .name = name,395                .head = source.head,396                .root = source.root,397                .conflicts = source.conflicts,398            };399        }400        std.debug.assert(cursor == names.len);401        return .{402            .allocator = self.allocator,403            .covered_length = covered_length,404            .entries = entries,405            .names = names,406        };407    }408};409410const RecordView = struct {411    kind: record_mod.RecordKind,412    expected: version.Hash,413    payload: []const u8,414};415416const ScanMode = enum {417    refs,418    commits,419    roots,420421    fn materializes(self: ScanMode, kind: record_mod.RecordKind) bool {422        return switch (self) {423            .refs => switch (kind) {424                .ref,425                .ref_delete,426                .fast_forward_prepare,427                .fast_forward_commit,428                .fast_forward_abort,429                .fast_forward_complete,430                => true,431                else => false,432            },433            .commits => kind == .commit,434            .roots => kind == .database_root,435        };436    }437};438439const RecordScanner = struct {440    allocator: Allocator,441    io: std.Io,442    file: std.Io.File,443    limit: usize,444    offset: usize,445    mode: ScanMode,446    control: sql.wal.Control,447    payload: std.ArrayList(u8) = .empty,448449    fn init(450        allocator: Allocator,451        io: std.Io,452        file: std.Io.File,453        start: usize,454        end: usize,455        mode: ScanMode,456        control: sql.wal.Control,457    ) RecordScanner {458        std.debug.assert(start <= end);459        return .{460            .allocator = allocator,461            .io = io,462            .file = file,463            .limit = end,464            .offset = start,465            .mode = mode,466            .control = control,467        };468    }469470    fn deinit(self: *RecordScanner) void {471        self.payload.deinit(self.allocator);472        self.* = undefined;473    }474475    fn next(self: *RecordScanner) !?RecordView {476        try self.control.check();477        if (self.offset == self.limit) return null;478        if (self.limit - self.offset < record_mod.record_header_size) {479            return error.InvalidHistory;480        }481        var header: [record_mod.record_header_size]u8 = undefined;482        try self.readExact(&header, self.offset);483        if (record_mod.readIntU32(header[0..4]) != record_mod.magic or484            record_mod.readIntU32(header[4..8]) != record_mod.format_version)485        {486            return error.InvalidHistory;487        }488        const kind_value = record_mod.readIntU32(header[8..12]);489        const kind = record_mod.recordKind(kind_value) orelse return error.InvalidHistory;490        const payload_len: usize = record_mod.readIntU32(header[12..16]);491        if (kind == .row_chunk and payload_len < version.hash_bytes) {492            return error.InvalidHistory;493        }494        const payload_offset = std.math.add(495            usize,496            self.offset,497            record_mod.record_header_size,498        ) catch return error.InvalidHistory;499        const payload_end = std.math.add(usize, payload_offset, payload_len) catch500            return error.InvalidHistory;501        if (payload_end > self.limit) return error.InvalidHistory;502        const materialize = self.mode.materializes(kind);503        if (materialize and payload_len > repair_metadata_record_bytes_max) {504            return error.StreamTooLong;505        }506        self.payload.clearRetainingCapacity();507        if (materialize) try self.payload.resize(self.allocator, payload_len);508        var hasher = std.crypto.hash.sha2.Sha256.init(.{});509        record_mod.writeHashU32(&hasher, record_mod.magic);510        record_mod.writeHashU32(&hasher, record_mod.format_version);511        record_mod.writeHashU32(&hasher, kind_value);512        record_mod.writeHashU32(&hasher, @intCast(payload_len));513        if (materialize) {514            try self.readExact(self.payload.items, payload_offset);515            hasher.update(self.payload.items);516        } else {517            var scratch: [repair_stream_bytes]u8 = undefined;518            var cursor = payload_offset;519            while (cursor < payload_end) {520                try self.control.check();521                const chunk_len = @min(scratch.len, payload_end - cursor);522                try self.readExact(scratch[0..chunk_len], cursor);523                hasher.update(scratch[0..chunk_len]);524                cursor += chunk_len;525            }526        }527        var actual: version.Hash = undefined;528        hasher.final(&actual);529        const expected = header[16..][0..version.hash_bytes].*;530        if (!version.same(actual, expected)) return error.InvalidHistory;531        self.offset = payload_end;532        return .{533            .kind = kind,534            .expected = expected,535            .payload = self.payload.items,536        };537    }538539    fn readExact(self: *RecordScanner, target: []u8, start: usize) !void {540        var filled: usize = 0;541        while (filled < target.len) {542            try self.control.check();543            const count = try self.file.readPositionalAll(544                self.io,545                target[filled..],546                start + filled,547            );548            if (count == 0) return error.InvalidHistory;549            filled += count;550        }551    }552};553554pub fn advance(555    allocator: Allocator,556    io: std.Io,557    dir: std.Io.Dir,558    history_path: []const u8,559    snapshot: *const Snapshot,560    history_length: u64,561    control: sql.wal.Control,562) !Snapshot {563    try control.check();564    if (snapshot.covered_length > history_length) return error.InvalidHistory;565    const start = std.math.cast(usize, snapshot.covered_length) orelse566        return error.StreamTooLong;567    const end = std.math.cast(usize, history_length) orelse return error.StreamTooLong;568    var file = try dir.openFile(io, history_path, .{});569    defer file.close(io);570    try validate_mod.validateSuffix(571        allocator,572        io,573        file,574        start,575        end,576        snapshot.entries,577        repair_limits,578        control,579    );580    var state = try RepairState.init(allocator, snapshot);581    defer state.deinit();582    try applySuffix(&state, allocator, io, file, start, end, control);583    try resolveHeads(&state, allocator, io, file, start, end, control);584    if (hasStage(&state, .head) and start != 0) {585        try resolveHeads(&state, allocator, io, file, 0, start, control);586    }587    try resolveRoots(&state, allocator, io, file, start, end, control);588    if (hasStage(&state, .root) and start != 0) {589        try resolveRoots(&state, allocator, io, file, 0, start, control);590    }591    return try state.finish(history_length);592}593594fn applySuffix(595    state: *RepairState,596    allocator: Allocator,597    io: std.Io,598    file: std.Io.File,599    start: usize,600    end: usize,601    control: sql.wal.Control,602) !void {603    var scanner = RecordScanner.init(allocator, io, file, start, end, .refs, control);604    defer scanner.deinit();605    while (try scanner.next()) |record| {606        if (state.fast_forward != null and !fastForwardKind(record.kind)) {607            return error.InvalidHistory;608        }609        switch (record.kind) {610            .ref => {611                try applyRef(state, record.payload);612            },613            .ref_delete => try applyRefDelete(state, record.payload),614            .fast_forward_prepare => try applyFastForwardPrepare(state, record),615            .fast_forward_commit => try applyFastForwardDecision(state, record.payload, .target),616            .fast_forward_abort => try applyFastForwardDecision(state, record.payload, .baseline),617            .fast_forward_complete => try applyFastForwardComplete(state, record.payload),618            else => {},619        }620    }621}622623fn fastForwardKind(kind: record_mod.RecordKind) bool {624    return switch (kind) {625        .fast_forward_prepare,626        .fast_forward_commit,627        .fast_forward_abort,628        .fast_forward_complete,629        => true,630        else => false,631    };632}633634fn applyRef(state: *RepairState, payload: []const u8) !void {635    var reader = record_mod.PayloadReader.init(payload);636    const target = try reader.hash();637    const name = try reader.readBytes();638    try reader.finish();639    try state.setRef(name, target);640}641642fn applyRefDelete(state: *RepairState, payload: []const u8) !void {643    var reader = record_mod.PayloadReader.init(payload);644    const name = try reader.readBytes();645    try reader.finish();646    state.deleteRef(name);647}648649fn applyFastForwardPrepare(650    state: *RepairState,651    record: RecordView,652) !void {653    if (state.fast_forward != null) return error.InvalidHistory;654    var reader = record_mod.PayloadReader.init(record.payload);655    const name = try reader.readBytes();656    const expected = try reader.hash();657    const target = try reader.hash();658    try reader.finish();659    const index = state.find(name) orelse return error.InvalidHistory;660    if (!version.same(state.entries.items[index].head, expected)) return error.InvalidHistory;661    state.fast_forward = .{662        .id = record.expected,663        .name = try state.allocator.dupe(u8, name),664        .expected = expected,665        .target = target,666    };667}668669fn applyFastForwardDecision(670    state: *RepairState,671    payload: []const u8,672    decision: FastForwardDecision,673) !void {674    var reader = record_mod.PayloadReader.init(payload);675    const id = try reader.hash();676    try reader.finish();677    const pending = if (state.fast_forward) |*value| value else return error.InvalidHistory;678    if (!version.same(pending.id, id) or pending.decision != .pending) {679        return error.InvalidHistory;680    }681    if (decision == .target) try state.setRef(pending.name, pending.target);682    pending.decision = decision;683}684685fn applyFastForwardComplete(state: *RepairState, payload: []const u8) !void {686    var reader = record_mod.PayloadReader.init(payload);687    const id = try reader.hash();688    try reader.finish();689    const pending = state.fast_forward orelse return error.InvalidHistory;690    if (!version.same(pending.id, id) or pending.decision == .pending) {691        return error.InvalidHistory;692    }693    const selected = if (pending.decision == .target) pending.target else pending.expected;694    const index = state.find(pending.name) orelse return error.InvalidHistory;695    if (!version.same(state.entries.items[index].head, selected)) return error.InvalidHistory;696    state.allocator.free(pending.name);697    state.fast_forward = null;698}699700fn hasStage(state: *const RepairState, stage: RepairStage) bool {701    std.debug.assert(state.entries.items.len <= max_entries);702    for (state.entries.items) |entry| if (entry.stage == stage) return true;703    return false;704}705706fn resolveHeads(707    state: *RepairState,708    allocator: Allocator,709    io: std.Io,710    file: std.Io.File,711    start: usize,712    end: usize,713    control: sql.wal.Control,714) !void {715    if (!hasStage(state, .head) or start == end) return;716    var scanner = RecordScanner.init(allocator, io, file, start, end, .commits, control);717    defer scanner.deinit();718    while (try scanner.next()) |record| {719        if (record.kind != .commit) continue;720        const identity = try commitIdentity(allocator, record.payload);721        for (state.entries.items) |*entry| {722            if (entry.stage != .head or !version.same(entry.head, identity.head)) continue;723            entry.root = identity.root;724            entry.stage = .root;725        }726        if (!hasStage(state, .head)) break;727    }728}729730const CommitIdentity = struct {731    head: version.Hash,732    root: version.Hash,733};734735fn commitIdentity(allocator: Allocator, payload: []const u8) !CommitIdentity {736    var reader = record_mod.PayloadReader.init(payload);737    const root = try reader.hash();738    const parent_count = try reader.readU32();739    const parent_bytes = std.math.mul(usize, parent_count, version.hash_bytes) catch740        return error.InvalidHistory;741    if (parent_bytes != reader.remaining()) return error.InvalidHistory;742    const parents = try allocator.alloc(version.Hash, parent_count);743    defer allocator.free(parents);744    for (parents) |*parent| parent.* = try reader.hash();745    try reader.finish();746    return .{ .head = version.Commit.init(root, parents).hash, .root = root };747}748749fn resolveRoots(750    state: *RepairState,751    allocator: Allocator,752    io: std.Io,753    file: std.Io.File,754    start: usize,755    end: usize,756    control: sql.wal.Control,757) !void {758    if (!hasStage(state, .root) or start == end) return;759    var scanner = RecordScanner.init(allocator, io, file, start, end, .roots, control);760    defer scanner.deinit();761    while (try scanner.next()) |record| {762        if (record.kind != .database_root) continue;763        const identity = try databaseRootIdentity(allocator, record.payload);764        for (state.entries.items) |*entry| {765            if (entry.stage != .root or !version.same(entry.root, identity.root)) continue;766            entry.conflicts = identity.conflicts;767            entry.stage = .current;768        }769        if (!hasStage(state, .root)) break;770    }771}772773const DatabaseRootIdentity = struct {774    root: version.Hash,775    conflicts: version.Hash,776};777778fn databaseRootIdentity(allocator: Allocator, payload: []const u8) !DatabaseRootIdentity {779    var reader = record_mod.PayloadReader.init(payload);780    const conflicts = try reader.hash();781    const entry_count = try reader.readU32();782    const fixed_bytes = std.math.mul(usize, entry_count, 4 + version.hash_bytes) catch783        return error.InvalidHistory;784    if (fixed_bytes > reader.remaining()) return error.InvalidHistory;785    const entries = try allocator.alloc(version.RelationEntry, entry_count);786    defer allocator.free(entries);787    for (entries) |*entry| {788        entry.* = .{ .name = try reader.readBytes(), .hash = try reader.hash() };789    }790    try reader.finish();791    std.mem.sort(version.RelationEntry, entries, {}, relationEntryLessThan);792    const root = version.DatabaseRoot.init(entries, .{ .hash = conflicts });793    return .{ .root = root.hash, .conflicts = conflicts };794}795796fn relationEntryLessThan(797    _: void,798    left: version.RelationEntry,799    right: version.RelationEntry,800) bool {801    return simd.order(Bytes, left.name, right.name) == .lt;802}803804const testing_io = std.Options.debug_io;805806test "refs snapshot round-trips entries and covered length" {807    var tmp = std.testing.tmpDir(.{});808    defer tmp.cleanup();809810    const head = @as([version.hash_bytes]u8, @splat(0xaa));811    const root = @as([version.hash_bytes]u8, @splat(0xbb));812    const conflicts = @as([version.hash_bytes]u8, @splat(0xcc));813    const entries = [_]Entry{814        .{ .name = "main", .head = head, .root = root, .conflicts = conflicts },815        .{ .name = "feature/wide", .head = root, .root = head, .conflicts = root },816    };817    try store(std.testing.allocator, testing_io, tmp.dir, "log.history", 12_345, entries[0..]);818819    var loaded = (try load(std.testing.allocator, testing_io, tmp.dir, "log.history")) orelse820        return error.SnapshotMissing;821    defer loaded.deinit();822823    try std.testing.expectEqual(@as(u64, 12_345), loaded.covered_length);824    try std.testing.expectEqual(@as(usize, 2), loaded.entries.len);825    const found = loaded.find("main") orelse return error.RefMissing;826    try std.testing.expectEqualSlices(u8, head[0..], found.head[0..]);827    try std.testing.expectEqualSlices(u8, root[0..], found.root[0..]);828    try std.testing.expectEqualSlices(u8, conflicts[0..], found.conflicts[0..]);829    try std.testing.expect(loaded.find("missing") == null);830}831832test "refs snapshot load rejects a corrupted digest" {833    var tmp = std.testing.tmpDir(.{});834    defer tmp.cleanup();835836    const head = @as([version.hash_bytes]u8, @splat(0x11));837    const entries = [_]Entry{.{838        .name = "main",839        .head = head,840        .root = head,841        .conflicts = head,842    }};843    try store(std.testing.allocator, testing_io, tmp.dir, "log.history", 7, entries[0..]);844845    {846        var file = try tmp.dir.openFile(testing_io, "log.history" ++ suffix, .{ .mode = .read_write });847        defer file.close(testing_io);848        try file.writePositionalAll(testing_io, "?", 9);849    }850851    try std.testing.expect((try load(std.testing.allocator, testing_io, tmp.dir, "log.history")) == null);852}853854test "refs snapshot load returns null when the sidecar is missing" {855    var tmp = std.testing.tmpDir(.{});856    defer tmp.cleanup();857    try std.testing.expect((try load(std.testing.allocator, testing_io, tmp.dir, "absent.history")) == null);858}859860test "refs snapshot rejects an empty branch name" {861    var tmp = std.testing.tmpDir(.{});862    defer tmp.cleanup();863    const bytes = try encode(std.testing.allocator, 3, &.{});864    defer std.testing.allocator.free(bytes);865    var snapshot = (try decode(std.testing.allocator, bytes)) orelse return error.SnapshotMissing;866    defer snapshot.deinit();867    try std.testing.expectEqual(@as(usize, 0), snapshot.entries.len);868    try std.testing.expectEqual(@as(u64, 3), snapshot.covered_length);869}870871test "refs advance resolves stale suffix and pre-boundary targets" {872    var tmp = std.testing.tmpDir(.{});873    defer tmp.cleanup();874875    const history_path = "repair.history";876    const conflicts = version.ConflictRoot.empty();877    const first_root = version.DatabaseRoot.init(&.{}, conflicts);878    const first_commit = version.Commit.init(first_root.hash, &.{});879    const second_root = version.DatabaseRoot.init(&.{.{880        .name = "items",881        .hash = version.emptyHash("repair.items"),882    }}, conflicts);883    var second_parents = [_]version.Hash{first_commit.hash};884    const second_commit = version.Commit.init(second_root.hash, &second_parents);885886    var history = try sql.History.open(std.testing.allocator, tmp.dir, .{887        .path = history_path,888        .recovery = .reject,889    });890    defer history.deinit();891    try history.putDatabaseRoot(first_root);892    try history.putCommit(first_commit);893    try history.putRef(.{ .name = "main", .target = first_commit.hash });894895    var stale = (try loadExisting(896        std.testing.allocator,897        testing_io,898        tmp.dir,899        history_path,900    )).?;901    defer stale.deinit();902903    try history.putDatabaseRoot(second_root);904    try history.putCommit(second_commit);905    try history.putRef(.{ .name = "main", .target = second_commit.hash });906    try history.putRef(.{ .name = "archive", .target = first_commit.hash });907    const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});908    try std.testing.expect(stale.covered_length < history_stat.size);909910    var repaired = try advance(911        std.testing.allocator,912        testing_io,913        tmp.dir,914        history_path,915        &stale,916        history_stat.size,917        .{},918    );919    defer repaired.deinit();920    try std.testing.expectEqual(history_stat.size, repaired.covered_length);921    const main = repaired.find("main") orelse return error.RefMissing;922    try std.testing.expect(version.same(second_commit.hash, main.head));923    try std.testing.expect(version.same(second_root.hash, main.root));924    try std.testing.expect(version.same(second_root.conflicts, main.conflicts));925    const archive = repaired.find("archive") orelse return error.RefMissing;926    try std.testing.expect(version.same(first_commit.hash, archive.head));927    try std.testing.expect(version.same(first_root.hash, archive.root));928    try std.testing.expect(version.same(first_root.conflicts, archive.conflicts));929}930931test "refs advance rejects a ref before its target commit" {932    var tmp = std.testing.tmpDir(.{});933    defer tmp.cleanup();934935    const history_path = "future-ref.history";936    const conflicts = version.ConflictRoot.empty();937    const first_root = version.DatabaseRoot.init(&.{}, conflicts);938    const first_commit = version.Commit.init(first_root.hash, &.{});939    const future_root = version.DatabaseRoot.init(&.{.{940        .name = "future",941        .hash = version.emptyHash("future-ref.relation"),942    }}, conflicts);943    var parents = [_]version.Hash{first_commit.hash};944    const future_commit = version.Commit.init(future_root.hash, &parents);945    var stale: Snapshot = undefined;946    {947        var history = try sql.History.open(std.testing.allocator, tmp.dir, .{948            .path = history_path,949            .recovery = .reject,950        });951        defer history.deinit();952        try history.putDatabaseRoot(first_root);953        try history.putCommit(first_commit);954        try history.putRef(.{ .name = "main", .target = first_commit.hash });955        stale = (try loadExisting(956            std.testing.allocator,957            testing_io,958            tmp.dir,959            history_path,960        )).?;961        try history.putDatabaseRoot(future_root);962        var payload: std.ArrayList(u8) = .empty;963        defer payload.deinit(std.testing.allocator);964        try record_mod.appendHash(std.testing.allocator, &payload, future_commit.hash);965        try record_mod.appendBytes(std.testing.allocator, &payload, "main");966        try history.appendRecord(.ref, payload.items);967        try history.putCommit(future_commit);968    }969    defer stale.deinit();970    const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});971    try std.testing.expectError(972        error.InvalidHistory,973        advance(974            std.testing.allocator,975            testing_io,976            tmp.dir,977            history_path,978            &stale,979            history_stat.size,980            .{},981        ),982    );983    try std.testing.expectError(984        error.TruncatedHistory,985        sql.History.open(std.testing.allocator, tmp.dir, .{986            .path = history_path,987            .recovery = .reject,988        }),989    );990}991992test "refs advance rejects a hash-valid malformed non-ref suffix" {993    var tmp = std.testing.tmpDir(.{});994    defer tmp.cleanup();995996    const history_path = "malformed-conflict.history";997    const conflicts = version.ConflictRoot.empty();998    const root = version.DatabaseRoot.init(&.{}, conflicts);999    const commit = version.Commit.init(root.hash, &.{});1000    var stale: Snapshot = undefined;1001    {1002        var history = try sql.History.open(std.testing.allocator, tmp.dir, .{1003            .path = history_path,1004            .recovery = .reject,1005        });1006        defer history.deinit();1007        try history.putDatabaseRoot(root);1008        try history.putCommit(commit);1009        try history.putRef(.{ .name = "main", .target = commit.hash });1010        stale = (try loadExisting(1011            std.testing.allocator,1012            testing_io,1013            tmp.dir,1014            history_path,1015        )).?;1016        try history.appendRecord(.conflict, &.{});1017    }1018    defer stale.deinit();1019    const refs_before = try tmp.dir.readFileAlloc(1020        testing_io,1021        "malformed-conflict.history.refs",1022        std.testing.allocator,1023        .unlimited,1024    );1025    defer std.testing.allocator.free(refs_before);1026    const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});1027    try std.testing.expectError(1028        error.InvalidHistory,1029        advance(1030            std.testing.allocator,1031            testing_io,1032            tmp.dir,1033            history_path,1034            &stale,1035            history_stat.size,1036            .{},1037        ),1038    );1039    const refs_after = try tmp.dir.readFileAlloc(1040        testing_io,1041        "malformed-conflict.history.refs",1042        std.testing.allocator,1043        .unlimited,1044    );1045    defer std.testing.allocator.free(refs_after);1046    try std.testing.expectEqualSlices(u8, refs_before, refs_after);1047    try std.testing.expectError(1048        error.TruncatedHistory,1049        sql.History.open(std.testing.allocator, tmp.dir, .{1050            .path = history_path,1051            .recovery = .reject,1052        }),1053    );1054}10551056test "refs advance streams opaque records above the metadata bound" {1057    var tmp = std.testing.tmpDir(.{});1058    defer tmp.cleanup();10591060    const history_path = "opaque.history";1061    const payload = try std.testing.allocator.alloc(u8, repair_metadata_record_bytes_max);1062    defer std.testing.allocator.free(payload);1063    @memset(payload, 0);1064    const kind = record_mod.RecordKind.row_chunk;1065    const expected = record_mod.recordHash(@backingInt(kind), payload);1066    var header: [record_mod.record_header_size]u8 = @splat(0);1067    std.mem.writeInt(u32, header[0..4], record_mod.magic, .big);1068    std.mem.writeInt(u32, header[4..8], record_mod.format_version, .big);1069    std.mem.writeInt(u32, header[8..12], @backingInt(kind), .big);1070    std.mem.writeInt(u32, header[12..16], @intCast(payload.len), .big);1071    @memcpy(header[16..][0..version.hash_bytes], &expected);1072    var file = try tmp.dir.createFile(testing_io, history_path, .{ .read = true });1073    defer file.close(testing_io);1074    try file.writePositionalAll(testing_io, &header, 0);1075    try file.writePositionalAll(testing_io, payload, header.len);1076    const history_length = header.len + payload.len;10771078    var empty = Snapshot{1079        .allocator = std.testing.allocator,1080        .covered_length = 0,1081        .entries = try std.testing.allocator.alloc(Entry, 0),1082        .names = try std.testing.allocator.alloc(u8, 0),1083    };1084    defer empty.deinit();1085    var repaired = try advance(1086        std.testing.allocator,1087        testing_io,1088        tmp.dir,1089        history_path,1090        &empty,1091        history_length,1092        .{},1093    );1094    defer repaired.deinit();1095    try std.testing.expectEqual(@as(usize, 0), repaired.entries.len);1096    try std.testing.expectEqual(@as(u64, history_length), repaired.covered_length);1097}10981099test "refs advance rejects metadata above the repair bound" {1100    var tmp = std.testing.tmpDir(.{});1101    defer tmp.cleanup();11021103    const history_path = "oversized.history";1104    var header: [record_mod.record_header_size]u8 = @splat(0);1105    std.mem.writeInt(u32, header[0..4], record_mod.magic, .big);1106    std.mem.writeInt(u32, header[4..8], record_mod.format_version, .big);1107    std.mem.writeInt(u32, header[8..12], @backingInt(record_mod.RecordKind.commit), .big);1108    const oversized_payload_bytes = repair_metadata_record_bytes_max + 1;1109    std.mem.writeInt(u32, header[12..16], oversized_payload_bytes, .big);1110    var file = try tmp.dir.createFile(testing_io, history_path, .{ .read = true });1111    defer file.close(testing_io);1112    try file.writePositionalAll(testing_io, &header, 0);1113    const history_length = record_mod.record_header_size + oversized_payload_bytes;1114    try file.setLength(testing_io, history_length);11151116    var empty = Snapshot{1117        .allocator = std.testing.allocator,1118        .covered_length = 0,1119        .entries = try std.testing.allocator.alloc(Entry, 0),1120        .names = try std.testing.allocator.alloc(u8, 0),1121    };1122    defer empty.deinit();1123    try std.testing.expectError(1124        error.StreamTooLong,1125        advance(1126            std.testing.allocator,1127            testing_io,1128            tmp.dir,1129            history_path,1130            &empty,1131            history_length,1132            .{},1133        ),1134    );1135}11361137test "refs snapshot rejects the retired format version" {1138    const head = @as([version.hash_bytes]u8, @splat(0x33));1139    const entries = [_]Entry{.{1140        .name = "main",1141        .head = head,1142        .root = head,1143        .conflicts = head,1144    }};1145    const bytes = try encode(std.testing.allocator, 9, &entries);1146    defer std.testing.allocator.free(bytes);1147    std.mem.writeInt(u32, bytes[4..8], 1, .big);1148    std.crypto.hash.sha2.Sha256.hash(1149        bytes[0 .. bytes.len - digest_size],1150        bytes[bytes.len - digest_size ..][0..digest_size],1151        .{},1152    );1153    try std.testing.expect((try decode(std.testing.allocator, bytes)) == null);1154}

1pub const refs = refs_mod;