fix(terminal): unsubscribe when the SSE header write fails after attaching
Adversarial review of the previous commit found a leak it introduced. `attach_terminal()` subscribes and only then are the response headers written. A client that dropped in that instant makes `end_headers()` raise BrokenPipeError/ConnectionResetError, and because the `try/finally` started *after* those writes, the exception escaped `_handle_terminal_output` with the queue still in `_subscribers`. That pins `unwatched_since` at None — only `unsubscribe()` on an empty subscriber list resets it — so the terminal became permanently unreapable: the exact fd/thread leak (#4633) this reaper exists to prevent, and strictly worse than the TOCTOU the previous commit fixed, since that at least still allowed cleanup. The try now starts immediately after the attach, so every exit path unsubscribes. Regression test drives a handler whose `end_headers()` raises and asserts the subscriber list is empty, `unwatched_since` is restored, and the terminal is reapable again once the grace elapses. It fails against the previous ordering.
This commit is contained in:
1 parent
07d110a064
commit
a31cdb7765
2 files changed
+64
-7
No files matched your search
+14
-7
@@ -17865,14 +17865,21 @@ def _handle_terminal_output(handler, parsed):
|
||||
return j(handler, {"error": "terminal not running"}, status=404)
|
||||
term, output = attached
|
||||
|
||||
handler.send_response(200)
|
||||
handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||||
handler.send_header("Cache-Control", "no-cache")
|
||||
handler.send_header("X-Accel-Buffering", "no")
|
||||
handler.send_header("Connection", "close")
|
||||
end_sse_headers(handler)
|
||||
_sse_set_write_deadline(handler) # Defect A: slow tab can't pin this thread
|
||||
# The subscription is live from here on, so EVERY exit path — including a
|
||||
# failure while writing the response headers — must unsubscribe. Writing
|
||||
# headers to a client that already dropped raises BrokenPipeError, and if
|
||||
# that escaped before the try block the queue would stay in
|
||||
# `_subscribers` forever, pinning `unwatched_since` at None and making the
|
||||
# terminal permanently unreapable: the exact fd/thread leak this reaper
|
||||
# exists to prevent. Hence the try starts immediately after the attach.
|
||||
try:
|
||||
handler.send_response(200)
|
||||
handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||||
handler.send_header("Cache-Control", "no-cache")
|
||||
handler.send_header("X-Accel-Buffering", "no")
|
||||
handler.send_header("Connection", "close")
|
||||
end_sse_headers(handler)
|
||||
_sse_set_write_deadline(handler) # Defect A: slow tab can't pin this thread
|
||||
while True:
|
||||
try:
|
||||
event_seq, event, data = output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)
|
||||
|
||||
@@ -340,3 +340,53 @@ def test_teardown_runs_without_holding_the_registry_lock(monkeypatch):
|
||||
terminal._reap_idle_terminals(time.time())
|
||||
|
||||
assert observed.get("free") is True, "_LOCK was held across process teardown"
|
||||
|
||||
|
||||
def test_a_header_write_failure_after_attach_still_unsubscribes(monkeypatch):
|
||||
"""Attaching before the headers must not be able to leak the subscriber.
|
||||
|
||||
`attach_terminal()` subscribes, and only then are the response headers
|
||||
written. A client that dropped in that instant makes `end_headers()` raise
|
||||
BrokenPipeError; if that escaped before the try/finally, the queue would
|
||||
stay in `_subscribers` forever, pin `unwatched_since` at None, and make the
|
||||
terminal permanently unreapable — the very leak this reaper exists to
|
||||
prevent, self-inflicted.
|
||||
"""
|
||||
import io
|
||||
from types import SimpleNamespace
|
||||
|
||||
import api.routes as routes
|
||||
|
||||
sid = "sse-header-failure"
|
||||
term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
|
||||
|
||||
class _Handler:
|
||||
headers = {}
|
||||
|
||||
def __init__(self):
|
||||
self.wfile = io.BytesIO()
|
||||
self.status = None
|
||||
|
||||
def send_response(self, status):
|
||||
self.status = status
|
||||
|
||||
def send_header(self, _name, _value):
|
||||
pass
|
||||
|
||||
def end_headers(self):
|
||||
raise ConnectionResetError("client vanished")
|
||||
|
||||
monkeypatch.setattr(routes, "_embedded_terminal_gate_allows", lambda _handler: True)
|
||||
monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: None)
|
||||
|
||||
routes._handle_terminal_output(
|
||||
_Handler(), SimpleNamespace(query=f"session_id={sid}")
|
||||
)
|
||||
|
||||
assert term._subscribers == [], "the subscriber leaked on a header-write failure"
|
||||
assert term.unwatched_since is not None, "the terminal was pinned as watched forever"
|
||||
# The grace restarts from the detach, which is correct — but the terminal
|
||||
# is reapable again once it elapses, instead of being pinned forever.
|
||||
assert terminal._reap_idle_terminals(
|
||||
term.unwatched_since + terminal._TERMINAL_IDLE_GRACE_SECONDS + 1
|
||||
) == 1
|
||||
Reference in new issue
Block a user