diff --git a/src/main.zig b/src/main.zig index 39b373c..5f179ae 100644 --- a/src/main.zig +++ b/src/main.zig @@ -200,6 +200,7 @@ pub fn main() !void { 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); // install signal handlers (including SIGPIPE ignore) @@ -266,6 +267,7 @@ pub fn main() !void { .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, diff --git a/src/slurper.zig b/src/slurper.zig index 3cc301d..bb84e7f 100644 --- a/src/slurper.zig +++ b/src/slurper.zig @@ -1,6 +1,6 @@ //! slurper — multi-host PDS crawl manager //! -//! manages one Subscriber thread per tracked PDS host. handles: +//! manages one Subscriber fiber per tracked PDS host. handles: //! - loading known hosts from DB on startup //! - spawning/stopping subscriber workers //! - processing crawl requests (adding new hosts) @@ -29,6 +29,10 @@ pub const Options = struct { max_message_size: usize = 5 * 1024 * 1024, frame_workers: u16 = 16, frame_queue_capacity: u16 = 4096, + /// max hosts to connect concurrently during initial startup ramp. + /// prevents TLS handshake storm from starving the event loop. + /// 0 = unlimited (legacy behavior). + startup_batch_size: u16 = 50, }; // --- host validation --- @@ -510,11 +514,11 @@ pub const Slurper = struct { self.allocator.destroy(sub); } - /// background thread: load hosts from DB and spawn all workers. + /// background fiber: load hosts from DB and spawn all workers. /// runs in background so HTTP server + probes come up immediately. /// Go relay: ResubscribeAllHosts loops with 1ms sleep per host (goroutines). - /// we spawn all at once — the brief memory spike from concurrent TLS handshakes - /// is shorter than a throttled ramp (many hosts fail-fast, freeing memory quickly). + /// we batch-spawn with yields between batches to keep the event loop responsive + /// for health checks and metrics during the initial TLS handshake ramp. fn spawnWorkers(self: *Slurper) void { // pull hosts from seed relay first — idempotent (getOrCreateHost skips existing) if (self.options.seed_host.len > 0) { @@ -538,11 +542,24 @@ pub const Slurper = struct { self.allocator.free(hosts); } + const batch: usize = if (self.options.startup_batch_size > 0) + self.options.startup_batch_size + else + hosts.len; // 0 = unlimited + + var spawned: usize = 0; for (hosts) |host| { if (self.shutdown.load(.acquire)) break; self.spawnWorker(host.id, host.hostname) catch |err| { log.warn("failed to spawn worker for {s}: {s}", .{ host.hostname, @errorName(err) }); }; + spawned += 1; + + // yield between batches so the event loop can service health checks + if (spawned % batch == 0 and spawned < hosts.len) { + log.info("startup: spawned {d}/{d} hosts, yielding...", .{ spawned, hosts.len }); + self.io.sleep(Io.Duration.fromMilliseconds(100), .awake) catch break; + } } log.info("startup complete: {d} host(s) spawned", .{hosts.len});