Something went wrong. Try again.
search for standard sites pub-search.waow.tech
search zig blog atproto
Something went wrong. Try again.
35 kB · 823 lines
Zig
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824//! Local SQLite read replica using zqlite//! Provides fast FTS5 queries while Turso remains source of truth
const std = @import("std");const Io = std.Io;const zqlite = @import("zqlite");const Allocator = std.mem.Allocator;const logfire = @import("logfire");const pubkey = @import("../server/pubkey.zig");
const LocalDb = @This();
/// Local replica schema generation. Bump when createSchema changes shape/// (new column/table/index the serving code depends on) — AND when the FTS/// tokenizer or query normalization changes: a snapshot tokenized one way/// searched by a server normalizing another serves wrong results with no/// error (typeahead's normalize_version lesson — the stamp is the only/// thing that surfaces the divergence). The builder stamps it into the/// manifest; the promote watcher rejects a mismatch — a snapshot built by/// an out-of-date builder image must stall freshness, not break serving./// (Scheduled builder machines pin their creation image: recreate them/// after bumping this.)pub const SCHEMA_VERSION: u32 = 5; // v5: external-content documents_fts + content-last column order (replica 816→518MB)
const READ_POOL_SIZE = 12;
// Per-thread checkout state. A thread holds at most one pool connection; nested// and sibling queries on the same thread share it, counted by `tl_depth`. The// connection is acquired when depth goes 0→1 and returned when it falls 1→0.threadlocal var tl_conn_idx: ?usize = null;threadlocal var tl_depth: usize = 0;
conn: ?zqlite.Conn = null,read_conn: ?zqlite.Conn = null, // legacy single read conn (kept as pool[0] alias)// READ POOL — one connection per thread, reference-counted.//// A single SQLite connection runs one *thread's* statements; a query() holds// its connection open for the whole row iteration. The old round-robin scheme// let two threads share a connection and serialize (a PK lookup measured at 10s// behind a content scan, 2026-06-25). A naive per-query checkout fixed that but// DEADLOCKED: search holds its result rows open while running per-result// is_bridgyfed / content queries, so one request needs two connections at once;// 12 such requests each grabbed one and waited forever for a second.//// The fix removes hold-and-wait entirely: a thread checks out ONE connection on// its outermost query() and every nested or sibling query on that thread reuses// it (SQLite is happy to run several statements on one connection from one// thread). The connection returns to the pool when the thread's last Rows// closes. A thread therefore blocks (in acquireRead) only while holding zero// connections, so a deadlock cycle cannot form. Distinct threads get distinct// connections → true WAL-parallel reads.read_pool: [READ_POOL_SIZE]?zqlite.Conn = @splat(null),read_in_use: [READ_POOL_SIZE]bool = @splat(false),pool_mutex: Io.Mutex = Io.Mutex.init, // guards read_in_usepool_cond: Io.Condition = Io.Condition.init, // signaled when a connection is returnedallocator: Allocator,io: Io,is_ready: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),needs_resync: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),mutex: Io.Mutex = Io.Mutex.init, // protects write conn onlypath: []const u8 = "",consecutive_errors: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
pub fn init(allocator: Allocator, io: Io) LocalDb { return .{ .allocator = allocator, .io = io };}
/// Check database integrity and return false if corrupt. quick_check, not/// integrity_check: the full check scans every index b-tree, which on a/// 365MB+ file on a fly volume is minutes of is_ready=false at every boot/// (keyword falls back to slow turso the whole time). Adopted snapshots are/// already sha256-gated before they become local.db, so the full scan is/// redundant; quick_check still catches page-level corruption in seconds.fn checkIntegrity(self: *LocalDb) bool { const c = self.conn orelse return false; const row = c.row("PRAGMA quick_check", .{}) catch return false; if (row) |r| { defer r.deinit(); const result = r.text(0); if (std.mem.eql(u8, result, "ok")) { return true; } std.debug.print("local db: integrity check failed: {s}\n", .{result}); return false; } return false;}
/// Delete the database file and WAL/SHM filesfn deleteDbFiles(path: []const u8) void { unlinkPath(path); // also delete WAL and SHM files var wal_buf: [260]u8 = undefined; var shm_buf: [260]u8 = undefined; if (path.len < 252) { const wal_path = std.fmt.bufPrint(&wal_buf, "{s}-wal", .{path}) catch return; const shm_path = std.fmt.bufPrint(&shm_buf, "{s}-shm", .{path}) catch return; unlinkPath(wal_path); unlinkPath(shm_path); }}
fn hasManifestSidecar(path_env: []const u8) bool { var buf: [280]u8 = undefined; const sidecar = std.mem.printSentinel(&buf, "{s}.manifest.json", .{path_env}, 0) catch return false; const fd = std.c.open(sidecar.ptr, .{ .ACCMODE = .RDONLY }, @as(std.c.mode_t, 0)); if (fd < 0) return false; _ = std.c.close(fd); return true;}
/// Delete a file by path using C unlink (std.fs.cwd removed in 0.16)fn unlinkPath(path: []const u8) void { var buf: [260]u8 = undefined; if (path.len >= buf.len) return; @memcpy(buf[0..path.len], path); buf[path.len] = 0; _ = std.c.unlink(@ptrCast(&buf));}
pub fn open(self: *LocalDb) !void { const path_env = if (std.c.getenv("LOCAL_DB_PATH")) |p| std.mem.span(p) else "/data/local.db"; try self.openAt(path_env);}
/// Open at an explicit path — the snapshot builder targets a scratch file,/// never the serving replica.pub fn openAt(self: *LocalDb, path: []const u8) !void { self.path = path; adoptPending(path); try self.openDb(path, false);}
/// Snapshot adoption: if `<path>.new` exists at boot, rename it over the live/// file (plus clear stale WAL/SHM) before opening. This is how offline-built/// replicas ship: stage as .new while the old process serves, restart, and/// the swap is atomic — the serving path never coexists with a bulk writer./// The displaced live file is kept as `<path>.prev` (with its manifest/// sidecar) so rollback is `mv .prev .new` + restart, no re-download.fn adoptPending(path_env: []const u8) void { var new_buf: [256]u8 = undefined; const new_path = std.mem.printSentinel(&new_buf, "{s}.new", .{path_env}, 0) catch return; var cur_buf: [256]u8 = undefined; const cur_path = std.mem.printSentinel(&cur_buf, "{s}", .{path_env}, 0) catch return;
const fd = std.c.open(new_path.ptr, .{ .ACCMODE = .RDONLY }, @as(std.c.mode_t, 0)); if (fd < 0) return; // nothing pending _ = std.c.close(fd);
var aux_buf: [264]u8 = undefined; inline for (.{ "-wal", "-shm" }) |suffix| { const aux = std.mem.printSentinel(&aux_buf, "{s}{s}", .{ path_env, suffix }, 0) catch return; _ = std.c.unlink(aux.ptr); }
// keep the displaced snapshot for rollback: live -> .prev (best-effort) var prev_buf: [256]u8 = undefined; if (std.mem.printSentinel(&prev_buf, "{s}.prev", .{path_env}, 0)) |prev_path| { _ = std.c.rename(cur_path.ptr, prev_path.ptr); } else |_| {}
// manifest sidecars follow their snapshots (promote watcher writes them; // manual swaps without sidecars are fine — these renames just fail) var sc_a: [280]u8 = undefined; var sc_b: [280]u8 = undefined; if (std.mem.printSentinel(&sc_a, "{s}.manifest.json", .{path_env}, 0)) |live_sc| { if (std.mem.printSentinel(&sc_b, "{s}.prev.manifest.json", .{path_env}, 0)) |prev_sc| { _ = std.c.rename(live_sc.ptr, prev_sc.ptr); } else |_| {} } else |_| {} if (std.mem.printSentinel(&sc_a, "{s}.new.manifest.json", .{path_env}, 0)) |new_sc| { if (std.mem.printSentinel(&sc_b, "{s}.manifest.json", .{path_env}, 0)) |live_sc| { _ = std.c.rename(new_sc.ptr, live_sc.ptr); } else |_| {} } else |_| {}
if (std.c.rename(new_path.ptr, cur_path.ptr) == 0) { std.debug.print("local db: adopted pending snapshot {s}\n", .{new_path}); @import("../server/memo.zig").bumpGeneration(); } else { std.debug.print("local db: snapshot adopt FAILED, serving existing file\n", .{}); }}
fn openDb(self: *LocalDb, path_env: []const u8, is_retry: bool) !void { // convert to null-terminated for zqlite var path_buf: [256]u8 = undefined; if (path_env.len >= path_buf.len) return error.PathTooLong; @memcpy(path_buf[0..path_env.len], path_env); path_buf[path_env.len] = 0; const path: [*:0]const u8 = path_buf[0..path_env.len :0];
std.debug.print("local db: opening {s}\n", .{path_env});
const flags = zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite; self.conn = zqlite.open(path, flags) catch |err| { std.debug.print("local db: failed to open write conn: {}\n", .{err}); return err; };
// enable WAL for better concurrency _ = self.conn.?.exec("PRAGMA journal_mode=WAL", .{}) catch {}; _ = self.conn.?.exec("PRAGMA busy_timeout=5000", .{}) catch {};
// open the read POOL — WAL mode allows concurrent reads + writes for (&self.read_pool) |*slot| { const rc = zqlite.open(path, zqlite.OpenFlags.ReadOnly) catch |err| { std.debug.print("local db: failed to open read conn: {}\n", .{err}); return err; }; _ = rc.exec("PRAGMA busy_timeout=1000", .{}) catch {}; // mmap for fast reads — avoids pread() syscalls, uses OS page cache directly _ = rc.exec("PRAGMA mmap_size=268435456", .{}) catch {}; // 8MB page cache per conn. Smaller than before because the pool is now // 12 connections (12×8=96MB heap); mmap already serves hot pages from // the shared OS page cache, so the per-conn cache matters less. _ = rc.exec("PRAGMA cache_size=-8000", .{}) catch {}; slot.* = rc; } self.read_conn = self.read_pool[0];
// integrity: if the manifest sidecar exists, this exact file was // sha256-verified + sentinel-checked at stage time and is never written // between adoptions (sync disabled; adoption is an atomic rename), so a // boot-time scan re-reads 350MB of cold volume for nothing — measured // ~3.5-4 min of is_ready=false per adoption (2026-06-11 first prod // adopt), during which keyword falls back to turso. Attested files skip // it; unattested files (manual swaps, fresh creates) keep the check. if (hasManifestSidecar(path_env)) { std.debug.print("local db: integrity attested by manifest sidecar; skipping boot quick_check\n", .{}); } else if (!self.checkIntegrity()) { if (is_retry) { std.debug.print("local db: still corrupt after recreation, giving up\n", .{}); return error.DatabaseCorrupt; } std.debug.print("local db: corrupt, deleting and recreating\n", .{}); if (self.conn) |c| c.close(); self.conn = null; deleteDbFiles(path_env); return self.openDb(path_env, true); }
try self.createSchema(); std.debug.print("local db: initialized\n", .{});}
pub fn deinit(self: *LocalDb) void { for (&self.read_pool) |*slot| { if (slot.*) |c| c.close(); slot.* = null; } self.read_conn = null; if (self.conn) |c| c.close(); self.conn = null;}
/// Retire a heap-allocated instance displaced by a hot snapshot swap: wait/// out requests that loaded this instance before the pointer swap, then close/// and free it off-thread. If the thread can't spawn, the instance is leaked/// — one stale read pool is recoverable, a close under a live reader is not.pub fn retireAsync(self: *LocalDb) void { const t = std.Thread.spawn(.{}, retireThread, .{self}) catch { logfire.warn("local db: retire thread spawn failed; leaking displaced instance", .{}); return; }; t.detach();}
fn retireThread(self: *LocalDb) void { const io = self.io; self.setReady(false); // grace: a request that fetched this instance just before the swap may // not have checked out a pool connection yet io.sleep(Io.Duration.fromSeconds(60), .awake) catch {}; // drain: wait for every checked-out read connection to come home var waited: usize = 0; while (waited < 300) : (waited += 1) { self.pool_mutex.lockUncancelable(io); var busy = false; for (self.read_in_use) |in_use| busy = busy or in_use; self.pool_mutex.unlock(io); if (!busy) break; io.sleep(Io.Duration.fromSeconds(1), .awake) catch break; } self.deinit(); logfire.info("local db: retired displaced snapshot instance", .{}); self.allocator.destroy(self);}
pub fn isReady(self: *LocalDb) bool { return self.is_ready.load(.acquire);}
pub fn setReady(self: *LocalDb, ready: bool) void { self.is_ready.store(ready, .release);}
fn createSchema(self: *LocalDb) !void { const c = self.conn orelse return error.NotOpen;
// documents table (no embedding/embedded_at — vectors are in turbopuffer). // `content` is deliberately the LAST column (schema v5): search's inner // ranking pass probes rows by rowid for created_at/policy flags, and with // the body last those probes never read a document's overflow pages (the // 2026-08-11 fat-row regression; this ordering replaced the covering // index). Keep any new column BEFORE content. c.exec( \\CREATE TABLE IF NOT EXISTS documents ( \\ uri TEXT PRIMARY KEY, \\ did TEXT NOT NULL, \\ rkey TEXT NOT NULL, \\ title TEXT NOT NULL, \\ created_at TEXT, \\ publication_uri TEXT, \\ platform TEXT DEFAULT 'leaflet', \\ source_collection TEXT, \\ path TEXT, \\ base_path TEXT DEFAULT '', \\ has_publication INTEGER DEFAULT 0, \\ indexed_at TEXT, \\ cover_image TEXT DEFAULT '', \\ source_cid TEXT DEFAULT '', \\ content TEXT NOT NULL \\) , .{}) catch |err| { std.debug.print("local db: failed to create documents table: {}\n", .{err}); return err; };
// FTS5 index (unicode61 tokenizer to match Turso). External content // (schema v5): the index reads title/content from documents by rowid // instead of storing a second copy of every body — 286MB of the 816MB // replica was that duplicate (2026-08-16). Writers must key fts rows to // documents.rowid (they already did pre-v5) and delete via the fts5 // 'delete' command with the OLD row values, never plain DELETE. c.exec( \\CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts USING fts5( \\ uri UNINDEXED, \\ title, \\ content, \\ content='documents', \\ content_rowid='rowid', \\ tokenize='unicode61' \\) , .{}) catch |err| { std.debug.print("local db: failed to create documents_fts: {}\n", .{err}); return err; };
// publications table (no created_at - matches Turso schema) c.exec( \\CREATE TABLE IF NOT EXISTS publications ( \\ uri TEXT PRIMARY KEY, \\ did TEXT NOT NULL, \\ rkey TEXT NOT NULL, \\ name TEXT NOT NULL, \\ description TEXT, \\ base_path TEXT, \\ platform TEXT DEFAULT 'leaflet', \\ source_collection TEXT, \\ indexed_at TEXT, \\ show_in_discover INTEGER NOT NULL DEFAULT 1 \\) , .{}) catch |err| { std.debug.print("local db: failed to create publications table: {}\n", .{err}); return err; };
// publications FTS c.exec( \\CREATE VIRTUAL TABLE IF NOT EXISTS publications_fts USING fts5( \\ uri UNINDEXED, \\ name, \\ description, \\ base_path, \\ tokenize='unicode61' \\) , .{}) catch |err| { std.debug.print("local db: failed to create publications_fts: {}\n", .{err}); return err; };
// document_tags table c.exec( \\CREATE TABLE IF NOT EXISTS document_tags ( \\ document_uri TEXT NOT NULL, \\ tag TEXT NOT NULL, \\ PRIMARY KEY (document_uri, tag) \\) , .{}) catch |err| { std.debug.print("local db: failed to create document_tags table: {}\n", .{err}); return err; };
// index for tag queries c.exec("CREATE INDEX IF NOT EXISTS idx_document_tags_tag ON document_tags(tag)", .{}) catch {};
// index for base_path join (documents → publications via publication_uri) c.exec("CREATE INDEX IF NOT EXISTS idx_documents_publication_uri ON documents(publication_uri)", .{}) catch {};
// index for per-author lookups (/wrapped author lens). without this every // author query full-scans the whole documents table — 13s on /wrapped. c.exec("CREATE INDEX IF NOT EXISTS idx_documents_did ON documents(did)", .{}) catch {};
// index for the (did, rkey) publication join used by /wrapped, /subscribed, // /subscribers. The join matches a publication to an at-uri by did+rkey // PARSED out of the uri (collection-agnostic dual-write), so it can't use // the `uri` PRIMARY KEY. Without this index every such join FULL-SCANS // publications: one query is sub-ms cached, but 16 concurrent full-scans // saturate the single vCPU → 45s (measured). With the index: 0.39s. c.exec("CREATE INDEX IF NOT EXISTS idx_publications_did_rkey ON publications(did, rkey)", .{}) catch {};
// sync metadata table c.exec( \\CREATE TABLE IF NOT EXISTS sync_meta ( \\ key TEXT PRIMARY KEY, \\ value TEXT \\) , .{}) catch |err| { std.debug.print("local db: failed to create sync_meta table: {}\n", .{err}); return err; };
// stats table for local counters c.exec( \\CREATE TABLE IF NOT EXISTS stats ( \\ id INTEGER PRIMARY KEY CHECK (id = 1), \\ total_searches INTEGER DEFAULT 0, \\ total_errors INTEGER DEFAULT 0, \\ service_started_at INTEGER \\) , .{}) catch {}; c.exec("INSERT OR IGNORE INTO stats (id) VALUES (1)", .{}) catch {};
// popular searches c.exec( \\CREATE TABLE IF NOT EXISTS popular_searches ( \\ query TEXT PRIMARY KEY, \\ count INTEGER DEFAULT 1 \\) , .{}) catch {};
// recommends (schema v2): lets the top-author cascade and (eventually) // the recommended/curators cache refreshes run against the replica // instead of walking ~30k turso rows per request c.exec( \\CREATE TABLE IF NOT EXISTS recommends ( \\ uri TEXT PRIMARY KEY, \\ did TEXT NOT NULL, \\ rkey TEXT NOT NULL, \\ document_uri TEXT NOT NULL, \\ created_at TEXT, \\ indexed_at TEXT \\) , .{}) catch |err| { std.debug.print("local db: failed to create recommends table: {}\n", .{err}); }; c.exec("CREATE INDEX IF NOT EXISTS idx_recommends_document_uri ON recommends(document_uri)", .{}) catch {}; c.exec("CREATE INDEX IF NOT EXISTS idx_recommends_did ON recommends(did)", .{}) catch {};
// subscriptions (schema v3): the /subscribed leaderboard + /subscribers // drill-down serve from the replica, same as recommends. Empty on pre-v3 // snapshots — endpoints fall back to turso until the first v3 build. c.exec( \\CREATE TABLE IF NOT EXISTS subscriptions ( \\ uri TEXT PRIMARY KEY, \\ did TEXT NOT NULL, \\ rkey TEXT NOT NULL, \\ publication_uri TEXT NOT NULL, \\ created_at TEXT, \\ indexed_at TEXT \\) , .{}) catch |err| { std.debug.print("local db: failed to create subscriptions table: {}\n", .{err}); }; c.exec("CREATE INDEX IF NOT EXISTS idx_subscriptions_publication_uri ON subscriptions(publication_uri)", .{}) catch {}; c.exec("CREATE INDEX IF NOT EXISTS idx_subscriptions_did ON subscriptions(did)", .{}) catch {};
// Idempotent column-add for existing local DBs. SQLite returns an error // when the column already exists ("duplicate column name"), which is // the common case — we'd flood logs if every boot warned on that. So // detect the column first via PRAGMA table_info and skip if present, // and on the rare ACTUAL failure log loudly. addColumnIfMissing(c, "documents", "indexed_at", "TEXT"); addColumnIfMissing(c, "documents", "embedded_at", "TEXT"); addColumnIfMissing(c, "documents", "cover_image", "TEXT DEFAULT ''"); addColumnIfMissing(c, "documents", "is_bridgyfed", "INTEGER DEFAULT 0"); addColumnIfMissing(c, "documents", "url_dead", "INTEGER DEFAULT 0"); addColumnIfMissing(c, "publications", "indexed_at", "TEXT"); addColumnIfMissing(c, "publications", "show_in_discover", "INTEGER NOT NULL DEFAULT 1");
// The pre-v5 covering index (idx_documents_search_cover) is obsolete: // v5 stores `content` last, so the rowid probes the search SQL now does // never read past a body. Pre-v5 snapshots may still carry the index; // harmless, and it dies with them.
// Materialized (did, rkey) for subscriptions so the publication join is a // sargable indexed equijoin instead of parsing publication_uri per row (see // pubkey.joinOnStored). createSchema runs at every boot before serving, so // backfilling here makes the columns correct on the currently-adopted // snapshot immediately — no multi-step rollout. New rows are populated at // build time (sync.zig); this UPDATE only touches rows still NULL. addColumnIfMissing(c, "subscriptions", "publication_did", "TEXT"); addColumnIfMissing(c, "subscriptions", "publication_rkey", "TEXT"); const sub_backfill_sql = comptime "UPDATE subscriptions SET publication_did = " ++ pubkey.didExpr("publication_uri") ++ ", publication_rkey = " ++ pubkey.rkeyExpr("publication_uri") ++ " WHERE publication_did IS NULL AND publication_uri LIKE 'at://%/%/%'"; c.exec(sub_backfill_sql, .{}) catch |err| std.debug.print("local db: subscriptions pubkey backfill failed: {}\n", .{err}); c.exec("CREATE INDEX IF NOT EXISTS idx_subscriptions_pub_did_rkey ON subscriptions(publication_did, publication_rkey)", .{}) catch {};
// Materialized per-document recommender counts. Tag browse ranks by // ln(1+recommenders); computing that with a GROUP BY over ALL of // recommends on every request was the last O(corpus) request path // (docs/scale-300k-plan.md §3a — typeahead's rule: ranking work belongs // at build time). The replica is frozen between adoptions and adoption // always restarts through here, so one rebuild per boot keeps the table // exactly consistent with the snapshot's recommends at request-time cost // zero. No schema-version bump: the table is server-generated, never // shipped in snapshots. c.exec("CREATE TABLE IF NOT EXISTS recommend_counts (document_uri TEXT PRIMARY KEY, rc INTEGER NOT NULL)", .{}) catch |err| { std.debug.print("local db: failed to create recommend_counts table: {}\n", .{err}); }; c.exec("DELETE FROM recommend_counts", .{}) catch {}; c.exec("INSERT INTO recommend_counts SELECT document_uri, COUNT(DISTINCT did) FROM recommends GROUP BY document_uri", .{}) catch |err| { std.debug.print("local db: recommend_counts rebuild failed: {}\n", .{err}); };}
fn hasColumn(c: zqlite.Conn, table: []const u8, column: []const u8) bool { var buf: [128]u8 = undefined; const sql = std.fmt.bufPrint(&buf, "PRAGMA table_info({s})", .{table}) catch return false; var rows = c.rows(sql, .{}) catch return false; defer rows.deinit(); while (rows.next()) |row| { if (std.mem.eql(u8, row.text(1), column)) return true; } return false;}
fn addColumnIfMissing(c: zqlite.Conn, table: []const u8, column: []const u8, decl: []const u8) void { if (hasColumn(c, table, column)) return; var buf: [256]u8 = undefined; const sql = std.fmt.bufPrint(&buf, "ALTER TABLE {s} ADD COLUMN {s} {s}", .{ table, column, decl }) catch { std.debug.print("local db: ALTER format failed for {s}.{s}\n", .{ table, column }); return; }; c.exec(sql, .{}) catch |err| { // Hit only on actual failures (lock contention, disk full) since // we pre-checked for duplicate-column. Worth knowing about. std.debug.print("local db: ALTER TABLE {s} ADD COLUMN {s} failed: {}\n", .{ table, column, err }); };}
/// Row adapter matching result.Row interface (column-indexed access)pub const Row = struct { stmt: zqlite.Row,
pub fn text(self: Row, index: usize) []const u8 { return self.stmt.text(index); }
pub fn int(self: Row, index: usize) i64 { return self.stmt.int(index); }
pub fn float(self: Row, index: usize) f64 { return self.stmt.float(index); }};
/// Iterator for query resultspub const Rows = struct { inner: zqlite.Rows, db: ?*LocalDb = null, // non-null => deinit must call endQuery() to balance beginQuery()
pub fn next(self: *Rows) ?Row { if (self.inner.next()) |r| { return .{ .stmt = r }; } return null; }
pub fn deinit(self: *Rows) void { self.inner.deinit(); // balance the beginQuery() that produced this Rows, exactly once if (self.db) |d| { d.endQuery(); self.db = null; } }
pub fn err(self: *Rows) ?anyerror { return self.inner.err; }};
/// Check out an idle read connection, blocking until one is free. Only ever/// called while this thread holds NO connection (tl_depth 0→1), so a blocked/// thread holds nothing and no deadlock cycle can form.fn acquireRead(self: *LocalDb) usize { self.pool_mutex.lockUncancelable(self.io); defer self.pool_mutex.unlock(self.io); while (true) { for (0..READ_POOL_SIZE) |i| { if (self.read_pool[i] != null and !self.read_in_use[i]) { self.read_in_use[i] = true; return i; } } self.pool_cond.waitUncancelable(self.io, &self.pool_mutex); // releases mutex while parked }}
fn releaseRead(self: *LocalDb, idx: usize) void { self.pool_mutex.lockUncancelable(self.io); self.read_in_use[idx] = false; self.pool_mutex.unlock(self.io); self.pool_cond.signal(self.io);}
/// Begin a query on this thread's connection, acquiring one from the pool if/// this is the outermost query. Returns the connection to run on.fn beginQuery(self: *LocalDb) zqlite.Conn { if (tl_depth == 0) tl_conn_idx = self.acquireRead(); tl_depth += 1; return self.read_pool[tl_conn_idx.?].?;}
/// End a query; return this thread's connection to the pool when its last/// open Rows closes (depth 1→0).fn endQuery(self: *LocalDb) void { tl_depth -= 1; if (tl_depth == 0) { if (tl_conn_idx) |idx| { self.releaseRead(idx); tl_conn_idx = null; } }}
/// Execute a SELECT on this thread's pool connection (shared across the thread's/// nested/sibling queries, never across threads). The returned Rows holds the/// connection until deinit.pub fn query(self: *LocalDb, comptime sql: []const u8, args: anytype) !Rows { const span = logfire.span("db.local.query", .{ .sql = truncateSql(sql), }); defer span.end();
const c = self.beginQuery(); const rows = c.rows(sql, args) catch |e| { self.endQuery(); // unwind the depth bump; release if it was the outermost logfire.err("db.local.query failed: {s} | sql: {s}", .{ @errorName(e), truncateSql(sql) }); return e; }; return .{ .inner = rows, .db = self };}
/// Execute a statement (INSERT, UPDATE, DELETE)pub fn exec(self: *LocalDb, comptime sql: []const u8, args: anytype) !void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
const c = self.conn orelse return error.NotOpen; c.exec(sql, args) catch |e| { logfire.err("db.local.exec failed: {s} | sql: {s}", .{ @errorName(e), truncateSql(sql) }); return e; };}
/// Get raw connection for batch operations (caller must handle locking)pub fn getConn(self: *LocalDb) ?zqlite.Conn { return self.conn;}
/// Lock for batch operationspub fn lock(self: *LocalDb) void { self.mutex.lockUncancelable(self.io);}
/// Unlock after batch operationspub fn unlock(self: *LocalDb) void { self.mutex.unlock(self.io);}
fn truncateSql(sql: []const u8) []const u8 { const max_len = 100; return if (sql.len > max_len) sql[0..max_len] else sql;}
// remove a sqlite db file and its WAL/SHM sidecars (test helper)fn unlinkDbFiles(path: [:0]const u8) void { _ = std.c.unlink(path.ptr); var b: [80]u8 = undefined; inline for (.{ "-wal", "-shm" }) |sfx| { const z = std.mem.printSentinel(&b, "{s}{s}", .{ path, sfx }, 0) catch return; _ = std.c.unlink(z.ptr); }}
test "read pool: concurrent nested queries never deadlock and never leak" { // Real OS threads contending for the pool, each running the NESTED pattern // (outer rows held open while inner queries run) that deadlocked a per-query // checkout. With far more threads than connections, a hold-and-wait design // would wedge here forever; the thread-local reuse must let every thread // finish. (If the design regressed, this test HANGS — that is the signal.) var threaded = Io.Threaded.init(std.testing.allocator, .{}); defer threaded.deinit(); const tio = threaded.io();
// Per-process path + defer cleanup: never shares state with a concurrent // run or a prior crashed run, and tidies up even if an assertion fails. var path_buf: [64]u8 = undefined; const zpath = std.mem.printSentinel(&path_buf, "/tmp/leaflet-pool-{d}.db", .{std.c.getpid()}, 0) catch unreachable; unlinkDbFiles(zpath); defer unlinkDbFiles(zpath); { const w = try zqlite.open(zpath.ptr, zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); defer w.close(); try w.exec("CREATE TABLE t(x INTEGER)", .{}); try w.exec("INSERT INTO t(x) VALUES (1),(2),(3),(4),(5)", .{}); }
var ldb = LocalDb.init(std.testing.allocator, tio); for (&ldb.read_pool) |*slot| slot.* = try zqlite.open(zpath.ptr, zqlite.OpenFlags.ReadOnly); defer for (&ldb.read_pool) |*slot| { if (slot.*) |c| c.close(); slot.* = null; };
const Worker = struct { fn run(db_: *LocalDb, ok: *std.atomic.Value(u32)) void { var iter: usize = 0; while (iter < 25) : (iter += 1) { var outer = db_.query("SELECT x FROM t ORDER BY x", .{}) catch return; defer outer.deinit(); var count: usize = 0; while (outer.next()) |row| { const x = row.int(0); // nested query while `outer` is still open var inner = db_.query("SELECT x FROM t WHERE x = ?", .{x}) catch return; defer inner.deinit(); if (inner.next()) |r2| { if (r2.int(0) == x) count += 1; } } if (count != 5) return; // wrong result → leave ok unbumped, test fails } _ = ok.fetchAdd(1, .monotonic); } };
const N = 4 * READ_POOL_SIZE; // 4× more nesting threads than connections var ok = std.atomic.Value(u32).init(0); var threads: [N]std.Thread = undefined; for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Worker.run, .{ &ldb, &ok }); for (threads) |t| t.join();
try std.testing.expectEqual(@as(u32, N), ok.load(.monotonic)); // all finished correctly for (ldb.read_in_use) |u| try std.testing.expect(!u); // every connection returned}
test "adoptPending keeps previous snapshot and moves manifest sidecars" { const tio = std.Options.debug_io;
// Per-process dir + defer cleanup: hermetic against concurrent runs, prior // crashed runs, and assertion failures partway through this test. var dir_buf: [64]u8 = undefined; const dir = std.mem.printSentinel(&dir_buf, "/tmp/leaflet-adopt-{d}", .{std.c.getpid()}, 0) catch unreachable; _ = std.c.mkdir(dir.ptr, 0o755); const jp = struct { fn f(buf: []u8, d: []const u8, suffix: []const u8) [:0]const u8 { return std.mem.printSentinel(buf, "{s}/local.db{s}", .{ d, suffix }, 0) catch unreachable; } }.f; defer { var b: [160]u8 = undefined; inline for (.{ "", ".new", ".prev", ".manifest.json", ".new.manifest.json", ".prev.manifest.json" }) |sfx| { _ = std.c.unlink(jp(&b, dir, sfx).ptr); } _ = std.c.rmdir(dir.ptr); } var pb: [160]u8 = undefined; // paths are used sequentially, so one reused buffer is safe
const writeStr = struct { fn f(io_: Io, path: []const u8, content: []const u8) !void { const file = try Io.Dir.createFileAbsolute(io_, path, .{ .truncate = true }); defer file.close(io_); var wbuf: [64]u8 = undefined; var fw = Io.File.Writer.init(file, io_, &wbuf); try fw.interface.writeAll(content); try fw.interface.flush(); } }.f; const readStr = struct { fn f(io_: Io, path: []const u8, buf: []u8) ![]const u8 { const file = try Io.Dir.openFileAbsolute(io_, path, .{}); defer file.close(io_); const n = file.readStreaming(io_, &.{buf}) catch |err| switch (err) { error.EndOfStream => return buf[0..0], else => return err, }; return buf[0..n]; } }.f;
try writeStr(tio, jp(&pb, dir, ""), "OLD"); try writeStr(tio, jp(&pb, dir, ".manifest.json"), "{\"old\":1}"); try writeStr(tio, jp(&pb, dir, ".new"), "NEW"); try writeStr(tio, jp(&pb, dir, ".new.manifest.json"), "{\"new\":1}");
adoptPending(jp(&pb, dir, ""));
var buf: [64]u8 = undefined; try std.testing.expectEqualStrings("NEW", try readStr(tio, jp(&pb, dir, ""), &buf)); try std.testing.expectEqualStrings("OLD", try readStr(tio, jp(&pb, dir, ".prev"), &buf)); try std.testing.expectEqualStrings("{\"new\":1}", try readStr(tio, jp(&pb, dir, ".manifest.json"), &buf)); try std.testing.expectEqualStrings("{\"old\":1}", try readStr(tio, jp(&pb, dir, ".prev.manifest.json"), &buf));
// .new is consumed const fd = std.c.open(jp(&pb, dir, ".new").ptr, .{ .ACCMODE = .RDONLY }, @as(std.c.mode_t, 0)); try std.testing.expect(fd < 0);}