From 157765bf637d74718c67ef88c0b56beacf71fb66 Mon Sep 17 00:00:00 2001 From: wei-hai Date: Tue, 21 Jul 2026 21:49:53 -0700 Subject: [PATCH 1/2] fix: read-idle timeout on SSE so a stalled stream can't hang events() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live-events SSE reader used timeout=None, so a connection that goes silent-but-open (no frames, no keepalive, no close — e.g. a gateway idle timeout on a long job) blocked iter_lines() forever. The high-level events() reconnect/poll-fallback only runs when the stream *ends*, so it never recovered — events() hung indefinitely on jobs longer than a few minutes, even though the job itself completed fine (poll/wait/run stayed correct). Give get_job_events (sync + async) a read-idle timeout well above the server's keepalive interval: a stalled stream now raises ReadTimeout, which events() already catches and turns into a poll/reconnect. Adds a threaded stub "stall" mode + a regression test proving recovery instead of a hang. Co-Authored-By: Claude Opus 4.8 --- src/comfy_low/transport.py | 25 ++++++++++++++++++------ tests/conftest.py | 20 +++++++++++++++---- tests/test_sse_idle.py | 39 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 10 deletions(-) create mode 100644 tests/test_sse_idle.py diff --git a/src/comfy_low/transport.py b/src/comfy_low/transport.py index f906377..55fd563 100644 --- a/src/comfy_low/transport.py +++ b/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,)) diff --git a/tests/conftest.py b/tests/conftest.py index d553e98..0e17a40 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/test_sse_idle.py b/tests/test_sse_idle.py new file mode 100644 index 0000000..71d5d8e --- /dev/null +++ b/tests/test_sse_idle.py @@ -0,0 +1,39 @@ +"""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 Comfy, Progress, StatusChange + + +def _wf(client: Comfy): + 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)" From 8d217ec82cfcd26c99d3b89effb0b3dd5b14f5b1 Mon Sep 17 00:00:00 2001 From: wei-hai Date: Tue, 21 Jul 2026 22:01:26 -0700 Subject: [PATCH 2/2] test: cover async SSE recovery + timeout=None opt-out Extend the zombie-stream regression to the async get_job_events path (same idle timeout + poll fallback) and add a test proving timeout=None disables the idle timeout (rides out the stall until close instead of raising). Co-Authored-By: Claude Opus 4.8 --- tests/test_sse_idle.py | 43 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 41 insertions(+), 2 deletions(-) diff --git a/tests/test_sse_idle.py b/tests/test_sse_idle.py index 71d5d8e..0231146 100644 --- a/tests/test_sse_idle.py +++ b/tests/test_sse_idle.py @@ -9,10 +9,10 @@ import httpx from comfy_low import transport -from comfy_sdk import Comfy, Progress, StatusChange +from comfy_sdk import AsyncComfy, Comfy, Progress, StatusChange -def _wf(client: Comfy): +def _wf(client): return client.workflows.from_json({"3": {"class_type": "KSampler", "inputs": {}}}) @@ -37,3 +37,42 @@ def test_events_recovers_from_zombie_stream(server, monkeypatch) -> None: # 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