diff --git a/src/internal/broadcaster.zig b/src/internal/broadcaster.zig index 66c3020..f95676c 100644 --- a/src/internal/broadcaster.zig +++ b/src/internal/broadcaster.zig @@ -92,9 +92,6 @@ pub const Stats = struct { subscriber_disconnect_tls_init: std.atomic.Value(u64) = .{ .raw = 0 }, subscriber_disconnect_ws_handshake: std.atomic.Value(u64) = .{ .raw = 0 }, subscriber_disconnect_read_loop: std.atomic.Value(u64) = .{ .raw = 0 }, - /// workers torn down by the silent-worker sweep: alive, believed healthy, - /// but delivering nothing. climbing = hosts are going deaf (2026-08-18). - silent_worker_reconnects: std.atomic.Value(u64) = .{ .raw = 0 }, // frame pool memory pressure pool_queued_bytes: std.atomic.Value(u64) = .{ .raw = 0 }, // persist/broadcast pipeline contention @@ -1218,9 +1215,6 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib \\relay_subscriber_disconnect_total{{phase="ws_handshake"}} {d} \\relay_subscriber_disconnect_total{{phase="read_loop"}} {d} \\ - \\# TYPE relay_silent_worker_reconnects_total counter - \\# HELP relay_silent_worker_reconnects_total workers torn down by the silent-worker sweep — connected and believed healthy but delivering no frames. any sustained rate means hosts are going deaf. - \\relay_silent_worker_reconnects_total {d} \\ , .{ stats.failed_bad_did.load(.acquire), @@ -1244,7 +1238,6 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib stats.subscriber_disconnect_tls_init.load(.acquire), stats.subscriber_disconnect_ws_handshake.load(.acquire), stats.subscriber_disconnect_read_loop.load(.acquire), - stats.silent_worker_reconnects.load(.acquire), }) catch return w.buffered(); // memory attribution — internal capacities help identify what's consuming RSS diff --git a/src/internal/slurper.zig b/src/internal/slurper.zig index fc8457c..275dc94 100644 --- a/src/internal/slurper.zig +++ b/src/internal/slurper.zig @@ -41,15 +41,6 @@ pub const Options = struct { /// prevents TLS handshake storm from starving the event loop. /// 0 = unlimited (legacy behavior). startup_batch_size: u16 = 50, - /// tear down and respawn a worker that has been connected but silent for - /// this long. The backstop for deafness of ANY cause: a worker parked - /// forever on a read that will never complete looks perfectly healthy - /// from every other angle (see the 2026-08-18 incident, where 102 of 120 - /// sending hosts were silent for ~15 hours and every fleet-health check - /// passed). 0 disables the sweep. - silent_worker_timeout_sec: u32 = 900, - /// how often the sweep runs. - silent_worker_sweep_interval_sec: u32 = 60, }; const WorkerEntry = struct { @@ -82,7 +73,6 @@ pub const Slurper = struct { workers_mutex: Io.Mutex = Io.Mutex.init, // crawl request queue - sweep_future: ?Io.Future(void) = null, crawl_queue: std.ArrayListUnmanaged([]const u8) = .empty, crawl_mutex: Io.Mutex = Io.Mutex.init, crawl_cond: Io.Condition = Io.Condition.init, @@ -145,9 +135,6 @@ pub const Slurper = struct { // pullHosts + listActiveHosts + spawnWorker all happen in the background thread. self.startup_future = try self.io.concurrent(spawnWorkers, .{self}); self.crawl_future = try self.io.concurrent(processCrawlQueue, .{self}); - if (self.options.silent_worker_timeout_sec > 0) { - self.sweep_future = try self.io.concurrent(sweepSilentWorkers, .{self}); - } } /// pull PDS host list from the seed relay's com.atproto.sync.listHosts endpoint. @@ -679,72 +666,12 @@ pub const Slurper = struct { return .reconnected; } - /// periodically tear down workers that are connected but delivering - /// nothing. - /// - /// This is the backstop the 2026-08-18 incident did not have. The - /// mechanism that caused it is fixed in zio, but the reason it lasted 15 - /// hours rather than seconds is that nothing in the relay ever asked "has - /// this connection actually produced anything lately?" -- and the - /// websocket layer's own answer to that question is inert under zio, - /// because readLoopWithHeartbeat implements its timeout with - /// SO_RCVTIMEO, which does nothing on the non-blocking fds the fiber - /// domain requires. So a subscriber read has no deadline at all. - /// - /// This sweep gives it one from the outside, and it is deliberately - /// cause-agnostic: it recovers a silent worker whether the wake was lost - /// in the poller, the peer went quiet without sending FIN, or something - /// we have not diagnosed yet. - fn sweepSilentWorkers(self: *Slurper) void { - const timeout_ms: i64 = @as(i64, self.options.silent_worker_timeout_sec) * std.time.ms_per_s; - var candidates: std.ArrayListUnmanaged(u64) = .empty; - defer candidates.deinit(self.allocator); - - while (!self.shutdown.load(.acquire)) { - self.io.sleep(Io.Duration.fromSeconds(self.options.silent_worker_sweep_interval_sec), .awake) catch {}; - if (self.shutdown.load(.acquire)) return; - - candidates.clearRetainingCapacity(); - { - self.workers_mutex.lockUncancelable(self.io); - defer self.workers_mutex.unlock(self.io); - const now_ms = milliTimestamp(self.io); - var it = self.workers.iterator(); - while (it.next()) |entry| { - const last = entry.value_ptr.subscriber.last_progress_ms.load(.acquire); - // addHost stamps at spawn and run() stamps every connect - // attempt, so 0 only occurs if a worker was somehow never - // stamped; treating it as silent is the safe reading. - if (last != 0 and now_ms - last < timeout_ms) continue; - candidates.append(self.allocator, entry.key_ptr.*) catch break; - } - } - - // act outside the lock: forceReconnect takes it, and cancelling a - // worker future while holding it would deadlock against runWorker. - for (candidates.items) |host_id| { - if (self.shutdown.load(.acquire)) return; - switch (self.forceReconnect(host_id, timeout_ms)) { - .reconnected => { - _ = self.bc.stats.silent_worker_reconnects.fetchAdd(1, .monotonic); - log.warn("silent-worker sweep: host_id={d} delivered nothing for {d}s, reconnecting", .{ host_id, self.options.silent_worker_timeout_sec }); - }, - // it spoke between the scan and the call, or it is already - // gone. either way there is nothing to recover. - .skipped_recent_frame, .no_worker => {}, - .respawn_failed => log.err("silent-worker sweep: host_id={d} torn down but respawn enqueue failed", .{host_id}), - } - } - } - } - /// shutdown all workers and clean up pub fn deinit(self: *Slurper) void { // cancel background tasks if (self.startup_future) |*f| f.cancel(self.io); self.crawl_cond.signal(self.io); if (self.crawl_future) |*f| f.cancel(self.io); - if (self.sweep_future) |*f| f.cancel(self.io); // collect futures to cancel (can't cancel while holding workers_mutex) var futures_to_cancel: std.ArrayListUnmanaged(Io.Future(void)) = .empty;