tiny.smg.storage.store

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

1const std = @import("std");2const sql = @import("sql");3const sys = @import("sys");4const smg = @import("../root.zig");5const storage = smg.storage;6const database = storage.database;7const paths = storage.paths;8const sync = storage.sync;9const text = smg.text;1011const fs_io = std.Options.debug_io;12const gitignore_content = paths.database_file_name ++ "\n" ++ paths.wal_file_name ++13    "\nnames.bin\nnames.bin.tmp\nsearch.sql\nsearch.sql.wal\nsearch.sql.tmp\nsearch.sql.tmp.wal\nsummary.json\nsummary.json.tmp\n" ++14    paths.lock_file_name ++ "\n" ++ paths.refresh_file_name ++ "\n" ++15    paths.source_file_name ++ "\n" ++ paths.source_temp_file_name ++ ".*\n";1617pub const Store = struct {18    allocator: std.mem.Allocator,19    dir_path: []const u8,20    lock: std.Io.File,21    database: database.Database,2223    pub fn close(self: *Store) void {24        self.database.deinit();25        releaseLock(&self.lock);26        self.allocator.free(self.dir_path);27        self.* = undefined;28    }29};3031pub const ReadStore = struct {32    allocator: std.mem.Allocator,33    dir_path: []const u8,34    lock: std.Io.File,35    reader: database.Reader,3637    pub fn close(self: *ReadStore) void {38        self.reader.deinit();39        releaseLock(&self.lock);40        self.allocator.free(self.dir_path);41        self.* = undefined;42    }43};4445pub const Publisher = struct {46    store: Store,47    root: []const u8,4849    pub fn finish(self: *Publisher) !void {50        try sync.finishRefresh(self.store.allocator, self.root);51    }5253    pub fn close(self: *Publisher) void {54        self.store.close();55        self.* = undefined;56    }57};5859pub fn open(allocator: std.mem.Allocator, root: []const u8, limits: smg.StorageLimits) !Store {60    return try openWriter(allocator, root, limits, false);61}6263pub fn openMutation(allocator: std.mem.Allocator, root: []const u8, limits: smg.StorageLimits) !Store {64    return try openWriter(allocator, root, limits, true);65}6667pub fn beginRepairPublish(allocator: std.mem.Allocator, root: []const u8, limits: smg.StorageLimits) !Publisher {68    var opened = try openWriter(allocator, root, limits, false);69    errdefer opened.close();70    try sync.beginRefresh(allocator, root);71    return .{ .store = opened, .root = root };72}7374pub fn beginPublish(75    allocator: std.mem.Allocator,76    root: []const u8,77    expected_head: sql.Hash,78    limits: smg.StorageLimits,79) !Publisher {80    var opened = try openWriter(allocator, root, limits, true);81    errdefer opened.close();82    const current_head = (try opened.database.connection.checkout()).head;83    if (!sql.version.same(current_head, expected_head)) return error.StaleSourceHead;84    try sync.beginRefresh(allocator, root);85    return .{ .store = opened, .root = root };86}8788fn openWriter(89    allocator: std.mem.Allocator,90    root: []const u8,91    limits: smg.StorageLimits,92    require_complete: bool,93) !Store {94    const dir_path = try openDirPath(allocator, root);95    errdefer allocator.free(dir_path);96    var lock = try acquireExclusiveLock(allocator, dir_path);97    errdefer releaseLock(&lock);98    if (require_complete and try sync.refreshIncomplete(allocator, root)) return error.GraphRefreshIncomplete;99    try ensureGitignore(allocator, dir_path, limits);100    var dir = try std.Io.Dir.openDirAbsolute(fs_io, dir_path, .{});101    defer dir.close(fs_io);102    var opened = try database.open(allocator, dir, limits);103    errdefer opened.deinit();104    try database.checkpointIfWalLarge(&opened);105106    return .{107        .allocator = allocator,108        .dir_path = dir_path,109        .lock = lock,110        .database = opened,111    };112}113114pub fn openRead(allocator: std.mem.Allocator, root: []const u8, limits: smg.StorageLimits) !ReadStore {115    return readStore(allocator, root, limits, true);116}117118/// Opens the existing graph without waiting for a publisher or repairing storage.119pub fn tryOpenRead(120    allocator: std.mem.Allocator,121    root: []const u8,122    limits: smg.StorageLimits,123) !ReadStore {124    return readStore(allocator, root, limits, false);125}126127fn readStore(128    allocator: std.mem.Allocator,129    root: []const u8,130    limits: smg.StorageLimits,131    wait: bool,132) !ReadStore {133    const dir_path = try existingDirPath(allocator, root);134    errdefer allocator.free(dir_path);135    var lock = try sharedLock(allocator, dir_path, wait);136    errdefer releaseLock(&lock);137    if (try sync.refreshIncomplete(allocator, root)) return error.GraphRefreshIncomplete;138139    var dir = try std.Io.Dir.openDirAbsolute(fs_io, dir_path, .{});140    defer dir.close(fs_io);141    var reader = switch (try database.openReader(allocator, dir, limits)) {142        .ready => |ready| ready,143        .repair_required => |reason| return switch (reason) {144            .stale_refs, .root_mismatch => error.GraphRefreshIncomplete,145            .file => |file_reason| switch (file_reason) {146                .wal_too_large => error.WalRecoveryLimitBytesExceeded,147                else => error.GraphStoreRepairRequired,148            },149            .missing_refs, .invalid_refs, .missing_branch => error.GraphStoreRepairRequired,150        },151    };152    errdefer reader.deinit();153    if (!try database.readerSchemaReady(&reader)) return error.GraphStoreMigrationRequired;154155    return .{156        .allocator = allocator,157        .dir_path = dir_path,158        .lock = lock,159        .reader = reader,160    };161}162163fn existingDirPath(allocator: std.mem.Allocator, root: []const u8) ![]const u8 {164    return try std.fs.path.join(allocator, &.{ root, paths.smg_dir_name });165}166167fn openDirPath(allocator: std.mem.Allocator, root: []const u8) ![]const u8 {168    const dir_path = try std.fs.path.join(allocator, &.{ root, paths.smg_dir_name });169    errdefer allocator.free(dir_path);170    try sys.fs.createDirPath(dir_path);171    return dir_path;172}173174fn acquireExclusiveLock(allocator: std.mem.Allocator, dir_path: []const u8) !std.Io.File {175    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });176    defer allocator.free(lock_path);177    var lock = try std.Io.Dir.createFileAbsolute(fs_io, lock_path, .{178        .truncate = false,179        .read = true,180        .permissions = @fromBackingInt(@intCast(0o600)),181    });182    errdefer lock.close(fs_io);183    try lock.lock(fs_io, .exclusive);184    try validateLock(lock, lock_path);185    return lock;186}187188fn sharedLock(allocator: std.mem.Allocator, dir_path: []const u8, wait: bool) !std.Io.File {189    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });190    defer allocator.free(lock_path);191    var lock = std.Io.Dir.openFileAbsolute(fs_io, lock_path, .{}) catch |err| switch (err) {192        error.FileNotFound => return error.NoProject,193        else => return err,194    };195    errdefer lock.close(fs_io);196    if (wait) {197        try lock.lock(fs_io, .shared);198    } else if (!try lock.tryLock(fs_io, .shared)) return error.GraphStoreBusy;199    try validateLock(lock, lock_path);200    return lock;201}202203fn validateLock(lock: std.Io.File, lock_path: []const u8) !void {204    const current = std.Io.Dir.openFileAbsolute(fs_io, lock_path, .{}) catch |err| switch (err) {205        error.FileNotFound => return error.GraphStoreBusy,206        else => return err,207    };208    defer current.close(fs_io);209    const held_identity = try sys.fd.identity(lock.handle);210    const current_identity = try sys.fd.identity(current.handle);211    if (!std.meta.eql(held_identity, current_identity)) return error.GraphStoreBusy;212}213214fn releaseLock(lock: *std.Io.File) void {215    lock.unlock(fs_io);216    lock.close(fs_io);217}218219fn ensureGitignore(220    allocator: std.mem.Allocator,221    dir_path: []const u8,222    limits: smg.StorageLimits,223) !void {224    const path = try std.fs.path.join(allocator, &.{ dir_path, ".gitignore" });225    defer allocator.free(path);226    if (sys.fs.readFileAlloc(allocator, path, limits.max_gitignore_bytes)) |existing| {227        defer allocator.free(existing);228        if (std.mem.eql(u8, existing, gitignore_content)) return;229    } else |_| {}230    try text.writeFile(path, gitignore_content);231}232233test "store open creates directory gitignore and lock" {234    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);235    defer arena.deinit();236    const allocator = arena.allocator();237    const root = try testRoot(allocator, 0);238    defer sys.fs.deleteTree(root) catch {};239240    var opened = try open(allocator, root, smg.default_limits.storage);241    const dir_path = try allocator.dupe(u8, opened.dir_path);242    opened.close();243244    const gitignore_path = try std.fs.path.join(allocator, &.{ dir_path, ".gitignore" });245    const gitignore = try sys.fs.readFileAlloc(allocator, gitignore_path, 4096);246    try std.testing.expectEqualStrings(gitignore_content, gitignore);247    try std.testing.expect(std.mem.indexOf(u8, gitignore, "search.sql.version") == null);248249    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });250    try std.testing.expect(sys.fs.exists(lock_path));251}252253test "read store opens after writer closes" {254    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);255    defer arena.deinit();256    const allocator = arena.allocator();257    const root = try testRoot(allocator, 1);258    defer sys.fs.deleteTree(root) catch {};259260    var writer = try open(allocator, root, smg.default_limits.storage);261    writer.close();262    var reader = try openRead(allocator, root, smg.default_limits.storage);263    reader.close();264}265266test "read store rejects an incomplete graph refresh until a writer completes it" {267    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);268    defer arena.deinit();269    const allocator = arena.allocator();270    const root = try testRoot(allocator, 2);271    defer sys.fs.deleteTree(root) catch {};272273    var writer = try open(allocator, root, smg.default_limits.storage);274    writer.close();275    var interrupted = try beginRepairPublish(allocator, root, smg.default_limits.storage);276    interrupted.close();277    try std.testing.expectError(278        error.GraphRefreshIncomplete,279        openRead(allocator, root, smg.default_limits.storage),280    );281    try std.testing.expectError(282        error.GraphRefreshIncomplete,283        openMutation(allocator, root, smg.default_limits.storage),284    );285    var publisher = try beginRepairPublish(allocator, root, smg.default_limits.storage);286    errdefer publisher.close();287    try publisher.finish();288    publisher.close();289    var reader = try openRead(allocator, root, smg.default_limits.storage);290    reader.close();291}292293test "expected publication rejects stale heads before arming the refresh marker" {294    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);295    defer arena.deinit();296    const allocator = arena.allocator();297    const root = try testRoot(allocator, 6);298    defer sys.fs.deleteTree(root) catch {};299300    var writer = try open(allocator, root, smg.default_limits.storage);301    const source_head = (try writer.database.connection.checkout()).head;302    writer.close();303304    var matching = try beginPublish(305        allocator,306        root,307        source_head,308        smg.default_limits.storage,309    );310    try std.testing.expect(try sync.refreshIncomplete(allocator, root));311    try matching.finish();312    matching.close();313    try std.testing.expect(!try sync.refreshIncomplete(allocator, root));314315    var winner = try open(allocator, root, smg.default_limits.storage);316    try database.edits.apply(&winner.database, .{ .concept_document_puts = &.{.{317        .rowid = 1,318        .name = "winner",319        .document = "{\"name\":\"winner\"}",320    }} });321    const winner_head = try database.commit(&winner.database);322    winner.close();323    try std.testing.expect(!sql.version.same(source_head, winner_head));324325    try std.testing.expectError(326        error.StaleSourceHead,327        beginPublish(allocator, root, source_head, smg.default_limits.storage),328    );329    try std.testing.expect(!try sync.refreshIncomplete(allocator, root));330331    var reopened = try open(allocator, root, smg.default_limits.storage);332    try std.testing.expect(sql.version.same(winner_head, (try reopened.database.connection.checkout()).head));333    reopened.close();334}335336test "read store shared locks coexist and exclude a publisher" {337    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);338    defer arena.deinit();339    const allocator = arena.allocator();340    const root = try testRoot(allocator, 3);341    defer sys.fs.deleteTree(root) catch {};342343    var writer = try open(allocator, root, smg.default_limits.storage);344    writer.close();345    const dir_path = try existingDirPath(allocator, root);346    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });347    {348        var first = try sharedLock(allocator, dir_path, true);349        defer releaseLock(&first);350        var second = try sharedLock(allocator, dir_path, true);351        defer releaseLock(&second);352        var exclusive = try std.Io.Dir.openFileAbsolute(fs_io, lock_path, .{});353        defer exclusive.close(fs_io);354        try std.testing.expect(!try exclusive.tryLock(fs_io, .exclusive));355    }356}357358test "store lock validation accepts the current shared and exclusive lock" {359    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);360    defer arena.deinit();361    const allocator = arena.allocator();362    const root = try testRoot(allocator, 7);363    defer sys.fs.deleteTree(root) catch {};364    const dir_path = try openDirPath(allocator, root);365    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });366    for ([_]bool{ false, true }) |shared| {367        var lock = if (shared)368            try sharedLock(allocator, dir_path, false)369        else370            try acquireExclusiveLock(allocator, dir_path);371        defer releaseLock(&lock);372        try validateLock(lock, lock_path);373        var contender = try std.Io.Dir.openFileAbsolute(fs_io, lock_path, .{});374        defer contender.close(fs_io);375        try std.testing.expect(!try contender.tryLock(fs_io, .exclusive));376    }377}378379test "store lock validation refuses retirement while the replacement remains exclusive" {380    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);381    defer arena.deinit();382    const allocator = arena.allocator();383    const root = try testRoot(allocator, 8);384    defer sys.fs.deleteTree(root) catch {};385    const dir_path = try openDirPath(allocator, root);386    const retired_path = try std.fs.path.join(allocator, &.{ root, "retired" });387    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });388    for ([_]bool{ false, true }) |shared| {389        var lock = if (shared)390            try sharedLock(allocator, dir_path, false)391        else392            try acquireExclusiveLock(allocator, dir_path);393        defer releaseLock(&lock);394        try sys.fs.rename(dir_path, retired_path);395        defer sys.fs.deleteTree(retired_path) catch {};396        try sys.fs.createDirPath(dir_path);397        var replacement = try acquireExclusiveLock(allocator, dir_path);398        defer releaseLock(&replacement);399        try std.testing.expectError(error.GraphStoreBusy, validateLock(lock, lock_path));400        try std.testing.expectError(error.GraphStoreBusy, sharedLock(allocator, dir_path, false));401        try validateLock(replacement, lock_path);402    }403}404405test "store lock validation refuses retirement without creating a replacement" {406    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);407    defer arena.deinit();408    const allocator = arena.allocator();409    const root = try testRoot(allocator, 9);410    defer sys.fs.deleteTree(root) catch {};411    const dir_path = try openDirPath(allocator, root);412    const retired_path = try std.fs.path.join(allocator, &.{ root, "retired" });413    const lock_path = try std.fs.path.join(allocator, &.{ dir_path, paths.lock_file_name });414    var lock = try acquireExclusiveLock(allocator, dir_path);415    defer releaseLock(&lock);416    try sys.fs.rename(dir_path, retired_path);417    try std.testing.expectError(error.GraphStoreBusy, validateLock(lock, lock_path));418    try std.testing.expect(!sys.fs.exists(dir_path));419}420421test "read store observation preserves absence and refuses a held writer" {422    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);423    defer arena.deinit();424    const allocator = arena.allocator();425    const root = try testRoot(allocator, 10);426    defer sys.fs.deleteTree(root) catch {};427    try std.testing.expectError(error.NoProject, tryOpenRead(allocator, root, smg.default_limits.storage));428    try std.testing.expectError(error.NoProject, openRead(allocator, root, smg.default_limits.storage));429    try std.testing.expect(!sys.fs.exists(root));430    {431        var writer = try open(allocator, root, smg.default_limits.storage);432        defer writer.close();433        try std.testing.expectError(434            error.GraphStoreBusy,435            tryOpenRead(allocator, root, smg.default_limits.storage),436        );437    }438    var reader = try tryOpenRead(allocator, root, smg.default_limits.storage);439    defer reader.close();440}441442test "read store returns typed migration and repair requirements" {443    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);444    defer arena.deinit();445    const allocator = arena.allocator();446    const migration_root = try testRoot(allocator, 4);447    defer sys.fs.deleteTree(migration_root) catch {};448449    {450        var writer = try open(allocator, migration_root, smg.default_limits.storage);451        defer writer.close();452        var dropped = try writer.database.connection.execute(allocator, "DROP TABLE nodes", .{ .durability = .buffered });453        dropped.deinit(allocator);454        _ = try database.commit(&writer.database);455    }456    try std.testing.expectError(457        error.GraphStoreMigrationRequired,458        openRead(allocator, migration_root, smg.default_limits.storage),459    );460461    const repair_root = try testRoot(allocator, 5);462    defer sys.fs.deleteTree(repair_root) catch {};463    {464        var writer = try open(allocator, repair_root, smg.default_limits.storage);465        writer.close();466    }467    const database_path = try std.fs.path.join(allocator, &.{ repair_root, paths.smg_dir_name, paths.database_file_name });468    const wal_path = try std.fs.path.join(allocator, &.{ repair_root, paths.smg_dir_name, paths.wal_file_name });469    try sys.fs.deleteFile(database_path);470    try sys.fs.deleteFile(wal_path);471    try std.testing.expectError(472        error.GraphStoreRepairRequired,473        openRead(allocator, repair_root, smg.default_limits.storage),474    );475}476477fn testRoot(allocator: std.mem.Allocator, offset: i64) ![]const u8 {478    return try std.fmt.allocPrint(allocator, "/tmp/smg-storage-store-test-{x}", .{@as(u64, @intCast(@max(0, sys.time.realMilliTimestamp() + offset)))});479}