Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Zig
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438//! Index builder. Runs offline against Turso source-of-truth, produces a//! standalone SQLite snapshot DB at /data/build-<id>/index.db with the//! three immutable tables populated.//!//! Pipeline (single pass over actors, no extra memory for the corpus)://!//! 1. open index.db, init schema, set pragmas, ATTACH a scratch staging.db//! 2. stream rows from Turso actors (or a local export via INDEX_REUSE_SOURCE)//! 3. per row: pack actor_compact blob, write to actor_compact_base (index.db)//! normalize handle + display, derive prefix keys,//! emit (key, did, score) rows to stage.emissions (staging.db)//! 4. sort stage.emissions by (key, score DESC) via an index in staging.db//! 5. iterate sorted emissions, group by key in-stream, take tier-aware//! top-N, pack as did_list blob, write to prefix_top_base (index.db)//! 6. DETACH + delete staging.db, write index_meta, ANALYZE, close//!//! Staging lives in a SEPARATE attached DB, never in index.db. emissions//! (~144M rows, ~8GB) + its index (~3GB) are pure scratch; in index.db they//! ballooned the shipped artifact to ~12GB of unreclaimed free pages. Deleting//! the scratch file frees them with no VACUUM, so index.db ships at its real//! ~2GB size. The CREATE INDEX sort is file-backed (temp_store=FILE) and//! steered onto the build volume via SQLITE_TMPDIR, not held in RAM.//!//! No in-memory accumulation of emissions or top-N lists across keys —//! everything streams. Memory usage is bounded by one key's top-N//! candidate buffer (≤300 entries × ~32 bytes ≈ 10 KB at peak) plus the//! 200MB page cache — flat regardless of corpus size.
const std = @import("std");const builtin = @import("builtin");const zqlite = @import("zqlite");const Allocator = std.mem.Allocator;
const TursoClient = @import("../db/TursoClient.zig");const schema = @import("schema.zig");const normalize = @import("normalize.zig");const prefixes = @import("prefixes.zig");const pack = @import("pack.zig");const scoring = @import("scoring.zig");const manifest = @import("manifest.zig");const r2 = @import("r2.zig");
const log = std.log.scoped(.indexer);const io = std.Options.debug_io;
// std.c omits setenv in 0.16's curated libc list (getenv is present).// Declared here so we can steer SQLite's temp_store=FILE scratch onto the// build volume via SQLITE_TMPDIR.extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
// Linux glibc `struct statvfs` (64-bit): every field we touch is an// `unsigned long` (8 bytes). Used only on the Fly build VM (Linux); the// heartbeat reports free space so an operator never has to ssh + `df` to// tell a long phase apart from a dead one. macOS has a different layout,// so diskFreeBytes is a no-op off Linux (tests don't depend on it).const Statvfs = extern struct { f_bsize: c_ulong, f_frsize: c_ulong, f_blocks: c_ulong, f_bfree: c_ulong, f_bavail: c_ulong, f_files: c_ulong, f_ffree: c_ulong, f_favail: c_ulong, f_fsid: c_ulong, f_flag: c_ulong, f_namemax: c_ulong, __spare: [6]c_int,};extern "c" fn statvfs(path: [*:0]const u8, buf: *Statvfs) c_int;
/// Free bytes available on the filesystem holding `path`. 0 on any failure/// or off Linux. Cheap (one syscall) — safe to call on each heartbeat tick.fn diskFreeBytes(path: []const u8) u64 { if (builtin.os.tag != .linux) return 0; var zbuf: [256]u8 = undefined; if (path.len + 1 > zbuf.len) return 0; @memcpy(zbuf[0..path.len], path); zbuf[path.len] = 0; var st: Statvfs = undefined; if (statvfs(zbuf[0..path.len :0], &st) != 0) return 0; return @as(u64, st.f_bavail) *% @as(u64, st.f_frsize);}
/// Rows per Turso roundtrip during the streaming fetch. Matches the/// pattern in `sync.zig`'s bulk pagination. 2000 is small enough that/// each batch is ≤ ~1 MB JSON over the wire and decode is fast, but/// big enough to amortize roundtrip latency.const BATCH_SIZE: usize = 2000;
/// Source for the actor row stream. `.auto` picks the best available/// path. `export_http` (GET /info + /export/{gen} → local SQLite, then/// ATTACH) is the primary — ~60× faster than the v3 cursor for bulk/// reads. `hrana_v2_paginated` stays as a fallback for one cycle.pub const SourceMode = enum { auto, export_http, hrana_v2_paginated,};
pub const Config = struct { /// Filesystem root where build-<id>/ directories are created. /// Indexer machine's volume mount. build_root: []const u8 = "/data", /// 0 = unlimited (full corpus); >0 = LIMIT N actors (sampling). sample_limit: usize = 1000, /// Where the actor stream comes from. See SourceMode. source_mode: SourceMode = .auto, /// Top-N caps per prefix-length tier. top_n_short: usize = 300, // prefix len 1-3 top_n_medium: usize = 200, // prefix len 4-6 top_n_long: usize = 100, // prefix len 7+ /// Reported in index_meta; not enforced by builder. overlay_top_n: usize = 100, hydrate_cap: usize = 50, /// Stamped into index_meta as builder_version. builder_version: []const u8 = "dev",};
pub fn readConfigFromEnv() Config { return .{ .build_root = cGetenv("INDEX_BUILD_ROOT") orelse "/data", .sample_limit = envUsize("INDEX_SAMPLE_LIMIT", 1000), .source_mode = parseSourceMode(cGetenv("INDEX_SOURCE")), .top_n_short = envUsize("INDEX_TOP_N_SHORT", 300), .top_n_medium = envUsize("INDEX_TOP_N_MEDIUM", 200), .top_n_long = envUsize("INDEX_TOP_N_LONG", 100), .overlay_top_n = envUsize("INDEX_OVERLAY_TOP_N", 100), .hydrate_cap = envUsize("INDEX_HYDRATE_CAP", 50), .builder_version = cGetenv("INDEXER_VERSION") orelse "dev", };}
fn parseSourceMode(s: ?[]const u8) SourceMode { const v = s orelse return .auto; if (std.mem.eql(u8, v, "auto")) return .auto; if (std.mem.eql(u8, v, "export") or std.mem.eql(u8, v, "export_http")) return .export_http; if (std.mem.eql(u8, v, "hrana_v2") or std.mem.eql(u8, v, "hrana_v2_paginated")) return .hrana_v2_paginated; log.warn("INDEX_SOURCE='{s}' unrecognized; defaulting to auto", .{v}); return .auto;}
fn cGetenv(name: [*:0]const u8) ?[]const u8 { if (std.c.getenv(name)) |p| return std.mem.span(p); return null;}
fn envUsize(name: [*:0]const u8, default: usize) usize { const v = cGetenv(name) orelse return default; return std.fmt.parseInt(usize, v, 10) catch default;}
/// Truthy env flag: unset / "" / "0" / "false" / "no" → false, anything else/// → true. So `INDEX_ALLOW_PROD=0` actually leaves prod disarmed — a bare/// presence check would treat "0" as on and arm a prod write by surprise.fn envFlag(name: [*:0]const u8) bool { const v = cGetenv(name) orelse return false; if (v.len == 0 or std.mem.eql(u8, v, "0")) return false; if (std.ascii.eqlIgnoreCase(v, "false") or std.ascii.eqlIgnoreCase(v, "no")) return false; return true;}
const BuildStats = struct { /// Number of rows the builder advanced past in Turso. Used for the /// cursor (sample_limit, progress logs) and to know we've reached the /// watermark's tail. NOT the same as rows in actor_compact_base — /// hidden actors are skipped by the builder but still counted here so /// the cursor can advance past them. rows_seen: usize = 0, /// Number of rows ACTUALLY written to actor_compact_base. This is what /// the manifest contract calls `actor_count` and what /health should /// report. Excludes hidden actors and any rows where packActor failed. actor_count: usize = 0, emission_count: usize = 0, prefix_key_count: usize = 0, /// CJK n: keys emitted (any key with a non-ASCII byte) and actors that /// emitted ≥1. Manifest-reported; the promote gate floors them for /// v2-cjk snapshots so a CJK-blind build can't silently attach. cjk_key_count: usize = 0, cjk_actor_count: usize = 0, /// quality_score of every actor written, for the quantile summary. ~44 MB /// at 11M actors; the build box has headroom and this runs once a build. scores: std.ArrayListUnmanaged(f32) = .empty, build_start_ms: i64 = 0, build_end_ms: i64 = 0, /// max(updated_at) seen in Turso actors at build start. Persisted to /// index_meta so the serving layer can answer "how stale is this /// snapshot" without re-querying Turso. source_watermark: i64 = 0, phases: Phases = .{},};
/// Per-phase cumulative time. Bracketed via `nowMs()` at coarse boundaries/// (one bracket pair around each logical phase, not per-row) so the/// overhead of the timestamp syscall stays well under 1% of total build/// cost. The breakdown is what tells us whether build-time growth is/// driven by Turso fetch, emission inserts, or the group/pack pass.const Phases = struct { turso_fetch_ms: i64 = 0, derive_ms: i64 = 0, // normalize + deriveAll per actor (cumulative) actor_insert_ms: i64 = 0, // pack + INSERT actor_compact_base (cumulative) emission_insert_ms: i64 = 0, // INSERT emissions inner loop (cumulative) index_build_ms: i64 = 0, // CREATE INDEX idx_emissions_key_score group_pack_ms: i64 = 0, // buildPrefixTopBase whole pass meta_ms: i64 = 0, // writeMeta analyze_ms: i64 = 0, // ANALYZE manifest_ms: i64 = 0, // sha256 + write manifest.json};
fn nowMs() i64 { return @intCast(@divFloor(std.Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_ms));}
fn nowSeconds() i64 { return @intCast(@divFloor(std.Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s));}
/// Heartbeat cadence for the silent blocking phases. 30s ≈ 2 lines/min —/// enough to prove liveness across a tens-of-minutes phase without spamming.const HEARTBEAT_INTERVAL_S: u32 = 30;
/// Liveness logger for phases SQLite (or R2) runs as a single opaque/// blocking call — `CREATE INDEX` over ~250M emission rows, the multi-GB/// Turso export, the multi-GB index.db upload. The calling thread can't/// emit progress from inside those, so a side thread prints elapsed +/// watched-file size + free disk every tick until finish() joins it. This/// is what turns "is it alive?" (manual ssh + df) into a log line.////// Construct on the caller's stack, then start()/finish() around the/// blocking call — `self` must stay put for the spawned thread's lifetime,/// which the start..finish scope guarantees.const Heartbeat = struct { phase: []const u8, watch_path: []const u8, vol_path: []const u8, interval_s: u32 = HEARTBEAT_INTERVAL_S, start_ms: i64 = 0, stop: std.atomic.Value(bool) = .init(false), thread: ?std.Thread = null,
fn start(self: *Heartbeat) void { self.start_ms = nowMs(); self.stop.store(false, .release); log.info("phase start: {s}", .{self.phase}); self.thread = std.Thread.spawn(.{}, loop, .{self}) catch |err| blk: { log.warn("heartbeat spawn failed ({s}): {s}", .{ self.phase, @errorName(err) }); break :blk null; }; }
fn loop(self: *Heartbeat) void { while (true) { var slept: u32 = 0; while (slept < self.interval_s) : (slept += 1) { if (self.stop.load(.acquire)) return; io.sleep(.{ .nanoseconds = std.time.ns_per_s }, .real) catch return; } if (self.stop.load(.acquire)) return; self.emit("phase heartbeat"); } }
fn emit(self: *Heartbeat, label: []const u8) void { const elapsed_s = @divFloor(nowMs() - self.start_ms, std.time.ms_per_s); log.info("{s}: phase={s} elapsed_s={d} {s}={d}B disk_free={d}B", .{ label, self.phase, elapsed_s, std.fs.path.basename(self.watch_path), fileSize(self.watch_path), diskFreeBytes(self.vol_path), }); }
fn finish(self: *Heartbeat) void { self.stop.store(true, .release); if (self.thread) |t| { t.join(); self.thread = null; } self.emit("phase done"); }};
/// Snapshot publish channel. Default (unset INDEX_CHANNEL) is `local` — a/// build that forgets to declare its channel CANNOT touch prod's/// `latest.json`. See resolvePublish for the per-channel rules.const Channel = enum { local, staging, prod };
fn parseChannel(s: ?[]const u8) Channel { const v = s orelse return .local; if (std.mem.eql(u8, v, "local")) return .local; if (std.mem.eql(u8, v, "staging")) return .staging; if (std.mem.eql(u8, v, "prod")) return .prod; log.warn("INDEX_CHANNEL='{s}' unrecognized; defaulting to local (no upload)", .{v}); return .local;}
const Publish = struct { prefix: []const u8, latest_key: []const u8 };
/// Resolve where a build publishes, enforcing the safety rules. Pure (no/// env reads) so the rules are unit-testable:/// - local → never publishes (error.LocalNoPublish; caller gates first)/// - staging → requires a non-empty name; namespaced to/// `staging/<name>/` + `latest.<name>.json`/// - prod → `builds/` + `latest.json`, but ONLY when armed/// (allow_prod). Declaring the channel is not enough; prod/// writes must be explicitly armed so a scratch build can't/// clobber the prod pointer by accident./// staging prefix/latest are written into the caller-provided buffers.fn resolvePublish( channel: Channel, allow_prod: bool, staging_name: ?[]const u8, prefix_buf: []u8, latest_buf: []u8,) !Publish { switch (channel) { .local => return error.LocalNoPublish, .prod => { if (!allow_prod) return error.ProdNotArmed; return .{ .prefix = "builds/", .latest_key = "latest.json" }; }, .staging => { const name = staging_name orelse return error.StagingNameMissing; if (name.len == 0) return error.StagingNameMissing; return .{ .prefix = try std.fmt.bufPrint(prefix_buf, "staging/{s}/", .{name}), .latest_key = try std.fmt.bufPrint(latest_buf, "latest.{s}.json", .{name}), }; }, }}
/// SQLite single-quoted string literal: wrap in '' and double any/// embedded single quote. Writes into `buf`, returns the quoted slice/// (including the surrounding quotes). Defensive for paths interpolated/// into ATTACH — our build dirs don't contain quotes, but this is/// becoming general infrastructure.fn sqlQuote(buf: []u8, s: []const u8) ![]const u8 { var w: usize = 0; if (w >= buf.len) return error.PathTooLong; buf[w] = '\''; w += 1; for (s) |c| { if (c == '\'') { if (w + 2 > buf.len) return error.PathTooLong; buf[w] = '\''; buf[w + 1] = '\''; w += 2; } else { if (w + 1 > buf.len) return error.PathTooLong; buf[w] = c; w += 1; } } if (w >= buf.len) return error.PathTooLong; buf[w] = '\''; w += 1; return buf[0..w];}
/// Delete a build's source.db + sidecars. Called on a SUCCESSFUL export/// build to reclaim ~3+ GB. On failure we leave them so an operator can/// inspect / re-run with INDEX_REUSE_SOURCE.fn deleteSourceFiles(build_dir: []const u8) void { const names = [_][]const u8{ "source.db", "source.db-wal", "source.db-shm", "source.db.part" }; for (names) |name| { var path_buf: [320]u8 = undefined; const p = std.fmt.bufPrint(&path_buf, "{s}/{s}", .{ build_dir, name }) catch continue; std.Io.Dir.deleteFile(.cwd(), io, p) catch |err| switch (err) { error.FileNotFound => {}, else => log.warn("deleteSourceFiles({s}) failed: {s}", .{ p, @errorName(err) }), }; }}
/// Delete a build's staging.db + sidecars after DETACH. The scratch DB/// holds the ~11GB emissions table + its index; dropping the file frees it/// without a VACUUM rewrite of index.db.fn deleteStagingFiles(build_dir: []const u8) void { const names = [_][]const u8{ "staging.db", "staging.db-journal", "staging.db-wal", "staging.db-shm" }; for (names) |name| { var path_buf: [320]u8 = undefined; const p = std.fmt.bufPrint(&path_buf, "{s}/{s}", .{ build_dir, name }) catch continue; std.Io.Dir.deleteFile(.cwd(), io, p) catch |err| switch (err) { error.FileNotFound => {}, else => log.warn("deleteStagingFiles({s}) failed: {s}", .{ p, @errorName(err) }), }; }}
/// File size in bytes, 0 if it can't be opened. Used for build observability/// (final index.db size, peak staging footprint).fn fileSize(path: []const u8) u64 { const f = std.Io.Dir.openFileAbsolute(io, path, .{}) catch return 0; defer f.close(io); const st = f.stat(io) catch return 0; return st.size;}
/// Peak resident set size (VmHWM) from /proc/self/status, in kB. 0 when/// unavailable (macOS has no such file). The CREATE INDEX sort is the/// memory high-water mark, so this is the number that sizes the machine.fn peakRssKb() u64 { const f = std.Io.Dir.openFileAbsolute(io, "/proc/self/status", .{}) catch return 0; defer f.close(io); var buf: [4096]u8 = undefined; const n = f.readStreaming(io, &.{&buf}) catch return 0; const content = buf[0..n]; const idx = std.mem.indexOf(u8, content, "VmHWM:") orelse return 0; var rest = content[idx + "VmHWM:".len ..]; var i: usize = 0; while (i < rest.len and (rest[i] == ' ' or rest[i] == '\t')) : (i += 1) {} rest = rest[i..]; var j: usize = 0; while (j < rest.len and rest[j] >= '0' and rest[j] <= '9') : (j += 1) {} return std.fmt.parseInt(u64, rest[0..j], 10) catch 0;}
/// Fetch max(updated_at) from Turso actors. Used as source_watermark/// in index_meta. Cheap (single aggregate) compared to the streaming/// pass. Used by the hrana fallback path; the export path reads the/// watermark from the local snapshot DB instead.fn readSourceWatermark(turso: *TursoClient) !i64 { var result = try turso.query("SELECT COALESCE(MAX(updated_at), 0) FROM actors", &.{}); defer result.deinit(); if (result.rows.len == 0) return 0; return result.rows[0].int(0);}
/// Resolve `.auto` to the preferred source: export_http. v2 paginated/// stays available as an explicit fallback via INDEX_SOURCE=hrana_v2.fn chooseSourceMode(cfg: Config) SourceMode { return switch (cfg.source_mode) { .auto => .export_http, .export_http => .export_http, .hrana_v2_paginated => .hrana_v2_paginated, };}
/// Random 16-bit hex suffix appended to the unix-second timestamp in the/// build_id. Two builds starting in the same wall-clock second would/// otherwise collide and contaminate each other's per-build directory./// Crash recovery in the same second after a failed build is the more/// realistic case; the suffix makes it impossible to step on a stale/// dir even then.fn randomHex4(io_: std.Io, buf: *[4]u8) void { var raw: [2]u8 = undefined; io_.random(&raw); const hex = "0123456789abcdef"; buf[0] = hex[(raw[0] >> 4) & 0xF]; buf[1] = hex[raw[0] & 0xF]; buf[2] = hex[(raw[1] >> 4) & 0xF]; buf[3] = hex[raw[1] & 0xF];}
/// Top-level entry. Reads config from env, runs the full build, writes/// metadata, exits. Errors propagate so the caller (main.zig) can decide/// whether to retry/exit-fail.pub fn run(allocator: Allocator) !void { const cfg = readConfigFromEnv();
// Republish a build that is already on disk. // // uploadArtifacts writes the latest pointer LAST, so a failure part-way // leaves the artifact uploaded and unreferenced — the safe outcome, but // recovering it meant rebuilding the whole snapshot and re-uploading 12.4GB // over a home connection to fix a 300-byte manifest. The comment above the // failure path always claimed "a retry/manual upload can still promote it"; // this is that path. if (cGetenv("INDEX_PUBLISH_ONLY")) |dir| { const bid = std.fs.path.basename(dir); const build_id = if (std.mem.startsWith(u8, bid, "build-")) bid["build-".len..] else bid; const db_path = try std.fmt.allocPrint(allocator, "{s}/index.db", .{dir}); defer allocator.free(db_path); log.info("INDEX_PUBLISH_ONLY={s} — skipping build, republishing build_id={s}", .{ dir, build_id }); const channel = parseChannel(cGetenv("INDEX_CHANNEL")); return uploadArtifacts(allocator, channel, build_id, db_path, dir); }
var stats: BuildStats = .{ .build_start_ms = nowMs() }; defer stats.scores.deinit(allocator); log.info("indexer start: sample_limit={d} top_n=({d}/{d}/{d})", .{ cfg.sample_limit, cfg.top_n_short, cfg.top_n_medium, cfg.top_n_long, });
// build_id = "b<unix_seconds>-<16-bit hex random>". The random suffix // makes accidental same-second collisions (mostly a retry-after-crash // scenario) impossible; without it, the retry would land in an // existing dir with stale partial state and INSERT into emissions // without conflict resolution (duplicating rows). var id_buf: [32]u8 = undefined; var rand_buf: [4]u8 = undefined; randomHex4(io, &rand_buf); const build_id = try std.fmt.bufPrint(&id_buf, "b{d}-{s}", .{ nowSeconds(), rand_buf });
// Create the build directory atomically (mkdir -p semantics; ignore EEXIST). var dir_buf: [256]u8 = undefined; const build_dir = try std.fmt.bufPrint(&dir_buf, "{s}/build-{s}", .{ cfg.build_root, build_id }); std.Io.Dir.createDirPath(.cwd(), io, build_dir) catch |err| { log.err("createDirPath({s}) failed: {}", .{ build_dir, err }); return err; };
// Point SQLite's file-backed temp scratch (temp_store=FILE, set below) // at the build dir. build_dir lives on the Fly volume (build_root // defaults to /data), so the CREATE INDEX sort over ~144M emission rows // spills to the volume instead of the root fs / RAM. setenv copies the // value, so the stack buffer can go out of scope safely. { var tmp_z_buf: [260]u8 = undefined; if (build_dir.len + 1 <= tmp_z_buf.len) { @memcpy(tmp_z_buf[0..build_dir.len], build_dir); tmp_z_buf[build_dir.len] = 0; const tmp_z: [*:0]const u8 = tmp_z_buf[0..build_dir.len :0]; _ = setenv("SQLITE_TMPDIR", tmp_z, 1); _ = setenv("TMPDIR", tmp_z, 1); } }
var path_buf: [320]u8 = undefined; const db_path = try std.fmt.bufPrint(&path_buf, "{s}/index.db", .{build_dir}); var path_z_buf: [320]u8 = undefined; if (db_path.len + 1 > path_z_buf.len) return error.PathTooLong; @memcpy(path_z_buf[0..db_path.len], db_path); path_z_buf[db_path.len] = 0; const db_path_z: [*:0]const u8 = path_z_buf[0..db_path.len :0];
log.info("snapshot path: {s}", .{db_path});
const flags = zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite; var conn = try zqlite.open(db_path_z, flags); defer conn.close();
// Build-time pragmas. journal_mode=DELETE matters: the snapshot // artifact transport ships ONLY index.db (no .db-wal sidecar). WAL // mode would leave committed pages in index.db-wal until a checkpoint, // so the search service fetching just index.db would silently get an incomplete // snapshot. With DELETE, every COMMIT flushes through to the main // file and the rollback journal is removed at COMMIT end. // synchronous=OFF skips fsync — a half-built snapshot is discarded // anyway (manifest hasn't been promoted), so torn writes don't matter. _ = conn.exec("PRAGMA journal_mode = DELETE", .{}) catch {}; _ = conn.exec("PRAGMA synchronous = OFF", .{}) catch {}; _ = conn.exec("PRAGMA temp_store = FILE", .{}) catch {}; // sort scratch → SQLITE_TMPDIR (volume), not RAM _ = conn.exec("PRAGMA cache_size = -200000", .{}) catch {}; // 200MB page cache for the build only
try schema.initSnapshot(conn);
// Staging lives in a SEPARATE attached DB, never in index.db. The // emissions table (~144M rows, ~8GB) plus its (key, score DESC) index // (~3GB) are pure build scratch. Kept in index.db they ballooned the // shipped artifact to ~12GB of mostly-free pages — DROP TABLE without // VACUUM never reclaims them. With a scratch DB we DETACH + delete it // at the end: index.db only ever holds the real artifact (~2GB), and we // never pay a VACUUM rewrite. var staging_buf: [320]u8 = undefined; const staging_path = try std.fmt.bufPrint(&staging_buf, "{s}/staging.db", .{build_dir}); { var q_buf: [400]u8 = undefined; const q = try sqlQuote(&q_buf, staging_path); var sql_buf: [480]u8 = undefined; const attach_sql = try std.fmt.bufPrint(&sql_buf, "ATTACH DATABASE {s} AS stage", .{q}); try conn.exec(attach_sql, .{}); } // Scratch DB is throwaway: no rollback journal, no fsync. A crash just // restarts the whole build from a fresh dir, so durability is moot. _ = conn.exec("PRAGMA stage.journal_mode = OFF", .{}) catch {}; _ = conn.exec("PRAGMA stage.synchronous = OFF", .{}) catch {}; try conn.exec( \\CREATE TABLE stage.emissions ( \\ key TEXT NOT NULL, \\ did TEXT NOT NULL, \\ score REAL NOT NULL \\) , .{});
// ── stream actors + emit ────────────────────────────────────────────── // // Source dispatch. export_http is primary (GET /info + /export → // local SQLite checkpoint, ATTACH, iterate — ~60× faster than the // hrana cursor for bulk reads). hrana_v2_paginated is the fallback. const effective_mode = chooseSourceMode(cfg); log.info("source mode: {s}", .{@tagName(effective_mode)});
switch (effective_mode) { .export_http => { // Dev affordance: build from an existing local SQLite file, // no Turso creds or network needed. Useful for iterating on // the build phase without re-downloading 3+ GB. if (cGetenv("INDEX_REUSE_SOURCE")) |reuse| { log.info("INDEX_REUSE_SOURCE set — skipping export, using {s}", .{reuse}); try buildFromAttached(allocator, conn, reuse, cfg, &stats); } else { var turso = try TursoClient.init(allocator); defer turso.deinit(); try streamFromExport(allocator, &turso, conn, build_dir, cfg, &stats); } }, .hrana_v2_paginated => { var turso = try TursoClient.init(allocator); defer turso.deinit(); stats.source_watermark = readSourceWatermark(&turso) catch |err| blk: { log.warn("source_watermark fetch failed: {s}; defaulting to 0", .{@errorName(err)}); break :blk 0; }; log.info("source_watermark: {d}", .{stats.source_watermark}); try streamActorsAndEmit(allocator, &turso, conn, cfg, &stats); }, .auto => unreachable, // chooseSourceMode resolves auto }
// Source is fully consumed once emissions are staged (buildFromAttached // detaches src on return). Reclaim its ~3GB now, before the disk-heavy // CREATE INDEX, so peak volume usage stays lower. Skipped in reuse mode // — that file isn't ours to delete. if (effective_mode == .export_http and cGetenv("INDEX_REUSE_SOURCE") == null) { deleteSourceFiles(build_dir); }
// ── group, pack, write prefix_top_base (per-group scratch arena inside) ── { // CREATE INDEX is one opaque blocking call over ~250M rows — the // single longest silent phase. Heartbeat it so staging.db growth + // free disk are visible while it runs. var hb: Heartbeat = .{ .phase = "create_index.emissions", .watch_path = staging_path, .vol_path = cfg.build_root }; hb.start(); defer hb.finish(); const t = nowMs(); // did is in the index so the group/pack scan is index-ONLY (covering). // Without it, the (key,score) index orders the scan but SQLite must do // a rowid lookup into the ~8GB emissions table for did on every one of // ~144M rows — random I/O that thrashed the volume for hours at full // scale. Covering it turns the scan into a sequential index read. try conn.exec("CREATE INDEX IF NOT EXISTS stage.idx_emissions_key_score ON emissions(key, score DESC, did)", .{}); stats.phases.index_build_ms = nowMs() - t; } { const t = nowMs(); try buildPrefixTopBase(allocator, conn, cfg, &stats); stats.phases.group_pack_ms = nowMs() - t; }
// ── cleanup + metadata ──────────────────────────────────────────────── // Peak disk footprint: staging (emissions+index) + index.db coexist here, // just before we drop staging. Logged for machine/volume sizing. log.info("peak artifact footprint: staging.db={d}B index.db={d}B", .{ fileSize(staging_path), fileSize(db_path) }); // Discard staging entirely: DETACH then delete the file. Frees the // ~11GB of emissions+index instantly with no VACUUM rewrite, and // guarantees index.db ships at real-artifact size. conn.exec("DETACH DATABASE stage", .{}) catch |err| { log.warn("DETACH stage failed: {s}", .{@errorName(err)}); }; deleteStagingFiles(build_dir); { const t = nowMs(); try writeMeta(allocator, conn, build_id, cfg, &stats); stats.phases.meta_ms = nowMs() - t; }
{ const t = nowMs(); _ = conn.exec("ANALYZE", .{}) catch {}; stats.phases.analyze_ms = nowMs() - t; } // No VACUUM needed: staging never lived in index.db, so there are no // freed pages to reclaim — index.db is already real-artifact size.
// Manifest LAST: index.db is final + ANALYZE'd. Streams its sha256 (the // ~30-60s pass over the artifact is counted in elapsed). The search service reads this // to gate promotion before downloading. R2 upload (Phase 2) consumes it. log.info("phase start: manifest sha256 + write ({d}B artifact)", .{fileSize(db_path)}); { const t = nowMs(); const m = try manifest.write(allocator, .{ .build_id = build_id, .db_path = db_path, .build_dir = build_dir, .promoted_at = nowSeconds(), .source_watermark = stats.source_watermark, .actor_count = stats.actor_count, .prefix_count = stats.prefix_key_count, .top_n_short = cfg.top_n_short, .top_n_medium = cfg.top_n_medium, .top_n_long = cfg.top_n_long, .overlay_top_n = cfg.overlay_top_n, .hydrate_cap = cfg.hydrate_cap, .builder_version = cfg.builder_version, .cjk_key_count = stats.cjk_key_count, .cjk_actor_count = stats.cjk_actor_count, }); stats.phases.manifest_ms = nowMs() - t; log.info("manifest written: byte_size={d} sha256={s}", .{ m.byte_size, m.sha256_hex[0..16] }); }
stats.build_end_ms = nowMs(); const elapsed_ms = stats.build_end_ms - stats.build_start_ms; const p = stats.phases; log.info("indexer done: build_id={s} actors={d} emissions={d} prefix_keys={d} cjk_keys={d} cjk_actors={d} elapsed_ms={d}", .{ build_id, stats.actor_count, stats.emission_count, stats.prefix_key_count, stats.cjk_key_count, stats.cjk_actor_count, elapsed_ms, }); log.info( "phase ms: turso_fetch={d} derive={d} actor_insert={d} emission_insert={d} index_build={d} group_pack={d} meta={d} analyze={d} manifest={d}", .{ p.turso_fetch_ms, p.derive_ms, p.actor_insert_ms, p.emission_insert_ms, p.index_build_ms, p.group_pack_ms, p.meta_ms, p.analyze_ms, p.manifest_ms }, ); log.info("indexer resources: index_db_bytes={d} peak_rss_kb={d}", .{ fileSize(db_path), peakRssKb() });
// Publish to R2 per INDEX_CHANNEL. `local` (the default) never uploads, // so local/sample/dry builds — and any build that forgets to declare a // channel — leave prod untouched. On failure we log loudly but keep the // on-disk build; a retry/manual upload can still promote it. const channel = parseChannel(cGetenv("INDEX_CHANNEL")); switch (channel) { .local => log.info("channel=local — R2 upload skipped; build on disk at {s}", .{build_dir}), // Propagate, do not swallow. // // This used to `catch |err| log.err(...)` and fall through, so the // process exited 0 and the orchestrator recorded a Completed run. On // 2026-07-31 a 12.4GB index.db uploaded fine and the tiny manifest.json // 403'd, and the flow reported success — a build that never promoted, // indistinguishable from one that did. The build is still retained on // disk for INDEX_PUBLISH_ONLY; what changes is that the run now fails. .staging, .prod => uploadArtifacts(allocator, channel, build_id, db_path, build_dir) catch |err| { log.err("R2 upload failed ({s}); build retained on disk at {s} — " ++ "republish with INDEX_PUBLISH_ONLY={s}", .{ @errorName(err), build_dir, build_dir }); return err; }, }}
/// Upload index.db + manifest.json under the channel's prefix, then copy the/// manifest to the latest key LAST — so the search service never sees a/// latest pointing at an index.db that isn't fully present.////// The publish target (and the prod-arming / staging-name rules) come from/// resolvePublish. prod search polls `latest.json` under `builds/`; only a/// `prod` channel armed with INDEX_ALLOW_PROD can write there.fn uploadArtifacts(gpa: Allocator, channel: Channel, build_id: []const u8, db_path: []const u8, build_dir: []const u8) !void { var prefix_buf: [160]u8 = undefined; var latest_buf: [160]u8 = undefined; const target = resolvePublish(channel, envFlag("INDEX_ALLOW_PROD"), cGetenv("INDEX_STAGING_NAME"), &prefix_buf, &latest_buf) catch |err| { switch (err) { error.ProdNotArmed => log.err("channel=prod but INDEX_ALLOW_PROD unset — refusing to overwrite prod latest.json. Build on disk at {s}; rerun with INDEX_ALLOW_PROD=1 to publish.", .{build_dir}), error.StagingNameMissing => log.err("channel=staging requires a non-empty INDEX_STAGING_NAME (e.g. cjk-search) — refusing to publish. Build on disk at {s}.", .{build_dir}), else => log.err("publish target unresolved ({s}); build on disk at {s}", .{ @errorName(err), build_dir }), } return err; }; const prefix = target.prefix; const latest_key = target.latest_key; log.info("publish channel={s} prefix={s} latest_key={s}", .{ @tagName(channel), prefix, latest_key });
const cfg = try r2.configure(gpa);
var kbuf: [320]u8 = undefined; const db_key = try std.fmt.bufPrint(&kbuf, "{s}{s}/index.db", .{ prefix, build_id }); { // index.db is multi-GB; the upload is one blocking call. Heartbeat it. var hb: Heartbeat = .{ .phase = "r2.upload.index_db", .watch_path = db_path, .vol_path = build_dir }; hb.start(); defer hb.finish(); try r2.upload(gpa, cfg, db_path, db_key); }
var mpath_buf: [320]u8 = undefined; const manifest_local = try std.fmt.bufPrint(&mpath_buf, "{s}/manifest.json", .{build_dir}); var mkey_buf: [320]u8 = undefined; const m_key = try std.fmt.bufPrint(&mkey_buf, "{s}{s}/manifest.json", .{ prefix, build_id }); try r2.upload(gpa, cfg, manifest_local, m_key);
try r2.upload(gpa, cfg, manifest_local, latest_key); log.info("R2 publish complete: channel={s} {s} -> {s}{s}/index.db", .{ @tagName(channel), latest_key, prefix, build_id });}
/// Primary source path. Export Turso → local SQLite checkpoint file via/// the export HTTP endpoints (TursoClient.exportToFile), ATTACH it/// read-only, validate, then iterate `src.actors` from local disk./// source_watermark comes from the exported file's own MAX(updated_at)/// — the exact watermark of this checkpoint snapshot. The exported file/// is deleted after a successful build.fn streamFromExport( gpa: Allocator, turso: *TursoClient, snapshot: zqlite.Conn, build_dir: []const u8, cfg: Config, stats: *BuildStats,) !void { var path_buf: [320]u8 = undefined; const source_path = try std.fmt.bufPrint(&path_buf, "{s}/source.db", .{build_dir});
const t_export = nowMs(); { // Multi-GB download, one blocking call. Heartbeat the falling free // disk so a stalled export is distinguishable from a slow one. var hb: Heartbeat = .{ .phase = "turso.export", .watch_path = source_path, .vol_path = build_dir }; hb.start(); defer hb.finish(); try turso.exportToFile(source_path); } stats.phases.turso_fetch_ms = nowMs() - t_export; log.info("export complete in {d}ms → {s}", .{ stats.phases.turso_fetch_ms, source_path });
try buildFromAttached(gpa, snapshot, source_path, cfg, stats);}
/// ATTACH a local SQLite file as `src`, validate it, then iterate/// `src.actors` into the snapshot. Split out so the export path and the/// INDEX_REUSE_SOURCE dev path share the build logic.fn buildFromAttached( gpa: Allocator, snapshot: zqlite.Conn, source_path: []const u8, cfg: Config, stats: *BuildStats,) !void { // Plain-path ATTACH (NOT a file: URI). The snapshot connection isn't // opened with SQLITE_OPEN_URI, so a 'file:...?mode=ro' string would // be treated as a literal filename and fail. We never write to src // (all writes target the main snapshot tables), so read-write attach // is harmless. Path is SQL-quoted (single quotes doubled) so it's // safe even if a build dir ever contains one. var quoted_buf: [512]u8 = undefined; const quoted_path = try sqlQuote("ed_buf, source_path); var attach_buf: [600]u8 = undefined; const attach_sql = try std.fmt.bufPrint(&attach_buf, "ATTACH DATABASE {s} AS src", .{quoted_path}); try snapshot.exec(attach_sql, .{}); defer snapshot.exec("DETACH DATABASE src", .{}) catch {};
try validateExportedDb(snapshot);
// source_watermark from the exported file — the checkpoint's own cutoff. { const row = try snapshot.row("SELECT COALESCE(MAX(updated_at), 0) FROM src.actors", .{}); if (row) |r| { defer r.deinit(); stats.source_watermark = r.int(0); } log.info("source_watermark from exported DB: {d}", .{stats.source_watermark}); }
var sctx = StreamCtx{ .gpa = gpa, .stats = stats, .row_arena = std.heap.ArenaAllocator.init(gpa), .stmt_actor = undefined, .stmt_emit = undefined, }; defer sctx.row_arena.deinit();
sctx.stmt_actor = try snapshot.prepare( "INSERT OR REPLACE INTO actor_compact_base (did, blob) VALUES (?, ?)", ); defer sctx.stmt_actor.deinit(); sctx.stmt_emit = try snapshot.prepare( "INSERT INTO stage.emissions (key, did, score) VALUES (?, ?, ?)", ); defer sctx.stmt_emit.deinit();
try snapshot.exec("BEGIN", .{}); errdefer _ = snapshot.exec("ROLLBACK", .{}) catch {};
var rows = if (cfg.sample_limit == 0) try snapshot.rows("SELECT did, handle, display_name, avatar_url, labels, pds, created_at, associated, hidden, quality_score FROM src.actors WHERE handle != '' ORDER BY did", .{}) else try snapshot.rows("SELECT did, handle, display_name, avatar_url, labels, pds, created_at, associated, hidden, quality_score FROM src.actors WHERE handle != '' ORDER BY did LIMIT ?", .{cfg.sample_limit}); defer rows.deinit();
while (rows.next()) |row| { try processActorRow(&sctx, row); stats.rows_seen += 1; if (stats.rows_seen - sctx.progress_at >= 50_000) { sctx.progress_at = stats.rows_seen; log.info("indexer progress (export): rows_seen={d} actors={d} emissions={d}", .{ stats.rows_seen, stats.actor_count, stats.emission_count, }); } }
try snapshot.exec("COMMIT", .{}); log.info("emissions phase done (export): rows_seen={d} actors={d} emissions={d} watermark={d}", .{ stats.rows_seen, stats.actor_count, stats.emission_count, stats.source_watermark, });}
/// Validate the exported DB before building from it. Cheap structural/// checks: quick_check (catches a torn/garbage file — belt-and-suspenders/// on top of the .part→rename guard) and a plausible actor count.fn validateExportedDb(snapshot: zqlite.Conn) !void { { const row = snapshot.row("PRAGMA src.quick_check", .{}) catch |err| { log.err("exported DB quick_check errored: {s}", .{@errorName(err)}); return error.ExportValidationFailed; }; if (row) |r| { defer r.deinit(); const result = r.text(0); if (!std.mem.eql(u8, result, "ok")) { log.err("exported DB quick_check: {s}", .{result}); return error.ExportCorrupt; } } else return error.ExportValidationFailed; } { const row = try snapshot.row("SELECT COUNT(*) FROM src.actors WHERE handle != ''", .{}); if (row) |r| { defer r.deinit(); const count = r.int(0); if (count < 1) { log.err("exported DB has no searchable actors", .{}); return error.ExportEmptyActors; } log.info("exported DB validated: {d} searchable actors", .{count}); } else return error.ExportEmptyActors; }}
/// Per-stream state threaded through processActorRow.const StreamCtx = struct { gpa: Allocator, stats: *BuildStats, row_arena: std.heap.ArenaAllocator, stmt_actor: zqlite.Stmt, stmt_emit: zqlite.Stmt, progress_at: usize = 0,};
/// Process one actor row. Per-row arena reset keeps memory flat. `row`/// is `anytype` so this works for both zqlite rows (export path, reading/// the ATTACHed file) and any other row type exposing text/int/float.fn processActorRow(ctx: *StreamCtx, row: anytype) !void { defer _ = ctx.row_arena.reset(.retain_capacity); const aa = ctx.row_arena.allocator();
const did = row.text(0); if (did.len == 0) return;
const raw_handle = row.text(1); const raw_display = row.text(2); const avatar = row.text(3); const labels_json = row.text(4); const pds = row.text(5); const created_at = row.text(6); const associated_json = row.text(7); const hidden = row.int(8) != 0; const quality_score = row.float(9);
if (hidden) return;
const t_actor = nowMs(); const blob = pack.packActor(aa, .{ .hidden = false, .quality_score = @floatCast(quality_score), .handle = raw_handle, .display = raw_display, .avatar = avatar, .labels = if (labels_json.len == 0) "[]" else labels_json, .pds = pds, .created_at = created_at, .associated = if (associated_json.len == 0) "{}" else associated_json, }) catch |err| { log.warn("packActor failed for {s}: {s}", .{ did, @errorName(err) }); return; }; try ctx.stmt_actor.reset(); try ctx.stmt_actor.bindValue(did, 0); try ctx.stmt_actor.bindValue(zqlite.blob(blob), 1); try ctx.stmt_actor.stepToCompletion(); ctx.stats.phases.actor_insert_ms += nowMs() - t_actor; ctx.stats.actor_count += 1; try ctx.stats.scores.append(ctx.gpa, @floatCast(quality_score));
const t_derive = nowMs(); const norm_handle = try normalize.handle(aa, raw_handle); const norm_display = try normalize.display(aa, raw_display); const keys = try prefixes.deriveAllUnique(aa, norm_handle, norm_display); ctx.stats.phases.derive_ms += nowMs() - t_derive;
const score: f32 = scoring.scoreOf(.{ .quality_score = @floatCast(quality_score) });
const t_emit = nowMs(); var actor_has_cjk = false; for (keys) |k| { try ctx.stmt_emit.reset(); try ctx.stmt_emit.bindValue(k, 0); try ctx.stmt_emit.bindValue(did, 1); try ctx.stmt_emit.bindValue(@as(f64, @floatCast(score)), 2); try ctx.stmt_emit.stepToCompletion(); ctx.stats.emission_count += 1; // CJK keys (display-name bigrams) carry non-ASCII bytes; ASCII h:/n:/ // e:/d: keys never do. Count them for the manifest + promote floor. for (k) |b| { if (b >= 0x80) { ctx.stats.cjk_key_count += 1; actor_has_cjk = true; break; } } } if (actor_has_cjk) ctx.stats.cjk_actor_count += 1; ctx.stats.phases.emission_insert_ms += nowMs() - t_emit;}
/// gpa is used for long-lived allocations (the per-actor scratch arena's/// backing, the last-DID cursor buffer). The per-actor scratch arena/// resets after each row so peak memory stays flat at full corpus scale/// instead of growing with N actors.////// Pagination: keyset cursor on `did` with batch size BATCH_SIZE. Each/// batch is a fresh Turso roundtrip + a fresh local transaction, so a/// crash mid-build loses at most one batch of work. The/// `updated_at <= source_watermark` filter pins the snapshot to a/// consistent point-in-time — actors written after the watermark are/// excluded from this build and pick up in the next one. Without this/// filter, an actor renamed during pagination could appear in the/// snapshot under both old and new handles.fn streamActorsAndEmit( gpa: Allocator, turso: *TursoClient, snapshot: zqlite.Conn, cfg: Config, stats: *BuildStats,) !void { const stmt_actor = try snapshot.prepare( "INSERT OR REPLACE INTO actor_compact_base (did, blob) VALUES (?, ?)", ); defer stmt_actor.deinit();
const stmt_emit = try snapshot.prepare( "INSERT INTO stage.emissions (key, did, score) VALUES (?, ?, ?)", ); defer stmt_emit.deinit();
// Per-row scratch arena lives across batches; reset per actor inside // the inner loop. var row_arena = std.heap.ArenaAllocator.init(gpa); defer row_arena.deinit(); const aa = row_arena.allocator();
// GPA-backed buffer for the cursor: row.text() slices borrow from // turso's parsed JSON, which is freed at result.deinit() each batch. // We dupe the last DID after each batch so the next iteration's // `did > ?` bind has a valid pointer. var last_did_buf: std.ArrayListUnmanaged(u8) = .empty; defer last_did_buf.deinit(gpa);
var watermark_buf: [24]u8 = undefined; const watermark_str = try std.fmt.bufPrint(&watermark_buf, "{d}", .{stats.source_watermark});
var batch_idx: usize = 0; paginate: while (true) { // Honor cfg.sample_limit by capping this batch's LIMIT. const remaining: usize = if (cfg.sample_limit == 0) BATCH_SIZE else if (stats.rows_seen >= cfg.sample_limit) break :paginate else @min(BATCH_SIZE, cfg.sample_limit - stats.rows_seen);
var limit_buf: [16]u8 = undefined; const limit_str = try std.fmt.bufPrint(&limit_buf, "{d}", .{remaining});
const t_fetch = nowMs(); var result = try turso.query( "SELECT did, handle, display_name, avatar_url, labels, pds, created_at, associated, hidden, quality_score FROM actors WHERE handle != '' AND updated_at <= ? AND did > ? ORDER BY did LIMIT ?", &.{ watermark_str, last_did_buf.items, limit_str }, ); stats.phases.turso_fetch_ms += nowMs() - t_fetch;
if (result.rows.len == 0) { result.deinit(); break :paginate; }
try snapshot.exec("BEGIN", .{}); errdefer _ = snapshot.exec("ROLLBACK", .{}) catch {};
var batch_last_did: []const u8 = "";
for (result.rows) |row| { defer _ = row_arena.reset(.retain_capacity);
const did = row.text(0); if (did.len == 0) continue; batch_last_did = did;
const raw_handle = row.text(1); const raw_display = row.text(2); const avatar = row.text(3); const labels_json = row.text(4); const pds = row.text(5); const created_at = row.text(6); const associated_json = row.text(7); const hidden = row.int(8) != 0; const quality_score = row.float(9);
// rows_seen advances for cursor / sample_limit / progress; // actor_count advances only after we actually write the row. stats.rows_seen += 1;
if (hidden) continue;
// 1. pack + insert actor blob const t_actor = nowMs(); const blob = pack.packActor(aa, .{ .hidden = false, .quality_score = @floatCast(quality_score), .handle = raw_handle, .display = raw_display, .avatar = avatar, .labels = if (labels_json.len == 0) "[]" else labels_json, .pds = pds, .created_at = created_at, .associated = if (associated_json.len == 0) "{}" else associated_json, }) catch |err| { log.warn("packActor failed for {s}: {s}", .{ did, @errorName(err) }); continue; }; try stmt_actor.reset(); try stmt_actor.bindValue(did, 0); try stmt_actor.bindValue(zqlite.blob(blob), 1); try stmt_actor.stepToCompletion(); stats.phases.actor_insert_ms += nowMs() - t_actor; stats.actor_count += 1; // bumps ONLY after the actor row is in the snapshot DB
// 2. derive prefix keys, deduped (see commit 36d969f rationale) const t_derive = nowMs(); const norm_handle = try normalize.handle(aa, raw_handle); const norm_display = try normalize.display(aa, raw_display); const keys = try prefixes.deriveAllUnique(aa, norm_handle, norm_display); stats.phases.derive_ms += nowMs() - t_derive;
const score: f32 = scoring.scoreOf(.{ .quality_score = @floatCast(quality_score) });
const t_emit = nowMs(); for (keys) |k| { try stmt_emit.reset(); try stmt_emit.bindValue(k, 0); try stmt_emit.bindValue(did, 1); try stmt_emit.bindValue(@as(f64, @floatCast(score)), 2); try stmt_emit.stepToCompletion(); stats.emission_count += 1; } stats.phases.emission_insert_ms += nowMs() - t_emit; }
// Capture rows-len for the short-batch check before deinit // invalidates the slice header. const this_batch_rows = result.rows.len;
// Save the last DID into gpa-owned memory BEFORE result.deinit // invalidates the slice. Cursor for the next iteration. last_did_buf.clearRetainingCapacity(); try last_did_buf.appendSlice(gpa, batch_last_did);
result.deinit(); try snapshot.exec("COMMIT", .{});
batch_idx += 1; log.info("indexer batch {d}: rows_seen={d} actors={d} emissions={d} cursor_did={s}", .{ batch_idx, stats.rows_seen, stats.actor_count, stats.emission_count, last_did_buf.items[0..@min(40, last_did_buf.items.len)], });
// Short-batch sentinel: if Turso returned fewer rows than asked // for, we've reached the watermark's tail. No need for another // roundtrip just to learn that. if (this_batch_rows < remaining) break :paginate; }
log.info("emissions phase done: rows_seen={d} actors={d} emissions={d}", .{ stats.rows_seen, stats.actor_count, stats.emission_count, });}
fn topNFor(key_len: usize, cfg: Config) usize { const inner_len = if (key_len >= 2) key_len - 2 else 0; // strip "h:"/"n:"/"e:" prefix if (inner_len <= 3) return cfg.top_n_short; if (inner_len <= 6) return cfg.top_n_medium; return cfg.top_n_long;}
fn buildPrefixTopBase( gpa: Allocator, snapshot: zqlite.Conn, cfg: Config, stats: *BuildStats,) !void { log.info("building prefix_top_base via grouped scan…", .{});
try snapshot.exec("BEGIN", .{}); errdefer _ = snapshot.exec("ROLLBACK", .{}) catch {};
const stmt_insert = try snapshot.prepare( "INSERT OR REPLACE INTO prefix_top_base (key, dids) VALUES (?, ?)", ); defer stmt_insert.deinit();
// Sequential scan in (key, score DESC) order. idx_emissions_key_score is // a COVERING index (key, score DESC, did) so this is index-only: one // ordered pass over the index, no per-row table lookups, no temp B-tree. var rows = try snapshot.rows( "SELECT key, did, score FROM stage.emissions ORDER BY key, score DESC", .{}, ); defer rows.deinit();
// Per-group scratch arena. Holds the current key's duped string, the // top-N DidEntry items, and the packed blob. Reset on each key change // after the previous group has been flushed to the DB — without this, // per-key allocations accumulate across the full key space (1.6M at // 100k actors → 10M+ at 5.9M) and the build OOMs late in the pass. var group_arena = std.heap.ArenaAllocator.init(gpa); defer group_arena.deinit(); const aa = group_arena.allocator();
var current_key: ?[]const u8 = null; var current_top: std.ArrayListUnmanaged(pack.DidEntry) = .empty; var current_top_n: usize = 0;
// Progress logging: this scan is the longest phase (tens of minutes at // full scale on shared CPU) and would otherwise sit silent. Log every // 10M rows against the known emission total. const total_emissions = stats.emission_count; var rows_scanned: u64 = 0; var progress_at: u64 = 0;
while (rows.next()) |row| { const key = row.text(0); const did = row.text(1); const score = row.float(2);
rows_scanned += 1; if (rows_scanned - progress_at >= 10_000_000) { progress_at = rows_scanned; log.info("group/pack progress: scanned={d}/{d} keys={d}", .{ rows_scanned, total_emissions, stats.prefix_key_count }); }
if (current_key == null or !std.mem.eql(u8, current_key.?, key)) { // flush previous key group, THEN reset the arena — writeGroup // binds the packed blob with SQLITE_STATIC, so the blob must // stay valid through stepToCompletion which lives inside it. if (current_key) |k| { try writeGroup(aa, &stmt_insert, k, current_top.items); _ = group_arena.reset(.retain_capacity); current_top = .empty; } // copy the new key into arena memory so it survives across iterations current_key = try aa.dupe(u8, key); current_top_n = topNFor(current_key.?.len, cfg); stats.prefix_key_count += 1; }
if (current_top.items.len < current_top_n) { try current_top.append(aa, .{ .did = try aa.dupe(u8, did), .score = @floatCast(score), }); } // else: cap reached, skip the rest of this key's tail }
// flush trailing group if (current_key) |k| try writeGroup(aa, &stmt_insert, k, current_top.items);
try snapshot.exec("COMMIT", .{}); log.info("prefix_top_base done: distinct_keys={d}", .{stats.prefix_key_count});}
fn writeGroup( aa: Allocator, stmt_insert: *const zqlite.Stmt, key: []const u8, top: []const pack.DidEntry,) !void { const blob = try pack.packDidList(aa, top); try stmt_insert.reset(); try stmt_insert.bindValue(key, 0); try stmt_insert.bindValue(zqlite.blob(blob), 1); try stmt_insert.stepToCompletion();}
fn writeMeta( gpa: Allocator, conn: zqlite.Conn, build_id: []const u8, cfg: Config, stats: *BuildStats,) !void { var buf: [64]u8 = undefined;
try schema.setMeta(conn, schema.meta.build_id, build_id); try schema.setMeta(conn, schema.meta.built_at, try std.fmt.bufPrint(&buf, "{d}", .{nowSeconds()})); try schema.setMeta(conn, schema.meta.source_watermark, try std.fmt.bufPrint(&buf, "{d}", .{stats.source_watermark})); try schema.setMeta(conn, schema.meta.builder_version, cfg.builder_version); try schema.setMeta(conn, schema.meta.normalize_version, normalize.VERSION); try schema.setMeta(conn, schema.meta.scoring_version, scoring.VERSION); try schema.setMeta(conn, schema.meta.top_n_short, try std.fmt.bufPrint(&buf, "{d}", .{cfg.top_n_short})); try schema.setMeta(conn, schema.meta.top_n_medium, try std.fmt.bufPrint(&buf, "{d}", .{cfg.top_n_medium})); try schema.setMeta(conn, schema.meta.top_n_long, try std.fmt.bufPrint(&buf, "{d}", .{cfg.top_n_long})); try schema.setMeta(conn, schema.meta.overlay_top_n, try std.fmt.bufPrint(&buf, "{d}", .{cfg.overlay_top_n})); try schema.setMeta(conn, schema.meta.hydrate_cap, try std.fmt.bufPrint(&buf, "{d}", .{cfg.hydrate_cap})); try schema.setMeta(conn, schema.meta.actor_count, try std.fmt.bufPrint(&buf, "{d}", .{stats.actor_count})); try schema.setMeta(conn, schema.meta.prefix_count, try std.fmt.bufPrint(&buf, "{d}", .{stats.prefix_key_count})); try schema.setMeta(conn, schema.meta.cjk_key_count, try std.fmt.bufPrint(&buf, "{d}", .{stats.cjk_key_count})); try schema.setMeta(conn, schema.meta.cjk_actor_count, try std.fmt.bufPrint(&buf, "{d}", .{stats.cjk_actor_count}));
const quantiles_json = try scoreQuantilesJson(gpa, stats.scores.items); defer gpa.free(quantiles_json); try schema.setMeta(conn, schema.meta.score_quantiles, quantiles_json);}
/// p0..p100 of `scores` as a JSON array (sorted in place). Empty input gives/// `[]`, which the reader treats as "no percentile available".fn scoreQuantilesJson(allocator: std.mem.Allocator, scores: []f32) ![]u8 { std.mem.sort(f32, scores, {}, std.sort.asc(f32)); var out: std.Io.Writer.Allocating = .init(allocator); errdefer out.deinit(); try out.writer.writeByte('['); if (scores.len > 0) { var p: usize = 0; while (p <= 100) : (p += 1) { const idx = @min(scores.len - 1, (scores.len - 1) * p / 100); if (p > 0) try out.writer.writeByte(','); try out.writer.print("{d:.2}", .{scores[idx]}); } } try out.writer.writeByte(']'); return out.toOwnedSlice();}
test "scoreQuantilesJson: 101 points, monotone, empty is []" { const t = std.testing; var scores = [_]f32{ 5, 1, 4, 2, 3 }; const json = try scoreQuantilesJson(t.allocator, &scores); defer t.allocator.free(json); try t.expect(std.mem.startsWith(u8, json, "[1.00,")); try t.expect(std.mem.endsWith(u8, json, ",5.00]")); try t.expectEqual(@as(usize, 101), std.mem.count(u8, json, ",") + 1); var none = [_]f32{}; const empty = try scoreQuantilesJson(t.allocator, &none); defer t.allocator.free(empty); try t.expectEqualStrings("[]", empty);}
// ─── tests ───────────────────────────────────────────────────────────────
const testing = std.testing;
test "topNFor classifies by prefix-text length" { const cfg: Config = .{}; // Keys are "h:abc" / "n:foo" / "e:bar" — strip the 2-byte ns prefix. try testing.expectEqual(cfg.top_n_short, topNFor("h:a".len, cfg)); try testing.expectEqual(cfg.top_n_short, topNFor("h:abc".len, cfg)); try testing.expectEqual(cfg.top_n_medium, topNFor("h:abcd".len, cfg)); try testing.expectEqual(cfg.top_n_medium, topNFor("h:abcdef".len, cfg)); try testing.expectEqual(cfg.top_n_long, topNFor("h:abcdefg".len, cfg)); try testing.expectEqual(cfg.top_n_long, topNFor("h:abcdefghij".len, cfg));}
test "Config defaults reflect design doc" { const cfg: Config = .{}; try testing.expectEqual(@as(usize, 300), cfg.top_n_short); try testing.expectEqual(@as(usize, 200), cfg.top_n_medium); try testing.expectEqual(@as(usize, 100), cfg.top_n_long); try testing.expectEqual(@as(usize, 100), cfg.overlay_top_n); try testing.expectEqual(@as(usize, 50), cfg.hydrate_cap); try testing.expectEqualStrings("/data", cfg.build_root);}
// Regression guard: the group/pack scan MUST be index-only. The index has to// include `did` (the third selected column) or SQLite does a rowid lookup// into the multi-GB emissions table per row — ~144M random reads that// thrashed the volume for hours at full scale. If someone drops `did` from// the index, this catches it: the plan degrades from COVERING INDEX to INDEX.test "group/pack index is covering (index-only scan)" { var conn = try zqlite.open(":memory:", zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); defer conn.close(); try conn.exec("CREATE TABLE emissions (key TEXT NOT NULL, did TEXT NOT NULL, score REAL NOT NULL)", .{}); try conn.exec("CREATE INDEX idx_emissions_key_score ON emissions(key, score DESC, did)", .{});
var rows = try conn.rows("EXPLAIN QUERY PLAN SELECT key, did, score FROM emissions ORDER BY key, score DESC", .{}); defer rows.deinit(); var covering = false; while (rows.next()) |row| { if (std.mem.indexOf(u8, row.text(3), "COVERING INDEX") != null) covering = true; } try testing.expect(covering);}
test "parseChannel defaults to local (safe) and maps known values" { try testing.expectEqual(Channel.local, parseChannel(null)); try testing.expectEqual(Channel.local, parseChannel("local")); try testing.expectEqual(Channel.staging, parseChannel("staging")); try testing.expectEqual(Channel.prod, parseChannel("prod")); // Unrecognized must fail closed to local — never silently to prod. try testing.expectEqual(Channel.local, parseChannel("production")); try testing.expectEqual(Channel.local, parseChannel(""));}
test "resolvePublish: prod requires explicit arming" { var pb: [160]u8 = undefined; var lb: [160]u8 = undefined; try testing.expectError(error.ProdNotArmed, resolvePublish(.prod, false, null, &pb, &lb)); const armed = try resolvePublish(.prod, true, null, &pb, &lb); try testing.expectEqualStrings("builds/", armed.prefix); try testing.expectEqualStrings("latest.json", armed.latest_key);}
test "resolvePublish: staging needs a name and is namespaced away from prod" { var pb: [160]u8 = undefined; var lb: [160]u8 = undefined; try testing.expectError(error.StagingNameMissing, resolvePublish(.staging, false, null, &pb, &lb)); try testing.expectError(error.StagingNameMissing, resolvePublish(.staging, false, "", &pb, &lb)); const s = try resolvePublish(.staging, false, "cjk-search", &pb, &lb); try testing.expectEqualStrings("staging/cjk-search/", s.prefix); try testing.expectEqualStrings("latest.cjk-search.json", s.latest_key); // arming prod has no bearing on a staging build's target const s2 = try resolvePublish(.staging, true, "cjk-search", &pb, &lb); try testing.expectEqualStrings("staging/cjk-search/", s2.prefix);}
test "resolvePublish: local never publishes" { var pb: [160]u8 = undefined; var lb: [160]u8 = undefined; try testing.expectError(error.LocalNoPublish, resolvePublish(.local, true, "anything", &pb, &lb));}