From b3ffac49e5edf5b81a8e8064bb3540dbd8fc7f56 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 19 Aug 2026 18:48:09 +0000 Subject: [PATCH] slurper: close the sweep's pre-handshake blind spot Review caught that the silent-worker sweep skipped workers with last_frame_ms == 0, so one that wedged before its first handshake was invisible to it. Checking the code, it is worse than invisible: DNS, TLS and the websocket handshake all park on the same read path that lost wakes in the 2026-08-18 incident, and a wedge inside connectAndRead means it never returns -- so run()'s reconnect loop never iterates and there is no backoff to fall back on either. That worker is stuck permanently, and nothing watches it. The signal is now progress, not just delivery: `last_frame_ms` becomes `last_progress_ms`, stamped by addHost at spawn, by run() on every connect attempt, on a completed handshake, and on every frame. So: - wedged pre-handshake -> stops stamping -> swept. - unreachable host looping in backoff -> re-stamps each attempt -> left alone, because it is already reconnecting and does not need help. - connected and delivering -> stamps per frame, as before. Stamping at spawn also removes the window where a worker sat in the map unstamped and would have read as infinitely idle. Unit tests cover both new cases: the pre-handshake wedge ages into candidacy, the backoff loop does not. Verified: zig fmt --check, zig build test, zig build test -Dbackend=zio, and -Dbackend=zio -Dtarget=x86_64-linux-gnu -Doptimize=ReleaseSafe. Co-Authored-By: Claude Opus 5 (1M context) --- src/internal/slurper.zig | 37 +++++++++++++++++++++++++++---------- src/internal/subscriber.zig | 28 +++++++++++++++++++--------- 2 file(s) changed, 46 insertion(s)(+), 19 deletion(s)(-) diff --git a/src/internal/slurper.zig b/src/internal/slurper.zig --- a/src/internal/slurper.zig +++ b/src/internal/slurper.zig @@ -419,6 +419,10 @@ .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. + sub.last_progress_ms.store(milliTimestamp(self.io), .release); sub.collection_index = self.collection_index; sub.resyncer = self.resyncer; sub.status_checker = self.status_checker; @@ -620,18 +624,19 @@ fn reconnectSkip(self: *Slurper, entry: WorkerEntry, min_idle_ms: i64) ?ReconnectOutcome { return reconnectSkipAt( milliTimestamp(self.io), - entry.subscriber.last_frame_ms.load(.acquire), + entry.subscriber.last_progress_ms.load(.acquire), min_idle_ms, ); } /// the guard decision, free of clock and lock so it can be tested. all - /// three arguments are milliseconds -- the field is `last_frame_ms` and + /// three arguments are milliseconds -- the field is `last_progress_ms` and /// `util.timestamp` returns SECONDS, which is a live mixing hazard here. - fn reconnectSkipAt(now_ms: i64, last_frame_ms: i64, min_idle_ms: i64) ?ReconnectOutcome { - // 0 = never saw a frame or a handshake, so it is a reconnect candidate. - if (last_frame_ms == 0) return null; - const idle_ms = now_ms - last_frame_ms; + fn reconnectSkipAt(now_ms: i64, last_progress_ms: i64, min_idle_ms: i64) ?ReconnectOutcome { + // 0 means the worker was never stamped. addHost stamps at spawn, so + // this is defensive only; treat it as a reconnect candidate. + if (last_progress_ms == 0) return null; + const idle_ms = now_ms - last_progress_ms; if (idle_ms < min_idle_ms) return .{ .skipped_recent_frame = idle_ms }; return null; } @@ -706,10 +711,11 @@ const now_ms = milliTimestamp(self.io); var it = self.workers.iterator(); while (it.next()) |entry| { - const last = entry.value_ptr.subscriber.last_frame_ms.load(.acquire); - // 0 = still connecting; leave it to the backoff loop. - if (last == 0) continue; - if (now_ms - last < timeout_ms) continue; + 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; } } @@ -801,4 +807,15 @@ // min_idle_seconds=0 disables the guard entirely, which is what an // operator wants when they are certain a host is deaf. 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 + // 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. + 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 --- a/src/internal/subscriber.zig +++ b/src/internal/subscriber.zig @@ -211,13 +211,19 @@ shutdown: *std.atomic.Value(bool), last_upstream_seq: ?u64 = null, last_cursor_flush: i64 = 0, - /// ms timestamp of the last sign of life from upstream: a frame, or the - /// handshake that opened the connection. Written by the subscriber fiber, - /// read by the admin fiber, hence atomic. This is the liveness signal - /// `Slurper.forceReconnect` uses so a bulk recovery sweep cannot tear - /// down workers that are merely quiet (see the 2026-08-18 incident: a - /// deaf host and a quiet host look identical from seq sampling alone). - last_frame_ms: std.atomic.Value(i64) = .{ .raw = 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. + /// + /// 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. + last_progress_ms: std.atomic.Value(i64) = .{ .raw = 0 }, rate_limiter: RateLimiter = .{}, // per-host shutdown (e.g. FutureCursor — stops only this subscriber) @@ -268,6 +274,10 @@ 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| { if (self.shouldStop()) return; @@ -395,7 +405,7 @@ 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_frame_ms.store(milliTimestamp(self.io), .release); + self.last_progress_ms.store(milliTimestamp(self.io), .release); // reset failures on successful connect (via host_ops queue) if (self.options.host_id > 0) { @@ -484,7 +494,7 @@ // count every successfully decoded event (matches Go relay's events_received_counter) _ = sub.bc.stats.frames_in.fetchAdd(1, .monotonic); const now_ms = milliTimestamp(io); - sub.last_frame_ms.store(now_ms, .release); + sub.last_progress_ms.store(now_ms, .release); // extract seq for cursor tracking (deferred until after pool accepts) const upstream_seq = payload.getUint("seq"); -- tangled.sh