Something went wrong. Try again.
atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749//! disk persistence matching indigo's diskpersist format//!//! append-only log files with relay-assigned sequence numbers.//! Postgres metadata index for fast cursor→file lookup.//!//! on-disk entry format (28-byte LE header + CBOR payload)://! [4B flags LE] [4B kind LE] [4B payload_len LE] [8B uid LE] [8B seq LE] [payload]//!//! file naming: evts-{startSeq} (rotated every events_per_file events)//!//! see: indigo cmd/relay/stream/persist/diskpersist/diskpersist.go
const std = @import("std");const Io = std.Io;const pg = @import("pg");const lru = @import("util/lru.zig");const util = @import("util/util.zig");
const getenv = util.getenv;const Allocator = std.mem.Allocator;const log = std.log.scoped(.relay);
// --- constants matching indigo ---
pub const header_size: usize = 28; // 4 + 4 + 4 + 8 + 8
pub const EvtKind = enum(u32) { commit = 1, handle = 2, // deprecated tombstone = 3, // deprecated identity = 4, account = 5, sync = 6,};
pub const EvtFlags = struct { pub const takedown: u32 = 1; pub const rebased: u32 = 2;};
const default_events_per_file: u32 = 10_000;const default_flush_interval_ms: u64 = 100;const default_flush_threshold: usize = 400;/// upper bound on a single record's payload, used to reject a corrupt length/// during recovery. firehose frames are well under this; a larger value means a/// torn/garbage header, not a real record.const max_record_len: u32 = 16 * 1024 * 1024;
// --- header ---
pub const EvtHeader = struct { flags: u32, kind: u32, len: u32, // payload length (not including header) uid: u64, seq: u64,
pub fn encode(self: EvtHeader, buf: *[header_size]u8) void { std.mem.writeInt(u32, buf[0..4], self.flags, .little); std.mem.writeInt(u32, buf[4..8], self.kind, .little); std.mem.writeInt(u32, buf[8..12], self.len, .little); std.mem.writeInt(u64, buf[12..20], self.uid, .little); std.mem.writeInt(u64, buf[20..28], self.seq, .little); }
pub fn decode(buf: *const [header_size]u8) EvtHeader { return .{ .flags = std.mem.readInt(u32, buf[0..4], .little), .kind = std.mem.readInt(u32, buf[4..8], .little), .len = std.mem.readInt(u32, buf[8..12], .little), .uid = std.mem.readInt(u64, buf[12..20], .little), .seq = std.mem.readInt(u64, buf[20..28], .little), }; }};
// --- persist job (write buffer entry) ---
const PersistJob = struct { data: []u8, // header + payload, owned seq: u64, // assigned seq (for broadcast ordering)};
// --- disk persistence ---
/// bounds on a single playback call. a consumer connect with an old cursor/// must never materialize the whole retained backlog in one shot — that is/// an O(retention) allocation per connect, and a flapping consumer turns it/// into an OOM (see docs/handoffs/HANDOFF-2026-08-06-consumer-flap-oom.md)./// callers replay in bounded chunks instead, advancing `since` between calls.pub const PlaybackLimits = struct { max_entries: usize = std.math.maxInt(usize), max_bytes: usize = std.math.maxInt(usize),};
/// playback request — caller posts, pool_io worker executespub const PlaybackRequest = struct { since: u64, allocator: Allocator, limits: PlaybackLimits = .{}, entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty, err: ?anyerror = null, done: std.atomic.Value(bool) = .{ .raw = false }, next: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },};
/// DB request — caller posts, pool_io worker executes./// callers define typed structs embedding DbRequest + @fieldParentPtr.pub const DbRequest = struct { callback: *const fn (*DbRequest, *DiskPersist) void, done: std.atomic.Value(bool) = .{ .raw = false }, err: ?anyerror = null,
/// spin-wait for completion, yielding via io.sleep when available. /// never returns until `done` is true — the request struct lives on the /// caller's stack, so we must outlive the worker's callback execution. /// the queue's shutdown drain sets done+err on unprocessed requests, /// and workers set done after the callback returns, so this always terminates. pub fn wait(self: *DbRequest, io: Io, shutdown: *std.atomic.Value(bool)) void { _ = shutdown; while (!self.done.load(.acquire)) { io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch { // sleep failed (Io shutting down) — busy-wait for worker to finish while (!self.done.load(.acquire)) { std.atomic.spinLoopHint(); } break; }; } }};
/// MPSC FIFO ring buffer for general DB traffic./// multiple producers push via spinlock,/// multiple consumers (pool_io worker threads) pop via CAS.pub const DbRequestQueue = struct { const CAPACITY = 4096;
items: [CAPACITY]*DbRequest = undefined, head: std.atomic.Value(u32) = .{ .raw = 0 }, tail: std.atomic.Value(u32) = .{ .raw = 0 }, push_lock: std.atomic.Value(u32) = .{ .raw = 0 }, // producer spinlock /// total requests handled across all worker threads. exposed as /// relay_db_queue_handled_total — if this stops advancing while /// the queue has depth, the workers are dead or stuck. handled: std.atomic.Value(u64) = .{ .raw = 0 }, shutdown: *std.atomic.Value(bool), persist: *DiskPersist,
/// current depth = tail - head. exposed as relay_db_queue_depth /// so operators can correlate handler hangs with queue backup. pub fn depth(self: *const DbRequestQueue) u32 { const t = self.tail.load(.acquire); const h = self.head.load(.acquire); return t -% h; }
pub fn push(self: *DbRequestQueue, req: *DbRequest) void { // acquire producer spinlock while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { std.atomic.spinLoopHint(); } defer self.push_lock.store(0, .release);
const tail = self.tail.load(.monotonic); const head = self.head.load(.acquire); // if full, spin until space opens (workers are fast, this shouldn't happen) if (tail -% head >= CAPACITY) { self.push_lock.store(0, .release); while (self.tail.load(.monotonic) -% self.head.load(.acquire) >= CAPACITY) { std.atomic.spinLoopHint(); } while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { std.atomic.spinLoopHint(); } }
self.items[self.tail.load(.monotonic) % CAPACITY] = req; self.tail.store(self.tail.load(.monotonic) +% 1, .release); }
pub fn pop(self: *DbRequestQueue) ?*DbRequest { while (true) { const head = self.head.load(.acquire); if (head == self.tail.load(.acquire)) return null; const item = self.items[head % CAPACITY]; if (self.head.cmpxchgWeak(head, head +% 1, .acq_rel, .acquire) == null) return item; } }
pub fn run(self: *DbRequestQueue, pool_io: Io) void { log.info("db_queue worker: starting", .{}); var consecutive_sleep_errors: u32 = 0; while (!self.shutdown.load(.acquire)) { if (self.pop()) |req| { req.callback(req, self.persist); req.done.store(true, .release); _ = self.handled.fetchAdd(1, .monotonic); consecutive_sleep_errors = 0; } else { pool_io.sleep(Io.Duration.fromMilliseconds(5), .awake) catch |e| { // previous behavior: silently `return`, killing the worker // forever. one of these errors during a transient io hiccup // would wedge requestCrawl + every other DB-bound handler. // log + continue. spin-yield to avoid a tight error loop. consecutive_sleep_errors += 1; if (consecutive_sleep_errors == 1 or consecutive_sleep_errors % 1000 == 0) { log.warn("db_queue worker: pool_io.sleep failed ({s}), continuing (consecutive={d})", .{ @errorName(e), consecutive_sleep_errors }); } std.atomic.spinLoopHint(); }; } } // shutdown drain — signal error on unprocessed requests while (self.pop()) |req| { req.err = error.ShuttingDown; req.done.store(true, .release); } log.info("db_queue worker: exiting (shutdown=true, handled_total={d})", .{self.handled.load(.acquire)}); }};
pub const DiskPersist = struct { allocator: Allocator, dir_path: []const u8, dir: Io.Dir, db: *pg.Pool, current_file: ?Io.File = null, current_file_path: ?[]const u8 = null, current_file_pos: u64 = 0,
// sequence state cur_seq: u64 = 1, event_counter: u64 = 0,
// config events_per_file: u32 = default_events_per_file, history_enabled: bool = true, retention_seconds: ?u64 = null, retention_hours: u64 = 72, // 3 days max_dir_bytes: u64 = 100 * 1024 * 1024 * 1024, // 100 GB
// DID → UID cache (matches indigo's bidirectional ARC cache) did_cache: lru.LruCache(u64),
// UID → host_id cache. host_id mutates ~5 orders of magnitude slower than // it's read (every event hits getAccountHostId) and setAccountHostId is its // sole writer, so a read-through cache eliminates a per-event point read. // keyed by the uid's raw bytes; only nonzero host_ids are cached (0 means // "not yet assigned", a transient state that resolves on the next commit). host_id_cache: lru.LruCache(u64),
// write buffer (flushed periodically or when threshold hit) outbuf: std.ArrayListUnmanaged(u8) = .empty, evtbuf: std.ArrayListUnmanaged(PersistJob) = .empty, mutex: Io.Mutex = Io.Mutex.init,
// flush task flush_future: ?Io.Future(void) = null, alive: std.atomic.Value(bool) = .{ .raw = true },
// highest seq that is durably on disk. advanced only by flushLocked, after the // write. the broadcaster fiber refuses to emit a queued frame until its seq is // <= committed_seq, so a frame never reaches consumers before it is durable — // which means the seq recovered on restart can't trail an already-broadcast // seq (the post-crash rewind). see docs/incident-2026-05-31-seq-rewind.md. committed_seq: std.atomic.Value(u64) = .{ .raw = 0 },
io: Io,
/// last successful DB interaction (epoch seconds, set by Threaded workers). /// read by metrics server to report health without direct pg.Pool access. last_db_success: std.atomic.Value(i64) = .{ .raw = 0 },
/// MPSC queue for playback requests (caller → pool_io worker) playback_head: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
/// current evtbuf entry count (for metrics — non-blocking, returns 0 if lock is contended) pub fn evtbufLen(self: *DiskPersist) usize { if (!self.mutex.tryLock()) return 0; defer self.mutex.unlock(self.io); return self.evtbuf.items.len; }
/// current DID cache entry count (for metrics) pub fn didCacheLen(self: *DiskPersist) u32 { return self.did_cache.count(); }
/// DID cache hashmap backing capacity (for memory attribution) pub fn didCacheMapCap(self: *DiskPersist) u32 { return self.did_cache.mapCapacity(); }
/// evtbuf allocated capacity in jobs (for memory attribution — non-blocking) pub fn evtbufCap(self: *DiskPersist) usize { if (!self.mutex.tryLock()) return 0; defer self.mutex.unlock(self.io); return self.evtbuf.capacity; }
/// outbuf allocated capacity in bytes (for memory attribution — non-blocking) pub fn outbufCap(self: *DiskPersist) usize { if (!self.mutex.tryLock()) return 0; defer self.mutex.unlock(self.io); return self.outbuf.capacity; }
pub fn init(allocator: Allocator, dir_path: []const u8, database_url: []const u8, db_pool_size: u16, io: Io) !DiskPersist { // ensure directory exists try Io.Dir.cwd().createDirPath(io, dir_path);
var dir = try Io.Dir.cwd().openDir(io, dir_path, .{ .iterate = true }); errdefer dir.close(io);
// connect to Postgres const uri = std.Uri.parse(database_url) catch return error.InvalidDatabaseUrl; const pool = try pg.Pool.initUri(io, allocator, uri, .{ .size = db_pool_size }); errdefer pool.deinit();
// create tables (matching indigo's Go relay schema) _ = try pool.exec( \\CREATE TABLE IF NOT EXISTS log_file_refs ( \\ id BIGSERIAL PRIMARY KEY, \\ path TEXT NOT NULL, \\ archived BOOLEAN NOT NULL DEFAULT false, \\ seq_start BIGINT NOT NULL, \\ created_at TIMESTAMPTZ NOT NULL DEFAULT now() \\) , .{});
_ = try pool.exec( \\CREATE TABLE IF NOT EXISTS account ( \\ uid BIGSERIAL PRIMARY KEY, \\ did TEXT NOT NULL UNIQUE, \\ host_id BIGINT NOT NULL DEFAULT 0, \\ status TEXT NOT NULL DEFAULT 'active', \\ upstream_status TEXT NOT NULL DEFAULT 'active', \\ created_at TIMESTAMPTZ NOT NULL DEFAULT now() \\) , .{});
// migration: add columns if they don't exist (for existing deployments) _ = pool.exec("ALTER TABLE account ADD COLUMN IF NOT EXISTS host_id BIGINT NOT NULL DEFAULT 0", .{}) catch {}; _ = pool.exec("ALTER TABLE account ADD COLUMN IF NOT EXISTS upstream_status TEXT NOT NULL DEFAULT 'active'", .{}) catch {};
// index host_id so the per-host account COUNT(*) (new-account admission) // is an index scan instead of an 8.7M-row seq scan. CONCURRENTLY avoids // locking the live account table during the build; IF NOT EXISTS makes // re-runs (and resuming a failed build) safe. _ = pool.exec("CREATE INDEX CONCURRENTLY IF NOT EXISTS account_host_id_idx ON account (host_id)", .{}) catch {};
_ = try pool.exec( \\CREATE TABLE IF NOT EXISTS account_repo ( \\ uid BIGINT PRIMARY KEY REFERENCES account(uid), \\ rev TEXT NOT NULL, \\ commit_data_cid TEXT NOT NULL \\) , .{});
_ = try pool.exec("ALTER TABLE account_repo ADD COLUMN IF NOT EXISTS commit_cid TEXT NOT NULL DEFAULT ''", .{}); _ = try pool.exec("CREATE INDEX CONCURRENTLY IF NOT EXISTS account_repo_known_commit_idx ON account_repo (uid) WHERE commit_cid <> '' AND rev <> ''", .{});
_ = try pool.exec( \\CREATE TABLE IF NOT EXISTS domain_ban ( \\ id BIGSERIAL PRIMARY KEY, \\ domain TEXT NOT NULL UNIQUE, \\ created_at TIMESTAMPTZ NOT NULL DEFAULT now() \\) , .{});
_ = try pool.exec( \\CREATE TABLE IF NOT EXISTS host ( \\ id BIGSERIAL PRIMARY KEY, \\ hostname TEXT NOT NULL UNIQUE, \\ status TEXT NOT NULL DEFAULT 'active', \\ last_seq BIGINT NOT NULL DEFAULT 0, \\ failed_attempts INTEGER NOT NULL DEFAULT 0, \\ account_limit BIGINT, \\ created_at TIMESTAMPTZ NOT NULL DEFAULT now(), \\ updated_at TIMESTAMPTZ NOT NULL DEFAULT now() \\) , .{});
// migration: add account_limit column to existing host tables _ = try pool.exec("ALTER TABLE host ADD COLUMN IF NOT EXISTS account_limit BIGINT", .{});
_ = try pool.exec( \\CREATE TABLE IF NOT EXISTS backfill_progress ( \\ collection TEXT NOT NULL, \\ source TEXT NOT NULL, \\ cursor TEXT NOT NULL DEFAULT '', \\ imported_count BIGINT NOT NULL DEFAULT 0, \\ completed_at TIMESTAMPTZ, \\ created_at TIMESTAMPTZ NOT NULL DEFAULT now(), \\ PRIMARY KEY (collection, source) \\) , .{});
// migrate: old schema had collection as sole PK — add source to composite PK _ = pool.exec( \\DO $$ BEGIN \\ IF EXISTS ( \\ SELECT 1 FROM pg_constraint \\ WHERE conname = 'backfill_progress_pkey' \\ AND conrelid = 'backfill_progress'::regclass \\ AND array_length(conkey, 1) = 1 \\ ) THEN \\ ALTER TABLE backfill_progress DROP CONSTRAINT backfill_progress_pkey; \\ ALTER TABLE backfill_progress ADD PRIMARY KEY (collection, source); \\ END IF; \\END $$ , .{}) catch {};
var self = DiskPersist{ .allocator = allocator, .dir_path = try allocator.dupe(u8, dir_path), .dir = dir, .db = pool, .did_cache = lru.LruCache(u64).init(allocator, 500_000, io), .host_id_cache = lru.LruCache(u64).init(allocator, 500_000, io), .io = io, };
// recover from existing log files try self.resumeLog();
// everything already on disk is committed; cur_seq points one past it. self.committed_seq.store(self.cur_seq -| 1, .monotonic);
return self; }
pub fn deinit(self: *DiskPersist) void { // stop flush task (loop checks alive after each sleep interval) self.alive.store(false, .release); if (self.flush_future) |*f| f.cancel(self.io);
// flush remaining self.mutex.lockUncancelable(self.io); self.flushLocked() catch |err| log.err("final event flush failed: {s}", .{@errorName(err)}); self.mutex.unlock(self.io);
// free write buffer for (self.evtbuf.items) |job| self.allocator.free(job.data); self.evtbuf.deinit(self.allocator); self.outbuf.deinit(self.allocator);
self.did_cache.deinit(); self.host_id_cache.deinit();
if (self.current_file) |f| f.close(self.io); if (self.current_file_path) |p| self.allocator.free(p); self.dir.close(self.io); self.db.deinit(); self.allocator.free(self.dir_path); }
/// start the background flush thread pub fn start(self: *DiskPersist) !void { self.flush_future = try self.io.concurrent(flushLoop, .{self}); }
pub const UidResult = struct { uid: u64, host_changed: bool = false, is_new: bool = false, };
/// resolve a DID to a numeric UID, checking host association. /// does NOT set host_id — caller must verify authority first via DID doc /// resolution, then call setAccountHostId on success. /// returns is_new=true when no host_id is set yet (first encounter), /// host_changed=true when host_id differs from stored value. pub fn uidForDidFromHost(self: *DiskPersist, did: []const u8, host_id: u64) !UidResult { const uid = try self.uidForDid(did); if (host_id > 0) { const current_host = self.getAccountHostId(uid) catch 0; if (current_host == 0) { return .{ .uid = uid, .is_new = true }; } else if (current_host != host_id) { log.info("account {s} (uid={d}) host mismatch: current={d} new={d}", .{ did, uid, current_host, host_id }); return .{ .uid = uid, .host_changed = true }; } } return .{ .uid = uid }; }
/// check DB health without touching pg.Pool — safe from any Io context. /// returns true if a Threaded worker successfully queried the DB within the last 30s. pub fn isDbHealthy(self: *DiskPersist) bool { const last = self.last_db_success.load(.acquire); if (last == 0) return false; var ts: std.c.timespec = undefined; _ = std.c.clock_gettime(.REALTIME, &ts); return (@as(i64, ts.sec) - last) < 30; }
fn markDbSuccess(self: *DiskPersist) void { var ts: std.c.timespec = undefined; _ = std.c.clock_gettime(.REALTIME, &ts); self.last_db_success.store(@as(i64, ts.sec), .release); }
/// resolve a DID to a numeric UID. creates a new account row on first encounter. /// matches indigo's Relay.DidToUid → Account.UID mapping. pub fn uidForDid(self: *DiskPersist, did: []const u8) !u64 { // fast path: check in-memory cache if (self.did_cache.get(did)) |uid| return uid;
// check database if (try self.db.rowUnsafe( "SELECT uid FROM account WHERE did = $1", .{did}, )) |row| { var r = row; defer r.deinit() catch {}; const uid: u64 = @intCast(r.get(i64, 0)); self.didCachePut(did, uid); self.markDbSuccess(); return uid; }
// create new account row (ignore if already exists from concurrent insert) _ = self.db.exec( "INSERT INTO account (did) VALUES ($1) ON CONFLICT (did) DO NOTHING", .{did}, ) catch |err| { log.warn("failed to create account for {s}: {s}", .{ did, @errorName(err) }); return err; };
// read back the UID (whether we just created it or it already existed) var row = try self.db.rowUnsafe( "SELECT uid FROM account WHERE did = $1", .{did}, ) orelse return error.AccountCreationFailed; defer row.deinit() catch {}; const uid: u64 = @intCast(row.get(i64, 0));
self.didCachePut(did, uid); return uid; }
/// insert into did_cache (LRU handles eviction at capacity). fn didCachePut(self: *DiskPersist, did: []const u8, uid: u64) void { self.did_cache.put(did, uid) catch {}; }
/// per-DID sync state for chain tracking pub const AccountState = struct { rev: []const u8, data_cid: []const u8, };
/// get stored sync state for a user (from account_repo table) pub fn getAccountState(self: *DiskPersist, uid: u64, allocator: Allocator) !?AccountState { var row = (try self.db.rowUnsafe( "SELECT rev, commit_data_cid FROM account_repo WHERE uid = $1", .{@as(i64, @intCast(uid))}, )) orelse return null; defer row.deinit() catch {}; const rev = row.get([]const u8, 0); const data_cid = row.get([]const u8, 1); // A rev-only row is still prior repo state. Older zlay versions and // signature-only #sync handling stored an empty data CID; discarding // those rows disabled stale-rev and missing-prevData checks entirely. if (rev.len == 0) return null; return .{ .rev = try allocator.dupe(u8, rev), .data_cid = try allocator.dupe(u8, data_cid), }; }
/// update stored sync state after a verified commit (conditional upsert into account_repo). /// uses WHERE clause to ensure rev only moves forward, preventing race conditions /// when concurrent workers process commits for the same DID. /// returns true if the state was updated, false if a newer rev was already stored. pub fn updateAccountState(self: *DiskPersist, uid: u64, rev: []const u8, data_cid: []const u8, commit_cid: []const u8) !bool { const rows_affected = try self.db.exec( "INSERT INTO account_repo (uid, rev, commit_data_cid, commit_cid) VALUES ($1, $2, $3, $4) ON CONFLICT (uid) DO UPDATE SET rev = EXCLUDED.rev, commit_data_cid = EXCLUDED.commit_data_cid, commit_cid = EXCLUDED.commit_cid WHERE account_repo.rev < EXCLUDED.rev", .{ @as(i64, @intCast(uid)), rev, data_cid, commit_cid }, ); return (rows_affected orelse 0) > 0; }
// --- account status ---
/// raw little-endian bytes of a uid, for use as an LRU key. fn uidKey(uid: u64) [8]u8 { return @bitCast(std.mem.nativeToLittle(u64, uid)); }
/// get the host_id for an account. returns 0 if not set. /// read-through cache: only nonzero host_ids are cached (0 is the transient /// pre-assignment state, which resolves via setAccountHostId on next commit). pub fn getAccountHostId(self: *DiskPersist, uid: u64) !u64 { const key = uidKey(uid); if (self.host_id_cache.get(&key)) |hid| return hid;
var row = (try self.db.rowUnsafe( "SELECT host_id FROM account WHERE uid = $1", .{@as(i64, @intCast(uid))}, )) orelse return 0; defer row.deinit() catch {}; const hid = row.get(i64, 0); if (hid > 0) { const host_id: u64 = @intCast(hid); self.host_id_cache.put(&key, host_id) catch {}; return host_id; } return 0; }
/// count accounts on a host (Threaded pool) pub fn getHostAccountCount(self: *DiskPersist, host_id: u64) u64 { return getHostAccountCountImpl(host_id, self.db); }
fn getHostAccountCountImpl(host_id: u64, db: *pg.Pool) u64 { var row = (db.rowUnsafe( "SELECT COUNT(*) FROM account WHERE host_id = $1", .{@as(i64, @intCast(host_id))}, ) catch return 0) orelse return 0; defer row.deinit() catch {}; const count = row.get(i64, 0); return if (count > 0) @intCast(count) else 0; }
/// effective account count (Threaded pool) pub fn getEffectiveAccountCount(self: *DiskPersist, host_id: u64) u64 { return getEffectiveAccountCountImpl(host_id, self.db); }
/// uses admin-configured limit if set, otherwise actual COUNT(*). /// CASE (not COALESCE) so the COUNT subquery is only evaluated when no /// account_limit is set — SQL doesn't short-circuit a COALESCE's aggregate. fn getEffectiveAccountCountImpl(host_id: u64, db: *pg.Pool) u64 { var row = (db.rowUnsafe( "SELECT CASE WHEN h.account_limit IS NOT NULL THEN h.account_limit ELSE (SELECT COUNT(*) FROM account a WHERE a.host_id = h.id) END FROM host h WHERE h.id = $1", .{@as(i64, @intCast(host_id))}, ) catch return 0) orelse return 0; defer row.deinit() catch {}; const count = row.get(i64, 0); return if (count > 0) @intCast(count) else 0; }
/// set host account limit (Threaded pool) pub fn setHostAccountLimit(self: *DiskPersist, host_id: u64, limit: ?u64) !void { return setHostAccountLimitImpl(host_id, limit, self.db); }
/// pass null to clear the override and revert to actual COUNT(*). fn setHostAccountLimitImpl(host_id: u64, limit: ?u64, db: *pg.Pool) !void { if (limit) |l| { const clamped: i64 = if (l > @as(u64, @intCast(std.math.maxInt(i64)))) std.math.maxInt(i64) else @intCast(l); _ = try db.exec( "UPDATE host SET account_limit = $2, updated_at = now() WHERE id = $1", .{ @as(i64, @intCast(host_id)), clamped }, ); } else { _ = try db.exec( "UPDATE host SET account_limit = NULL, updated_at = now() WHERE id = $1", .{@as(i64, @intCast(host_id))}, ); } }
/// set the host_id for an account (first encounter or migration) pub fn setAccountHostId(self: *DiskPersist, uid: u64, host_id: u64) !void { _ = try self.db.exec( "UPDATE account SET host_id = $2 WHERE uid = $1", .{ @as(i64, @intCast(uid)), @as(i64, @intCast(host_id)) }, ); // write-through: sole writer, so the cache stays coherent with the DB. const key = uidKey(uid); if (host_id > 0) { self.host_id_cache.put(&key, host_id) catch {}; } else { _ = self.host_id_cache.remove(&key); } }
/// update the upstream status for an account (from #account events). /// Go relay: account.go UpdateAccountUpstreamStatus pub fn updateAccountUpstreamStatus(self: *DiskPersist, uid: u64, upstream_status: []const u8) !void { _ = try self.db.exec( "UPDATE account SET upstream_status = $2 WHERE uid = $1", .{ @as(i64, @intCast(uid)), upstream_status }, ); }
/// check if an account is active (both local status and upstream status). /// Go relay: models.Account.IsActive() pub const AccountActivity = struct { local_ok: bool, upstream_ok: bool,
pub fn active(self: AccountActivity) bool { return self.local_ok and self.upstream_ok; } };
/// account activity split by source: local (relay takedown) vs upstream /// (PDS-reported via #account events). callers that drop events for /// upstream-inactive accounts use the distinction to trigger a /// StatusChecker re-check — local takedowns must never be re-checked away. pub fn accountActivity(self: *DiskPersist, uid: u64) !AccountActivity { var row = (try self.db.rowUnsafe( "SELECT status, upstream_status FROM account WHERE uid = $1", .{@as(i64, @intCast(uid))}, )) orelse return .{ .local_ok = false, .upstream_ok = false }; defer row.deinit() catch {}; const status = row.get([]const u8, 0); const upstream = row.get([]const u8, 1); return .{ .local_ok = std.mem.eql(u8, status, "active"), .upstream_ok = std.mem.eql(u8, upstream, "active"), }; }
pub fn isAccountActive(self: *DiskPersist, uid: u64) !bool { return (try self.accountActivity(uid)).active(); }
// --- host management ---
pub const Host = struct { id: u64, hostname: []const u8, status: []const u8, last_seq: u64, failed_attempts: u32, account_limit: ?u64 = null, };
const HostResult = struct { id: u64, last_seq: u64 };
/// get or create a host row (Threaded pool) pub fn getOrCreateHost(self: *DiskPersist, hostname: []const u8) !HostResult { return getOrCreateHostImpl(hostname, self.db); }
fn getOrCreateHostImpl(hostname: []const u8, db: *pg.Pool) !HostResult { _ = db.exec( "INSERT INTO host (hostname) VALUES ($1) ON CONFLICT (hostname) DO NOTHING", .{hostname}, ) catch |err| { log.warn("failed to create host {s}: {s}", .{ hostname, @errorName(err) }); return err; };
var row = try db.rowUnsafe( "SELECT id, last_seq FROM host WHERE hostname = $1", .{hostname}, ) orelse return error.HostCreationFailed; defer row.deinit() catch {}; return .{ .id = @intCast(row.get(i64, 0)), .last_seq = @intCast(row.get(i64, 1)), }; }
/// check if a host is banned or blocked by status (Threaded pool) pub fn isHostBanned(self: *DiskPersist, hostname: []const u8) bool { return isHostBannedImpl(hostname, self.db); }
fn isHostBannedImpl(hostname: []const u8, db: *pg.Pool) bool { var row = db.rowUnsafe( "SELECT status FROM host WHERE hostname = $1", .{hostname}, ) catch return false; if (row) |*r| { defer r.deinit() catch {}; const status = r.get([]const u8, 0); return std.mem.eql(u8, status, "banned") or std.mem.eql(u8, status, "blocked"); } return false; }
/// update cursor position for a host pub fn updateHostSeq(self: *DiskPersist, host_id: u64, seq: u64) !void { _ = try self.db.exec( "UPDATE host SET last_seq = $2, updated_at = now() WHERE id = $1", .{ @as(i64, @intCast(host_id)), @as(i64, @intCast(seq)) }, ); }
/// look up host ID by hostname (Threaded pool) pub fn getHostIdForHostname(self: *DiskPersist, hostname: []const u8) !?u64 { return getHostIdForHostnameImpl(hostname, self.db); }
fn getHostIdForHostnameImpl(hostname: []const u8, db: *pg.Pool) !?u64 { var row = (try db.rowUnsafe( "SELECT id FROM host WHERE hostname = $1", .{hostname}, )) orelse return null; defer row.deinit() catch {}; return @intCast(row.get(i64, 0)); }
/// update host status (Threaded pool) pub fn updateHostStatus(self: *DiskPersist, host_id: u64, status: []const u8) !void { return updateHostStatusImpl(host_id, status, self.db); }
fn updateHostStatusImpl(host_id: u64, status: []const u8, db: *pg.Pool) !void { _ = try db.exec( "UPDATE host SET status = $2, updated_at = now() WHERE id = $1", .{ @as(i64, @intCast(host_id)), status }, ); }
/// list all active hosts (Threaded pool) pub fn listActiveHosts(self: *DiskPersist, allocator: Allocator) ![]Host { return listActiveHostsImpl(allocator, self.db); }
fn listActiveHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host { var hosts: std.ArrayListUnmanaged(Host) = .empty; errdefer { for (hosts.items) |host| { allocator.free(host.hostname); allocator.free(host.status); } hosts.deinit(allocator); }
var result = try db.query( "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host WHERE status = 'active' ORDER BY id ASC", .{}, ); defer result.deinit();
while (result.nextUnsafe() catch null) |row| { try hosts.append(allocator, .{ .id = @intCast(row.get(i64, 0)), .hostname = try allocator.dupe(u8, row.get([]const u8, 1)), .status = try allocator.dupe(u8, row.get([]const u8, 2)), .last_seq = @intCast(row.get(i64, 3)), .failed_attempts = @intCast(row.get(i32, 4)), .account_limit = if (row.get(?i64, 5)) |v| @as(?u64, @intCast(v)) else null, }); }
return try hosts.toOwnedSlice(allocator); }
/// list all hosts (Threaded pool) pub fn listAllHosts(self: *DiskPersist, allocator: Allocator) ![]Host { return listAllHostsImpl(allocator, self.db); }
fn listAllHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host { var hosts: std.ArrayListUnmanaged(Host) = .empty; errdefer { for (hosts.items) |host| { allocator.free(host.hostname); allocator.free(host.status); } hosts.deinit(allocator); }
var result = try db.query( "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host ORDER BY id ASC", .{}, ); defer result.deinit();
while (result.nextUnsafe() catch null) |row| { try hosts.append(allocator, .{ .id = @intCast(row.get(i64, 0)), .hostname = try allocator.dupe(u8, row.get([]const u8, 1)), .status = try allocator.dupe(u8, row.get([]const u8, 2)), .last_seq = @intCast(row.get(i64, 3)), .failed_attempts = @intCast(row.get(i32, 4)), .account_limit = if (row.get(?i64, 5)) |v| @as(?u64, @intCast(v)) else null, }); }
return try hosts.toOwnedSlice(allocator); }
/// increment failure count and return new value pub fn incrementHostFailures(self: *DiskPersist, host_id: u64) !u32 { _ = try self.db.exec( "UPDATE host SET failed_attempts = failed_attempts + 1, updated_at = now() WHERE id = $1", .{@as(i64, @intCast(host_id))}, ); var row = try self.db.rowUnsafe( "SELECT failed_attempts FROM host WHERE id = $1", .{@as(i64, @intCast(host_id))}, ) orelse return 0; defer row.deinit() catch {}; return @intCast(row.get(i32, 0)); }
/// check if a hostname (or any parent domain) is banned (Threaded pool). pub fn isDomainBanned(self: *DiskPersist, hostname: []const u8) bool { return self.isDomainBannedImpl(hostname, self.db); }
/// Go relay: domain_ban.go DomainIsBanned — suffix-based check. fn isDomainBannedImpl(_: *DiskPersist, hostname: []const u8, db: *pg.Pool) bool { // check each suffix: "pds.host.example.com", "host.example.com", "example.com" var offset: usize = 0; while (offset < hostname.len) { const suffix = hostname[offset..]; var row = db.rowUnsafe( "SELECT 1 FROM domain_ban WHERE domain = $1", .{suffix}, ) catch return false; if (row) |*r| { r.deinit() catch {}; return true; } // advance past next dot if (std.mem.indexOfScalarPos(u8, hostname, offset, '.')) |dot| { offset = dot + 1; } else break; } return false; }
/// reset failure count (Threaded pool) pub fn resetHostFailures(self: *DiskPersist, host_id: u64) !void { return resetHostFailuresImpl(host_id, self.db); }
fn resetHostFailuresImpl(host_id: u64, db: *pg.Pool) !void { _ = try db.exec( "UPDATE host SET failed_attempts = 0, updated_at = now() WHERE id = $1", .{@as(i64, @intCast(host_id))}, ); }
/// enqueue a playback request for the pool_io worker to execute. /// uses MPSC push (lock-free Treiber stack). pub fn enqueuePlayback(self: *DiskPersist, req: *PlaybackRequest) void { while (true) { const head = self.playback_head.load(.acquire); req.next.store(head, .monotonic); if (self.playback_head.cmpxchgWeak(@as(?*PlaybackRequest, head), req, .release, .monotonic) == null) { return; } } }
/// pop all pending playback requests (single-consumer — pool_io worker only). /// returns a singly-linked list via .next pointers, or null if empty. pub fn popPlaybackBatch(self: *DiskPersist) ?*PlaybackRequest { return self.playback_head.swap(null, .acq_rel); }
/// persist an event. assigns a sequence number. returns the assigned seq. /// the event is buffered and will be flushed to disk asynchronously. pub fn persist(self: *DiskPersist, kind: EvtKind, uid: u64, payload: []const u8) !u64 { if (!self.history_enabled) { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const seq = self.cur_seq; self.cur_seq += 1; self.committed_seq.store(seq, .release); return seq; } // build the on-disk record: header + payload const total_len = header_size + payload.len; const data = try self.allocator.alloc(u8, total_len); errdefer self.allocator.free(data);
// write header (seq filled in later under lock) const header = EvtHeader{ .flags = 0, .kind = @backingInt(kind), .len = @intCast(payload.len), .uid = uid, .seq = 0, // placeholder }; header.encode(data[0..header_size]); @memcpy(data[header_size..], payload);
self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
if (self.evtbuf.items.len >= default_flush_threshold) try self.flushLocked(); try self.evtbuf.ensureUnusedCapacity(self.allocator, 1); try self.outbuf.ensureUnusedCapacity(self.allocator, data.len);
// assign seq const seq = self.cur_seq; self.cur_seq += 1; std.mem.writeInt(u64, data[20..28], seq, .little);
self.evtbuf.appendAssumeCapacity(.{ .data = data, .seq = seq }); self.outbuf.appendSliceAssumeCapacity(data); self.event_counter += 1;
// flush if threshold hit if (self.evtbuf.items.len >= default_flush_threshold) { self.flushLocked() catch |err| log.warn("event flush deferred: {s}", .{@errorName(err)}); }
return seq; }
/// playback events with seq > since, bounded by `limits`. a truncated /// result is detected by the caller via entries hitting the limits; /// resume by calling again with since = last returned seq. /// /// runs without `mutex`. holding it across the file reads blocked persist /// and flush (so `committed_seq` froze and the broadcaster held the ring) /// for the length of every replay chunk (2026-10-18). the reads are safe /// without it: sealed files are immutable, the current file only grows by /// whole records appended in seq order (a torn tail reads as a short read /// and ends the chunk; the caller resumes from the last complete seq), and /// a file gc removes between the ref query and the open is skipped. the /// one writer that mutates records in place, takedown, sets the flag word /// before zeroing the payload, so a concurrent reader either skips the /// record or still sees the original payload. pub fn playback(self: *DiskPersist, since: u64, allocator: Allocator, entries: *std.ArrayListUnmanaged(PlaybackEntry), limits: PlaybackLimits) !void { if (!self.history_enabled) return error.ReplayDisabled; const since_i: i64 = @intCast(since);
// find the log file containing `since` var start_files: std.ArrayListUnmanaged(LogFileRef) = .empty; defer start_files.deinit(allocator);
if (since > 0) { // find file whose seq_start is just before `since` if (try self.db.rowUnsafe( "SELECT id, path, seq_start FROM log_file_refs WHERE seq_start <= $1 ORDER BY seq_start DESC LIMIT 1", .{since_i}, )) |row| { var r = row; defer r.deinit() catch {}; try start_files.append(allocator, .{ .path = try allocator.dupe(u8, r.get([]const u8, 1)), .seq_start = @intCast(r.get(i64, 2)), }); } }
// find all subsequent files { var result = try self.db.query( "SELECT id, path, seq_start FROM log_file_refs WHERE seq_start > $1 ORDER BY seq_start ASC", .{since_i}, ); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { try start_files.append(allocator, .{ .path = try allocator.dupe(u8, r.get([]const u8, 1)), .seq_start = @intCast(r.get(i64, 2)), }); } }
defer for (start_files.items) |f| allocator.free(f.path);
// read events from each file until the limits are hit var bytes_used: usize = 0; for (start_files.items) |ref| { if (entries.items.len >= limits.max_entries or bytes_used >= limits.max_bytes) break; var file = self.dir.openFile(self.io, ref.path, .{}) catch continue; defer file.close(self.io); try readEventsFrom(allocator, file, self.io, since, entries, limits, &bytes_used); } }
/// last assigned sequence number, or null if empty pub fn lastSeq(self: *DiskPersist) ?u64 { if (self.cur_seq <= 1) return null; return self.cur_seq - 1; }
/// oldest available sequence number (Threaded pool) pub fn firstSeq(self: *DiskPersist) ?u64 { return firstSeqImpl(self.db); }
fn firstSeqImpl(db: *pg.Pool) ?u64 { if (db.rowUnsafe( "SELECT seq_start FROM log_file_refs ORDER BY seq_start ASC LIMIT 1", .{}, ) catch null) |row| { var r = row; defer r.deinit() catch {}; const v = r.get(i64, 0); return if (v > 0) @intCast(v) else null; } return null; }
/// garbage-collect log files older than the retention period pub fn gc(self: *DiskPersist) !void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
const cutoff_interval = if (self.retention_seconds) |seconds| try std.fmt.allocPrint(self.allocator, "{d} seconds", .{seconds}) else try std.fmt.allocPrint(self.allocator, "{d} hours", .{self.retention_hours}); defer self.allocator.free(cutoff_interval);
// find expired refs var expired: std.ArrayListUnmanaged(GcRef) = .empty; defer { for (expired.items) |e| self.allocator.free(e.path); expired.deinit(self.allocator); }
{ var result = try self.db.query( "SELECT id, path FROM log_file_refs WHERE created_at < now() - $1::interval", .{cutoff_interval}, ); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { try expired.append(self.allocator, .{ .id = r.get(i64, 0), .path = try self.allocator.dupe(u8, r.get([]const u8, 1)), }); } }
for (expired.items) |ref| { // skip current file if (self.current_file_path) |cur| { if (std.mem.eql(u8, ref.path, cur)) continue; }
// delete db record first (prevents playback from finding it) _ = self.db.exec("DELETE FROM log_file_refs WHERE id = $1", .{ref.id}) catch |err| { log.warn("gc: failed to delete db record {d}: {s}", .{ ref.id, @errorName(err) }); continue; };
// delete file self.dir.deleteFile(self.io, ref.path) catch |err| { log.warn("gc: failed to delete {s}: {s}", .{ ref.path, @errorName(err) }); }; }
if (expired.items.len > 0) { log.info("gc: removed {d} expired log files", .{expired.items.len}); }
// size-based GC: delete oldest files until total size is under max_dir_bytes if (self.max_dir_bytes > 0) { try self.gcBySize(); }
self.gcOrphans() catch |err| { log.warn("gc: orphan sweep failed: {s}", .{@errorName(err)}); };
self.markDbSuccess(); }
/// delete segment files that no log_file_refs row references and that are /// strictly older (by start seq, from the evts-{seq} name) than everything /// still registered. gc deletes the row before the file — deliberately, so /// playback can never resolve a missing file — which means a failed file /// delete leaves an orphan no later time-based pass can see: the row is /// gone, and the file leaks forever. that is how production accumulated /// ~100G / 10 weeks of unreferenced segments (found 2026-08-27). playback /// resolves every read through log_file_refs, so an unreferenced file /// below the retained window is unreachable and safe to remove while live. fn gcOrphans(self: *DiskPersist) !void { var min_registered: ?u64 = null; { var result = try self.db.query("SELECT min(seq_start) FROM log_file_refs", .{}); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { if (r.get(?i64, 0)) |v| min_registered = @intCast(v); } } // no registered segments at all: nothing is provably expired — a // fresh DB over an old data dir must not eat the recovery source const floor = min_registered orelse return;
var doomed: std.ArrayListUnmanaged([]u8) = .empty; defer { for (doomed.items) |p| self.allocator.free(p); doomed.deinit(self.allocator); } { var it = self.dir.iterate(); while (it.next(self.io) catch null) |entry| { if (entry.kind != .file) continue; if (!std.mem.startsWith(u8, entry.name, "evts-")) continue; const seq = std.fmt.parseInt(u64, entry.name["evts-".len..], 10) catch continue; if (seq >= floor) continue; if (self.current_file_path) |cur| { if (std.mem.eql(u8, entry.name, cur)) continue; } try doomed.append(self.allocator, try self.allocator.dupe(u8, entry.name)); } }
var removed: usize = 0; for (doomed.items) |path| { self.dir.deleteFile(self.io, path) catch |err| { log.warn("gc: failed to delete orphan {s}: {s}", .{ path, @errorName(err) }); continue; }; removed += 1; } if (removed > 0) { log.info("gc: removed {d} orphaned log files (unreferenced, below seq {d})", .{ removed, floor }); } }
/// delete oldest event files until total directory size is under max_dir_bytes fn gcBySize(self: *DiskPersist) !void { // query all refs ordered by creation time (oldest first) var refs: std.ArrayListUnmanaged(GcSizeRef) = .empty; defer { for (refs.items) |e| self.allocator.free(e.path); refs.deinit(self.allocator); }
var total_bytes: u64 = 0; { var result = try self.db.query( "SELECT id, path FROM log_file_refs ORDER BY id ASC", .{}, ); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { const path = r.get([]const u8, 1); const stat = self.dir.statFile(self.io, path, .{}) catch continue; total_bytes += stat.size; try refs.append(self.allocator, .{ .id = r.get(i64, 0), .path = try self.allocator.dupe(u8, path), .size = stat.size, }); } }
if (total_bytes <= self.max_dir_bytes) return;
log.info("gc: event dir size {d} MB exceeds limit {d} MB, pruning oldest files", .{ total_bytes / (1024 * 1024), self.max_dir_bytes / (1024 * 1024), });
var removed: u64 = 0; for (refs.items) |ref| { if (total_bytes <= self.max_dir_bytes) break;
// never delete the current file if (self.current_file_path) |cur| { if (std.mem.eql(u8, ref.path, cur)) continue; }
_ = self.db.exec("DELETE FROM log_file_refs WHERE id = $1", .{ref.id}) catch |err| { log.warn("gc: failed to delete db record {d}: {s}", .{ ref.id, @errorName(err) }); continue; };
self.dir.deleteFile(self.io, ref.path) catch |err| { log.warn("gc: failed to delete {s}: {s}", .{ ref.path, @errorName(err) }); continue; // don't subtract size if file delete failed };
total_bytes -= @min(total_bytes, ref.size); removed += 1; }
if (removed > 0) { log.info("gc: size-based pruning removed {d} files, dir now {d} MB", .{ removed, total_bytes / (1024 * 1024), }); } }
/// take down all events for a user (set flag + zero payload) pub fn takeDownUser(self: *DiskPersist, uid: u64) !void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
// iterate all log files var refs: std.ArrayListUnmanaged([]const u8) = .empty; defer { for (refs.items) |p| self.allocator.free(p); refs.deinit(self.allocator); }
{ var result = try self.db.query("SELECT path FROM log_file_refs ORDER BY seq_start DESC", .{}); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { try refs.append(self.allocator, try self.allocator.dupe(u8, r.get([]const u8, 0))); } }
for (refs.items) |path| { var file = self.dir.openFile(self.io, path, .{ .mode = .read_write }) catch continue; defer file.close(self.io); mutateUserEventsInFile(file, self.io, uid) catch |err| { log.warn("takedown: failed to process {s}: {s}", .{ path, @errorName(err) }); }; } }
// --- internals ---
fn resumeLog(self: *DiskPersist) !void { // find most recent log file if (try self.db.rowUnsafe( "SELECT id, path, seq_start FROM log_file_refs ORDER BY seq_start DESC LIMIT 1", .{}, )) |row| { var r = row; defer r.deinit() catch {}; const path = r.get([]const u8, 1); const seq_start: u64 = @intCast(r.get(i64, 2));
var file = self.dir.openFile(self.io, path, .{ .mode = .read_write }) catch { // file missing, start fresh try self.initLogFile(); return; };
// scan for last seq (validates records, reports the last good offset) const scan = scanForLastSeq(file, self.io, seq_start) catch { file.close(self.io); try self.initLogFile(); return; };
// heal a torn tail (partial write from an unclean kill) so subsequent // appends stay record-aligned and future scans don't desync. const full_size = (try file.stat(self.io)).size; if (scan.valid_end < full_size) { log.warn("recovered log {s} has a torn tail: truncating {d} -> {d} bytes", .{ path, full_size, scan.valid_end }); file.setLength(self.io, scan.valid_end) catch |err| { log.warn("failed to truncate torn log tail: {s}", .{@errorName(err)}); }; }
if (scan.last_seq) |ls| { self.cur_seq = ls + 1; log.info("recovered seq from disk: last_seq={d}", .{ls}); } else { self.cur_seq = if (seq_start > 0) seq_start else 1; }
// append at the end of the last valid record (post-truncation) self.current_file_pos = scan.valid_end;
self.current_file = file; self.current_file_path = try self.allocator.dupe(u8, path); } else { try self.initLogFile(); } }
fn initLogFile(self: *DiskPersist) !void { try self.createLogFile(self.cur_seq); }
fn createLogFile(self: *DiskPersist, start_seq: u64) !void { var name_buf: [64]u8 = undefined; const name = std.fmt.bufPrint(&name_buf, "evts-{d}", .{start_seq}) catch unreachable;
const path = try self.allocator.dupe(u8, name); errdefer self.allocator.free(path); const file = try self.dir.createFile(self.io, name, .{ .exclusive = true, .read = true }); errdefer { file.close(self.io); self.dir.deleteFile(self.io, name) catch {}; }
// register in Postgres _ = try self.db.exec( "INSERT INTO log_file_refs (path, seq_start) VALUES ($1, $2)", .{ name, @as(i64, @intCast(start_seq)) }, );
if (self.current_file) |f| f.close(self.io); if (self.current_file_path) |p| self.allocator.free(p); self.current_file = file; self.current_file_pos = 0; self.current_file_path = path; self.event_counter = 0; }
pub fn flush(self: *DiskPersist) !void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); try self.flushLocked(); }
fn flushLocked(self: *DiskPersist) !void { if (self.evtbuf.items.len == 0) return;
// write buffered bytes to current file const file = self.current_file orelse return; try file.writePositionalAll(self.io, self.outbuf.items, self.current_file_pos); self.current_file_pos += self.outbuf.items.len;
// clear write buffer (bytes are now durable) self.outbuf.clearRetainingCapacity();
// these events are now durable — publish the watermark so the broadcaster // fiber may emit them. evtbuf is append-ordered, so the last entry holds // the highest seq just written. this is the only place committed_seq moves. if (self.evtbuf.items.len > 0) { self.committed_seq.store(self.evtbuf.items[self.evtbuf.items.len - 1].seq, .release); }
// free job data for (self.evtbuf.items) |job| { self.allocator.free(job.data); } self.evtbuf.clearRetainingCapacity();
// check if we need to rotate if (self.event_counter >= self.events_per_file) { self.createLogFile(self.cur_seq) catch |err| { log.err("flush: log rotation failed: {s}", .{@errorName(err)}); }; } }
fn flushLoop(self: *DiskPersist) void { while (self.alive.load(.acquire)) { // sleep for flush interval (replaces timedWait — Io.Condition has no timed variant) self.io.sleep(Io.Duration.fromMilliseconds(@intCast(default_flush_interval_ms)), .awake) catch {}; self.mutex.lockUncancelable(self.io); self.flushLocked() catch |err| log.err("event flush failed: {s}", .{@errorName(err)}); self.mutex.unlock(self.io); } }};
// --- playback types ---
pub const PlaybackEntry = struct { seq: u64, kind: u32, uid: u64, data: []const u8, // payload only (owned by caller)};
const LogFileRef = struct { path: []const u8, seq_start: u64,};
const GcRef = struct { id: i64, path: []const u8,};
const GcSizeRef = struct { id: i64, path: []const u8, size: u64,};
// --- file-level operations ---
/// positional reads through a block buffer. the event log is parsed one/// 28-byte header at a time, and reading each header with its own pread cost/// a syscall per record (a 4,096-entry replay chunk was ~10^4 syscalls, with/// the seek from the start of the file on top). a block refill amortizes/// that to a syscall per `block_size` bytes.const PositionalReader = struct { const block_size: usize = 256 * 1024;
file: Io.File, io: Io, buf: []u8, base: u64 = 0, len: usize = 0,
fn init(allocator: Allocator, file: Io.File, io: Io) !PositionalReader { return .{ .file = file, .io = io, .buf = try allocator.alloc(u8, block_size) }; }
fn deinit(self: *PositionalReader, allocator: Allocator) void { allocator.free(self.buf); }
/// read `dst.len` bytes at `pos`. returns the number of bytes read; short /// at end of file (or at a torn tail still being written). fn readAt(self: *PositionalReader, pos: u64, dst: []u8) !usize { if (dst.len > self.buf.len) { return try self.file.readPositionalAll(self.io, dst, pos); } if (pos < self.base or pos + dst.len > self.base + self.len) { self.base = pos; self.len = try self.file.readPositionalAll(self.io, self.buf, pos); } const off: usize = @intCast(pos - self.base); const n = @min(dst.len, self.len - off); @memcpy(dst[0..n], self.buf[off .. off + n]); return n; }};
fn readEventsFrom(allocator: Allocator, file: Io.File, io: Io, since: u64, result: *std.ArrayListUnmanaged(PlaybackEntry), limits: PlaybackLimits, bytes_used: *usize) !void { const file_size = (try file.stat(io)).size; var reader = try PositionalReader.init(allocator, file, io); defer reader.deinit(allocator);
// if since > 0, scan to the right position var pos: u64 = 0; if (since > 0) { pos = seekToSeq(&reader, since, file_size) catch return; }
// read events while (pos + header_size <= file_size) { if (result.items.len >= limits.max_entries or bytes_used.* >= limits.max_bytes) return; var hdr_buf: [header_size]u8 = undefined; const n = reader.readAt(pos, &hdr_buf) catch break; if (n < header_size) break; pos += header_size;
const hdr = EvtHeader.decode(&hdr_buf);
// skip taken down / rebased events if (hdr.flags & (EvtFlags.takedown | EvtFlags.rebased) != 0) { pos += hdr.len; continue; }
if (hdr.seq <= since) { pos += hdr.len; continue; }
// read payload const data = allocator.alloc(u8, hdr.len) catch break; const read_n = reader.readAt(pos, data) catch { allocator.free(data); break; }; if (read_n < hdr.len) { allocator.free(data); break; } pos += hdr.len;
result.append(allocator, .{ .seq = hdr.seq, .kind = hdr.kind, .uid = hdr.uid, .data = data, }) catch { allocator.free(data); break; }; bytes_used.* += data.len; }}
/// scan file headers to find position of first event with seq > targetfn seekToSeq(reader: *PositionalReader, target: u64, file_size: u64) !u64 { var pos: u64 = 0; while (pos + header_size <= file_size) { var hdr_buf: [header_size]u8 = undefined; const n = try reader.readAt(pos, &hdr_buf); if (n < header_size) break;
const hdr = EvtHeader.decode(&hdr_buf); if (hdr.seq > target) { return pos; } pos += header_size + hdr.len; } return pos;}
const ScanResult = struct { /// highest seq among fully-valid records, or null if the file has none last_seq: ?u64, /// byte offset just past the last fully-valid record. anything beyond is a /// torn/corrupt tail (a partial final write after an unclean kill) and must /// be truncated before appending, or it desyncs every future scan. valid_end: u64,};
/// scan a log file for the last valid sequence number. each record is validated/// (known kind, plausible length, body fully present, AND a seq that matches the/// dense monotonic space the writer guarantees) before being accepted; the scan/// stops at the first record it can't trust rather than following a garbage/// length and silently mis-recovering. this is what bounds a torn tail to the/// torn record itself — see docs/incident-2026-05-31-seq-rewind.md.////// `seq_start` is the file's declared starting seq (from its name / log_file_refs/// row). the writer assigns seqs densely (seq = cur_seq; cur_seq += 1) and a file/// begins at exactly seq_start, so the first record must have seq == seq_start and/// every subsequent record seq == prev + 1. a record whose seq breaks that chain/// is a torn/garbage tail even if its kind+len happen to look plausible — this is/// what stops `scanForLastSeq` from accepting CBOR payload bytes as a seq and/// relocating the whole sequencer into a junk space (the 2026-06 quintillion-seq/// incident). pass 0 to skip the seq-chain check (genuinely unknown start).fn scanForLastSeq(file: Io.File, io: Io, seq_start: u64) !ScanResult { const file_size = (try file.stat(io)).size;
var last_seq: ?u64 = null; var pos: u64 = 0; while (pos + header_size <= file_size) { var hdr_buf: [header_size]u8 = undefined; const n = try file.readPositionalAll(io, &hdr_buf, pos); if (n < header_size) break;
const hdr = EvtHeader.decode(&hdr_buf);
// a record we can't trust — unknown kind, implausible length, or a body // that runs past EOF (partial final write) — marks the torn tail. stop; // `pos` is the end of the last good record. const valid_kind = hdr.kind >= @backingInt(EvtKind.commit) and hdr.kind <= @backingInt(EvtKind.sync); if (!valid_kind or hdr.len > max_record_len or pos + header_size + hdr.len > file_size) break;
// seq must follow the dense monotonic chain: the first record == seq_start, // each next == prev + 1. a break here is a torn/garbage tail. if (seq_start > 0) { const expected = if (last_seq) |ls| ls + 1 else seq_start; if (hdr.seq != expected) break; }
last_seq = hdr.seq; pos += header_size + hdr.len; } return .{ .last_seq = last_seq, .valid_end = pos };}
/// set takedown flag + zero payload for all events belonging to uidfn mutateUserEventsInFile(file: Io.File, io: Io, uid: u64) !void { const file_size = (try file.stat(io)).size; var pos: u64 = 0;
while (pos + header_size <= file_size) { var hdr_buf: [header_size]u8 = undefined; const n = try file.readPositionalAll(io, &hdr_buf, pos); if (n < header_size) break;
const hdr = EvtHeader.decode(&hdr_buf);
if (hdr.uid == uid and (hdr.flags & EvtFlags.takedown) == 0) { // set takedown flag const new_flags = hdr.flags | EvtFlags.takedown; var flags_buf: [4]u8 = undefined; std.mem.writeInt(u32, &flags_buf, new_flags, .little); try file.writePositionalAll(io, &flags_buf, pos);
// zero the payload const payload_start = pos + header_size; var zeros: [4096]u8 = @splat(0); var remaining: u64 = hdr.len; var write_pos: u64 = payload_start; while (remaining > 0) { const chunk: usize = @min(remaining, zeros.len); try file.writePositionalAll(io, zeros[0..chunk], write_pos); write_pos += chunk; remaining -= chunk; } }
pos += header_size + hdr.len; }}
// === tests ===
test "header encode/decode roundtrip" { const hdr = EvtHeader{ .flags = 0, .kind = @backingInt(EvtKind.commit), .len = 1234, .uid = 42, .seq = 99999, }; var buf: [header_size]u8 = undefined; hdr.encode(&buf); const decoded = EvtHeader.decode(&buf); try std.testing.expectEqual(hdr.flags, decoded.flags); try std.testing.expectEqual(hdr.kind, decoded.kind); try std.testing.expectEqual(hdr.len, decoded.len); try std.testing.expectEqual(hdr.uid, decoded.uid); try std.testing.expectEqual(hdr.seq, decoded.seq);}
test "header is little-endian" { const hdr = EvtHeader{ .flags = 0, .kind = 1, .len = 256, .uid = 0, .seq = 1 }; var buf: [header_size]u8 = undefined; hdr.encode(&buf); // len=256 in LE is 0x00 0x01 0x00 0x00 try std.testing.expectEqual(@as(u8, 0x00), buf[8]); try std.testing.expectEqual(@as(u8, 0x01), buf[9]);}
test "scanForLastSeq stops at a torn tail" { // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md): // an interrupted flush leaves a partial final record. recovery must stop at the // last fully-written record, not trust the torn record's length and desync. const io = std.testing.io; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
var file = try tmp.dir.createFile(io, "log", .{ .truncate = true, .read = true }); defer file.close(io);
var pos: u64 = 0; const writeRecord = struct { fn go(f: Io.File, tio: Io, at: *u64, seq: u64, payload: []const u8) !void { var hbuf: [header_size]u8 = undefined; (EvtHeader{ .flags = 0, .kind = @backingInt(EvtKind.commit), .len = @intCast(payload.len), .uid = 0, .seq = seq }).encode(&hbuf); try f.writePositionalAll(tio, &hbuf, at.*); at.* += header_size; try f.writePositionalAll(tio, payload, at.*); at.* += payload.len; } }.go;
// two complete records try writeRecord(file, io, &pos, 1, "aa"); try writeRecord(file, io, &pos, 2, "bbbb"); const valid_end = pos;
// torn record: header claims a payload of 999 bytes, only 2 are present var hbuf: [header_size]u8 = undefined; (EvtHeader{ .flags = 0, .kind = @backingInt(EvtKind.commit), .len = 999, .uid = 0, .seq = 3 }).encode(&hbuf); try file.writePositionalAll(io, &hbuf, pos); try file.writePositionalAll(io, "xy", pos + header_size);
const scan = try scanForLastSeq(file, io, 1); try std.testing.expectEqual(@as(?u64, 2), scan.last_seq); // torn record 3 rejected try std.testing.expectEqual(valid_end, scan.valid_end); // stops at end of record 2}
test "scanForLastSeq rejects a record with a garbage seq" { // regression for the 2026-06 quintillion-seq incident: a torn tail left bytes // that decode to a structurally-valid header (good kind, plausible len, body // present) but whose seq field is actually CBOR payload (0x6363a382...). the // old scan trusted that seq and relocated the whole sequencer; the seq-chain // check must reject it so recovery resumes in the real space. const io = std.testing.io; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
var file = try tmp.dir.createFile(io, "log", .{ .truncate = true, .read = true }); defer file.close(io);
var pos: u64 = 0; const writeRecord = struct { fn go(f: Io.File, tio: Io, at: *u64, seq: u64, payload: []const u8) !void { var hbuf: [header_size]u8 = undefined; (EvtHeader{ .flags = 0, .kind = @backingInt(EvtKind.commit), .len = @intCast(payload.len), .uid = 0, .seq = seq }).encode(&hbuf); try f.writePositionalAll(tio, &hbuf, at.*); at.* += header_size; try f.writePositionalAll(tio, payload, at.*); at.* += payload.len; } }.go;
// file declares seq_start = 100; two good records continue the dense chain try writeRecord(file, io, &pos, 100, "aa"); try writeRecord(file, io, &pos, 101, "bbbb"); const valid_end = pos;
// a record that is structurally valid (good kind, small len, body present) // but whose seq is the garbage CBOR value — breaks the chain (expected 102) try writeRecord(file, io, &pos, 0x6363a38200000000, "zz");
const scan = try scanForLastSeq(file, io, 100); try std.testing.expectEqual(@as(?u64, 101), scan.last_seq); // garbage-seq record rejected try std.testing.expectEqual(valid_end, scan.valid_end); // stops at end of record 101
// and a file whose very first record is already garbage recovers nothing — // resumeLog then falls back to seq_start rather than the junk value var f2 = try tmp.dir.createFile(io, "log2", .{ .truncate = true, .read = true }); defer f2.close(io); var p2: u64 = 0; try writeRecord(f2, io, &p2, 0x6363a38200000000, "aa"); const scan2 = try scanForLastSeq(f2, io, 100); try std.testing.expectEqual(@as(?u64, null), scan2.last_seq); try std.testing.expectEqual(@as(u64, 0), scan2.valid_end);}
fn requireDatabaseUrl() ![]const u8 { return getenv("DATABASE_URL") orelse return error.SkipZigTest;}
/// per-test database isolation. the test binary's runner executes tests/// concurrently against one server, and the relay tables (log_file_refs //// account / host) are global — so sharing one database lets tests see each/// other's rows (playback(0) returns another test's events). each test gets its/// own freshly-created database instead, dropped on teardown. unique across/// processes (pid) and within one (counter).var test_db_counter: std.atomic.Value(u32) = .{ .raw = 0 };
pub const TestDb = struct { base_url: []const u8, name: []u8, url: []u8,
pub fn create(base_url: []const u8) !TestDb { const pid: u64 = @intCast(std.c.getpid()); const n = test_db_counter.fetchAdd(1, .monotonic); const name = try std.fmt.allocPrint(std.testing.allocator, "zlay_test_{d}_{d}", .{ pid, n }); errdefer std.testing.allocator.free(name);
{ const uri = std.Uri.parse(base_url) catch return error.InvalidDatabaseUrl; const pool = try pg.Pool.initUri(std.testing.io, std.testing.allocator, uri, .{ .size = 1 }); defer pool.deinit(); var buf: [128]u8 = undefined; // name is generated from pid/counter — no injection risk const stmt = try std.fmt.bufPrint(&buf, "CREATE DATABASE \"{s}\"", .{name}); _ = try pool.exec(stmt, .{}); }
const slash = std.mem.lastIndexOfScalar(u8, base_url, '/') orelse return error.InvalidDatabaseUrl; const url = try std.fmt.allocPrint(std.testing.allocator, "{s}/{s}", .{ base_url[0..slash], name }); return .{ .base_url = base_url, .name = name, .url = url }; }
pub fn destroy(self: *TestDb) void { const uri = std.Uri.parse(self.base_url) catch return; const pool = pg.Pool.initUri(std.testing.io, std.testing.allocator, uri, .{ .size = 1 }) catch return; defer pool.deinit(); var buf: [128]u8 = undefined; const stmt = std.fmt.bufPrint(&buf, "DROP DATABASE IF EXISTS \"{s}\" WITH (FORCE)", .{self.name}) catch return; _ = pool.exec(stmt, .{}) catch {}; std.testing.allocator.free(self.name); std.testing.allocator.free(self.url); }};
test "persist and playback" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
// persist some events (sync flush, no background thread) const seq1 = try dp.persist(.commit, 1, "payload-one"); const seq2 = try dp.persist(.identity, 2, "payload-two"); const seq3 = try dp.persist(.commit, 1, "payload-three");
try std.testing.expectEqual(@as(u64, 1), seq1); try std.testing.expectEqual(@as(u64, 2), seq2); try std.testing.expectEqual(@as(u64, 3), seq3);
// flush manually { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// playback from cursor=0 → all events var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &entries, .{});
try std.testing.expectEqual(@as(usize, 3), entries.items.len); try std.testing.expectEqualStrings("payload-one", entries.items[0].data); try std.testing.expectEqual(@as(u64, 1), entries.items[0].seq); try std.testing.expectEqualStrings("payload-two", entries.items[1].data); try std.testing.expectEqualStrings("payload-three", entries.items[2].data);}
test "worker failure preserves upstream cursor and successful retry reaches disk" { const cbor = @import("zat").cbor; const broadcaster = @import("broadcaster.zig"); const validator_mod = @import("validator.zig"); const worker = @import("frame_worker.zig"); const ingest_ack = @import("ingest_ack.zig"); var tdb = try TestDb.create(try requireDatabaseUrl()); defer tdb.destroy(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path); var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io); defer dp.deinit(); var shutdown = std.atomic.Value(bool){ .raw = false }; var bc = broadcaster.Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown); defer bc.deinit(); var validator = validator_mod.Validator.init(std.testing.allocator, &bc.stats, std.testing.io); defer validator.deinit(); var progress: ingest_ack.Progress = .{}; var cursor: ?u64 = 10; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const header = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#identity" } }, } }); const payload = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{ .{ .key = "did", .value = .{ .text = "did:plc:cursor-test" } }, .{ .key = "seq", .value = .{ .unsigned = 11 } }, } });
for (0..2) |attempt| { var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 }); dp.allocator = if (attempt == 0) failing.allocator() else std.testing.allocator; defer dp.allocator = std.testing.allocator; var work: worker.FrameWork = .{ .data = try std.mem.concat(std.testing.allocator, u8, &.{ header, payload }), .hostname = try std.testing.allocator.dupe(u8, "example.com"), .host_id = 1, .allocator = std.testing.allocator, .io = std.testing.io, .bc = &bc, .validator = &validator, .persist = &dp, .collection_index = null, .resyncer = null, .receipt = try progress.reserve(std.testing.io, 11, &cursor), }; bc.stats.pool_queued_bytes.store(work.data.len, .monotonic); worker.processFrame(&work); if (attempt == 0) { try std.testing.expectError(error.FrameProcessingFailed, progress.drain(&cursor, true)); try std.testing.expectEqual(@as(?u64, 10), cursor); try std.testing.expectEqual(@as(u64, 1), dp.cur_seq); progress.reset(); } else { try progress.drain(&cursor, true); try std.testing.expectEqual(@as(?u64, 11), cursor); try std.testing.expectEqual(@as(u64, 1), dp.committed_seq.load(.acquire)); const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1); try std.testing.expectEqual(@as(?u64, 1), scan.last_seq); } }}
test "persist allocation failures preserve sequence and buffered records" { const Case = struct { fn run(allocator: Allocator) !void { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true }); defer file.close(std.testing.io); var dp: DiskPersist = .{ .allocator = allocator, .dir_path = "", .dir = tmp.dir, .db = undefined, .current_file = file, .did_cache = undefined, .host_id_cache = undefined, .io = std.testing.io, }; defer { for (dp.evtbuf.items) |job| allocator.free(job.data); dp.evtbuf.deinit(allocator); dp.outbuf.deinit(allocator); } const seq = dp.persist(.commit, 1, "payload") catch |err| { try std.testing.expectEqual(@as(u64, 1), dp.cur_seq); try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len); try std.testing.expectEqual(@as(usize, 0), dp.outbuf.items.len); return err; }; try std.testing.expectEqual(@as(u64, 1), seq); try dp.flushLocked(); const scan = try scanForLastSeq(file, std.testing.io, 1); try std.testing.expectEqual(@as(?u64, 1), scan.last_seq); } }; try std.testing.checkAllAllocationFailures(std.testing.allocator, Case.run, .{});}
test "failed threshold write retains ownership and bounds pending events until retry" { var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true }); defer file.close(std.testing.io); const read_only = try tmp.dir.openFile(std.testing.io, "events", .{}); defer read_only.close(std.testing.io); var dp: DiskPersist = .{ .allocator = std.testing.allocator, .dir_path = "", .dir = tmp.dir, .db = undefined, .current_file = read_only, .did_cache = undefined, .host_id_cache = undefined, .io = std.testing.io, }; defer { for (dp.evtbuf.items) |job| std.testing.allocator.free(job.data); dp.evtbuf.deinit(std.testing.allocator); dp.outbuf.deinit(std.testing.allocator); } for (0..default_flush_threshold) |_| _ = try dp.persist(.commit, 1, "payload"); try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire)); try std.testing.expectError(error.NotOpenForWriting, dp.persist(.commit, 1, "next")); try std.testing.expectEqual(default_flush_threshold, dp.evtbuf.items.len); try std.testing.expectEqual(default_flush_threshold + 1, dp.cur_seq); dp.current_file = file; try dp.flush(); const scan = try scanForLastSeq(file, std.testing.io, 1); try std.testing.expectEqual(@as(?u64, default_flush_threshold), scan.last_seq); try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len);}
test "failed rotation preserves the active file and retries after database recovery" { var tdb = try TestDb.create(try requireDatabaseUrl()); defer tdb.destroy(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path); var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io); defer dp.deinit();
_ = try dp.persist(.commit, 1, "before"); try dp.flushLocked(); const old_pos = dp.current_file_pos; { var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 }); dp.allocator = failing.allocator(); defer dp.allocator = std.testing.allocator; try std.testing.expectError(error.OutOfMemory, dp.createLogFile(2)); } try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); try tmp.dir.createDir(std.testing.io, "evts-2", .default_dir); if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {} try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); try std.testing.expectEqual(old_pos, dp.current_file_pos); try tmp.dir.deleteDir(std.testing.io, "evts-2");
_ = try dp.db.exec("ALTER TABLE log_file_refs ADD CONSTRAINT reject_rotation CHECK (seq_start < 2)", .{}); if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {} try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); try std.testing.expectError(error.FileNotFound, tmp.dir.openFile(std.testing.io, "evts-2", .{})); _ = try dp.persist(.commit, 1, "after"); try dp.flushLocked(); const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1); try std.testing.expectEqual(@as(?u64, 2), scan.last_seq);
_ = try dp.db.exec("ALTER TABLE log_file_refs DROP CONSTRAINT reject_rotation", .{}); try dp.createLogFile(3); _ = try dp.persist(.commit, 1, "rotated"); try dp.flushLocked(); var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |entry| std.testing.allocator.free(entry.data); entries.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &entries, .{}); try std.testing.expectEqual(@as(usize, 3), entries.items.len);}
test "gc deletes expired segments and their refs, keeps the current file" { // written 2026-08-27, when production surfaced 10 weeks of never-pruned // segments (132G/150G): gc() had no test at all. this exercises the whole // path against real postgres — the $1::interval bind, the row decode // (whose `nextUnsafe() catch null` would silently turn a query error into // "nothing expired"), the ref-before-file delete order, and the // current-file skip. const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy();
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io); defer dp.deinit(); dp.events_per_file = 2; // force rotation quickly dp.retention_hours = 1;
// 5 events at 2/file, flushing each (rotation happens at flush end once // the counter passes events_per_file) → at least 3 segments registered for (0..5) |i| { var payload_buf: [16]u8 = undefined; const payload = std.fmt.bufPrint(&payload_buf, "payload-{d}", .{i}) catch unreachable; _ = try dp.persist(.commit, 1, payload); dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
var ref_count: i64 = 0; { var result = try dp.db.query("SELECT count(*) FROM log_file_refs", .{}); defer result.deinit(); while (result.nextUnsafe() catch null) |r| ref_count = r.get(i64, 0); } try std.testing.expect(ref_count >= 3);
// nothing is expired yet: gc must be a no-op try dp.gc(); var after_noop: i64 = 0; { var result = try dp.db.query("SELECT count(*) FROM log_file_refs", .{}); defer result.deinit(); while (result.nextUnsafe() catch null) |r| after_noop = r.get(i64, 0); } try std.testing.expectEqual(ref_count, after_noop);
// age every ref past retention, then gc: everything but the current // file (and its ref... the ref of the skipped file survives) must go dp.retention_seconds = 60; _ = try dp.db.exec("UPDATE log_file_refs SET created_at = now() - interval '2 minutes'", .{}); try dp.gc();
var survivors: std.ArrayListUnmanaged([]u8) = .empty; defer { for (survivors.items) |s| std.testing.allocator.free(s); survivors.deinit(std.testing.allocator); } { var result = try dp.db.query("SELECT path FROM log_file_refs ORDER BY seq_start", .{}); defer result.deinit(); while (result.nextUnsafe() catch null) |r| { try survivors.append(std.testing.allocator, try std.testing.allocator.dupe(u8, r.get([]const u8, 0))); } } // only the current file's ref remains try std.testing.expectEqual(@as(usize, 1), survivors.items.len); try std.testing.expectEqualStrings(dp.current_file_path.?, survivors.items[0]);
// expired files are gone from disk; the current file remains var dir = try std.Io.Dir.cwd().openDir(std.testing.io, dir_path, .{ .iterate = true }); defer dir.close(std.testing.io); var files_on_disk: usize = 0; var it = dir.iterate(); while (try it.next(std.testing.io)) |entry| { if (entry.kind != .file) continue; if (!std.mem.startsWith(u8, entry.name, "evts-")) continue; files_on_disk += 1; try std.testing.expectEqualStrings(dp.current_file_path.?, entry.name); } try std.testing.expectEqual(@as(usize, 1), files_on_disk);
// orphan sweep: an unreferenced file below the retained window (the // production 100G leak: row deleted, file delete failed) is removed; // an unreferenced file at/above the retained floor is left alone { const old_orphan = try dir.createFile(std.testing.io, "evts-1", .{}); old_orphan.close(std.testing.io); const new_unregistered = try dir.createFile(std.testing.io, "evts-999999", .{}); new_unregistered.close(std.testing.io); } try dp.gc(); try std.testing.expectError(error.FileNotFound, dir.openFile(std.testing.io, "evts-1", .{})); const kept = try dir.openFile(std.testing.io, "evts-999999", .{}); kept.close(std.testing.io); dir.deleteFile(std.testing.io, "evts-999999") catch {};}
test "committed_seq advances only on durable flush" { // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md): // the broadcaster fiber only emits a frame once its seq <= committed_seq, and // committed_seq advances ONLY here, after the durable write. so a frame can // never reach consumers before it is on disk, and the seq recovered on restart // can't trail an already-broadcast seq. const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
// fresh log: nothing durable yet try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));
// persist below the flush threshold — assigned seqs but NOT yet committed, // so the broadcaster would hold all three back. _ = try dp.persist(.commit, 1, "a"); _ = try dp.persist(.commit, 2, "b"); _ = try dp.persist(.commit, 3, "c"); try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));
// flush → committed_seq jumps to the highest durable seq, releasing all three { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); } try std.testing.expectEqual(@as(u64, 3), dp.committed_seq.load(.acquire));}
test "playback with cursor" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
_ = try dp.persist(.commit, 1, "a"); _ = try dp.persist(.commit, 2, "b"); _ = try dp.persist(.commit, 3, "c"); { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// playback from cursor=2 → only seq 3 var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } try dp.playback(2, std.testing.allocator, &entries, .{});
try std.testing.expectEqual(@as(usize, 1), entries.items.len); try std.testing.expectEqual(@as(u64, 3), entries.items[0].seq); try std.testing.expectEqualStrings("c", entries.items[0].data);}
test "playback respects entry and byte limits, resumable by cursor" { // regression for the 2026-08-06 consumer-flap OOM: an old cursor must // not materialize the whole backlog in one playback call. const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
for (0..10) |_| _ = try dp.persist(.commit, 1, "0123456789"); { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// max_entries caps the chunk; resuming from the last seq gets the rest var chunk: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (chunk.items) |e| std.testing.allocator.free(e.data); chunk.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &chunk, .{ .max_entries = 4 }); try std.testing.expectEqual(@as(usize, 4), chunk.items.len); try std.testing.expectEqual(@as(u64, 4), chunk.items[3].seq);
var rest: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (rest.items) |e| std.testing.allocator.free(e.data); rest.deinit(std.testing.allocator); } try dp.playback(chunk.items[3].seq, std.testing.allocator, &rest, .{ .max_entries = 100 }); try std.testing.expectEqual(@as(usize, 6), rest.items.len); try std.testing.expectEqual(@as(u64, 5), rest.items[0].seq); try std.testing.expectEqual(@as(u64, 10), rest.items[5].seq);
// max_bytes caps the chunk: 10-byte payloads, 25-byte budget → the // check runs before each read, so the 3rd entry crosses the budget // and the 4th is never read var byte_chunk: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (byte_chunk.items) |e| std.testing.allocator.free(e.data); byte_chunk.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &byte_chunk, .{ .max_bytes = 25 }); try std.testing.expectEqual(@as(usize, 3), byte_chunk.items.len);}
test "seq recovery after reinit" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
// write some events { var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit(); _ = try dp.persist(.commit, 1, "x"); _ = try dp.persist(.commit, 2, "y"); _ = try dp.persist(.account, 3, "z"); dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// reinit — should recover seq { var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit(); try std.testing.expectEqual(@as(u64, 3), dp.lastSeq().?); const seq4 = try dp.persist(.commit, 1, "w"); try std.testing.expectEqual(@as(u64, 4), seq4); }}
test "takedown zeros payload" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
_ = try dp.persist(.commit, 42, "secret-data"); _ = try dp.persist(.commit, 99, "other-data"); { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// take down user 42 try dp.takeDownUser(42);
// playback should skip user 42's events var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &entries, .{});
try std.testing.expectEqual(@as(usize, 1), entries.items.len); try std.testing.expectEqualStrings("other-data", entries.items[0].data);}
test "uidForDid assigns and caches UIDs" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
// first call creates the account const uid1 = try dp.uidForDid("did:plc:alice"); try std.testing.expect(uid1 > 0);
// same DID returns same UID (from cache) const uid1_again = try dp.uidForDid("did:plc:alice"); try std.testing.expectEqual(uid1, uid1_again);
// different DID gets a different UID const uid2 = try dp.uidForDid("did:plc:bob"); try std.testing.expect(uid2 > 0); try std.testing.expect(uid1 != uid2);}
test "accountActivity distinguishes local takedown from stale upstream status" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy();
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io); defer dp.deinit();
const uid = try dp.uidForDid("did:plc:migrant");
// fresh account: fully active var activity = try dp.accountActivity(uid); try std.testing.expect(activity.local_ok and activity.upstream_ok); try std.testing.expect(activity.active());
// the migration-stale shape: locally active, upstream deactivated. // this is the only combination that may trigger a StatusChecker re-check. try dp.updateAccountUpstreamStatus(uid, "deactivated"); activity = try dp.accountActivity(uid); try std.testing.expect(activity.local_ok); try std.testing.expect(!activity.upstream_ok); try std.testing.expect(!activity.active()); try std.testing.expect(!try dp.isAccountActive(uid));
// recovery: the checker writes the PDS's answer back try dp.updateAccountUpstreamStatus(uid, "active"); activity = try dp.accountActivity(uid); try std.testing.expect(activity.active()); try std.testing.expect(try dp.isAccountActive(uid));
// local takedown: local_ok false — must never look like the re-check shape _ = try dp.db.exec("UPDATE account SET status = 'takendown' WHERE uid = $1", .{@as(i64, @intCast(uid))}); activity = try dp.accountActivity(uid); try std.testing.expect(!activity.local_ok); try std.testing.expect(activity.upstream_ok); try std.testing.expect(!activity.active());
// unknown uid: inactive on both axes activity = try dp.accountActivity(999_999); try std.testing.expect(!activity.local_ok and !activity.upstream_ok);}
test "getAccountState retains rev-only legacy state" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy();
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io); defer dp.deinit();
const uid = try dp.uidForDid("did:plc:revonly"); try std.testing.expect(try dp.updateAccountState(uid, "3k2abcdefghij", "", ""));
var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const state = (try dp.getAccountState(uid, arena.allocator())).?; try std.testing.expectEqualStrings("3k2abcdefghij", state.rev); try std.testing.expectEqualStrings("", state.data_cid);}
test "host_id cache stays coherent with setAccountHostId" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
const uid = try dp.uidForDid("did:plc:hostcache");
// unset host_id reads as 0 and is not cached (transient pre-assignment state) try std.testing.expectEqual(@as(u64, 0), try dp.getAccountHostId(uid));
// write-through: set populates the cache, read returns it try dp.setAccountHostId(uid, 42); try std.testing.expectEqual(@as(u64, 42), try dp.getAccountHostId(uid));
// a host change updates the cache in place — no stale read try dp.setAccountHostId(uid, 99); try std.testing.expectEqual(@as(u64, 99), try dp.getAccountHostId(uid));
// clearing to 0 evicts the entry; read falls through to the DB (0) try dp.setAccountHostId(uid, 0); try std.testing.expectEqual(@as(u64, 0), try dp.getAccountHostId(uid));
// value survives a cache miss: clear the cache, read repopulates from DB try dp.setAccountHostId(uid, 7); _ = dp.host_id_cache.remove(&DiskPersist.uidKey(uid)); try std.testing.expectEqual(@as(u64, 7), try dp.getAccountHostId(uid));}
test "uidForDid survives reinit" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var uid1: u64 = undefined; { var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit(); uid1 = try dp.uidForDid("did:plc:carol"); }
// reinit — UID should be the same from database { var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit(); const uid1_again = try dp.uidForDid("did:plc:carol"); try std.testing.expectEqual(uid1, uid1_again); }}
test "takedown with real UIDs" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
const alice_uid = try dp.uidForDid("did:plc:alice"); const bob_uid = try dp.uidForDid("did:plc:bob");
_ = try dp.persist(.commit, alice_uid, "alice-post"); _ = try dp.persist(.commit, bob_uid, "bob-post"); _ = try dp.persist(.commit, alice_uid, "alice-post-2"); { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); }
// take down alice try dp.takeDownUser(alice_uid);
// playback should only have bob's event var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } try dp.playback(0, std.testing.allocator, &entries, .{});
try std.testing.expectEqual(@as(usize, 1), entries.items.len); try std.testing.expectEqualStrings("bob-post", entries.items[0].data); try std.testing.expectEqual(bob_uid, entries.items[0].uid);}
fn tmpDirRealPath(allocator: Allocator, tmp: std.testing.TmpDir) ![]const u8 { var buf: [Io.Dir.max_path_bytes]u8 = undefined; const n = try tmp.dir.realPath(std.testing.io, &buf); return try allocator.dupe(u8, buf[0..n]);}
test "playback does not take the persist mutex" { // regression: 2026-10-18 — playback held DiskPersist.mutex for every // replay chunk, which blocked persist and flush and froze committed_seq. const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
for (0..20) |_| _ = try dp.persist(.commit, 1, "payload"); try dp.flush();
// hold the mutex from this thread; playback on another thread must finish anyway dp.mutex.lockUncancelable(dp.io); const Reader = struct { fn run(d: *DiskPersist, out: *std.ArrayListUnmanaged(PlaybackEntry), ok: *std.atomic.Value(bool)) void { d.playback(5, std.testing.allocator, out, .{}) catch return; ok.store(true, .release); } }; var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } var ok = std.atomic.Value(bool){ .raw = false }; const t = try std.Thread.spawn(.{}, Reader.run, .{ &dp, &entries, &ok }); t.join(); dp.mutex.unlock(dp.io);
try std.testing.expect(ok.load(.acquire)); try std.testing.expectEqual(@as(usize, 15), entries.items.len); try std.testing.expectEqual(@as(u64, 6), entries.items[0].seq); try std.testing.expectEqual(@as(u64, 20), entries.items[14].seq);}
test "buffered playback reads records that straddle block boundaries" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); const database_url = tdb.url;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit();
// 700 records of 1,000 bytes: ~720 KB, crossing the 256 KiB block // boundary several times, with one payload larger than a whole block const payload = try std.testing.allocator.alloc(u8, 1000); defer std.testing.allocator.free(payload); for (payload, 0..) |*b, i| b.* = @intCast(i % 251); const big = try std.testing.allocator.alloc(u8, PositionalReader.block_size + 17); defer std.testing.allocator.free(big); @memset(big, 0xAB);
const count: usize = 700; for (0..count) |i| { if (i == 350) { _ = try dp.persist(.commit, 2, big); } else { payload[0] = @intCast(i % 256); _ = try dp.persist(.commit, 1, payload); } } try dp.flush();
var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; defer { for (entries.items) |e| std.testing.allocator.free(e.data); entries.deinit(std.testing.allocator); } try dp.playback(100, std.testing.allocator, &entries, .{}); try std.testing.expectEqual(count - 100, entries.items.len); for (entries.items, 0..) |e, j| { const i = j + 100; try std.testing.expectEqual(@as(u64, i + 1), e.seq); if (i == 350) { try std.testing.expectEqual(big.len, e.data.len); try std.testing.expectEqual(@as(u8, 0xAB), e.data[e.data.len - 1]); } else { try std.testing.expectEqual(@as(usize, 1000), e.data.len); try std.testing.expectEqual(@as(u8, @intCast(i % 256)), e.data[0]); try std.testing.expectEqual(payload[999], e.data[999]); } }}
test "account commit migration preserves roots and updates both CIDs atomically" { const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); defer std.testing.allocator.free(dir_path);
{ var old = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io); defer old.deinit(); _ = try old.db.exec("ALTER TABLE account_repo DROP COLUMN commit_cid CASCADE", .{}); const uid = try old.uidForDid("did:plc:migration"); _ = try old.db.exec("INSERT INTO account_repo (uid, rev, commit_data_cid) VALUES ($1, $2, $3)", .{ @as(i64, @intCast(uid)), "3my46tkfps52u", "old-mst-root" }); }
var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io); defer dp.deinit(); const uid = try dp.uidForDid("did:plc:migration"); { var row = (try dp.db.rowUnsafe("SELECT rev, commit_data_cid, commit_cid FROM account_repo WHERE uid = $1", .{@as(i64, @intCast(uid))})).?; defer row.deinit() catch {}; try std.testing.expectEqualStrings("3my46tkfps52u", row.get([]const u8, 0)); try std.testing.expectEqualStrings("old-mst-root", row.get([]const u8, 1)); try std.testing.expectEqualStrings("", row.get([]const u8, 2)); } try std.testing.expect(try dp.updateAccountState(uid, "3my46tkfps62u", "new-mst-root", "new-commit")); try std.testing.expect(!try dp.updateAccountState(uid, "3my46tkfps52u", "stale-root", "stale-commit")); try std.testing.expect(!try dp.updateAccountState(uid, "3my46tkfps62u", "conflicting-root", "conflicting-commit")); { var row = (try dp.db.rowUnsafe("SELECT rev, commit_data_cid, commit_cid FROM account_repo WHERE uid = $1", .{@as(i64, @intCast(uid))})).?; defer row.deinit() catch {}; try std.testing.expectEqualStrings("3my46tkfps62u", row.get([]const u8, 0)); try std.testing.expectEqualStrings("new-mst-root", row.get([]const u8, 1)); try std.testing.expectEqualStrings("new-commit", row.get([]const u8, 2)); }}
test "live-only sequences do not buffer history or touch replay storage" { var dp: DiskPersist = .{ .allocator = std.testing.failing_allocator, .dir_path = "", .dir = undefined, .db = undefined, .did_cache = undefined, .host_id_cache = undefined, .history_enabled = false, .io = std.testing.io, }; try std.testing.expectEqual(@as(u64, 1), try dp.persist(.commit, 1, "first")); try std.testing.expectEqual(@as(u64, 2), try dp.persist(.commit, 1, "second")); try dp.flush(); try std.testing.expectEqual(@as(u64, 2), dp.committed_seq.load(.acquire)); try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len); try std.testing.expectEqual(@as(usize, 0), dp.outbuf.items.len); var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; try std.testing.expectError(error.ReplayDisabled, dp.playback(0, std.testing.allocator, &entries, .{}));}