fix: read-idle timeout on SSE so a stalled stream can't hang events()

src/comfy_low/transport.py

@@ -33,6 +33,14 @@
 _API = "/api/v2"
 _UNSET = object()
 
+# SSE read-idle timeout. The server sends keepalive comments roughly every 15s,
+# so no data at all for ~3x that means the connection has silently stalled
+# ("zombie": open but delivering nothing) rather than being legitimately idle.
+# Applying a read timeout makes that case raise a ReadTimeout, which
+# comfy_sdk.Job.events() catches and turns into a poll/reconnect — instead of
+# blocking iter_lines() forever. (Pass timeout=None to opt out of the timeout.)
+_SSE_IDLE_TIMEOUT = httpx.Timeout(10.0, read=45.0)
+
 _DEFAULT_PORTS = {"http": 80, "https": 443}
 
 
@@ -335,17 +343,19 @@ def get_job(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> Job:
         resp = self.raw_request("GET", path, timeout=timeout)
         return Job.model_validate(self._p.parse_or_raise(resp, (200,)))
 
-    def get_job_events(self, job_id_or_url: str, *, timeout: Any = None) -> Iterator[RawEvent]:
+    def get_job_events(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> Iterator[RawEvent]:
         """GET /api/v2/jobs/{id}/events — raw live SSE iterator (escape hatch).
 
         No reconnection here; a single connection's frames. ``comfy_sdk`` adds the
-        reconnect loop. ``timeout=None`` by default (an idle stream must not time
-        out mid-job).
+        reconnect loop. Defaults to a read-idle timeout (``_SSE_IDLE_TIMEOUT``, well
+        above the server's keepalive interval) so a silently-stalled connection
+        raises instead of blocking forever; pass ``timeout=None`` to disable.
         """
         path = job_id_or_url if _looks_like_path(job_id_or_url) else f"/jobs/{job_id_or_url}/events"
         headers = {"Accept": "text/event-stream"}
         decoder = SSEDecoder()
-        with self.open("GET", path, headers=headers, timeout=timeout) as resp:
+        eff_timeout = _SSE_IDLE_TIMEOUT if timeout is _UNSET else timeout
+        with self.open("GET", path, headers=headers, timeout=eff_timeout) as resp:
             if resp.status_code != 200:
                 resp.read()
                 self._p.parse_or_raise(resp, (200,))
@@ -562,12 +572,15 @@ async def get_job(self, job_id_or_url: str, *, timeout: Any = _UNSET) -> Job:
         return Job.model_validate(self._p.parse_or_raise(resp, (200,)))
 
     async def get_job_events(
-        self, job_id_or_url: str, *, timeout: Any = None
+        self, job_id_or_url: str, *, timeout: Any = _UNSET
     ) -> AsyncIterator[RawEvent]:
+        # Read-idle timeout by default (see the sync get_job_events); a stalled
+        # connection raises so comfy_sdk falls back to poll/reconnect.
         path = job_id_or_url if _looks_like_path(job_id_or_url) else f"/jobs/{job_id_or_url}/events"
         headers = {"Accept": "text/event-stream"}
         decoder = SSEDecoder()
-        async with self.open("GET", path, headers=headers, timeout=timeout) as resp:
+        eff_timeout = _SSE_IDLE_TIMEOUT if timeout is _UNSET else timeout
+        async with self.open("GET", path, headers=headers, timeout=eff_timeout) as resp:
             if resp.status_code != 200:
                 await resp.aread()
                 self._p.parse_or_raise(resp, (200,))

tests/conftest.py

@@ -11,8 +11,9 @@
 import json
 import re
 import threading
+import time
 from dataclasses import dataclass, field
-from http.server import BaseHTTPRequestHandler, HTTPServer
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any
 
 import pytest
@@ -38,8 +39,11 @@ class ServerState:
     polls_to_succeed: int = 1
     # Terminal status the job reaches.
     terminal_status: str = "succeeded"
-    # SSE behavior: "reconnect" drops the first stream before terminal.
+    # SSE behavior: "reconnect" drops the first stream before terminal;
+    # "stall" sends a couple frames then holds the connection open, silent
+    # (a "zombie": no terminal, no close) for `stall_seconds`.
     sse_mode: str = "normal"
+    stall_seconds: float = 2.0
     # If set, GET /assets/{id}/content responds 302 to this URL instead of
     # serving bytes directly (simulates a signed-URL redirect to another host).
     redirect_content_to: str | None = None
@@ -236,6 +240,14 @@ def frame(event: str, data: dict) -> None:
                 # First connection: a progress frame, then drop without terminal.
                 frame("progress", {"value": 0.4, "nodes_done": 4, "nodes_total": 10})
                 return
+            if state.sse_mode == "stall" and state.events_connect_count == 1:
+                # First connection: a couple frames, then hold the socket open and
+                # silent (no terminal, no close) — a "zombie" the client must
+                # recover from via its read-idle timeout + poll fallback.
+                frame("status", {"status": "running"})
+                frame("progress", {"value": 0.3, "nodes_done": 3, "nodes_total": 10})
+                time.sleep(state.stall_seconds)
+                return
             # Normal (or the reconnect's 2nd connection): full run to terminal.
             frame("status", {"status": "running"})
             frame("progress", {"value": 0.5, "nodes_done": 5, "nodes_total": 10})
@@ -318,7 +330,7 @@ def _post_jobs(self) -> None:
 
 
 class _Server:
-    def __init__(self, httpd: HTTPServer, state: ServerState) -> None:
+    def __init__(self, httpd: ThreadingHTTPServer, state: ServerState) -> None:
         self._httpd = httpd
         self.state = state
         host, port = httpd.server_address[:2]
@@ -327,7 +339,7 @@ def __init__(self, httpd: HTTPServer, state: ServerState) -> None:
 
 def _start_server() -> _Server:
     state = ServerState()
-    httpd = HTTPServer(("127.0.0.1", 0), _make_handler(state))
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _make_handler(state))
     thread = threading.Thread(target=httpd.serve_forever, daemon=True)
     thread.start()
     srv = _Server(httpd, state)

tests/test_sse_idle.py

@@ -0,0 +1,78 @@
+"""events() must recover from a silently-stalled ("zombie") SSE connection
+instead of blocking forever — a read-idle timeout trips, and the poll fallback
+takes over. Regression for the long-job SSE hang."""
+
+from __future__ import annotations
+
+import time
+
+import httpx
+
+from comfy_low import transport
+from comfy_sdk import AsyncComfy, Comfy, Progress, StatusChange
+
+
+def _wf(client):
+    return client.workflows.from_json({"3": {"class_type": "KSampler", "inputs": {}}})
+
+
+def test_events_recovers_from_zombie_stream(server, monkeypatch) -> None:
+    # Short read-idle timeout so the stalled stream trips it quickly.
+    monkeypatch.setattr(transport, "_SSE_IDLE_TIMEOUT", httpx.Timeout(5.0, read=0.3))
+    server.state.sse_mode = "stall"
+    # Socket held open + silent, well past the 0.3s read timeout.
+    server.state.stall_seconds = 3.0
+    server.state.polls_to_succeed = 1  # poll fallback sees terminal immediately
+
+    with Comfy(server.base_url) as client:
+        job = client.submit(_wf(client))
+        t0 = time.monotonic()
+        seen = list(job.events())  # MUST NOT hang
+        elapsed = time.monotonic() - t0
+
+    kinds = [type(e).__name__ for e in seen]
+    # Delivered the pre-stall frames, then recovered to a terminal StatusChange via poll.
+    assert any(isinstance(e, Progress) for e in seen), kinds
+    assert isinstance(seen[-1], StatusChange) and seen[-1].status == "succeeded", kinds
+    # Recovery must happen around the ~0.3s read timeout, NOT after the full 3s
+    # stall (which would mean it waited for the socket to close instead of timing out).
+    assert elapsed < 2.0, f"events() did not time out on the idle stream ({elapsed:.2f}s)"
+
+
+async def test_async_events_recovers_from_zombie_stream(server, monkeypatch) -> None:
+    # The async path uses the same idle timeout + poll fallback — must also recover.
+    monkeypatch.setattr(transport, "_SSE_IDLE_TIMEOUT", httpx.Timeout(5.0, read=0.3))
+    server.state.sse_mode = "stall"
+    server.state.stall_seconds = 3.0
+    server.state.polls_to_succeed = 1
+
+    async with AsyncComfy(server.base_url) as client:
+        job = await client.submit(_wf(client))
+        t0 = time.monotonic()
+        seen = [e async for e in job.events()]  # MUST NOT hang
+        elapsed = time.monotonic() - t0
+
+    kinds = [type(e).__name__ for e in seen]
+    assert any(isinstance(e, Progress) for e in seen), kinds
+    assert isinstance(seen[-1], StatusChange) and seen[-1].status == "succeeded", kinds
+    assert elapsed < 2.0, f"async events() did not time out on the idle stream ({elapsed:.2f}s)"
+
+
+def test_get_job_events_timeout_none_opts_out_of_idle_timeout(server, monkeypatch) -> None:
+    # The documented escape hatch: timeout=None disables the idle timeout, so the
+    # raw iterator rides out the stall until the socket closes rather than raising.
+    monkeypatch.setattr(transport, "_SSE_IDLE_TIMEOUT", httpx.Timeout(5.0, read=0.05))
+    server.state.sse_mode = "stall"
+    server.state.stall_seconds = 0.8  # held open, then the stub closes
+
+    with Comfy(server.base_url) as client:
+        job = client.submit(_wf(client))
+        events_url = job._model.urls.events
+        t0 = time.monotonic()
+        raws = list(client._low.get_job_events(events_url, timeout=None))
+        elapsed = time.monotonic() - t0
+
+    # Rode out the full stall (until close) instead of tripping the 0.05s default —
+    # proves timeout=None disabled the idle timeout. And it delivered both frames.
+    assert elapsed >= 0.7, f"timeout=None should not have timed out early ({elapsed:.2f}s)"
+    assert len(raws) == 2, raws