tiny.sql.history.resolver

1const std = @import("std");2const sql = @import("../root.zig");3const identity = @import("identity.zig");4const record_mod = @import("record.zig");56const version = sql.version;78pub const Error = std.Io.File.ReadPositionalError || error{9    CapacityExceeded,10    CapacityOverflow,11    InvalidHistory,12    WrongObject,13};1415pub const Limits = struct {16    metadata_bytes_max: usize = 4 * 1024 * 1024,17    stream_bytes: usize = 64 * 1024,18    relation_entries_max: usize = 1024,19    conflict_entries_max: usize = 1024,20};2122pub const Capacity = struct {23    bytes: usize,24    relation_entries: usize,25    conflict_entries: usize,2627    pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {28        if (limits.stream_bytes < version.hash_bytes) return error.CapacityOverflow;29        if (limits.metadata_bytes_max < version.hash_bytes) return error.CapacityOverflow;30        if (limits.metadata_bytes_max > std.math.maxInt(u32)) {31            return error.CapacityOverflow;32        }33        const bytes = std.math.add(34            usize,35            limits.metadata_bytes_max,36            limits.stream_bytes,37        ) catch return error.CapacityOverflow;38        _ = std.math.mul(39            usize,40            limits.relation_entries_max,41            @sizeOf(version.RelationEntry),42        ) catch return error.CapacityOverflow;43        _ = std.math.mul(44            usize,45            limits.conflict_entries_max,46            @sizeOf(version.ConflictEntry),47        ) catch return error.CapacityOverflow;48        return .{49            .bytes = bytes,50            .relation_entries = limits.relation_entries_max,51            .conflict_entries = limits.conflict_entries_max,52        };53    }54};5556pub const Storage = struct {57    bytes: []u8,58    relations: []version.RelationEntry,59    conflicts: []version.ConflictEntry,60};6162pub const Exhaustion = struct {63    name: []const u8,64    requested: usize,65    available: usize,66};6768pub const Workspace = struct {69    metadata: []u8,70    stream: []u8,71    relations: []version.RelationEntry,72    conflicts: []version.ConflictEntry,73    exhaustion: ?Exhaustion = null,7475    /// Activation borrows all storage before any record is read.76    pub fn activate(limits: Limits, storage: Storage) Error!Workspace {77        const capacity = try Capacity.derive(limits);78        if (storage.bytes.len < capacity.bytes) return error.CapacityExceeded;79        if (storage.relations.len < capacity.relation_entries) return error.CapacityExceeded;80        if (storage.conflicts.len < capacity.conflict_entries) return error.CapacityExceeded;81        return .{82            .metadata = storage.bytes[0..limits.metadata_bytes_max],83            .stream = storage.bytes[limits.metadata_bytes_max..capacity.bytes],84            .relations = storage.relations[0..limits.relation_entries_max],85            .conflicts = storage.conflicts[0..limits.conflict_entries_max],86        };87    }8889    pub fn scratch(self: *Workspace) identity.Scratch {90        return .{ .relations = self.relations, .conflicts = self.conflicts };91    }9293    pub fn each(94        self: *Workspace,95        kind: record_mod.RecordKind,96        payload: []const u8,97        location: identity.Location,98        context: anytype,99        comptime accept: fn (@TypeOf(context), identity.Entry) anyerror!void,100    ) anyerror!void {101        identity.each(kind, payload, location, self.scratch(), context, accept) catch |err| {102            if (err != error.KeyScratchExhausted) return err;103            const count_offset: usize = switch (kind) {104                .database_root => version.hash_bytes,105                .conflict_root => 0,106                else => unreachable,107            };108            if (payload.len -| count_offset < 4) return error.InvalidHistory;109            const requested = record_mod.readIntU32(payload[count_offset..][0..4]);110            const available = switch (kind) {111                .database_root => self.relations.len,112                .conflict_root => self.conflicts.len,113                else => unreachable,114            };115            try self.require("key_entries_max", requested, available);116            unreachable;117        };118    }119120    fn require(self: *Workspace, name: []const u8, requested: usize, available: usize) Error!void {121        if (requested <= available) return;122        self.exhaustion = .{123            .name = name,124            .requested = requested,125            .available = available,126        };127        return error.CapacityExceeded;128    }129};130131pub const Scanned = struct {132    kind: record_mod.RecordKind,133    location: identity.Location,134    payload: []const u8,135    end: u64,136};137138/// A file-order scanner retains only one metadata record or stream chunk.139pub const Scanner = struct {140    io: std.Io,141    file: std.Io.File,142    workspace: *Workspace,143    offset: u64,144    end: u64,145    chunk_key: version.Hash = undefined,146147    pub fn init(148        io: std.Io,149        file: std.Io.File,150        workspace: *Workspace,151        start: u64,152        end: u64,153    ) Error!Scanner {154        if (start > end) return error.InvalidHistory;155        return .{ .io = io, .file = file, .workspace = workspace, .offset = start, .end = end };156    }157158    pub fn next(self: *Scanner) Error!?Scanned {159        if (self.offset == self.end) return null;160        if (self.end - self.offset < record_mod.record_header_size) return error.InvalidHistory;161        var header_bytes: [record_mod.record_header_size]u8 = undefined;162        try readExact(self.io, self.file, &header_bytes, self.offset);163        const header = try record_mod.Header.decode(&header_bytes);164        const payload_start = std.math.add(165            u64,166            self.offset,167            record_mod.record_header_size,168        ) catch return error.InvalidHistory;169        const record_end = std.math.add(170            u64,171            payload_start,172            header.payload_len,173        ) catch return error.InvalidHistory;174        if (record_end > self.end) return error.InvalidHistory;175        const location = identity.Location{176            .record_offset = self.offset,177            .payload_len = header.payload_len,178            .envelope_hash = header.envelope_hash,179        };180        var hasher = record_mod.envelopeHasher(181            @backingInt(header.kind),182            header.payload_len,183        );184        const payload: []const u8 = if (header.kind == .row_chunk) chunk: {185            var cursor = payload_start;186            var copied: usize = 0;187            while (cursor < record_end) {188                const count: usize = @intCast(@min(self.workspace.stream.len, record_end - cursor));189                try readExact(self.io, self.file, self.workspace.stream[0..count], cursor);190                if (copied < version.hash_bytes) {191                    const take = @min(version.hash_bytes - copied, count);192                    @memcpy(self.chunk_key[copied..][0..take], self.workspace.stream[0..take]);193                    copied += take;194                }195                hasher.update(self.workspace.stream[0..count]);196                cursor += count;197            }198            break :chunk &self.chunk_key;199        } else metadata: {200            try self.workspace.require(201                "metadata_bytes_max",202                header.payload_len,203                self.workspace.metadata.len,204            );205            const bytes = self.workspace.metadata[0..header.payload_len];206            try readExact(self.io, self.file, bytes, payload_start);207            hasher.update(bytes);208            break :metadata bytes;209        };210        var actual: version.Hash = undefined;211        hasher.final(&actual);212        if (!version.same(actual, header.envelope_hash)) return error.InvalidHistory;213        const result = Scanned{214            .kind = header.kind,215            .location = location,216            .payload = payload,217            .end = record_end,218        };219        self.offset = record_end;220        return result;221    }222};223224pub const Resolver = struct {225    io: std.Io,226    file: std.Io.File,227    indexed_eof: u64,228    workspace: *Workspace,229230    /// The caller owns payload storage and keeps the returned slice alive.231    pub fn read(232        self: *Resolver,233        key: identity.Key,234        location: identity.Location,235        output: []u8,236    ) Error![]const u8 {237        const end = std.math.add(238            u64,239            location.record_offset,240            location.encodedLen(),241        ) catch return error.InvalidHistory;242        if (end > self.indexed_eof) return error.InvalidHistory;243        try self.workspace.require("record_payload", location.payload_len, output.len);244        var header_bytes: [record_mod.record_header_size]u8 = undefined;245        try readExact(self.io, self.file, &header_bytes, location.record_offset);246        const header = try record_mod.Header.decode(&header_bytes);247        if (header.payload_len != location.payload_len) return error.InvalidHistory;248        if (!version.same(header.envelope_hash, location.envelope_hash)) {249            return error.InvalidHistory;250        }251        try readExact(252            self.io,253            self.file,254            output[0..location.payload_len],255            location.record_offset + record_mod.record_header_size,256        );257        const payload = output[0..location.payload_len];258        if (!version.same(259            record_mod.recordHash(@backingInt(header.kind), payload),260            header.envelope_hash,261        )) return error.InvalidHistory;262        var found = false;263        const Context = struct {264            wanted: identity.Key,265            position: u32,266            found: *bool,267268            fn accept(context: *@This(), entry: identity.Entry) Error!void {269                if (entry.key.kind != context.wanted.kind) return;270                if (entry.location.node_position != context.position) return;271                context.found.* = version.same(entry.key.bytes, context.wanted.bytes);272            }273        };274        var context = Context{275            .wanted = key,276            .position = location.node_position,277            .found = &found,278        };279        self.workspace.each(280            header.kind,281            payload,282            location,283            &context,284            Context.accept,285        ) catch |err| switch (err) {286            error.CapacityExceeded => return error.CapacityExceeded,287            else => return error.InvalidHistory,288        };289        if (!found) return error.WrongObject;290        return payload;291    }292};293294fn readExact(io: std.Io, file: std.Io.File, output: []u8, offset: u64) Error!void {295    var filled: usize = 0;296    while (filled < output.len) {297        const position = std.math.add(u64, offset, filled) catch return error.InvalidHistory;298        const count = try file.readPositionalAll(io, output[filled..], position);299        if (count == 0) return error.InvalidHistory;300        filled += count;301    }302}303304test "history resolver keeps the first typed commit location" {305    const store_mod = @import("store.zig");306    const allocator = std.testing.allocator;307    const io = std.Options.debug_io;308    var tmp = std.testing.tmpDir(.{});309    defer tmp.cleanup();310    var file = try tmp.dir.createFile(io, "objects.history", .{ .read = true });311    defer file.close(io);312    var payload: std.ArrayList(u8) = .empty;313    defer payload.deinit(allocator);314    const root = version.emptyHash("resolver-commit");315    try record_mod.appendHash(allocator, &payload, root);316    try record_mod.appendU32(allocator, &payload, 0);317    const first_end = try store_mod.writeTestingRecord(file, 0, .commit, payload.items);318    const end = try store_mod.writeTestingRecord(file, first_end, .commit, payload.items);319    const limits = Limits{320        .metadata_bytes_max = 128,321        .stream_bytes = 64,322        .relation_entries_max = 2,323        .conflict_entries_max = 2,324    };325    const capacity = try Capacity.derive(limits);326    const bytes = try allocator.alloc(u8, capacity.bytes);327    defer allocator.free(bytes);328    const relations = try allocator.alloc(version.RelationEntry, capacity.relation_entries);329    defer allocator.free(relations);330    const conflicts = try allocator.alloc(version.ConflictEntry, capacity.conflict_entries);331    defer allocator.free(conflicts);332    var workspace = try Workspace.activate(limits, .{333        .bytes = bytes,334        .relations = relations,335        .conflicts = conflicts,336    });337    var scanner = try Scanner.init(io, file, &workspace, 0, end);338    const first = (try scanner.next()).?;339    try std.testing.expectEqual(@as(u64, 0), first.location.record_offset);340    const expected = version.Commit.init(root, &.{}).hash;341    var captured: ?identity.Entry = null;342    const Capture = struct {343        fn accept(target: *?identity.Entry, entry: identity.Entry) !void {344            target.* = entry;345        }346    };347    try workspace.each(348        first.kind,349        first.payload,350        first.location,351        &captured,352        Capture.accept,353    );354    try std.testing.expect(version.same(expected, captured.?.key.bytes));355    const second = (try scanner.next()).?;356    try std.testing.expectEqual(@as(u64, first_end), second.location.record_offset);357    try std.testing.expect((try scanner.next()) == null);358    var resolver = Resolver{ .io = io, .file = file, .indexed_eof = end, .workspace = &workspace };359    var output: [64]u8 = undefined;360    const read = try resolver.read(captured.?.key, captured.?.location, &output);361    try std.testing.expectEqualSlices(u8, payload.items, read);362}363364test "history resolver typed keys agree with replay across object kinds" {365    const store_mod = @import("store.zig");366    const allocator = std.testing.allocator;367    const io = std.Options.debug_io;368    var tmp = std.testing.tmpDir(.{});369    defer tmp.cleanup();370    const path = "typed.history";371    var relation = try store_mod.testingMerkleRelationRoot(372        allocator,373        .{ .branches = 1, .leaves = 2 },374        null,375        "typed",376    );377    defer relation.deinit();378    const rows = try store_mod.testingChunkRows(allocator, 300, 42);379    defer version.freeRelationRows(allocator, rows);380    const artifact = version.ConflictArtifact.init("metrics", 7, "old", "ours", "theirs");381    var database_root: version.DatabaseRoot = undefined;382    var commit: version.Commit = undefined;383    {384        var history = try store_mod.History.open(allocator, tmp.dir, .{385            .path = path,386            .recovery = .reject,387        });388        defer history.deinit();389        try history.putRelationRoot(relation);390        try history.putRelationRows(relation.hash, rows);391        try history.putConflict(artifact);392        const conflicts = try history.putConflictRoot(&.{artifact.entry()});393        database_root = try version.DatabaseRoot.initSorted(allocator, &.{394            .{ .name = "metrics", .hash = relation.hash },395        }, conflicts);396        try history.putDatabaseRoot(database_root);397        commit = version.Commit.init(database_root.hash, &.{});398        try history.putCommit(commit);399        try history.putRef(.{ .name = "main", .target = commit.hash });400        try history.putRef(.{ .name = "temporary", .target = commit.hash });401        try history.deleteRef("temporary");402        const chunk = history.row_chunks.items[0].payload;403        const duplicate = try allocator.alloc(u8, chunk.len);404        defer allocator.free(duplicate);405        try std.testing.expectEqual(406            chunk.len,407            try history.file.?.readPositionalAll(io, duplicate, chunk.offset),408        );409        try history.appendRecord(.row_chunk, duplicate);410        try history.flushSync();411    }412    defer database_root.deinit();413    var replay = try store_mod.History.open(allocator, tmp.dir, .{414        .path = path,415        .recovery = .reject,416        .read_only = true,417    });418    defer replay.deinit();419    const verify_mod = @import("verify.zig");420    const verify_limits = verify_mod.Limits{421        .metadata_bytes_max = 128 * 1024,422        .dependencies_max = 4096,423        .conflict_bytes_max = 4096,424        .working_bytes_max = 8 * 1024 * 1024,425    };426    const verify_capacity = try verify_mod.Capacity.derive(verify_limits);427    const verify_bytes = try allocator.alloc(u8, verify_capacity.bytes);428    defer allocator.free(verify_bytes);429    var verify_workspace = try verify_mod.Workspace.activate(430        verify_limits,431        .{ .bytes = verify_bytes },432    );433    const report = try verify_mod.verifyFile(434        &verify_workspace,435        io,436        replay.file.?,437        replay.len(),438        .{},439    );440    try std.testing.expect(report.bad == null);441    try std.testing.expectEqual(replay.len(), report.last_valid_boundary);442    const limits = Limits{443        .metadata_bytes_max = 128 * 1024,444        .stream_bytes = 4096,445        .relation_entries_max = 16,446        .conflict_entries_max = 16,447    };448    const capacity = try Capacity.derive(limits);449    const bytes = try allocator.alloc(u8, capacity.bytes);450    defer allocator.free(bytes);451    const relations = try allocator.alloc(version.RelationEntry, capacity.relation_entries);452    defer allocator.free(relations);453    const conflicts = try allocator.alloc(version.ConflictEntry, capacity.conflict_entries);454    defer allocator.free(conflicts);455    var workspace = try Workspace.activate(limits, .{456        .bytes = bytes,457        .relations = relations,458        .conflicts = conflicts,459    });460    var entries: std.ArrayList(identity.Entry) = .empty;461    defer entries.deinit(allocator);462    var duplicates: usize = 0;463    const Collect = struct {464        allocator: std.mem.Allocator,465        entries: *std.ArrayList(identity.Entry),466        duplicates: *usize,467468        fn accept(context: *@This(), entry: identity.Entry) !void {469            for (context.entries.items) |existing| {470                if (existing.key.kind != entry.key.kind) continue;471                if (!version.same(existing.key.bytes, entry.key.bytes)) continue;472                context.duplicates.* += 1;473                return;474            }475            try context.entries.append(context.allocator, entry);476        }477    };478    var collect = Collect{479        .allocator = allocator,480        .entries = &entries,481        .duplicates = &duplicates,482    };483    var scanner = try Scanner.init(io, replay.file.?, &workspace, 0, replay.len());484    while (try scanner.next()) |record| {485        try workspace.each(486            record.kind,487            record.payload,488            record.location,489            &collect,490            Collect.accept,491        );492    }493    try std.testing.expectEqual(@as(usize, 1), duplicates);494    const replay_count = replay.commits.items.len + replay.conflicts.items.len +495        replay.database_roots.items.len + replay.relation_roots.items.len +496        replay.relation_rows.items.len + replay.conflict_roots.items.len +497        replay.row_chunks.items.len + replay.index_pages.items.len +498        replay.tree_nodes.items.len + replay.relation_spans.items.len;499    try std.testing.expectEqual(replay_count, entries.items.len);500    var resolver = Resolver{501        .io = io,502        .file = replay.file.?,503        .indexed_eof = replay.len(),504        .workspace = &workspace,505    };506    const output = try allocator.alloc(u8, 128 * 1024);507    defer allocator.free(output);508    for (entries.items) |entry| {509        const key = entry.key.bytes;510        const present = switch (entry.key.kind) {511            .commit => replay.findCommit(key) != null,512            .conflict => replay.findConflict(key) != null,513            .database_root => replay.findDatabaseRoot(key) != null,514            .relation_root => replay.findRelationRoot(key) != null,515            .relation_rows => replay.hasRelationRows(key),516            .conflict_root => replay.findConflictRoot(key) != null,517            .row_chunk => replay.findRowChunk(key) != null,518            .chunk_index_page => replay.hasIndexPage(key),519            .tree_node => replay.findTreeNode(key) != null,520            .relation_spans => replay.lookup.relation_spans.get(key) != null,521        };522        try std.testing.expect(present);523        _ = try resolver.read(entry.key, entry.location, output);524    }525}

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