tiny.sql.history.verification

1pub const Inspection = struct {2    last_valid_boundary: usize,3    bad: ?BadRecord = null,4};

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

1const std = @import("std");2const sys = @import("sys");3const sql = @import("../root.zig");4const record_mod = @import("record.zig");5const batch_mod = @import("batch.zig");6const validate_mod = @import("validate.zig");7const version = sql.version;89const max_lanes: usize = 4;1011/// A complete scan uses caller storage and never opens the history for write.12pub const Limits = struct {13    metadata_bytes_max: usize = 4 * 1024 * 1024,14    dependencies_max: usize = 2 * 1024 * 1024,15    conflict_bytes_max: usize = 4 * 1024 * 1024,16    working_bytes_max: usize,17};1819pub const Capacity = struct {20    bytes: usize,2122    pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {23        if (limits.metadata_bytes_max == 0 or limits.dependencies_max == 0 or24            limits.conflict_bytes_max == 0 or limits.working_bytes_max == 0)25        {26            return error.CapacityOverflow;27        }28        if (limits.conflict_bytes_max > std.math.maxInt(u32)) {29            return error.CapacityOverflow;30        }31        _ = std.math.add(32            usize,33            limits.metadata_bytes_max,34            limits.conflict_bytes_max,35        ) catch return error.CapacityOverflow;36        return .{ .bytes = limits.working_bytes_max };37    }38};3940pub const Storage = struct {41    bytes: []u8,42};4344pub const Exhaustion = struct {45    name: []const u8,46    bytes_in_use: usize,47    requested: usize,48    available: usize,49};5051pub const Workspace = struct {52    storage: []u8,53    fixed: std.heap.FixedBufferAllocator,54    limits: Limits,55    exhaustion: ?Exhaustion = null,5657    fn allocator(self: *Workspace) std.mem.Allocator {58        return .{59            .ptr = self,60            .vtable = &.{61                .alloc = alloc,62                .resize = resize,63                .remap = remap,64                .free = free,65            },66        };67    }6869    fn alloc(ctx: *anyopaque, len: usize, alignment: std.mem.Alignment, ra: usize) ?[*]u8 {70        const self: *Workspace = @ptrCast(@alignCast(ctx));71        const underlying = self.fixed.allocator();72        const result = underlying.vtable.alloc(underlying.ptr, len, alignment, ra);73        if (result == null) self.noteExhaustion(len);74        return result;75    }7677    fn resize(78        ctx: *anyopaque,79        memory: []u8,80        alignment: std.mem.Alignment,81        len: usize,82        ra: usize,83    ) bool {84        const self: *Workspace = @ptrCast(@alignCast(ctx));85        const underlying = self.fixed.allocator();86        return underlying.vtable.resize(underlying.ptr, memory, alignment, len, ra);87    }8889    fn remap(90        ctx: *anyopaque,91        memory: []u8,92        alignment: std.mem.Alignment,93        len: usize,94        ra: usize,95    ) ?[*]u8 {96        const self: *Workspace = @ptrCast(@alignCast(ctx));97        const underlying = self.fixed.allocator();98        return underlying.vtable.remap(underlying.ptr, memory, alignment, len, ra);99    }100101    fn free(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, ra: usize) void {102        const self: *Workspace = @ptrCast(@alignCast(ctx));103        const underlying = self.fixed.allocator();104        underlying.vtable.free(underlying.ptr, memory, alignment, ra);105    }106107    fn noteExhaustion(self: *Workspace, requested: usize) void {108        const used = self.fixed.end_index;109        self.exhaustion = .{110            .name = "working_bytes_max",111            .bytes_in_use = used,112            .requested = requested,113            .available = self.storage.len - used,114        };115    }116117    pub fn activate(limits: Limits, storage: Storage) error{118        CapacityOverflow,119        CapacityExceeded,120    }!Workspace {121        const capacity = try Capacity.derive(limits);122        if (storage.bytes.len < capacity.bytes) return error.CapacityExceeded;123        const bytes = storage.bytes[0..capacity.bytes];124        return .{125            .storage = bytes,126            .fixed = std.heap.FixedBufferAllocator.init(bytes),127            .limits = limits,128        };129    }130131    pub fn reset(self: *Workspace) void {132        self.fixed.reset();133        self.exhaustion = null;134    }135};136137pub const Report = validate_mod.Inspection;138139/// Reports the earliest bad record while leaving the file untouched.140pub fn verifyFile(141    workspace: *Workspace,142    io: std.Io,143    file: std.Io.File,144    end: usize,145    control: sql.wal.Control,146) !Report {147    workspace.reset();148    return validate_mod.inspectWhole(149        workspace.allocator(),150        io,151        file,152        end,153        .{154            .metadata_bytes_max = workspace.limits.metadata_bytes_max,155            .dependencies_max = workspace.limits.dependencies_max,156            .conflict_bytes_max = workspace.limits.conflict_bytes_max,157        },158        control,159    ) catch |err| switch (err) {160        error.OutOfMemory => {161            std.debug.assert(workspace.exhaustion != null);162            return error.CapacityExceeded;163        },164        else => return err,165    };166}167168const Work = struct {169    bytes: []const u8 = &.{},170    records: []batch_mod.Record = &.{},171};172173pub const Workers = struct {174    handles: [max_lanes - 1]sys.thread.JoinHandle = undefined,175    started: usize = 0,176    mutex: sys.thread.Mutex = .{},177    ready: sys.thread.Condition = .{},178    done: sys.thread.Condition = .{},179    generation: usize = 0,180    completed: usize = 0,181    shutdown: bool = false,182    work: Work = .{},183    next_index: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),184185    pub fn start(self: *Workers, requested_lanes: usize) void {186        if (comptime !sys.thread.threadsSupported()) return;187        const background = @min(requested_lanes -| 1, self.handles.len);188        while (self.started < background) {189            self.handles[self.started] = sys.thread.spawn(workerLoop, .{self}) catch return;190            self.started += 1;191        }192    }193194    pub fn deinit(self: *Workers) void {195        self.mutex.lock();196        self.shutdown = true;197        self.ready.broadcast();198        self.mutex.unlock();199        for (self.handles[0..self.started]) |handle| handle.join();200        self.* = undefined;201    }202203    pub fn verify(self: *Workers, bytes: []const u8, records: []batch_mod.Record) void {204        if (records.len == 0) return;205        if (self.started == 0 or records.len == 1) {206            verifySerial(bytes, records);207            return;208        }209210        self.mutex.lock();211        self.work = .{ .bytes = bytes, .records = records };212        self.next_index.store(0, .monotonic);213        self.completed = 0;214        self.generation +%= 1;215        self.ready.broadcast();216        self.mutex.unlock();217218        runWork(self, self.work);219220        self.mutex.lock();221        while (self.completed != self.started) self.done.wait(&self.mutex);222        self.mutex.unlock();223    }224225    fn workerLoop(self: *Workers) void {226        var observed_generation: usize = 0;227        self.mutex.lock();228        while (true) {229            while (!self.shutdown and self.generation == observed_generation) self.ready.wait(&self.mutex);230            if (self.shutdown) {231                self.mutex.unlock();232                return;233            }234            observed_generation = self.generation;235            const work = self.work;236            self.mutex.unlock();237            runWork(self, work);238            self.mutex.lock();239            self.completed += 1;240            if (self.completed == self.started) self.done.signal();241        }242    }243};244245fn runWork(workers: *Workers, work: Work) void {246    while (true) {247        const index = workers.next_index.fetchAdd(1, .monotonic);248        if (index >= work.records.len) return;249        verifyRecord(work.bytes, &work.records[index]);250    }251}252253fn verifySerial(bytes: []const u8, records: []batch_mod.Record) void {254    for (records) |*record| verifyRecord(bytes, record);255}256257fn verifyRecord(bytes: []const u8, record: *batch_mod.Record) void {258    const payload = bytes[record.payload_start..][0..record.payload_len];259    record.valid_hash = version.same(record.expected, record_mod.recordHash(record.kind_value, payload));260}261262test "history verification reports a missing dependency without writing" {263    const store_mod = @import("store.zig");264    const allocator = std.testing.allocator;265    const io = std.Options.debug_io;266    var tmp = std.testing.tmpDir(.{});267    defer tmp.cleanup();268    var file = try tmp.dir.createFile(io, "verify.history", .{ .read = true });269    defer file.close(io);270    var payload: std.ArrayList(u8) = .empty;271    defer payload.deinit(allocator);272    const root = version.emptyHash("verify-root");273    try record_mod.appendHash(allocator, &payload, root);274    try record_mod.appendU32(allocator, &payload, 0);275    const boundary = try store_mod.writeTestingRecord(file, 0, .commit, payload.items);276    payload.clearRetainingCapacity();277    const missing = version.emptyHash("verify-missing");278    try record_mod.appendHash(allocator, &payload, missing);279    try record_mod.appendBytes(allocator, &payload, "main");280    const end = try store_mod.writeTestingRecord(file, boundary, .ref, payload.items);281    var before: [256]u8 = undefined;282    try std.testing.expect(end <= before.len);283    try std.testing.expectEqual(end, try file.readPositionalAll(io, before[0..end], 0));284    const limits = Limits{285        .metadata_bytes_max = 128,286        .dependencies_max = 16,287        .conflict_bytes_max = 128,288        .working_bytes_max = 1024 * 1024,289    };290    const capacity = try Capacity.derive(limits);291    const bytes = try allocator.alloc(u8, capacity.bytes);292    defer allocator.free(bytes);293    var workspace = try Workspace.activate(limits, .{ .bytes = bytes });294    const report = try verifyFile(&workspace, io, file, end, .{});295    try std.testing.expectEqual(boundary, report.last_valid_boundary);296    try std.testing.expectEqual(boundary, report.bad.?.offset);297    try std.testing.expectEqual(record_mod.RecordKind.ref, report.bad.?.kind.?);298    try std.testing.expect(version.same(missing, report.bad.?.dependency.?.key));299    var after: [256]u8 = undefined;300    try std.testing.expectEqual(end, try file.readPositionalAll(io, after[0..end], 0));301    try std.testing.expectEqualSlices(u8, before[0..end], after[0..end]);302    after[boundary] ^= 0xff;303    try file.writePositionalAll(io, after[boundary..][0..1], boundary);304    const malformed = try verifyFile(&workspace, io, file, end, .{});305    try std.testing.expectEqual(boundary, malformed.last_valid_boundary);306    try std.testing.expectEqual(boundary, malformed.bad.?.offset);307    try std.testing.expect(malformed.bad.?.kind == null);308    var unchanged: [256]u8 = undefined;309    try std.testing.expectEqual(end, try file.readPositionalAll(io, unchanged[0..end], 0));310    try std.testing.expectEqualSlices(u8, after[0..end], unchanged[0..end]);311312    const small_limits = Limits{313        .metadata_bytes_max = 128,314        .dependencies_max = 16,315        .conflict_bytes_max = 128,316        .working_bytes_max = 64,317    };318    var small_storage: [64]u8 = undefined;319    var small = try Workspace.activate(small_limits, .{ .bytes = &small_storage });320    try std.testing.expectError(error.CapacityExceeded, verifyFile(&small, io, file, end, .{}));321    const exhausted = small.exhaustion.?;322    try std.testing.expectEqualStrings("working_bytes_max", exhausted.name);323    try std.testing.expectEqual(small.fixed.end_index, exhausted.bytes_in_use);324    try std.testing.expectEqual(small_storage.len - exhausted.bytes_in_use, exhausted.available);325    try std.testing.expect(exhausted.requested > 0);326}