diff --git a/src/api/admin.zig b/src/api/admin.zig index 9bc786a..738f0f0 100644 --- a/src/api/admin.zig +++ b/src/api/admin.zig @@ -218,7 +218,7 @@ pub fn handleAdminChangeLimits(conn: *h.Conn, body: []const u8, headers: *const }; // update running subscriber's rate limits immediately - const effective = if (parsed.value.account_limit) |l| l else ctx.persist.getHostAccountCountEv(host_id); + const effective = if (parsed.value.account_limit) |l| l else ctx.persist.getHostAccountCountEv(host_id) catch 0; ctx.slurper.updateHostLimits(host_id, effective); if (parsed.value.account_limit) |limit| { diff --git a/src/api/xrpc.zig b/src/api/xrpc.zig index e360e7a..e8ef2dc 100644 --- a/src/api/xrpc.zig +++ b/src/api/xrpc.zig @@ -35,9 +35,14 @@ pub fn handleListRepos(conn: *h.Conn, query: []const u8, persist: *event_log_mod return; } + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + // query accounts with repo state, paginated by UID // includes both local status and upstream_status for combined active check - var result = persist.ensureEvDb().query( + var result = ev_db.query( \\SELECT a.uid, a.did, a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_data_cid, '') \\FROM account a LEFT JOIN account_repo r ON a.uid = r.uid \\WHERE a.uid > $1 ORDER BY a.uid ASC LIMIT $2 @@ -121,8 +126,13 @@ pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, persist: *event_log return; } + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + // look up account (includes both local and upstream status) - var row = (persist.ensureEvDb().rowUnsafe( + var row = (ev_db.rowUnsafe( "SELECT a.uid, a.status, a.upstream_status, COALESCE(r.rev, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1", .{did}, ) catch { @@ -180,8 +190,13 @@ pub fn handleGetRepo(conn: *h.Conn, query: []const u8, persist: *event_log_mod.D return; } + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + // look up the PDS hostname for this account - var row = (persist.ensureEvDb().rowUnsafe( + var row = (ev_db.rowUnsafe( "SELECT h.hostname FROM account a JOIN host h ON a.host_id = h.id WHERE a.did = $1 AND a.host_id > 0", .{did}, ) catch { @@ -214,8 +229,13 @@ pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, persist: *event_l return; } + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + // look up account + repo state (includes both local and upstream status) - var row = (persist.ensureEvDb().rowUnsafe( + var row = (ev_db.rowUnsafe( "SELECT a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_data_cid, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1", .{did}, ) catch { @@ -345,7 +365,12 @@ pub fn handleListHosts(conn: *h.Conn, query: []const u8, persist: *event_log_mod return; } - var result = persist.ensureEvDb().query( + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + + var result = ev_db.query( "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2", .{ cursor_val, limit }, ) catch { @@ -399,8 +424,13 @@ pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, persist: *event_log return; }; + const ev_db = persist.ensureEvDb() catch { + h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); + return; + }; + // look up host - var row = (persist.ensureEvDb().rowUnsafe( + var row = (ev_db.rowUnsafe( "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1", .{hostname}, ) catch { @@ -426,7 +456,7 @@ pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, persist: *event_log raw_status; // active, idle pass through // count accounts on this host - const account_count: i64 = if (persist.ensureEvDb().rowUnsafe( + const account_count: i64 = if (ev_db.rowUnsafe( "SELECT COUNT(*) FROM account WHERE host_id = $1", .{host_id}, ) catch null) |cnt_row| blk: { @@ -474,7 +504,7 @@ pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, slurper: *slurper_mod defer slurper.allocator.free(hostname); // fast validation: domain ban check (Evented fiber — use Ev pool) - if (slurper.persist.isDomainBannedEv(hostname)) { + if (slurper.persist.isDomainBannedEv(hostname) catch false) { log.warn("requestCrawl rejected '{s}': domain banned", .{hostname}); h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"domain is banned\"}"); return; diff --git a/src/backfill.zig b/src/backfill.zig index 0e6b99a..d62c533 100644 --- a/src/backfill.zig +++ b/src/backfill.zig @@ -31,7 +31,7 @@ pub const Backfiller = struct { source: []const u8, io: Io, - fn db(self: *Backfiller) *pg.Pool { + fn db(self: *Backfiller) !*pg.Pool { return self.persist.ensureEvDb(); } @@ -79,6 +79,11 @@ pub const Backfiller = struct { self.running.store(false, .release); } + const pool = self.db() catch |err| { + log.err("backfill: database unavailable: {s}", .{@errorName(err)}); + return; + }; + // discover collections const collections = self.discoverCollections() catch |err| { log.err("collection discovery failed: {s}", .{@errorName(err)}); @@ -93,7 +98,7 @@ pub const Backfiller = struct { // insert progress rows (skip existing) for (collections) |collection| { - _ = self.db().exec( + _ = pool.exec( "INSERT INTO backfill_progress (collection, source) VALUES ($1, $2) ON CONFLICT (collection, source) DO NOTHING", .{ collection, self.source }, ) catch |err| { @@ -207,12 +212,14 @@ pub const Backfiller = struct { } fn backfillCollection(self: *Backfiller, collection: []const u8) !void { + const pool = try self.db(); + // single query: check completion, get cursor + count for resume var cursor: ?[]const u8 = null; defer if (cursor) |c| self.allocator.free(c); var imported: i64 = 0; { - var row = (self.db().rowUnsafe( + var row = (pool.rowUnsafe( "SELECT completed_at IS NOT NULL, cursor, imported_count FROM backfill_progress WHERE collection = $1 AND source = $2", .{ collection, self.source }, ) catch return error.DatabaseError) orelse return; @@ -255,7 +262,7 @@ pub const Backfiller = struct { // update cursor in progress table const new_cursor = fetch_result.next_cursor orelse ""; - _ = self.db().exec( + _ = pool.exec( "UPDATE backfill_progress SET cursor = $1, imported_count = $2 WHERE collection = $3 AND source = $4", .{ new_cursor, imported, collection, self.source }, ) catch {}; @@ -269,7 +276,7 @@ pub const Backfiller = struct { self.io.sleep(Io.Duration.fromMilliseconds(100), .awake) catch {}; } else { // no more pages — mark complete - _ = self.db().exec( + _ = pool.exec( "UPDATE backfill_progress SET completed_at = now(), cursor = '', imported_count = $1 WHERE collection = $2 AND source = $3", .{ imported, collection, self.source }, ) catch {}; @@ -342,6 +349,8 @@ pub const Backfiller = struct { /// return status summary for the admin endpoint pub fn getStatus(self: *Backfiller, allocator: Allocator) ![]u8 { + const pool = try self.db(); + var aw: Io.Writer.Allocating = .init(allocator); defer aw.deinit(); const w = &aw.writer; @@ -351,7 +360,7 @@ pub const Backfiller = struct { var completed: i64 = 0; var total_imported: i64 = 0; { - var row = (self.db().rowUnsafe( + var row = (pool.rowUnsafe( "SELECT COUNT(*)::bigint, COUNT(completed_at)::bigint, COALESCE(SUM(imported_count), 0)::bigint FROM backfill_progress", .{}, ) catch null) orelse null; @@ -372,7 +381,7 @@ pub const Backfiller = struct { }) catch return error.FormatError; // per-collection detail - var result = self.db().query( + var result = pool.query( "SELECT collection, source, cursor, imported_count, completed_at IS NOT NULL FROM backfill_progress ORDER BY collection, source", .{}, ) catch return error.DatabaseError; diff --git a/src/broadcaster.zig b/src/broadcaster.zig index 2197f53..43e81b5 100644 --- a/src/broadcaster.zig +++ b/src/broadcaster.zig @@ -782,7 +782,7 @@ pub const Handler = struct { // OutdatedCursor: cursor older than oldest available — info, continue const oldest = blk: { if (ctx.persist) |dp| { - if (dp.firstSeqEv()) |s| break :blk s; + if (dp.firstSeqEv() catch null) |s| break :blk s; } break :blk ctx.history.oldestSeq() orelse 0; }; diff --git a/src/cleaner.zig b/src/cleaner.zig index 3846ad4..39ac826 100644 --- a/src/cleaner.zig +++ b/src/cleaner.zig @@ -23,7 +23,7 @@ pub const Cleaner = struct { scanned: std.atomic.Value(u64), removed: std.atomic.Value(u64), - fn db(self: *Cleaner) *pg.Pool { + fn db(self: *Cleaner) !*pg.Pool { return self.persist.ensureEvDb(); } @@ -69,12 +69,17 @@ pub const Cleaner = struct { log.info("cleanup started", .{}); + const pool = self.db() catch |err| { + log.err("cleanup: database unavailable: {s}", .{@errorName(err)}); + return; + }; + // page through inactive accounts by uid var last_uid: i64 = 0; while (true) { var batch_count: u64 = 0; { - var result = self.db().query( + var result = pool.query( "SELECT uid, did FROM account WHERE (status != 'active' OR upstream_status != 'active') AND uid > $1 ORDER BY uid LIMIT 500", .{last_uid}, ) catch |err| { diff --git a/src/event_log.zig b/src/event_log.zig index 0f4adbb..f7362d9 100644 --- a/src/event_log.zig +++ b/src/event_log.zig @@ -144,16 +144,17 @@ pub const DiskPersist = struct { /// pg.Pool.initUri does TCP connects via io_uring, so it can only run inside /// an Evented fiber (after the event loop is spinning). callers from main() /// init set ev_io/db_url/ev_db_pool_size; the actual pool is created here. - pub fn ensureEvDb(self: *DiskPersist) *pg.Pool { + /// on failure, resets state to uninit so the next call retries. + pub fn ensureEvDb(self: *DiskPersist) !*pg.Pool { // fast path — already initialized (single atomic load) if (self.ev_db) |db| return db; - const ev_io = self.ev_io orelse @panic("ensureEvDb: ev_io not set"); + const ev_io = self.ev_io orelse return error.EvDbNotConfigured; while (true) { const state: EvDbInit = @enumFromInt(self.ev_db_state.load(.acquire)); switch (state) { - .ready => return self.ensureEvDb(), + .ready => return self.ev_db orelse error.EvDbNotConfigured, .uninit => { if (self.ev_db_state.cmpxchgWeak( @intFromEnum(EvDbInit.uninit), @@ -163,7 +164,7 @@ pub const DiskPersist = struct { ) == null) { // won the race — create the pool const uri = std.Uri.parse(self.db_url) catch - @panic("ensureEvDb: invalid DATABASE_URL"); + return error.InvalidDatabaseUrl; self.ev_db = pg.Pool.initUri( self.allocator, ev_io, @@ -171,11 +172,13 @@ pub const DiskPersist = struct { .{ .size = self.ev_db_pool_size }, ) catch |err| { log.err("ensureEvDb: initUri failed: {s}", .{@errorName(err)}); - @panic("ensureEvDb: failed to create Evented pg.Pool"); + // reset to uninit so next call retries + self.ev_db_state.store(@intFromEnum(EvDbInit.uninit), .release); + return error.EvDbInitFailed; }; self.ev_db_state.store(@intFromEnum(EvDbInit.ready), .release); log.info("lazy-initialized Evented pg.Pool (size={d})", .{self.ev_db_pool_size}); - return self.ensureEvDb(); + return self.ev_db.?; } // lost CAS — another fiber is initializing, fall through }, @@ -499,8 +502,8 @@ pub const DiskPersist = struct { } /// count accounts on a host (Evented pool) - pub fn getHostAccountCountEv(self: *DiskPersist, host_id: u64) u64 { - return getHostAccountCountImpl(host_id, self.ensureEvDb()); + pub fn getHostAccountCountEv(self: *DiskPersist, host_id: u64) !u64 { + return getHostAccountCountImpl(host_id, try self.ensureEvDb()); } fn getHostAccountCountImpl(host_id: u64, db: *pg.Pool) u64 { @@ -519,8 +522,8 @@ pub const DiskPersist = struct { } /// effective account count (Evented pool) - pub fn getEffectiveAccountCountEv(self: *DiskPersist, host_id: u64) u64 { - return getEffectiveAccountCountImpl(host_id, self.ensureEvDb()); + pub fn getEffectiveAccountCountEv(self: *DiskPersist, host_id: u64) !u64 { + return getEffectiveAccountCountImpl(host_id, try self.ensureEvDb()); } /// uses admin-configured limit if set, otherwise actual COUNT(*). @@ -541,7 +544,7 @@ pub const DiskPersist = struct { /// set host account limit (Evented pool) pub fn setHostAccountLimitEv(self: *DiskPersist, host_id: u64, limit: ?u64) !void { - return setHostAccountLimitImpl(host_id, limit, self.ensureEvDb()); + return setHostAccountLimitImpl(host_id, limit, try self.ensureEvDb()); } /// pass null to clear the override and revert to actual COUNT(*). @@ -613,7 +616,7 @@ pub const DiskPersist = struct { /// get or create a host row (Evented pool) pub fn getOrCreateHostEv(self: *DiskPersist, hostname: []const u8) !HostResult { - return getOrCreateHostImpl(hostname, self.ensureEvDb()); + return getOrCreateHostImpl(hostname, try self.ensureEvDb()); } fn getOrCreateHostImpl(hostname: []const u8, db: *pg.Pool) !HostResult { @@ -642,8 +645,8 @@ pub const DiskPersist = struct { } /// check if a host is banned or blocked by status (Evented pool) - pub fn isHostBannedEv(self: *DiskPersist, hostname: []const u8) bool { - return isHostBannedImpl(hostname, self.ensureEvDb()); + pub fn isHostBannedEv(self: *DiskPersist, hostname: []const u8) !bool { + return isHostBannedImpl(hostname, try self.ensureEvDb()); } fn isHostBannedImpl(hostname: []const u8, db: *pg.Pool) bool { @@ -674,7 +677,7 @@ pub const DiskPersist = struct { /// look up host ID by hostname (Evented pool) pub fn getHostIdForHostnameEv(self: *DiskPersist, hostname: []const u8) !?u64 { - return getHostIdForHostnameImpl(hostname, self.ensureEvDb()); + return getHostIdForHostnameImpl(hostname, try self.ensureEvDb()); } fn getHostIdForHostnameImpl(hostname: []const u8, db: *pg.Pool) !?u64 { @@ -693,7 +696,7 @@ pub const DiskPersist = struct { /// update host status (Evented pool) pub fn updateHostStatusEv(self: *DiskPersist, host_id: u64, status: []const u8) !void { - return updateHostStatusImpl(host_id, status, self.ensureEvDb()); + return updateHostStatusImpl(host_id, status, try self.ensureEvDb()); } fn updateHostStatusImpl(host_id: u64, status: []const u8, db: *pg.Pool) !void { @@ -710,7 +713,7 @@ pub const DiskPersist = struct { /// list all active hosts (Evented pool) pub fn listActiveHostsEv(self: *DiskPersist, allocator: Allocator) ![]Host { - return listActiveHostsImpl(allocator, self.ensureEvDb()); + return listActiveHostsImpl(allocator, try self.ensureEvDb()); } fn listActiveHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host { @@ -750,7 +753,7 @@ pub const DiskPersist = struct { /// list all hosts (Evented pool) pub fn listAllHostsEv(self: *DiskPersist, allocator: Allocator) ![]Host { - return listAllHostsImpl(allocator, self.ensureEvDb()); + return listAllHostsImpl(allocator, try self.ensureEvDb()); } fn listAllHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host { @@ -803,8 +806,8 @@ pub const DiskPersist = struct { } /// check if a hostname (or any parent domain) is banned (Evented pool). - pub fn isDomainBannedEv(self: *DiskPersist, hostname: []const u8) bool { - return self.isDomainBannedImpl(hostname, self.ensureEvDb()); + pub fn isDomainBannedEv(self: *DiskPersist, hostname: []const u8) !bool { + return self.isDomainBannedImpl(hostname, try self.ensureEvDb()); } /// Go relay: domain_ban.go DomainIsBanned — suffix-based check. @@ -836,7 +839,7 @@ pub const DiskPersist = struct { /// reset failure count (Evented pool) pub fn resetHostFailuresEv(self: *DiskPersist, host_id: u64) !void { - return resetHostFailuresImpl(host_id, self.ensureEvDb()); + return resetHostFailuresImpl(host_id, try self.ensureEvDb()); } fn resetHostFailuresImpl(host_id: u64, db: *pg.Pool) !void { @@ -849,7 +852,7 @@ pub const DiskPersist = struct { /// resolve a DID to UID using the Evented pool. skips the DID cache /// (which uses pool_io mutex). only used from admin ban (rare path). pub fn uidForDidEv(self: *DiskPersist, did: []const u8) !u64 { - const db = self.ensureEvDb(); + const db = try self.ensureEvDb(); // check database if (try db.rowUnsafe( "SELECT uid FROM account WHERE did = $1", @@ -998,8 +1001,8 @@ pub const DiskPersist = struct { } /// oldest available sequence number (Evented pool) - pub fn firstSeqEv(self: *DiskPersist) ?u64 { - return firstSeqImpl(self.ensureEvDb()); + pub fn firstSeqEv(self: *DiskPersist) !?u64 { + return firstSeqImpl(try self.ensureEvDb()); } fn firstSeqImpl(db: *pg.Pool) ?u64 { diff --git a/src/slurper.zig b/src/slurper.zig index f667f46..059fd79 100644 --- a/src/slurper.zig +++ b/src/slurper.zig @@ -363,7 +363,7 @@ pub const Slurper = struct { defer self.allocator.free(normalized); // skip banned domains (Evented fiber — use Ev pool) - if (self.persist.isDomainBannedEv(normalized)) continue; + if (self.persist.isDomainBannedEv(normalized) catch true) continue; // insert into DB (no describeServer check — the seed relay already vetted them) _ = self.persist.getOrCreateHostEv(normalized) catch continue; @@ -419,14 +419,14 @@ pub const Slurper = struct { // step 2: domain ban check (Evented fiber — use Ev pool) // Go relay: domain_ban.go DomainIsBanned - if (self.persist.isDomainBannedEv(hostname)) { + if (self.persist.isDomainBannedEv(hostname) catch true) { log.warn("host {s}: domain is banned, rejecting", .{hostname}); return; } // step 3: check if host is banned/blocked in DB (Evented pool) // Go relay: crawl.go checks host.Status == HostStatusBanned - if (self.persist.isHostBannedEv(hostname)) { + if (self.persist.isHostBannedEv(hostname) catch true) { log.warn("host {s}: banned/blocked in DB, rejecting", .{hostname}); return; } @@ -468,7 +468,7 @@ pub const Slurper = struct { const sub = try self.allocator.create(subscriber_mod.Subscriber); errdefer self.allocator.destroy(sub); - const account_count: u64 = self.persist.getEffectiveAccountCountEv(host_id); + const account_count: u64 = self.persist.getEffectiveAccountCountEv(host_id) catch 0; sub.* = subscriber_mod.Subscriber.init( self.allocator,