From 1d9c0a260f2cb78ec9ecd55c60ae4b722e39aa04 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rafa=C5=82=20Leszko?= Date: Tue, 23 Jun 2026 15:00:03 +0200 Subject: [PATCH] fix(rtmg): reap dead-client sessions so a half-open WS can't wedge a pod MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A client whose transport half-opens (tab closed without a close frame, network drop, or — common in prod — a Cloudflare-tunnel half-open) leaves ws.send buffering instead of raising ConnectionClosed. The session then never tears down: the runner keeps generating at full GPU and holds the one-session-per-pod seat forever. The pod keeps heartbeating "healthy", so the pool keeps routing users to it, they get preempt/"another connection took over", and the only recovery is destroying the pod. The server already widened the keepalive ping_timeout to 90s (to avoid GIL-starvation false-disconnects) and its comment explicitly calls for an "app-level idle-session reaper" to pair with it. This adds that reaper. Signal: the client acks every received slice (monotonic byte count). The reaper runs only while we're actively SENDING slices (sent advancing) and the ack count has stalled for _DEAD_CLIENT_ACK_TIMEOUT_S (default 30s, env DEMON_DEAD_CLIENT_ACK_TIMEOUT_S). That's the exact dead-client fingerprint we observed in prod: 70-95% GPU generating, zero connected sockets, acks frozen. On detection it flips state.running (runner tears down on its next iteration) and closes the ws to unblock the hung send/recv. False-positive-proof: an idle-paused session sends no slices, so `sent` doesn't advance and the reaper never fires — idle+dead is left to the WS keepalive (GPU is free then, so the keepalive thread isn't starved). The reaper is a daemon thread that self-exits when state.running flips, so it adds no teardown coupling. Co-Authored-By: Claude Opus 4.8 (1M context) --- demos/realtime_motion_graph_web/ws_adapter.py | 73 ++++++++++++++++++- 1 file changed, 70 insertions(+), 3 deletions(-) diff --git a/demos/realtime_motion_graph_web/ws_adapter.py b/demos/realtime_motion_graph_web/ws_adapter.py index f5ac4bcf..115a25a0 100644 --- a/demos/realtime_motion_graph_web/ws_adapter.py +++ b/demos/realtime_motion_graph_web/ws_adapter.py @@ -257,6 +257,28 @@ def _loop() -> None: # (it only observes running=False between pipeline iterations). _PREEMPT_TEARDOWN_TIMEOUT_S = 45.0 +# Dead-client reaper (the "app-level idle-session reaper" the server's +# ws_serve ping_timeout comment calls for). When a client's transport +# half-opens — tab closed without a close frame, network drop, or a +# Cloudflare-tunnel half-open — ws.send buffers instead of raising +# ConnectionClosed, so the session never tears down: it keeps generating +# at full GPU and holds the one-session-per-pod seat forever, leaving the +# pod heartbeating "healthy" but unconnectable (only a destroy clears it). +# The keepalive ping is widened to 90s AND can be GIL-starved by a busy +# generation tick, so it can't be the sole backstop while streaming. +# +# Signal: the client acks every received slice (monotonic byte count). If +# we are actively SENDING slices (sent advancing) but the ack count has +# not moved for this long, the client is gone — flip state.running so the +# runner tears down on its next iteration. Gating on "sent advancing" +# means an idle-paused session (no slices, so no acks expected) is never +# falsely reaped; the keepalive covers idle+dead (GPU free → not starved). +_DEAD_CLIENT_ACK_TIMEOUT_S = float( + os.environ.get("DEMON_DEAD_CLIENT_ACK_TIMEOUT_S", "") or 30.0 +) +# How often the reaper samples slice flow. +_DEAD_CLIENT_POLL_S = 5.0 + def _windowed_slice_drop_reason( *, @@ -1177,11 +1199,14 @@ def _send_json(payload: dict) -> None: except ValueError: _SLICE_WINDOW_BYTES = 256 * 1024 # [bytes sent, bytes acked (None until first ack), drops since last - # log, last log wall]. Shared between the WS subscriber thread - # (writer of sent/drops) and the recv thread (writer of acked); - # single-field updates under the GIL, no torn reads that matter. + # log, last log wall, monotonic ts of last ack progress]. Shared + # between the WS subscriber thread (writer of sent/drops) and the recv + # thread (writer of acked/acked_ts); single-field updates under the + # GIL, no torn reads that matter. `acked_ts` seeds at session start so + # the dead-client reaper also catches a client that never acks at all. _slice_flow = { "sent": 0, "acked": None, "drops": 0, "log_wall": 0.0, + "acked_ts": time.monotonic(), } def _note_slice_drop(reason: str, detail: float) -> None: @@ -1378,6 +1403,45 @@ def on_event(event) -> None: streaming.bus.subscribe(on_event, name="ws") + # Dead-client reaper: a half-open transport (closed tab, dropped + # network, Cloudflare-tunnel half-open) lets ws.send keep buffering + # instead of raising ConnectionClosed, so without this the runner + # would generate forever and hold the one-session-per-pod seat. We + # only act while actively streaming (sent advancing) with a stalled + # ack clock — an idle-paused session sends nothing, so it's never + # falsely reaped; the WS keepalive covers idle+dead. Daemon thread + # self-exits when state.running flips (here or via normal teardown). + def _dead_client_reaper() -> None: + last_sent = -1 + while state.running: + time.sleep(_DEAD_CLIENT_POLL_S) + if not state.running: + return + sent = _slice_flow["sent"] + streaming_now = sent > last_sent + last_sent = sent + if not streaming_now: + continue # idle / not streaming → keepalive owns liveness + stalled_s = time.monotonic() - _slice_flow["acked_ts"] + if stalled_s > _DEAD_CLIENT_ACK_TIMEOUT_S: + logger.warning( + "dead_client_reap session_id={} stalled_s={:.0f} " + "sent={} acked={}", + session_id, stalled_s, sent, _slice_flow["acked"], + ) + state.running = False + try: + ws.close(1011, "client unresponsive") + except Exception: + pass + return + + threading.Thread( + target=_dead_client_reaper, + name="ws-dead-client-reaper", + daemon=True, + ).start() + # ---- Init handshake: ready + binary initial buffer + optional stems ---- # # These ship inline (not through the bus) because they're produced @@ -1575,6 +1639,9 @@ def _recv_binary_payload(fail_type: str): prev = _slice_flow["acked"] if prev is None or ack > prev: _slice_flow["acked"] = ack + # Ack advanced → client is alive; reset the + # dead-client reaper's stall clock. + _slice_flow["acked_ts"] = time.monotonic() streaming.set_knobs( data.get("raw") or {}, pp, origin=origin, client_time=ct, slice_lead_s=sl,