tiny.smg.storage.database.lifecycle

1const std = @import("std");2const sql = @import("sql");3const storage = @import("../root.zig");4const limits_mod = @import("../../limits/root.zig");5const paths = storage.paths;6const rows = storage.rows;78pub const Error = rows.Error || sql.lifecycle.Error;910const branch_name = "main";1112const checkpoint_wal_bytes: usize = 4 * 1024 * 1024;13const fs_io = std.Options.debug_io;14const database_header = sql.wal.Header{15    .sequence = 0x534d_4744,16    .salt = .{ .first = 0x534d_4744, .second = 0x4752_5048 },17};1819pub const Database = struct {20    allocator: std.mem.Allocator,21    workspace: *sql.LifecycleWorkspace,22    limits: limits_mod.Storage,23    file: *sql.FileDatabase,24    history: sql.LazyHistory,25    connection: sql.Connection,26    recovery: sql.LifecycleRecovery,2728    pub fn deinit(self: *Database) void {29        const allocator = self.allocator;30        const workspace = self.workspace;31        sql.lifecycle.close(self.file, &self.history, &self.connection);32        workspace.deallocate(allocator);33        allocator.destroy(workspace);34        self.* = undefined;35    }36};3738pub const Reader = struct {39    allocator: std.mem.Allocator,40    workspace: *sql.LifecycleWorkspace,41    file: *sql.FileReadOnlyDatabase,42    catalog: sql.CatalogReader,43    head: sql.Hash,44    root: sql.Hash,4546    pub fn deinit(self: *Reader) void {47        const allocator = self.allocator;48        const workspace = self.workspace;49        self.file.deinit();50        self.allocator.destroy(self.file);51        workspace.deallocate(allocator);52        allocator.destroy(workspace);53        self.* = undefined;54    }55};5657pub const ReaderOpenResult = union(enum) {58    ready: Reader,59    repair_required: sql.LifecycleReadRepairReason,60};6162pub fn open(63    allocator: std.mem.Allocator,64    dir: std.Io.Dir,65    limits: limits_mod.Storage,66) Error!Database {67    const workspace = try allocator.create(sql.LifecycleWorkspace);68    errdefer allocator.destroy(workspace);69    workspace.* = try sql.LifecycleWorkspace.allocate(allocator, .{70        .header = database_header,71        .max_wal_bytes = limits.wal_recovery_limit_bytes,72        .path_storage = sql.file.PathStorage.Limits.forDirect(.{73            .database = paths.database_file_name,74            .wal = paths.wal_file_name,75        }),76    });77    errdefer workspace.deallocate(allocator);78    var lifecycle = try sql.lifecycle.open(allocator, workspace, dir, .{79        .paths = .{ .database = paths.database_file_name, .wal = paths.wal_file_name },80        .history_path = paths.history_file_name,81        .branch = branch_name,82        .header = database_header,83        .max_wal_bytes = limits.wal_recovery_limit_bytes,84    });85    errdefer lifecycle.deinit();8687    var database = Database{88        .allocator = lifecycle.allocator,89        .workspace = workspace,90        .limits = limits,91        .file = lifecycle.file,92        .history = lifecycle.history,93        .connection = lifecycle.connection,94        .recovery = lifecycle.recovery,95    };96    try ensureSchema(&database);97    return database;98}99100pub fn openReader(101    allocator: std.mem.Allocator,102    dir: std.Io.Dir,103    limits: limits_mod.Storage,104) Error!ReaderOpenResult {105    const workspace = try allocator.create(sql.LifecycleWorkspace);106    errdefer allocator.destroy(workspace);107    workspace.* = try sql.LifecycleWorkspace.allocate(allocator, .{108        .header = database_header,109        .max_wal_bytes = limits.wal_recovery_limit_bytes,110        .path_storage = sql.file.PathStorage.Limits.forDirect(.{111            .database = paths.database_file_name,112            .wal = paths.wal_file_name,113        }),114    });115    errdefer workspace.deallocate(allocator);116    const opened = switch (try sql.lifecycle.openReadOnly(allocator, workspace, dir, .{117        .paths = .{ .database = paths.database_file_name, .wal = paths.wal_file_name },118        .history_path = paths.history_file_name,119        .branch = branch_name,120        .header = database_header,121        .max_wal_bytes = limits.wal_recovery_limit_bytes,122    })) {123        .ready => |ready| ready,124        .repair_required => |reason| {125            workspace.deallocate(allocator);126            allocator.destroy(workspace);127            return .{ .repair_required = reason };128        },129    };130    return .{ .ready = .{131        .allocator = opened.allocator,132        .workspace = workspace,133        .file = opened.file,134        .catalog = opened.catalog,135        .head = opened.head,136        .root = opened.root,137    } };138}139140pub fn readerSchemaReady(reader: *Reader) Error!bool {141    return try relationHasIndex(&reader.catalog, reader.allocator, rows.nodes_relation, rows.node_name_index) and142        try relationHasIndex(&reader.catalog, reader.allocator, rows.edges_relation, rows.edge_source_index) and143        try relationHasIndex(&reader.catalog, reader.allocator, rows.edges_relation, rows.edge_target_index) and144        try relationHasIndex(&reader.catalog, reader.allocator, rows.concept_documents_relation, rows.concept_name_index);145}146147pub fn commitIfDirty(opened: *Database) Error!void {148    if (!(try opened.connection.checkout()).working.dirty()) return;149    _ = try commit(opened);150    try checkpointIfWalLarge(opened);151}152153pub fn commit(opened: *Database) Error!sql.Hash {154    return try sql.lifecycle.commit(&opened.connection, &opened.history, opened.file);155}156157pub fn checkpointIfWalLarge(opened: *Database) Error!void {158    try sql.lifecycle.checkpointIfWalLarge(opened.file, database_header, checkpoint_wal_bytes);159}160161fn checkpoint(opened: *Database) Error!void {162    try sql.lifecycle.checkpoint(opened.file, database_header);163}164165fn ensureSchema(opened: *Database) Error!void {166    const statements = [_]struct { name: []const u8, source: []const u8 }{167        .{ .name = rows.nodes_relation, .source = "CREATE TABLE " ++ rows.nodes_relation ++ " (" ++ rows.node_columns ++ ")" },168        .{ .name = rows.edges_relation, .source = "CREATE TABLE " ++ rows.edges_relation ++ " (" ++ rows.edge_columns ++ ")" },169        .{ .name = rows.concept_documents_relation, .source = "CREATE TABLE " ++ rows.concept_documents_relation ++ " (" ++ rows.concept_document_columns ++ ")" },170    };171    var names = try opened.connection.catalog.relationNames(opened.allocator);172    defer names.deinit();173    var created = false;174    for (statements) |statement| {175        var exists = false;176        for (names.names) |name| {177            if (std.mem.eql(u8, name, statement.name)) {178                exists = true;179                break;180            }181        }182        if (exists) continue;183        var result = try opened.connection.execute(opened.allocator, statement.source, .{ .durability = .buffered });184        result.deinit(opened.allocator);185        created = true;186    }187    if (created) {188        _ = try commit(opened);189        try checkpoint(opened);190    }191    const indexes = [_]struct { relation: []const u8, name: []const u8, columns: []const u8 }{192        .{ .relation = rows.nodes_relation, .name = rows.node_name_index, .columns = "name" },193        .{ .relation = rows.edges_relation, .name = rows.edge_source_index, .columns = "source" },194        .{ .relation = rows.edges_relation, .name = rows.edge_target_index, .columns = "target" },195        .{ .relation = rows.concept_documents_relation, .name = rows.concept_name_index, .columns = "name" },196    };197    for (indexes) |index| {198        if (try relationHasIndex(&opened.connection.catalog, opened.allocator, index.relation, index.name)) continue;199        const source = try std.fmt.allocPrint(opened.allocator, "CREATE INDEX {s} ON {s} ({s})", .{ index.name, index.relation, index.columns });200        defer opened.allocator.free(source);201        var result = try opened.connection.execute(opened.allocator, source, .{ .durability = .buffered });202        result.deinit(opened.allocator);203        _ = try commit(opened);204        try checkpoint(opened);205    }206}207208fn relationHasIndex(catalog: anytype, allocator: std.mem.Allocator, relation: []const u8, index: []const u8) Error!bool {209    var handle = catalog.openRelation(allocator, relation) catch |err| switch (err) {210        error.RelationNotFound => return false,211        else => return err,212    };213    defer handle.deinit();214    for (handle.index_definitions) |definition| {215        if (std.mem.eql(u8, definition.name, index)) return true;216    }217    return false;218}219220test "smg database round-trips nodes and edges with history" {221    const database = @import("root.zig");222    const edits = database.edits;223    const read = database.read;224    const model = @import("../../root.zig").model;225    var tmp = std.testing.tmpDir(.{});226    defer tmp.cleanup();227228    const metadata = [_]model.Pair{229        .{ .key = "source", .value = "scan" },230        .{ .key = "metrics", .value = "{\"cyclomatic_complexity\":2}", .json = true },231    };232    const node = model.Node{233        .name = "app.main",234        .type = model.NodeType.function,235        .file = "src/app.zig",236        .line = 3,237        .end_line = 9,238        .docstring = "doc body",239        .metadata = &metadata,240    };241    const edge = model.Edge{242        .source = "app",243        .target = "app.main",244        .rel = model.RelType.contains,245    };246247    var first_head: sql.Hash = undefined;248    {249        var db = try open(std.testing.allocator, tmp.dir, testingLimits());250        defer db.deinit();251252        try edits.apply(&db, .{253            .node_puts = &.{.{ .rowid = 1, .node = node }},254            .edge_puts = &.{.{ .rowid = 1, .edge = edge }},255        });256        try std.testing.expect((try db.connection.checkout()).working.dirty());257        first_head = try commit(&db);258        try std.testing.expect(!(try db.connection.checkout()).working.dirty());259    }260261    var db = try open(std.testing.allocator, tmp.dir, testingLimits());262    defer db.deinit();263    try std.testing.expect(!(try db.connection.checkout()).working.dirty());264    try std.testing.expect(sql.version.same(first_head, (try db.connection.checkout()).head));265    {266        var node_handle = try db.connection.catalog.openRelation(std.testing.allocator, rows.nodes_relation);267        defer node_handle.deinit();268        try std.testing.expectEqual(@as(usize, 1), node_handle.index_definitions.len);269        try std.testing.expectEqualStrings(rows.node_name_index, node_handle.index_definitions[0].name);270    }271    {272        var edge_handle = try db.connection.catalog.openRelation(std.testing.allocator, rows.edges_relation);273        defer edge_handle.deinit();274        try std.testing.expectEqual(@as(usize, 2), edge_handle.index_definitions.len);275        try std.testing.expectEqualStrings(rows.edge_source_index, edge_handle.index_definitions[0].name);276        try std.testing.expectEqualStrings(rows.edge_target_index, edge_handle.index_definitions[1].name);277    }278279    const nodes = try read.databaseNodes(&db, std.testing.allocator, std.testing.allocator);280    defer rows.freeNodeRowSlice(std.testing.allocator, std.testing.allocator, nodes);281    try std.testing.expectEqual(@as(usize, 1), nodes.len);282    try std.testing.expectEqual(@as(i64, 1), nodes[0].rowid);283    try std.testing.expectEqualStrings("app.main", nodes[0].node.name);284    try std.testing.expectEqual(@as(?i64, 3), nodes[0].node.line);285    try std.testing.expectEqual(@as(?i64, 9), nodes[0].node.end_line);286    try std.testing.expectEqualStrings("doc body", nodes[0].node.docstring.?);287    try std.testing.expectEqualStrings("scan", model.pairValue(nodes[0].node.metadata, "source").?);288    try std.testing.expectEqualStrings("{\"cyclomatic_complexity\":2}", model.pairValue(nodes[0].node.metadata, "metrics").?);289    try std.testing.expect(nodes[0].node.metadata[1].json);290291    const edges = try read.databaseEdges(&db, std.testing.allocator, std.testing.allocator);292    defer rows.freeEdgeRowSlice(std.testing.allocator, std.testing.allocator, edges);293    try std.testing.expectEqual(@as(usize, 1), edges.len);294    try std.testing.expectEqualStrings("app", edges[0].edge.source);295    try std.testing.expectEqualStrings("app.main", edges[0].edge.target);296    try std.testing.expectEqual(@as(usize, 0), edges[0].edge.metadata.len);297}298299test "smg database updates and deletes survive checkpointed reopen" {300    const database = @import("root.zig");301    const edits = database.edits;302    const read = database.read;303    const model = @import("../../root.zig").model;304    var tmp = std.testing.tmpDir(.{});305    defer tmp.cleanup();306307    const kept = model.Node{ .name = "app.kept", .type = model.NodeType.function };308    const dropped = model.Node{ .name = "app.dropped", .type = model.NodeType.function };309310    {311        var db = try open(std.testing.allocator, tmp.dir, testingLimits());312        defer db.deinit();313        try edits.apply(&db, .{314            .node_puts = &.{ .{ .rowid = 1, .node = kept }, .{ .rowid = 2, .node = dropped } },315        });316        _ = try commit(&db);317318        var renamed = kept;319        renamed.line = 42;320        try edits.apply(&db, .{321            .node_puts = &.{.{ .rowid = 1, .node = renamed }},322            .node_deletes = &.{2},323        });324        _ = try commit(&db);325        try checkpoint(&db);326    }327328    var db = try open(std.testing.allocator, tmp.dir, testingLimits());329    defer db.deinit();330    const nodes = try read.databaseNodes(&db, std.testing.allocator, std.testing.allocator);331    defer rows.freeNodeRowSlice(std.testing.allocator, std.testing.allocator, nodes);332    try std.testing.expectEqual(@as(usize, 1), nodes.len);333    try std.testing.expectEqualStrings("app.kept", nodes[0].node.name);334    try std.testing.expectEqual(@as(?i64, 42), nodes[0].node.line);335}336337test "smg database rebuilds unreadable history from live state" {338    const database = @import("root.zig");339    const edits = database.edits;340    const read = database.read;341    const model = @import("../../root.zig").model;342    var tmp = std.testing.tmpDir(.{});343    defer tmp.cleanup();344345    const node = model.Node{ .name = "app.rebuild", .type = model.NodeType.function };346347    {348        var db = try open(std.testing.allocator, tmp.dir, testingLimits());349        defer db.deinit();350        try edits.apply(&db, .{ .node_puts = &.{.{ .rowid = 1, .node = node }} });351        _ = try commit(&db);352        try checkpoint(&db);353    }354355    {356        var file = try tmp.dir.createFile(fs_io, paths.history_file_name, .{ .read = true, .truncate = false });357        var stale_version: [4]u8 = undefined;358        std.mem.writeInt(u32, stale_version[0..], 1, .big);359        try file.writePositionalAll(fs_io, stale_version[0..], 4);360        file.close(fs_io);361    }362363    var db = try open(std.testing.allocator, tmp.dir, testingLimits());364    defer db.deinit();365    try std.testing.expect(!(try db.connection.checkout()).working.dirty());366367    const nodes = try read.databaseNodes(&db, std.testing.allocator, std.testing.allocator);368    defer rows.freeNodeRowSlice(std.testing.allocator, std.testing.allocator, nodes);369    try std.testing.expectEqual(@as(usize, 1), nodes.len);370    try std.testing.expectEqualStrings("app.rebuild", nodes[0].node.name);371372    const next = model.Node{ .name = "app.next", .type = model.NodeType.function };373    try edits.apply(&db, .{ .node_puts = &.{.{ .rowid = 2, .node = next }} });374    _ = try commit(&db);375}376377test "smg database tolerates empty graph edits" {378    const database = @import("root.zig");379    const edits = database.edits;380    var tmp = std.testing.tmpDir(.{});381    defer tmp.cleanup();382383    var db = try open(std.testing.allocator, tmp.dir, testingLimits());384    defer db.deinit();385386    try edits.apply(&db, .{});387    try std.testing.expect(!(try db.connection.checkout()).working.dirty());388}389390test "smg database installs query indexes on existing unindexed relations" {391    var tmp = std.testing.tmpDir(.{});392    defer tmp.cleanup();393    {394        var raw = try sql.lifecycle.openForTesting(std.testing.allocator, tmp.dir, .{395            .paths = .{ .database = paths.database_file_name, .wal = paths.wal_file_name },396            .history_path = paths.history_file_name,397            .branch = branch_name,398            .header = database_header,399            .max_wal_bytes = testingLimits().wal_recovery_limit_bytes,400        });401        defer raw.deinit();402        var nodes_result = try raw.connection.execute(std.testing.allocator, "CREATE TABLE nodes (name, type, file, line, end_line, docstring, metadata)", .{ .durability = .buffered });403        nodes_result.deinit(std.testing.allocator);404        var edges_result = try raw.connection.execute(std.testing.allocator, "CREATE TABLE edges (source, rel, target, metadata)", .{ .durability = .buffered });405        edges_result.deinit(std.testing.allocator);406        _ = try sql.lifecycle.commit(&raw.connection, &raw.history, raw.file);407    }408    var upgraded = try open(std.testing.allocator, tmp.dir, testingLimits());409    defer upgraded.deinit();410    try std.testing.expect(try relationHasIndex(&upgraded.connection.catalog, std.testing.allocator, rows.nodes_relation, rows.node_name_index));411    try std.testing.expect(try relationHasIndex(&upgraded.connection.catalog, std.testing.allocator, rows.edges_relation, rows.edge_source_index));412    try std.testing.expect(try relationHasIndex(&upgraded.connection.catalog, std.testing.allocator, rows.edges_relation, rows.edge_target_index));413}414415fn testingLimits() limits_mod.Storage {416    return @import("../../root.zig").default_limits.storage;417}

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