Something went wrong. Try again.
atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735//! zat relay — AT Protocol firehose relay server//!//! crawls PDS instances directly via the Slurper (one subscriber per host),//! validates frames via DID resolution and signature verification, persists//! to disk with relay-assigned seq numbers, and rebroadcasts to downstream//! consumers over WebSocket.//!//! port 3000 (RELAY_PORT): WebSocket firehose + HTTP API (via httpFallback)//! /xrpc/com.atproto.sync.subscribeRepos — firehose WebSocket (supports ?cursor=N)//! /xrpc/com.atproto.sync.listRepos — paginated account listing//! /xrpc/com.atproto.sync.getRepoStatus — single account status//! /xrpc/com.atproto.sync.getLatestCommit — latest commit CID + rev//! /xrpc/com.atproto.sync.listReposByCollection — repos with records in a collection//! /xrpc/com.atproto.sync.listHosts — paginated active host listing//! /xrpc/com.atproto.sync.getHostStatus — single host status//! /xrpc/com.atproto.sync.requestCrawl — request PDS crawl (POST)//! /admin/hosts — list all hosts (GET, admin)//! /admin/hosts/block — block a host (POST, admin)//! /admin/hosts/unblock — unblock a host (POST, admin)//! /admin/ingest-stall — liveness stall detector knobs (GET/POST, admin)//! /admin/ws-ping — websocket keepalive pinger knobs (GET/POST, admin)//! /_health, /_stats — health, stats//!//! port 3001 (RELAY_METRICS_PORT): internal metrics + health//! /metrics — prometheus metrics//! /_healthz — liveness probe (503 on process-wide ingest stall)//! /_health, /_readyz — readiness probe (DB check)
const std = @import("std");const Io = std.Io;const http = std.http;const websocket = @import("websocket");const broadcaster = @import("internal/broadcaster.zig");const validator_mod = @import("internal/validator.zig");const slurper_mod = @import("internal/slurper.zig");const event_log_mod = @import("internal/event_log.zig");const collection_index_mod = @import("internal/collection_index/index.zig");const backfill_mod = @import("internal/collection_index/backfill.zig");const cleaner_mod = @import("internal/collection_index/cleaner.zig");const resync_mod = @import("internal/collection_index/resync.zig");const status_checker_mod = @import("internal/status_checker.zig");const host_ops_mod = @import("internal/host_ops.zig");const ingest_liveness_mod = @import("internal/ingest_liveness.zig");const ws_ping_mod = @import("internal/ws_ping.zig");const api = @import("internal/api/main.zig");const util = @import("internal/util/util.zig");const build_options = @import("build_options");
const getenv = util.getenv;const parseEnvInt = util.parseEnvInt;const malloc_trim: ?*const fn (pad: usize) callconv(.c) c_int = if (builtin.os.tag == .linux) @extern(*const fn (pad: usize) callconv(.c) c_int, .{ .name = "malloc_trim" })else null;
const log = std.log.scoped(.relay);
pub const default_stack_size = 8 * 1024 * 1024;
// -- Io backend selection --// Io.Threaded: one OS thread per io.concurrent() call (~2,800 for PDS subscribers).// Higher thread count than Evented (~35) but proven stable at 99%+ coverage on 0.15.//// Evented (io_uring fibers) shelved: 8 crash classes from cross-Io violations,// ReleaseSafe GPF from fiber context-switch codegen bug, persistent ~10-15% coverage// degradation, zig team marks Evented as experimental. See docs/evented-attempt.md.// Can revisit when zig's Evented runtime matures.const backend_config = @import("backend_config");const zio = @import("zio");
/// -Dbackend=zio builds the same source against lalinsky/zio: stackful/// coroutines scheduled across executor threads with work stealing, driven by/// io_uring (epoll fallback) on linux and kqueue on macOS. unlike Io.Threaded/// it is a complete std.Io of its own, so a blocking libc call made from a/// fiber stalls an executor thread rather than a dedicated one.var threaded_backend: Io.Threaded = undefined;var zio_runtime: *zio.Runtime = undefined;var debug_threaded_io: Io.Threaded = undefined;
/// override single-threaded debug_io default — required for std.debug.print safety/// when multiple OS threads exist (frame worker pool, websocket server).pub const std_options_debug_threaded_io: ?*Io.Threaded = &debug_threaded_io;
var shutdown_flag: std.atomic.Value(bool) = .{ .raw = false };
/// the zio scheduler block of /metrics. module-level (not inline in the/// handler) so the fits-in-buffer test below can render it: on 2026-08-27 a/// block like this outgrew its fixed buffer and a `catch ""` fallback silently/// dropped every runtime metric from the scrape. growth in HELP text must fail/// the suite, not blank the scrape.const zio_metrics_fmt = \\# HELP zio_sched_executors executor threads running fibers \\# TYPE zio_sched_executors gauge \\zio_sched_executors {d} \\# HELP zio_sched_parks_total executor parks (doze = capped grace park, full = indefinite). all executors parked while ingest is flat = nothing runnable, not a wedge \\# TYPE zio_sched_parks_total counter \\zio_sched_parks_total{{kind="doze"}} {d} \\zio_sched_parks_total{{kind="full"}} {d} \\# HELP zio_sched_park_elections_total park exits where a pusher had claimed the executor's idle bit \\# TYPE zio_sched_park_elections_total counter \\zio_sched_park_elections_total {d} \\# HELP zio_sched_steals_total work-stealing scans by outcome (hint_hit = first probe of the hinted victim) \\# TYPE zio_sched_steals_total counter \\zio_sched_steals_total{{result="attempt"}} {d} \\zio_sched_steals_total{{result="hit"}} {d} \\zio_sched_steals_total{{result="hint_hit"}} {d} \\# HELP zio_sched_drain_batches_total cross-executor wake queue drains that woke at least one fiber \\# TYPE zio_sched_drain_batches_total counter \\zio_sched_drain_batches_total {d} \\# HELP zio_sched_drain_woken_total fibers woken by cross-executor drains (flat while frames flow = wakes are being lost) \\# TYPE zio_sched_drain_woken_total counter \\zio_sched_drain_woken_total {d} \\# HELP zio_sched_batch_wake_claims_total sleeping executors woken by batched wake decisions \\# TYPE zio_sched_batch_wake_claims_total counter \\zio_sched_batch_wake_claims_total {d} \\;const zio_metrics_buf_len = 4096;
test "zio metrics block fits its buffer at maximum counter widths" { var buf: [zio_metrics_buf_len]u8 = undefined; const m = std.math.maxInt(u64); const rendered = try std.fmt.bufPrint(&buf, zio_metrics_fmt, .{ m, m, m, m, m, m, m, m, m, m, }); // headroom for future HELP growth before the next widening is forced try std.testing.expect(rendered.len + 1024 <= zio_metrics_buf_len);}
/// metrics-only server on the internal portconst MetricsServer = struct { server: Io.net.Server, io: Io, stats: *broadcaster.Stats, validator: *validator_mod.Validator, data_dir: []const u8, persist: *event_log_mod.DiskPersist, bc: *broadcaster.Broadcaster, slurper: *slurper_mod.Slurper, collection_index: *collection_index_mod.CollectionIndex, ingest_liveness: *ingest_liveness_mod.IngestLiveness,
fn run(self: *MetricsServer) void { while (!shutdown_flag.load(.acquire)) { const stream = self.server.accept(self.io) catch |err| { if (shutdown_flag.load(.acquire)) return; log.debug("metrics accept error: {s}", .{@errorName(err)}); continue; }; self.handleMetricsConn(stream); } }
fn handleMetricsConn(self: *MetricsServer, stream: Io.net.Stream) void { defer stream.close(self.io);
var recv_buf: [4096]u8 = undefined; var send_buf: [4096]u8 = undefined; var connection_reader = stream.reader(self.io, &recv_buf); var connection_writer = stream.writer(self.io, &send_buf); var server = http.Server.init(&connection_reader.interface, &connection_writer.interface);
var request = server.receiveHead() catch return; const path = request.head.target;
if (std.mem.eql(u8, path, "/_healthz")) { // liveness: the process is up AND process-wide ingest is not stalled const v = self.ingest_liveness.verdict(util.milliTimestamp(self.io)); var healthz_buf: [512]u8 = undefined; const status: http.Status = if (v.stalled) .service_unavailable else .ok; const body = if (v.stalled) ingest_liveness_mod.formatHealthzStalled(&healthz_buf, v) else "{\"status\":\"ok\"}"; request.respond(body, .{ .status = status, .keep_alive = false, .extra_headers = &.{ .{ .name = "content-type", .value = "application/json" }, .{ .name = "server", .value = "zlay (atproto-relay)" }, } }) catch {}; } else if (std.mem.eql(u8, path, "/_health") or std.mem.eql(u8, path, "/_readyz")) { // use atomic health flag — pg.Pool runs on Threaded, metrics server is Evented const db_ok = self.persist.isDbHealthy(); const status: http.Status = if (db_ok) .ok else .internal_server_error; const body = if (db_ok) "{\"status\":\"ok\"}" else "{\"status\":\"error\",\"msg\":\"database unavailable\"}"; request.respond(body, .{ .status = status, .keep_alive = false, .extra_headers = &.{ .{ .name = "content-type", .value = "application/json" }, .{ .name = "server", .value = "zlay (atproto-relay)" }, } }) catch {}; } else if (std.mem.eql(u8, path, "/metrics")) { const cache_entries = self.validator.cacheSize(); const attribution = broadcaster.AttributionMetrics{ .history_entries = self.bc.history.count(), .evtbuf_entries = self.persist.evtbufLen(), .did_cache_entries = self.persist.didCacheLen(), .host_authority_cache_entries = self.validator.hostAuthorityCacheSize(), .resolve_queue_len = self.validator.resolveQueueLen(), .resolve_queued_set_count = self.validator.resolveQueuedSetCount(), .validator_cache_map_cap = self.validator.cacheMapCapacity(), .did_cache_map_cap = self.persist.didCacheMapCap(), .queued_set_map_cap = self.validator.resolveQueuedSetCapacity(), .evtbuf_cap = self.persist.evtbufCap(), .outbuf_cap = self.persist.outbufCap(), .workers_count = self.slurper.workerCount(), .db_queue_depth = if (self.bc.db_queue) |q| q.depth() else 0, .db_queue_handled_total = if (self.bc.db_queue) |q| q.handled.load(.acquire) else 0, };
var metrics_buf: [65536]u8 = undefined; var base_body = broadcaster.formatPrometheusMetrics(self.stats, cache_entries, attribution, self.data_dir, &metrics_buf, self.io);
var memory_writer: Io.Writer = .fixed(metrics_buf[base_body.len..]); const cache_bytes = .{ self.validator.cache.logicalBytes(), self.persist.did_cache.logicalBytes(), self.persist.host_id_cache.logicalBytes() }; inline for (cache_bytes, .{ "signing_keys", "did_uids", "host_ids" }) |bytes, component| { if (bytes) |value| memory_writer.print("relay_cache_logical_bytes{{component=\"{s}\"}} {d}\n", .{ component, value }) catch {}; } if (self.collection_index.memoryUsage()) |usage| { memory_writer.print( "relay_zkv_snapshot_available 1\nrelay_zkv_block_cache_bytes {d}\nrelay_zkv_memtable_bytes {d}\n" ++ "relay_zkv_unflushed_segments {d}\nrelay_zkv_l0_tables {d}\nrelay_zkv_compaction_active {d}\n" ++ "relay_zkv_changed_commits_total {d}\nrelay_zkv_flushed_segments_total {d}\n" ++ "relay_zkv_compaction_steps_total {d}\nrelay_zkv_log_gc_steps_total {d}\n", .{ usage.blocks, usage.memtable, usage.unflushed_segments, usage.l0_tables, @intFromBool(usage.compaction_active), usage.changed_commits, usage.flushed_segments, usage.compaction_steps, usage.log_gc_steps }, ) catch {}; } else memory_writer.writeAll("relay_zkv_snapshot_available 0\n") catch {}; base_body = metrics_buf[0 .. base_body.len + memory_writer.buffered().len];
// zio scheduler counters: the scrape is the one forensic channel // that survives a pod dying. the counters are per-executor plain // integers read without synchronization, so a sample can be a // few events stale — fine for rates. var zio_buf: [zio_metrics_buf_len]u8 = undefined; const zio_body = if (comptime backend_config.use_zio) blk: { const m = zio_runtime.schedulerMetrics(); const extra = std.fmt.bufPrint(&zio_buf, zio_metrics_fmt, .{ zio_runtime.executors.items.len, m.parks_doze, m.parks_full, m.park_elections, m.steal_attempts, m.steal_hits, m.steal_hint_hits, m.drain_batches, m.drain_woken, m.batch_wake_claims, }) catch // overflow must be visible on the scrape, never a silent // blanking of the whole block (2026-08-27) "zio_metrics_format_error 1\n"; if (base_body.len + extra.len > metrics_buf.len) break :blk base_body; @memcpy(metrics_buf[base_body.len..][0..extra.len], extra); break :blk metrics_buf[0 .. base_body.len + extra.len]; } else base_body;
var liveness_buf: [1024]u8 = undefined; const liveness = self.ingest_liveness.formatPrometheus(&liveness_buf, util.milliTimestamp(self.io)); const body = if (zio_body.len + liveness.len <= metrics_buf.len) blk: { @memcpy(metrics_buf[zio_body.len..][0..liveness.len], liveness); break :blk metrics_buf[0 .. zio_body.len + liveness.len]; } else zio_body;
request.respond(body, .{ .status = .ok, .keep_alive = false, .extra_headers = &.{ .{ .name = "content-type", .value = "text/plain; version=0.0.4; charset=utf-8" }, .{ .name = "server", .value = "zlay (atproto-relay)" }, } }) catch {}; } else { request.respond("not found", .{ .status = .not_found, .keep_alive = false, .extra_headers = &.{ .{ .name = "content-type", .value = "text/plain" }, .{ .name = "server", .value = "zlay (atproto-relay)" }, } }) catch {}; } }};
pub fn main() !void { try runRelay(); // clean shutdown reached. runRelay's defers (incl. the event-log flush in // DiskPersist.deinit) have already run, so all durable state is persisted. // now skip glibc's exit-time teardown (atexit handlers / locale free): with // ~2,800 Io.Threaded backend threads still live (the backends are never // deinit'd), that teardown races a worker mid-getaddrinfo/strtold and GPFs on // every shutdown (libc.so.6+0x50f). the raw exit_group(2) syscall terminates // all threads atomically in-kernel before any of them can fault — note we // must NOT use std.process.exit here, which calls libc exit(3) (running the // very atexit/locale teardown we're avoiding) when linked against libc. see // docs/shutdown-gpf-analysis.md. if (builtin.os.tag == .linux) std.os.linux.exit_group(0) else std.process.exit(0);}
fn runRelay() !void { // exp-002: optional GPA wrapper for leak detection. // build with -Duse_gpa=true to enable. on clean shutdown (SIGTERM), // GPA logs every allocation that was never freed, with stack traces. var gpa: std.heap.DebugAllocator(.{ .stack_trace_frames = if (build_options.use_gpa) 8 else 0, // catch use-after-free at the use site with the free site's stack: // freed regions are kept unmapped-but-tracked instead of recycled. .retain_metadata = build_options.use_gpa, .never_unmap = build_options.use_gpa, }) = .init; defer if (build_options.use_gpa) { log.info("GPA: checking for leaks...", .{}); const status = gpa.deinit(); if (status == .leak) { log.err("GPA: leaks detected! see stderr for details", .{}); } else { log.info("GPA: no leaks detected", .{}); } }; const allocator = if (build_options.use_gpa) gpa.allocator() else std.heap.c_allocator;
// shared options for both debug and primary runtimes const io_opts: Io.Threaded.InitOptions = .{ .stack_size = default_stack_size, // 8MB (default is 16MB) .concurrent_limit = Io.Limit.limited(4096), // safety rail (steady-state ~2,800 hosts) };
// init debug io (for std.debug.print thread safety) debug_threaded_io = Io.Threaded.init(allocator, io_opts);
const io = if (comptime backend_config.use_zio) blk: { // 0 = one executor per usable cpu (cgroup quota aware) const executors = parseEnvInt(u8, "RELAY_IO_EXECUTORS", 4); zio_runtime = try zio.Runtime.init(allocator, .{ .executors = if (executors == 0) .auto else .exact(@min(executors, 64)), // zio caps at 64 }); log.info("io backend: zio ({d} executors)", .{zio_runtime.executors.items.len}); break :blk zio_runtime.io(); } else blk: { threaded_backend = Io.Threaded.init(allocator, io_opts); log.info("io backend: Threaded", .{}); break :blk threaded_backend.io(); };
// dedicated Threaded runtime for the frame worker pool. // historically needed to isolate Threaded work from Evented fibers (cross-Io // crash class). now redundant since Backend is also Threaded, but harmless. // used for: persist ordering mutex, timestamps, validator cache, // DID resolution HTTP, and thread pool internal sync. var pool_io_backend = Io.Threaded.init(allocator, .{ .stack_size = default_stack_size, }); const pool_io = pool_io_backend.io();
// parse config from env const port = parseEnvInt(u16, "RELAY_PORT", 3000); const metrics_port = parseEnvInt(u16, "RELAY_METRICS_PORT", 3001); const upstream = normalizeSeedHost(getenv("RELAY_UPSTREAM") orelse "bsky.network"); const data_dir = getenv("RELAY_DATA_DIR") orelse "data/events"; const retention_hours = parseEnvInt(u64, "RELAY_RETENTION_HOURS", 72); const max_events_gb = parseEnvInt(u64, "RELAY_MAX_EVENTS_GB", 100); const frame_workers = parseEnvInt(u16, "FRAME_WORKERS", 16); const frame_queue_capacity = parseEnvInt(u16, "FRAME_QUEUE_CAPACITY", 4096); const startup_batch_size = parseEnvInt(u16, "STARTUP_BATCH_SIZE", 50); const db_pool_size = parseEnvInt(u16, "DB_POOL_SIZE", 20); const index_profile = std.meta.stringToEnum(enum { default, bounded }, getenv("RELAY_INDEX_PROFILE") orelse "default") orelse return error.InvalidIndexProfile;
// install signal handlers (including SIGPIPE ignore) installSignalHandlers();
// init components — pass io to network-facing modules var bc = broadcaster.Broadcaster.init(allocator, io, &shutdown_flag); bc.retain_ram_history = !std.mem.eql(u8, getenv("RELAY_RAM_HISTORY") orelse "true", "false"); defer bc.deinit();
var ingest_liveness = ingest_liveness_mod.IngestLiveness.fromEnv(util.milliTimestamp(io)); log.info("ingest stall liveness: enabled={} threshold={d}fps window={d}s startup_grace={d}s", .{ ingest_liveness.enabled.load(.acquire), ingest_liveness.threshold_fps.load(.acquire), ingest_liveness.window_sec.load(.acquire), ingest_liveness.startup_grace_sec.load(.acquire) });
// validator uses pool_io — its cache LRU and host resolvers are called from worker threads var ws_ping = ws_ping_mod.WsPing.fromEnv(); log.info("ws keepalive pinger: enabled={} interval={d}s max_failures={d}", .{ ws_ping.enabled.load(.acquire), ws_ping.interval_sec.load(.acquire), ws_ping.max_failures.load(.acquire) });
var val = validator_mod.Validator.initWithConfig(allocator, &bc.stats, .{ .plc_url = getenv("RELAY_PLC_URL") orelse "https://plc.directory", }, pool_io); defer val.deinit(); try val.start();
// init disk persistence (indigo-compatible diskpersist format + Postgres index) const database_url = getenv("DATABASE_URL") orelse "postgres://relay:relay@localhost:5432/relay"; var dp = event_log_mod.DiskPersist.init(allocator, data_dir, database_url, db_pool_size, pool_io) catch |err| { log.err("failed to init disk persist at {s}: {s}", .{ data_dir, @errorName(err) }); return err; }; defer dp.deinit(); dp.retention_hours = retention_hours; if (getenv("RELAY_RETENTION_SECONDS") != null) dp.retention_seconds = parseEnvInt(u64, "RELAY_RETENTION_SECONDS", retention_hours * 3600); dp.events_per_file = @max(1, parseEnvInt(u32, "RELAY_EVENTS_PER_FILE", dp.events_per_file)); dp.history_enabled = !std.mem.eql(u8, getenv("RELAY_REPLAY_HISTORY") orelse "true", "false"); if (!dp.history_enabled) log.warn("event replay history disabled: live-only sequences are not durable across restart", .{}); dp.max_dir_bytes = parseEnvInt(u64, "RELAY_MAX_EVENTS_BYTES", max_events_gb * 1024 * 1024 * 1024);
// DbRequestQueue — routes DB requests through pool_io worker threads. // originally needed to bridge Evented fibers to Threaded pg.Pool; now // redundant under all-Threaded but still functional. cleanup candidate. var queue_shutdown = std.atomic.Value(bool){ .raw = false }; var db_queue: event_log_mod.DbRequestQueue = .{ .shutdown = &queue_shutdown, .persist = &dp, };
if (dp.lastSeq()) |last| { log.info("event log recovered: last_seq={d}", .{last}); }
// start flush thread try dp.start();
// spawn 2 DbRequestQueue worker threads on pool_io const db_worker_1 = std.Thread.spawn(.{}, event_log_mod.DbRequestQueue.run, .{ &db_queue, pool_io }) catch |err| { log.err("failed to start db queue worker 1: {s}", .{@errorName(err)}); return err; }; const db_worker_2 = std.Thread.spawn(.{}, event_log_mod.DbRequestQueue.run, .{ &db_queue, pool_io }) catch |err| { log.err("failed to start db queue worker 2: {s}", .{@errorName(err)}); return err; };
// wire persist into broadcaster for cursor replay and validator for migration checks bc.persist = &dp; bc.db_queue = &db_queue; val.persist = &dp;
// init collection index (zkv — inspired by lightrail/microcosm.blue) const ci_dir = getenv("COLLECTION_INDEX_DIR") orelse "data/collection-index"; var ci = collection_index_mod.CollectionIndex.openWithReaderIo(allocator, pool_io, io, ci_dir) catch |err| { log.err("failed to init collection index at {s}: {s}", .{ ci_dir, @errorName(err) }); return err; }; defer ci.deinit(); ci.db.options.table_encoding = std.meta.stringToEnum(@TypeOf(ci.db.options.table_encoding), getenv("ZKV_TABLE_ENCODING") orelse "legacy") orelse return error.InvalidIndexEncoding; log.info("collection index encoding: {s}", .{@tagName(ci.db.options.table_encoding)}); switch (index_profile) { .default => try ci.enableBatching(.{}), .bounded => { ci.maintenance_policy = .{ .max_segments = 256, .step_bytes = 64 << 10, .log_gc_names = 8 }; try ci.enableBatching(.{ .delay_ns = 0, .maintenance_after_ack = true, .background_writes = true }); try ci.enablePageWorker(16); }, } log.info("collection index profile: {s}", .{@tagName(index_profile)});
// init backfiller (collection index backfill from source relay) // uses pool_io (Threaded) — backfiller spawns std.Thread, DNS works var backfiller = backfill_mod.Backfiller.init(allocator, &ci, &dp, pool_io, &shutdown_flag);
// init cleaner (removes stale entries from collection index) // uses pool_io (Threaded) — cleaner spawns std.Thread, checks shutdown var cleaner = cleaner_mod.Cleaner.init(allocator, pool_io, &ci, &dp, &shutdown_flag);
// init resyncer (updates collection index on #sync events) // runs on pool_io — enqueue() is called from frame worker threads, // background worker is a plain std.Thread. var resyncer = resync_mod.Resyncer.init(allocator, pool_io, &ci); try resyncer.start(); defer resyncer.deinit();
// init status checker (lazy getRepoStatus re-check for stale upstream // account status — recovers accounts muted by a lost migration activation) var status_checker = status_checker_mod.StatusChecker.init(allocator, pool_io, &dp); try status_checker.start(); defer status_checker.deinit();
// init cursor map + host ops queue — subscriber threads write cursor seqs // to the coalescing map (atomic store, no lock) and push rare ops (failures, // status) to the MPSC queue. a background thread sweeps cursors every 5s // and drains rare ops immediately. var cursor_map: host_ops_mod.CursorMap = .{}; var host_ops_queue: host_ops_mod.HostOpsQueue = .{ .persist = &dp, .cursor_map = &cursor_map, .shutdown = &queue_shutdown, .max_consecutive_failures = 15, .bc = &bc, }; const host_ops_thread = std.Thread.spawn(.{}, host_ops_mod.HostOpsQueue.run, .{ &host_ops_queue, pool_io }) catch |err| { log.err("failed to start host ops thread: {s}", .{@errorName(err)}); return err; };
// init slurper (multi-host crawl manager) var slurper = slurper_mod.Slurper.init( allocator, &bc, &val, &dp, &shutdown_flag, .{ .seed_host = upstream, .private_pds_host = getenv("RELAY_PRIVATE_PDS_HOST"), .max_message_size = 5 * 1024 * 1024, .frame_workers = frame_workers, .frame_queue_capacity = frame_queue_capacity, .startup_batch_size = startup_batch_size, }, io, pool_io, ); var slurper_alive = true; defer if (slurper_alive) slurper.deinit(); slurper.collection_index = &ci; slurper.resyncer = &resyncer; slurper.status_checker = &status_checker; slurper.host_ops = &host_ops_queue; slurper.cursor_map = &cursor_map; slurper.db_queue = &db_queue; slurper.ws_ping = &ws_ping;
// start: loads active hosts from DB, spawns subscriber threads. // pullHosts runs on its own std.Thread (parallel, not gating spawnWorkers). try slurper.start();
// spawn pullHosts on a separate thread — runs in parallel with spawnWorkers, // uses pool_io for HTTP (DNS works), writes to DB via Threaded pool directly. const pull_hosts_thread = if (slurper.options.seed_host.len > 0) blk: { log.info("pulling hosts from {s} (background thread)...", .{slurper.options.seed_host}); break :blk std.Thread.spawn(.{}, slurper_mod.Slurper.pullHosts, .{&slurper}) catch |err| { log.warn("failed to spawn pullHosts thread: {s}", .{@errorName(err)}); break :blk null; }; } else blk: { log.info("no seed host configured, skipping bootstrap", .{}); break :blk null; };
const ingest_liveness_thread = std.Thread.spawn(.{}, ingest_liveness_mod.IngestLiveness.run, .{ &ingest_liveness, pool_io, &bc.stats, &shutdown_flag }) catch |err| { log.err("failed to start ingest liveness thread: {s}", .{@errorName(err)}); return err; };
// start broadcast loop — drains broadcast queue, owns all consumer state. // frame workers push results to the queue, this thread does the fan-out. var broadcast_future = try io.concurrent(broadcaster.Broadcaster.runBroadcastLoop, .{&bc}); defer _ = broadcast_future.cancel(io);
// start GC loop on a plain thread — dp.gc() uses pool_io mutex and pg.Pool. const gc_thread = std.Thread.spawn(.{}, gcLoop, .{ &dp, pool_io }) catch |err| { log.err("failed to start GC thread: {s}", .{@errorName(err)}); return err; };
// wire HTTP fallback into broadcaster (all API endpoints served on WS port) var http_context = api.HttpContext{ .stats = &bc.stats, .persist = &dp, .slurper = &slurper, .collection_index = &ci, .backfiller = &backfiller, .cleaner = &cleaner, .resyncer = &resyncer, .bc = &bc, .validator = &val, .host_ops = &host_ops_queue, .pool_io = pool_io, .io = io, .db_queue = &db_queue, .shutdown = &shutdown_flag, .ingest_liveness = &ingest_liveness, .ws_ping = &ws_ping, }; bc.http_fallback = api.handleHttpRequest; bc.http_fallback_ctx = @ptrCast(&http_context);
// start metrics-only server (internal port) const metrics_address = Io.net.Ip4Address.unspecified(metrics_port); var metrics_srv = MetricsServer{ .server = (Io.net.IpAddress{ .ip4 = metrics_address }).listen(io, .{ .reuse_address = true }) catch |err| { log.err("metrics server failed to listen on :{d}: {s}", .{ metrics_port, @errorName(err) }); return err; }, .io = io, .stats = &bc.stats, .validator = &val, .data_dir = data_dir, .persist = &dp, .bc = &bc, .slurper = &slurper, .collection_index = &ci, .ingest_liveness = &ingest_liveness, }; var metrics_future = try io.concurrent(MetricsServer.run, .{&metrics_srv}); defer _ = metrics_future.cancel(io);
// start downstream WebSocket server (also serves HTTP API via httpFallback) log.info("relay listening on :{d} (ws+http), :{d} (metrics)", .{ port, metrics_port }); log.info("seed host: {s}", .{upstream}); if (dp.retention_seconds) |seconds| { log.info("data dir: {s} (retention: {d}s, max: {d} bytes)", .{ data_dir, seconds, dp.max_dir_bytes }); } else { log.info("data dir: {s} (retention: {d}h, max: {d} bytes)", .{ data_dir, retention_hours, dp.max_dir_bytes }); }
var server = try websocket.Server(broadcaster.Handler).init(allocator, io, .{ .port = port, .address = "0.0.0.0", .max_conn = 4096, .max_message_size = 5 * 1024 * 1024, .handshake = .{ .max_size = 8192, .max_headers = 64 }, }); defer server.deinit();
// Io-native accept loop: fiber-based under Evented, thread-based under Threaded const ws_address = Io.net.Ip4Address.unspecified(port); var ws_listener = (Io.net.IpAddress{ .ip4 = ws_address }).listen(io, .{ .reuse_address = true }) catch |err| { log.err("websocket server failed to listen on :{d}: {s}", .{ port, @errorName(err) }); return err; }; var server_future = try io.concurrent(runWsServer, .{ &server, &ws_listener, &bc }); defer _ = server_future.cancel(io);
var replay_error: ?anyerror = null; if (comptime build_options.offline_replay) { @import("relay_replay.zig").run(allocator, io, pool_io, &bc, &val, &dp, &ci, &slurper.frame_pool.?, &shutdown_flag) catch |err| { replay_error = err; shutdown_flag.store(true, .release); }; }
// wait for shutdown signal while (!shutdown_flag.load(.acquire)) { io.sleep(Io.Duration.fromMilliseconds(100), .awake) catch break; }
log.info("shutdown signal received, stopping...", .{});
// stop WebSocket server: cancel the accept task, then close its listener. // never the other way round — see util.net.interrupt. server_future.cancel(io); ws_listener.deinit(io);
// join GC thread — it checks shutdown_flag and will exit its sleep loop. // must complete before dp.deinit() runs (dp is stack-owned). gc_thread.join(); ingest_liveness_thread.join();
// join pullHosts thread if running if (pull_hosts_thread) |t| t.join();
metrics_future.cancel(io); metrics_srv.server.deinit(io);
slurper.deinit(); slurper_alive = false; queue_shutdown.store(true, .release);
// join db queue workers after their producers have stopped db_worker_1.join(); db_worker_2.join();
// join host ops thread — drains remaining ops before dp.deinit() host_ops_thread.join();
// join backfiller/cleaner threads if running — they touch dp and ci backfiller.waitForCompletion(); cleaner.waitForCompletion();
const flush_result = dp.flush(); bc.producers_done.store(true, .release); broadcast_future.await(io);
try flush_result; ci.mutex.lockUncancelable(pool_io); log.info("collection index final: commits={d} flushed_segments={d} compactions={d} compaction_steps={d} log_gc_steps={d} unflushed={d} l0={d}", .{ ci.maintenance_detail.changed_commits, ci.maintenance_detail.flushed_segments, ci.maintenance_detail.compactions, ci.maintenance_detail.compaction_steps, ci.maintenance_detail.log_gc_steps, ci.db.unflushedSegments(), ci.db.version.tables[0].len, }); ci.mutex.unlock(pool_io); log.info("relay stopped cleanly", .{}); if (replay_error) |err| return err;}
const builtin = @import("builtin");
/// concrete wrapper for runIo — io.concurrent needs ArgsTuple, which can't handle anytypefn runWsServer(server: *websocket.Server(broadcaster.Handler), listener: *Io.net.Server, bc: *broadcaster.Broadcaster) void { server.runIo(listener, bc);}
fn gcLoop(dp: *event_log_mod.DiskPersist, io: Io) void { const gc_interval = @max(1, parseEnvInt(u64, "RELAY_GC_INTERVAL_SECONDS", 10 * 60)); while (!shutdown_flag.load(.acquire)) { // sleep in small increments to check shutdown var remaining: u64 = gc_interval; while (remaining > 0 and !shutdown_flag.load(.acquire)) { const chunk = @min(remaining, 1); io.sleep(Io.Duration.fromSeconds(@intCast(chunk)), .awake) catch return; remaining -= chunk; } if (shutdown_flag.load(.acquire)) return;
dp.gc() catch |err| { log.warn("event log GC failed: {s}", .{@errorName(err)}); };
// return freed pages to OS (glibc-specific, no-op on other platforms) if (comptime malloc_trim) |trim| { _ = trim(0); log.info("gc: malloc_trim complete", .{}); } }}
fn signalHandler(_: std.posix.SIG) callconv(.c) void { shutdown_flag.store(true, .release);}
fn installSignalHandlers() void { const act: std.posix.Sigaction = .{ .handler = .{ .handler = signalHandler }, .mask = std.posix.sigemptyset(), .flags = 0, }; std.posix.sigaction(std.posix.SIG.INT, &act, null); std.posix.sigaction(std.posix.SIG.TERM, &act, null);
// ignore SIGPIPE — writing to disconnected consumers must not crash the process const ignore_act: std.posix.Sigaction = .{ .handler = .{ .handler = std.posix.SIG.IGN }, .mask = std.posix.sigemptyset(), .flags = 0, }; std.posix.sigaction(std.posix.SIG.PIPE, &ignore_act, null);}
/// normalize seed host: empty string or "none" means disabled (no bootstrap).fn normalizeSeedHost(raw: []const u8) []const u8 { if (raw.len == 0 or std.mem.eql(u8, raw, "none")) return ""; return raw;}
test "normalizeSeedHost" { // normal hostnames pass through try std.testing.expectEqualStrings("bsky.network", normalizeSeedHost("bsky.network")); try std.testing.expectEqualStrings("relay.example.com", normalizeSeedHost("relay.example.com"));
// empty string and "none" disable bootstrap try std.testing.expectEqualStrings("", normalizeSeedHost("")); try std.testing.expectEqualStrings("", normalizeSeedHost("none"));}