Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724//! Turso → overlay projection.//!//! The search box holds the prefix-index snapshot (base, attached read-only as//! `idx`) plus the live overlay. Every 5 minutes this loop pages Turso for rows//! whose `updated_at` is newer than the watermark, projects each one into the//! overlay, and persists the watermark per page. Recovery from any outage is//! "replay rows newer than the attached snapshot" — bounded by the promote//! cadence, resumable by construction. There is no local copy of the corpus://! docs/replica-removal-plan.md phase 3 removed the `actors` replica whose//! 11-hour resyncs produced the 2026-07/08 retros.
const std = @import("std");const mem = std.mem;const io = std.Options.debug_io;const zqlite = @import("zqlite");const logfire = @import("logfire");const fatal = @import("../fatal.zig");const Allocator = std.mem.Allocator;const TursoClient = @import("TursoClient.zig");const LocalDb = @import("LocalDb.zig");const overlay = @import("../index/overlay.zig");const index_schema = @import("../index/schema.zig");const pack = @import("../index/pack.zig");
const log = std.log.scoped(.sync);
/// column list for the Turso actor rows this loop readspub const actor_cols = "did, handle, display_name, avatar_url, hidden, labels, created_at, associated, pds, followers_count, follows_count, posts_count, last_activity_at, quality_score";
/// named column indexes matching actor_cols orderpub const Col = struct { pub const did: usize = 0; pub const handle: usize = 1; pub const display_name: usize = 2; pub const avatar_url: usize = 3; pub const hidden: usize = 4; pub const labels: usize = 5; pub const created_at: usize = 6; pub const associated: usize = 7; pub const pds: usize = 8; pub const followers_count: usize = 9; pub const follows_count: usize = 10; pub const posts_count: usize = 11; pub const last_activity_at: usize = 12; pub const quality_score: usize = 13;};
/// index of the first column appended after actor_cols (updated_at)const trailing_col: usize = Col.quality_score + 1;
/// Per-row projection context. `arena` is reset by the caller after each row;/// `updated_at` is the source timestamp; `local` is for reading the old row/// from overlay-then-base through a read lease (only read-pool conns have the/// `idx` snapshot attached).const OverlayCtx = struct { arena: Allocator, updated_at: i64, local: *LocalDb,};
fn timestamp() i64 { return @intCast(@divFloor(std.Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s));}
/// Keyset page ≥2 of the incremental drain. The `updated_at >= ?1` anchor is/// load-bearing: it gives the planner an index SEARCH on/// idx_actors_updated_at even with bound params. The plain OR keyset form/// `(updated_at > ? OR (updated_at = ? AND did > ?))` degrades to a full/// index SCAN of the ~8M-row corpus when the values are parameters — that/// scan blew the 60s fetch deadline on every page-2 query and starved the/// drain to one page per cycle (Aug 3–14 2026 staleness incident). Guarded/// by the planner test at the bottom of this file.const incremental_paged_sql = "SELECT " ++ actor_cols ++ ", updated_at FROM actors WHERE handle != '' AND updated_at >= ?1 AND (updated_at > ?1 OR did > ?2) ORDER BY updated_at, did LIMIT 500";const incremental_first_sql = "SELECT " ++ actor_cols ++ ", updated_at FROM actors WHERE handle != '' AND updated_at > ?1 ORDER BY updated_at, did LIMIT 500";
// 500-row pages keep responses ~300KB (the 2026-08-03 wedge bit on a 1.1MiB page).const BATCH_SIZE = 500;const SYNC_INTERVAL_S = 300;/// Turso prunes tombstones after this long (must match cron.ts). A watermark/// older than this can miss deletions until the next promote bakes them in.const TOMBSTONE_RETENTION_S = 7 * 24 * 3600;
// Liveness watchdog, mirroring the ingester's.//// On 2026-07-28 the sync loop stopped making progress and stayed stopped for// 48 hours. It did not crash and did not log: syncLoop catches every error and// continues, so a hang INSIDE incrementalSync (a Turso read that never returns)// is indistinguishable from a healthy idle loop. Exiting lets fly restart us,// which is exactly how the same failure was handled for the ingester.//// Fed by every applied page, every compaction chunk, and every completed pass.const SYNC_STALL_SECS: i64 = 20 * 60;/// overlay compaction chunks per incremental pass (~7s each on the fly volume/// at 20k rows: ≈ 12 min worst case, inside one watchdog window even without/// the per-chunk progress note).const INCREMENTAL_COMPACT_CHUNKS: u64 = 100;pub var last_progress_at: std.atomic.Value(i64) = .init(0);
pub fn noteProgress() void { last_progress_at.store(timestamp(), .monotonic);}
pub fn watchdogLoop() void { last_progress_at.store(timestamp(), .monotonic); while (true) { io.sleep(std.Io.Duration.fromSeconds(60), .awake) catch return; const age = timestamp() - last_progress_at.load(.monotonic); if (age >= SYNC_STALL_SECS) { log.err("sync watchdog: no progress in {d}s, exiting for restart", .{age}); fatal.exitForRestart(); } }}
/// Where a pass starts reading Turso from. The attached snapshot already/// carries every row at or below its watermark, so a persisted `last_sync`/// behind it (a long outage, a fresh volume) resumes at the snapshot instead of/// paging rows the base will only skip.fn resumePoint(last_sync: i64, snapshot_watermark: i64) i64 { return @max(last_sync, snapshot_watermark);}
fn readLastSync(local: *LocalDb, conn: zqlite.Conn) i64 { const s = logfire.span("sync.read_last_sync", .{}); defer s.end(); local.lock(); defer local.unlock(); const row = conn.row("SELECT value FROM sync_meta WHERE key = 'last_sync'", .{}) catch { s.setStatus(.@"error", "query failed"); return 0; }; if (row) |r| { defer r.deinit(); const val = r.text(0); const ts = if (val.len == 0) 0 else std.fmt.parseInt(i64, val, 10) catch 0; s.setAttribute("last_sync", ts); s.setStatus(.ok, null); return ts; } s.setStatus(.ok, null); return 0;}
fn writeLastSync(local: *LocalDb, conn: zqlite.Conn, ts: i64) void { local.lock(); defer local.unlock(); var buf: [20]u8 = undefined; const s = std.fmt.bufPrint(&buf, "{d}", .{ts}) catch return; conn.exec("INSERT OR REPLACE INTO sync_meta (key, value) VALUES ('last_sync', ?)", .{s}) catch {};}
/// One pass: page Turso rows newer than the watermark into the overlay, then/// the actors that became unsearchable and the tombstones. Only advances the/// watermark to "now" when every query and write succeeded; page progress is/// persisted as it goes.pub fn incrementalSync(turso: *TursoClient, local: *LocalDb) !void { const span = logfire.span("sync.incremental", .{}); defer span.end();
const conn = local.getConn() orelse { span.setStatus(.@"error", "LocalNotOpen"); return error.LocalNotOpen; };
// Deltas are gated on the attached snapshot's watermark: only rows NEWER // than the base get projected (older state is already baked in). No // snapshot → nothing to project into; the promote watcher is fetching one. const snapshot_wm: i64 = @intCast(local.attachedMetaInt(index_schema.meta.source_watermark)); if (snapshot_wm == 0) { log.info("incremental: no snapshot attached, waiting for promote", .{}); span.setAttribute("skipped", "no_snapshot"); span.setStatus(.ok, null); return; }
const last_sync_ts = readLastSync(local, conn); const from = resumePoint(last_sync_ts, snapshot_wm); log.info("incremental: last_sync={d} snapshot_watermark={d} from={d}", .{ last_sync_ts, snapshot_wm, from }); span.setAttribute("from", from);
const now = timestamp(); if (now - from > TOMBSTONE_RETENTION_S) { log.warn("incremental: watermark is {d}s old, past tombstone retention — deletions before it are only recovered by the next promote", .{now - from}); span.setAttribute("past_tombstone_retention", true); }
// rows read from Turso but NOT projected because their timestamp is at or // below the snapshot watermark — the base already reflects them. Non-zero // only right after a promote whose watermark passed `last_sync`. var projection_skipped: usize = 0; var ov_arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer ov_arena.deinit();
// buffer: subtract 300s to catch stragglers (matches pub-search) const since_ts = from - 300; var since_buf: [20]u8 = undefined; const since_str = std.fmt.bufPrint(&since_buf, "{d}", .{since_ts}) catch return;
var updated: usize = 0; var error_count: usize = 0; var had_turso_error = false; { var cursor_ts_buf: [20]u8 = undefined; var cursor_did_buf: [128]u8 = undefined; @memcpy(cursor_ts_buf[0..since_str.len], since_str); var cursor_ts: []const u8 = cursor_ts_buf[0..since_str.len]; var cursor_did_len: usize = 0; // one retry per pass: a single transient turso failure shouldn't abort // the whole drain (each aborted pass caps catch-up at one page per cycle) var page_retries: u2 = 1;
while (true) { const cursor_did: []const u8 = if (cursor_did_len > 0) cursor_did_buf[0..cursor_did_len] else ""; const args: []const []const u8 = if (cursor_did_len > 0) &.{ cursor_ts, cursor_did } else &.{cursor_ts}; const sql = if (cursor_did_len > 0) incremental_paged_sql else incremental_first_sql;
log.info("incremental: querying updated actors (cursor_ts={s})", .{cursor_ts}); const q_span = logfire.span("sync.query.updated", .{});
var result = turso.query(sql, args) catch |err| { if (err == error.Canceled) { q_span.end(); return err; } log.err("incremental query failed: {}", .{err}); if (page_retries > 0) { page_retries -= 1; log.info("incremental: retrying page after {s}", .{@errorName(err)}); q_span.setStatus(.@"error", "query failed; retrying"); q_span.end(); continue; } q_span.setStatus(.@"error", "query failed"); q_span.end(); had_turso_error = true; break; }; defer result.deinit();
q_span.setAttribute("rows", @as(i64, @intCast(result.rows.len))); q_span.setStatus(.ok, null); q_span.end();
if (result.rows.len == 0) break;
log.info("incremental: projecting {d} updated actors", .{result.rows.len}); { const a_span = logfire.span("sync.apply.updated", .{}); defer a_span.end();
local.lock(); defer local.unlock(); conn.exec("BEGIN", .{}) catch {}; for (result.rows) |row| { const ua = row.int(trailing_col); if (ua <= snapshot_wm) { projection_skipped += 1; continue; } projectRow(conn, row, .{ .arena = ov_arena.allocator(), .updated_at = ua, .local = local }) catch { error_count += 1; continue; }; _ = ov_arena.reset(.retain_capacity); updated += 1; } conn.exec("COMMIT", .{}) catch {};
a_span.setAttribute("rows", @as(i64, @intCast(result.rows.len))); a_span.setStatus(.ok, null); }
const last_row = result.rows[result.rows.len - 1]; const last_ua_int = last_row.int(trailing_col); const last_did = last_row.text(Col.did);
const ua_str = std.fmt.bufPrint(&cursor_ts_buf, "{d}", .{last_ua_int}) catch break; cursor_ts = ua_str; if (last_did.len > 0 and last_did.len <= cursor_did_buf.len) { @memcpy(cursor_did_buf[0..last_did.len], last_did); cursor_did_len = last_did.len; }
// persist progress per page: "all updates <= last_ua_int are // applied" holds because the keyset is ordered by updated_at. // Without this, a page whose fetch wedges (2026-08-03: the read // of a >1MiB response hung forever; watchdog restarts replayed // the cycle from the same watermark and re-wedged on the same // page) turns into an infinite restart loop. With it, each // watchdog cycle advances at least one page. writeLastSync(local, conn, last_ua_int); noteProgress();
if (result.rows.len < BATCH_SIZE) break; } }
// actors that became unsearchable (handle cleared) var cleared: usize = 0; unsearchable: { log.info("incremental: querying unsearchable actors", .{}); const q_span = logfire.span("sync.query.unsearchable", .{});
var result = turso.query( "SELECT did, updated_at FROM actors WHERE updated_at > ? AND handle = ''", &.{since_str}, ) catch |err| { log.err("unsearchable query failed: {}", .{err}); q_span.setStatus(.@"error", "query failed"); q_span.end(); if (err == error.Canceled) return err; had_turso_error = true; break :unsearchable; }; defer result.deinit();
q_span.setAttribute("rows", @as(i64, @intCast(result.rows.len))); q_span.setStatus(.ok, null); q_span.end();
if (result.rows.len > 0) { log.info("incremental: clearing {d} unsearchable actors", .{result.rows.len}); local.lock(); defer local.unlock(); conn.exec("BEGIN", .{}) catch {}; for (result.rows) |row| { const ua = row.int(1); if (ua <= snapshot_wm) { projection_skipped += 1; continue; } projectRemoval(conn, row.text(0), "incremental.cleared", .{ .arena = ov_arena.allocator(), .updated_at = ua, .local = local }) catch { error_count += 1; continue; }; _ = ov_arena.reset(.retain_capacity); cleared += 1; } conn.exec("COMMIT", .{}) catch {}; } }
// tombstones var deleted: usize = 0; tombstone: { log.info("incremental: querying tombstones", .{}); const q_span = logfire.span("sync.query.tombstones", .{});
var tomb_result = turso.query( "SELECT did, deleted_at FROM tombstones WHERE deleted_at > ?", &.{since_str}, ) catch |err| { log.err("tombstone query failed: {}", .{err}); q_span.setStatus(.@"error", "query failed"); q_span.end(); if (err == error.Canceled) return err; had_turso_error = true; break :tombstone; }; defer tomb_result.deinit();
q_span.setAttribute("rows", @as(i64, @intCast(tomb_result.rows.len))); q_span.setStatus(.ok, null); q_span.end();
if (tomb_result.rows.len > 0) { log.info("incremental: deleting {d} tombstoned actors", .{tomb_result.rows.len}); local.lock(); defer local.unlock(); conn.exec("BEGIN", .{}) catch {}; for (tomb_result.rows) |row| { const del_at = row.int(1); if (del_at <= snapshot_wm) { projection_skipped += 1; continue; } projectRemoval(conn, row.text(0), "incremental.tombstone", .{ .arena = ov_arena.allocator(), .updated_at = del_at, .local = local }) catch { error_count += 1; continue; }; _ = ov_arena.reset(.retain_capacity); deleted += 1; } conn.exec("COMMIT", .{}) catch {}; } }
// only advance the watermark to now when everything succeeded const had_error = had_turso_error or error_count > 0; if (had_error) { span.setAttribute("error_count", @as(i64, @intCast(error_count))); span.setStatus(.@"error", "incremental sync had errors"); log.err("incremental: done with errors (updated={d}, cleared={d}, deleted={d}, errors={d})", .{ updated, cleared, deleted, error_count, }); } else { writeLastSync(local, conn, timestamp()); { local.lock(); defer local.unlock(); // TRUNCATE (not PASSIVE): the read pool nearly always has at least // one slot active under live load, and PASSIVE never reclaims frames // while a reader exists — WAL grows unbounded (we observed 748MB // bloat in ~16min, dragging per-query latency from <50ms to >1s). // TRUNCATE waits up to the writer's busy_timeout for readers to // release their snapshots, then truncates the WAL file to zero. // If it can't (sustained read pressure), it returns busy without // throwing; we'll retry next cycle. conn.exec("PRAGMA wal_checkpoint(TRUNCATE)", .{}) catch {}; } span.setStatus(.ok, null); log.info("incremental: done (updated={d}, cleared={d}, deleted={d})", .{ updated, cleared, deleted }); }
// Prune overlay rows the attached base already covers, a bounded slice per // pass, so the overlay never ratchets up to a compaction longer than the // watchdog (the 2026-08-19..29 loop). Each chunk feeds the watchdog. const r = local.compactOverlay(snapshot_wm, INCREMENTAL_COMPACT_CHUNKS, noteProgress); span.setAttribute("compact_deleted", @as(i64, @intCast(r.deleted))); span.setAttribute("compact_complete", r.complete);
span.setAttribute("updated", @as(i64, @intCast(updated))); span.setAttribute("deleted", @as(i64, @intCast(deleted))); span.setAttribute("cleared", @as(i64, @intCast(cleared))); span.setAttribute("projection_skipped", @as(i64, @intCast(projection_skipped)));}
/// A row assembled from JSON rather than read back out of Turso.////// projectRow takes `anytype` and only ever calls .text/.int/.float, so the/// identical overlay projection the sync loop uses works unchanged on a row/// handed to us over HTTP. No second projection implementation to drift.pub const InlineRow = struct { vals: [14][]const u8, ints: [5]i64 = @splat(0), qs: f64 = 0,
pub fn text(self: *const InlineRow, i: usize) []const u8 { return self.vals[i]; } pub fn int(self: *const InlineRow, i: usize) i64 { return switch (i) { Col.hidden => self.ints[0], Col.followers_count => self.ints[1], Col.follows_count => self.ints[2], Col.posts_count => self.ints[3], Col.last_activity_at => self.ints[4], else => 0, }; } pub fn float(self: *const InlineRow, i: usize) f64 { return if (i == Col.quality_score) self.qs else 0; }};
/// Apply one actor the caller already has, immediately.////// The point of this path is that an explicit "index this account" must not/// wait for the next 5-minute sync tick. The first version took only a DID and/// re-read the row FROM TURSO, which put the fast path through the slowest,/// most contended component in the system: measured 2026-07-31, an indexed/// point lookup took 28s while a bulk job was running, so "instant" became/// "hung". The worker already holds the row — it wrote it moments earlier — so/// it sends the row and nothing here touches Turso at all.////// Idempotent against the sync loop, which will later project the same state.pub fn applyInline(local: *LocalDb, row: *const InlineRow, updated_at: i64) !void { const conn = local.getConn() orelse return error.NoConn; const snapshot_wm: i64 = @intCast(local.attachedMetaInt(index_schema.meta.source_watermark)); if (snapshot_wm == 0) return error.NoSnapshot;
// A freshly-indexed actor is always newer than the base, so in practice // this is always on; the guard exists so a replayed old row cannot re-add // a prefix the base already accounts for. if (updated_at <= snapshot_wm) { log.info("inline: {s} at {d} is not newer than the snapshot ({d}); nothing to project", .{ row.text(Col.did), updated_at, snapshot_wm }); return; }
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit();
local.lock(); defer local.unlock();
conn.exec("BEGIN", .{}) catch {}; projectRow(conn, row, .{ .arena = arena.allocator(), .updated_at = updated_at, .local = local }) catch |err| { conn.exec("ROLLBACK", .{}) catch {}; return err; }; conn.exec("COMMIT", .{}) catch {}; log.info("inline: projected {s}", .{row.text(Col.did)});}
pub fn syncLoop(allocator: Allocator, local: *LocalDb) void { log.info("sync loop started", .{}); var turso = TursoClient.init(allocator) catch |err| { log.err("turso client init failed: {}, sync disabled", .{err}); return; }; defer turso.deinit();
// Nothing to project into until the promote watcher has attached a // snapshot (a fresh volume downloads one first, ~15 min). Keep the // watchdog fed while waiting: this is expected idleness, not a wedge. while (!local.isReady()) { noteProgress(); io.sleep(.{ .nanoseconds = 5 * std.time.ns_per_s }, .real) catch return; }
incrementalSync(&turso, local) catch |err| { if (err == error.Canceled) return; log.err("initial sync failed: {}", .{err}); };
while (true) { io.sleep(.{ .nanoseconds = SYNC_INTERVAL_S * std.time.ns_per_s }, .real) catch return;
incrementalSync(&turso, local) catch |err| { if (err == error.Canceled) return; log.err("periodic sync failed: {}", .{err}); }; // a completed pass is progress even when it applied nothing; a FAILED // pass counts too — the watchdog is for silence, and a loop that errors // every 5 minutes is loud enough to alert on separately noteProgress(); }}
/// Old (handle, display_name) of an actor as serving currently knows it —/// the values whose prefix keys a projection must suppress.const OldRow = struct { handle_buf: [256]u8 = undefined, name_buf: [512]u8 = undefined, handle: ?[]const u8 = null, name: ?[]const u8 = null,};
/// Old handle/display from the overlay blob, falling back to the attached/// base blob — the same bytes serving uses, so the keys diffed away are/// exactly the keys that exist. Goes through a read lease because only/// read-pool conns have the snapshot attached. A NULL overlay blob/// (tombstone) or an actor in neither table yields an empty old row: there/// are no live prefix keys to diff away. Returns false only when the lookup/// itself failed (no lease, oversized blob), in which case the caller/// projects with an empty old row and the stale keys wait for the next/// promote to bake them out.////// Phase 2 of docs/replica-removal-plan.md ran this alongside the `actors`/// replica for two days; every disagreement class traced to the replica being/// the wrong source (docs/search-replica-research.md, addendum).fn snapshotOldFromOverlayBase(local: *LocalDb, did: []const u8, out: *OldRow) bool { const lease = local.checkoutRead() catch return false; defer lease.release();
var arena_state = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena_state.deinit(); return oldFromLease(arena_state.allocator(), lease, did, out);}
fn oldFromLease(arena: std.mem.Allocator, lease: LocalDb.ReadLease, did: []const u8, out: *OldRow) bool { var blob: ?[]const u8 = null; { var rows = lease.query("SELECT blob FROM actor_overlay WHERE did = ?", .{did}) catch return false; defer rows.deinit(); if (rows.next()) |row| { if (row.nullableBlob(0)) |b| { if (b.len > 16384) return false; blob = arena.dupe(u8, b) catch return false; } else return true; } } if (blob == null) blob = lease.baseActor(arena, did) catch return false; const b = blob orelse return true; if (b.len > 16384) return false; const ac = pack.unpackActor(b) catch return false; if (ac.handle.len > 0 and ac.handle.len <= out.handle_buf.len) { @memcpy(out.handle_buf[0..ac.handle.len], ac.handle); out.handle = out.handle_buf[0..ac.handle.len]; } if (ac.display.len > 0 and ac.display.len <= out.name_buf.len) { @memcpy(out.name_buf[0..ac.display.len], ac.display); out.name = out.name_buf[0..ac.display.len]; } return true;}
test "unified base old actor suppresses renamed prefixes and preserves large sync limit" { const bytes = @embedFile("../index/unified_fixture.bin"); var digest: [32]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(bytes, &digest, .{}); const hex = std.fmt.bytesToHex(digest, .lower); const map = try @import("../index/flat_prefix.zig").Map.validate(bytes, &hex); var conn = try zqlite.open(":memory:", zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); defer conn.close(); try @import("../index/schema.zig").initOverlay(conn); var local = LocalDb.init(std.testing.allocator); const lease: LocalDb.ReadLease = .{ .db = &local, .slot = 0, .conn = conn, .prefix_map = map }; var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); var old: OldRow = .{}; try std.testing.expect(oldFromLease(arena, lease, "did:arbitrary:small", &old)); try std.testing.expectEqualStrings("old.example", old.handle.?); try overlay.projectUpdate(arena, conn, .{ .did = "did:arbitrary:small", .handle = "new.example", .display_name = "New Name", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 2, }, old.handle.?, old.name.?, 2); const suppressed = (try conn.row("SELECT op FROM prefix_overlay WHERE key='h:old' AND did='did:arbitrary:small'", .{})).?; defer suppressed.deinit(); try std.testing.expectEqual(@as(i64, 0), suppressed.int(0)); old = .{}; try std.testing.expect(oldFromLease(arena, lease, "did:arbitrary:small", &old)); try std.testing.expectEqualStrings("new.example", old.handle.?); old = .{}; try std.testing.expect(!oldFromLease(arena, lease, "did:web:custom.example/path", &old)); try conn.exec("UPDATE actor_overlay SET blob=NULL WHERE did='did:arbitrary:small'", .{}); old = .{}; try std.testing.expect(oldFromLease(arena, lease, "did:arbitrary:small", &old)); try std.testing.expect(old.handle == null);}
/// Project one updated actor into the overlay, inside the caller's/// transaction on the write conn. `row` only needs .text/.int/.float.fn projectRow(conn: zqlite.Conn, row: anytype, ctx: OverlayCtx) !void { const did = row.text(Col.did); var old: OldRow = .{}; if (!snapshotOldFromOverlayBase(ctx.local, did, &old)) { log.warn("old-row lookup failed for {s}; projecting without suppression", .{did}); } try overlay.projectUpdate(ctx.arena, conn, .{ .did = did, .handle = row.text(Col.handle), .display_name = row.text(Col.display_name), .avatar_url = row.text(Col.avatar_url), .labels = row.text(Col.labels), .associated = row.text(Col.associated), .pds = row.text(Col.pds), .created_at = row.text(Col.created_at), .hidden = row.int(Col.hidden) != 0, .quality_score = @floatCast(row.float(Col.quality_score)), }, old.handle orelse "", old.name orelse "", ctx.updated_at);}
/// Global tombstone in the overlay so serving suppresses the actor/// immediately. Used for both "became unsearchable" and "tombstoned".fn projectRemoval(conn: zqlite.Conn, did: []const u8, caller: []const u8, ctx: OverlayCtx) !void { var old: OldRow = .{}; if (!snapshotOldFromOverlayBase(ctx.local, did, &old)) { log.warn("{s}: old-row lookup failed for {s}; tombstoning without suppression", .{ caller, did }); } try overlay.projectDelete(ctx.arena, conn, did, old.handle orelse "", old.name orelse "", ctx.updated_at);}
// ─── tests ────────────────────────────────────────────────────────────────
test "resumePoint never starts behind the attached snapshot" { const t = std.testing; // fresh volume: no last_sync → replay from the snapshot watermark try t.expectEqual(@as(i64, 1_788_000_000), resumePoint(0, 1_788_000_000)); // long outage: last_sync behind the snapshot → the base already has those rows try t.expectEqual(@as(i64, 1_788_000_000), resumePoint(1_787_000_000, 1_788_000_000)); // steady state: last_sync ahead of the snapshot → resume where we left off try t.expectEqual(@as(i64, 1_788_500_000), resumePoint(1_788_500_000, 1_788_000_000));}
// Regression guard for the Aug 3–14 2026 staleness incident. Turso's// server-side planner degraded the plain OR keyset form to a full index// SCAN when the values were bound params; stock local SQLite plans both// forms fine, so the live planner guard is smoke.py's turso EXPLAIN check.// Here we pin the seek-friendly `updated_at >= ?1` anchor shape and verify// the query stays valid + SEARCHable against the mirrored schema.test "incremental paged keyset query keeps its seek anchor and plans as SEARCH" { try std.testing.expect(std.mem.indexOf(u8, incremental_paged_sql, "updated_at >= ?1") != null); const testing = std.testing; const conn = try zqlite.open(":memory:", zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); defer conn.close();
// mirror the remote actors schema (columns referenced by actor_cols) // and the composite index the keyset walk relies on try conn.exec( \\CREATE TABLE actors ( \\ did TEXT PRIMARY KEY, \\ handle TEXT NOT NULL DEFAULT '', \\ display_name TEXT DEFAULT '', \\ avatar_url TEXT DEFAULT '', \\ updated_at INTEGER NOT NULL DEFAULT 0, \\ hidden INTEGER NOT NULL DEFAULT 0, \\ labels TEXT NOT NULL DEFAULT '[]', \\ created_at TEXT DEFAULT '', \\ associated TEXT DEFAULT '{}', \\ pds TEXT DEFAULT '', \\ followers_count INTEGER DEFAULT 0, \\ follows_count INTEGER DEFAULT 0, \\ posts_count INTEGER DEFAULT 0, \\ last_activity_at INTEGER DEFAULT 0, \\ quality_score REAL DEFAULT 0 \\) , .{}); try conn.exec("CREATE INDEX idx_actors_updated_at ON actors(updated_at, did)", .{});
var rows = try conn.rows( "EXPLAIN QUERY PLAN " ++ incremental_paged_sql, .{ "1786740000", "did:plc:aaaa" }, ); defer rows.deinit();
var saw_search = false; while (rows.next()) |row| { const detail = row.text(3); try testing.expect(std.mem.indexOf(u8, detail, "SCAN") == null); if (std.mem.indexOf(u8, detail, "SEARCH") != null) saw_search = true; } try testing.expect(saw_search);}