From 9730b66343e6ff0f271132ef1500d158bfe7c434 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 19 Aug 2026 15:53:08 -0500 Subject: [PATCH] Revert the silent-worker sweep: silence is the normal case, not a fault The sweep took production down. Deployed 13:47Z, RSS growth alert 14:21, scrape down 14:52, liveness kill 14:56, then an hour of failed probes with the HTTP server not answering inside 15s. Rolled back to 95b9ec6. The defect is the premise. The sweep treats "delivered no frames" as evidence a connection is broken, but on this network only 113-120 of ~3000 connected hosts deliver anything in a given 30s window. Silence is the normal, healthy state of a small PDS. So the sweep tore down thousands of working connections, and since a rebuilt worker only re-stamps on handshake, every quiet host entered a permanent 15-minute destroy-and-rebuild cycle -- sustained fleet-wide connection churn, which is the same variable the 08-12 soak exists to accelerate. It manufactured the trigger we are hunting. The evidence was already in the incident capture I worked from: 102 of 120 sending hosts out of 3092 connected. I read that as "102 hosts went deaf" and not as "2900 hosts are silent and fine", then built a backstop on the second reading being false. I had also just warned the operator that a bulk reconnect loop over a wrong silent-set would inflict exactly this churn, and then shipped one on a 60s timer. b3ffac4 does not rescue it: stamping spawn, connect attempts and handshakes closed the pre-handshake blind spot, but a connected-and-quiet worker still never re-stamps between frames, so the mass teardown is unchanged. Reverted rather than tuned. No threshold makes silence mean deafness. Kept: `last_progress_ms` and its stamping, which the /admin/hosts/reconnect guard uses, and that endpoint stays manual and guarded. Removed: the sweep fiber, its options, and relay_silent_worker_reconnects_total. A real backstop needs a signal that separates "nothing to say" from "cannot hear" -- a protocol ping with a response, as indigo does with a pinger goroutine and a pong-refreshed read deadline. Whatever replaces this must also have its timeout and an off switch settable at runtime; the operator had no lever short of an image rollback. Co-Authored-By: Claude Opus 5 (1M context) --- src/internal/broadcaster.zig | 7 ---- src/internal/slurper.zig | 73 ------------------------------------ 2 files changed, 80 deletions(-) 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; -- 2.51.2