diff --git a/src/internal/api/admin.zig b/src/internal/api/admin.zig index 372bf65..f78f83b 100644 --- a/src/internal/api/admin.zig +++ b/src/internal/api/admin.zig @@ -313,7 +313,7 @@ pub fn handleAdminUnblockHost(conn: *h.Conn, body: []const u8, headers: *const w /// /// `min_idle_seconds` guards against tearing down a healthy worker: a deaf /// host and a merely quiet host are indistinguishable from seq sampling, so -/// a bulk sweep would otherwise manufacture the connection churn it is +/// reconnecting on silence alone manufactures the connection churn it is /// trying to recover from. It is a query parameter, not a constant, so the /// window can be widened mid-incident without a redeploy. pub fn handleAdminReconnectHost(conn: *h.Conn, query: []const u8, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { diff --git a/src/internal/broadcaster.zig b/src/internal/broadcaster.zig index f95676c..894fd60 100644 --- a/src/internal/broadcaster.zig +++ b/src/internal/broadcaster.zig @@ -1215,7 +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} \\ - \\ , .{ stats.failed_bad_did.load(.acquire), stats.failed_bad_rev.load(.acquire), diff --git a/src/internal/slurper.zig b/src/internal/slurper.zig index 275dc94..82f2614 100644 --- a/src/internal/slurper.zig +++ b/src/internal/slurper.zig @@ -406,9 +406,8 @@ pub const Slurper = struct { .ca_bundle = self.ca_bundle, }, ); - // stamp before the worker fiber can run: a worker must be visible to - // the silent-worker sweep from the moment it exists, including one - // that wedges before its first handshake ever completes. + // stamp before the worker fiber can run: a worker that wedges before + // its first handshake must still age into reconnect candidacy. sub.last_progress_ms.store(milliTimestamp(self.io), .release); sub.collection_index = self.collection_index; sub.resyncer = self.resyncer; @@ -600,9 +599,9 @@ pub const Slurper = struct { /// outlives its fiber until awaited or cancelled exactly once. /// /// `min_idle_ms` is the guard. A deaf host and a merely quiet host look - /// identical from seq sampling, so a bulk sweep driven by a silent-host - /// probe would otherwise tear down healthy connections -- manufacturing - /// exactly the connection churn that this class of bug feeds on. Any + /// identical from seq sampling, so a reconnect driven by silence alone + /// would tear down healthy connections -- manufacturing exactly the + /// connection churn that this class of bug feeds on (2026-08-18). Any /// worker that has seen a frame (or completed a handshake) more recently /// than this is left alone. /// non-null when this worker should be left alone. Shared by @@ -736,13 +735,13 @@ test "forceReconnect guard: only workers idle past the window are torn down" { try std.testing.expectEqual(@as(?S.ReconnectOutcome, null), S.reconnectSkipAt(now, now, 0)); // a worker wedged before its first handshake stamped only at spawn, so it - // ages past the window and IS a candidate. This is the case the sweep - // would otherwise never see: connectAndRead that never returns means + // ages past the window and IS a candidate. This is the case the idle + // guard must not hide: connectAndRead that never returns means // run()'s reconnect loop never iterates, so there is no backoff to // recover it either. try std.testing.expectEqual(@as(?S.ReconnectOutcome, null), S.reconnectSkipAt(now, now - 30 * 60 * std.time.ms_per_s, min_idle_ms)); // ...whereas an unreachable host looping in backoff re-stamps on every - // attempt, so it stays out of the sweep's way. + // attempt, so it stays inside the idle guard. try std.testing.expect(S.reconnectSkipAt(now, now - 5 * std.time.ms_per_s, min_idle_ms) != null); } diff --git a/src/internal/subscriber.zig b/src/internal/subscriber.zig index 76045e4..24829ae 100644 --- a/src/internal/subscriber.zig +++ b/src/internal/subscriber.zig @@ -213,16 +213,15 @@ pub const Subscriber = struct { last_cursor_flush: i64 = 0, /// ms timestamp of the last sign of life from this worker: a frame, a /// completed handshake, or the start of a connect attempt. Written by the - /// subscriber fiber, read by the sweep and admin fibers, hence atomic. + /// subscriber fiber, read by the admin reconnect fiber, hence atomic. /// /// It stamps connect *attempts*, not just delivered frames, because a /// worker can wedge inside connectAndRead -- DNS, TLS and the websocket /// handshake all park on the same read path that lost wakes in the /// 2026-08-18 incident. connectAndRead never returning means run()'s /// reconnect loop never iterates either, so such a worker has no backoff - /// to fall back on. Stamping attempts makes it visible to the sweep while - /// still leaving a genuinely unreachable host (which keeps attempting, - /// and so keeps stamping) alone. + /// to fall back on. A genuinely unreachable host keeps attempting, so it + /// keeps stamping and stays inside the idle guard. last_progress_ms: std.atomic.Value(i64) = .{ .raw = 0 }, rate_limiter: RateLimiter = .{}, @@ -274,9 +273,6 @@ pub const Subscriber = struct { while (!self.shouldStop()) { log.info("host {s}: connecting...", .{self.options.hostname}); - // stamp the attempt, not just its success: a wedge inside - // connectAndRead never returns here, so without this the worker - // would look permanently idle to nobody and be swept by no one. self.last_progress_ms.store(milliTimestamp(self.io), .release); self.connectAndRead() catch |err| { @@ -403,8 +399,6 @@ pub const Subscriber = struct { return e; }; log.info("host {s}: connected", .{self.options.hostname}); - // a fresh connection counts as liveness: a worker that just completed - // its handshake must not look idle to a recovery sweep. self.last_progress_ms.store(milliTimestamp(self.io), .release); // reset failures on successful connect (via host_ops queue)