[PR #214] fix(jobs): back off the events() reconnect pause on streams that keep dropping #214

New issue

Open

opened 2026-10-31 22:02:24 +00:00 by bsakaguchi · 0 comments

bsakaguchi commented

2026-10-31 22:02:24 +00:00

Owner

📋 Pull Request Information

Original PR: https://github.com/Comfy-Org/comfy-python-sdk/pull/214
Author: @mattmillerai
Created: 10/9/2026
Status: 🔄 Open

Base: main ← Head: matt/be-19980-events-reconnect-backoff

📝 Commits (2)

82a7e87 fix(jobs): back off the events() reconnect pause on streams that keep dropping
64fa27d fix(jobs): reset the events() reconnect backoff on connection lifetime, not delivered frames

📊 Changes

5 files changed (+182 additions, -5 deletions)

View changed files

📝 CHANGELOG.md (+10 -0)
📝 src/comfy_sdk/jobs.py (+27 -2)
📝 tests/conftest.py (+16 -1)
📝 tests/test_events.py (+127 -0)
📝 uv.lock (+2 -2)

📄 Description

ELI-5

When the live event stream for a job closes without saying the job is finished, the SDK checks the job's status and then reconnects. Until now it always waited 0.1 s before reconnecting. If a server or proxy keeps closing the stream straight away, that turns into about 10 reconnects and 10 status checks every second for as long as the job runs. That happens whether it closes with no data or right after the status snapshot every connection starts with. Now each short-lived reconnect waits twice as long as the one before, up to 5 s. Once a connection stays open for at least 5 s, the wait goes back to 0.1 s.

What changed

src/comfy_sdk/jobs.py: Job.events() and AsyncJob.events() now create a single _core.backoff_schedule(_RECONNECT_PAUSE, 2.0, _MAX_RECONNECT_PAUSE) per call, so the pauses go 0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0, … The new constant _MAX_RECONNECT_PAUSE = 5.0 matches the default cap in backoff_schedule. The code comment explains the cap: events() has no deadline of its own, so a dropped healthy stream still has to notice a terminal status within seconds.
The schedule starts over only when the connection that just ended stayed open for at least _HEALTHY_STREAM_SECONDS (= the 5 s cap), measured from connect until the stream ended. Receiving frames does not count as a sign of health: per the getJobEvents contract every connection opens with a status/progress snapshot, so resetting after any frame (this PR's first revision) left the snapshot-then-close storm unfixed. A connection that lasted at least the cap cannot reconnect faster than a capped pause would allow.
Only the drop path is affected: a clean close without a terminal frame, or an httpx.HTTPError / httpx.StreamError mid-stream. The ApiError branch is untouched, so 501 still ends iteration silently and other statuses, 429 included, still raise right away. The first pause is still 0.1 s.
tests/conftest.py: a new stub mode, sse_mode = "drop". Every connection gets 200 text/event-stream and closes without a terminal frame, by default with no frames at all. sse_drop_snapshot = True sends the on-connect status: running + progress snapshot before closing. sse_hold_on_connect = N / sse_hold_seconds hold connection N open first, using threading.Event().wait because tests fake time.sleep.
tests/test_events.py: five new tests:
- sync and async backoff on empty drops: record pauses [0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0]; the sync one also checks that connects and polls are each at most pauses + 1
- sync and async backoff on snapshot-then-close: same pause schedule, even though every connection delivered frames
- reset: 3 short snapshot-then-close connects, then one held past the threshold (patched to 0.3 s), then more short ones; the pauses come out as 0.1, 0.2, 0.4, 0.1, 0.2, 0.4
CHANGELOG.md: new entry under Unreleased → Fixed.
uv.lock: resynced to pyproject.toml. Main's mypy specifier bump (#203) did not update the lock, so uv run rewrote it. This PR has no dependency changes of its own.

Judgment calls

The async tests patch asyncio.sleep on the module, because jobs.py imports asyncio inside the function. The fake records every call and still yields to the loop with real_sleep(0). The exact-list assertion would fail if anything else in that test path called asyncio.sleep.
The reset test covers only the sync path. The async body is a line-for-line mirror.
Health is judged by how long the connection stayed open, not by frame count, because the on-connect snapshot (status, latest progress, latest preview) has no fixed size, so "frames beyond the snapshot" can't be told apart reliably.
No jitter on the pause. This matches the SDK's other backoff (wait()/result() polling), and the 5 s cap already bounds a client at one connect plus one poll every 5 s.

Provenance

Authored by: agent-work loop
Verified:
- Tests: full suite green on 3.12, 1270 passed and 9 skipped (live-gateway integration tests skip when their env vars are not set)
- Lint/format/mypy: ruff check ., ruff format --check . and mypy src all clean
- Public-repo hygiene script: clean
- Mutation-tested: forcing the reset to always fire fails all 5 new tests; forcing it to never fire fails the reset test
- Not run locally: the 3.10/3.11/3.13 matrix legs, build-check, codegen-drift (no generated or vendored files touched)
Deviations: the reset rule changed from "the connection delivered a frame" to "the connection stayed open ≥ 5 s", after review showed the on-connect snapshot defeats the frame-based rule

Residual

None for the behavior asked for. The TypeScript SDK has the same reconnect loop, and that change is tracked separately. It is not part of this repo.

_{🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.}

## 📋 Pull Request Information **Original PR:** https://github.com/Comfy-Org/comfy-python-sdk/pull/214 **Author:** [@mattmillerai](https://github.com/mattmillerai) **Created:** 10/9/2026 **Status:** 🔄 Open **Base:** `main` ← **Head:** `matt/be-19980-events-reconnect-backoff` --- ### 📝 Commits (2) - [`82a7e87`](https://github.com/Comfy-Org/comfy-python-sdk/commit/82a7e877c9141920d3007922f2437dd6d2a607d1) fix(jobs): back off the events() reconnect pause on streams that keep dropping - [`64fa27d`](https://github.com/Comfy-Org/comfy-python-sdk/commit/64fa27d60c8e80c0ab2fad0a54ea87e3d0b6e190) fix(jobs): reset the events() reconnect backoff on connection lifetime, not delivered frames ### 📊 Changes **5 files changed** (+182 additions, -5 deletions) <details> <summary>View changed files</summary> 📝 `CHANGELOG.md` (+10 -0) 📝 `src/comfy_sdk/jobs.py` (+27 -2) 📝 `tests/conftest.py` (+16 -1) 📝 `tests/test_events.py` (+127 -0) 📝 `uv.lock` (+2 -2) </details> ### 📄 Description ## ELI-5 When the live event stream for a job closes without saying the job is finished, the SDK checks the job's status and then reconnects. Until now it always waited 0.1 s before reconnecting. If a server or proxy keeps closing the stream straight away, that turns into about 10 reconnects and 10 status checks every second for as long as the job runs. That happens whether it closes with no data or right after the status snapshot every connection starts with. Now each short-lived reconnect waits twice as long as the one before, up to 5 s. Once a connection stays open for at least 5 s, the wait goes back to 0.1 s. ## What changed - `src/comfy_sdk/jobs.py`: `Job.events()` and `AsyncJob.events()` now create a single `_core.backoff_schedule(_RECONNECT_PAUSE, 2.0, _MAX_RECONNECT_PAUSE)` per call, so the pauses go 0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0, … The new constant `_MAX_RECONNECT_PAUSE = 5.0` matches the default cap in `backoff_schedule`. The code comment explains the cap: `events()` has no deadline of its own, so a dropped healthy stream still has to notice a terminal status within seconds. - The schedule starts over only when the connection that just ended stayed open for at least `_HEALTHY_STREAM_SECONDS` (= the 5 s cap), measured from connect until the stream ended. Receiving frames does not count as a sign of health: per the `getJobEvents` contract every connection opens with a `status`/`progress` snapshot, so resetting after any frame (this PR's first revision) left the snapshot-then-close storm unfixed. A connection that lasted at least the cap cannot reconnect faster than a capped pause would allow. - Only the drop path is affected: a clean close without a terminal frame, or an `httpx.HTTPError` / `httpx.StreamError` mid-stream. The `ApiError` branch is untouched, so 501 still ends iteration silently and other statuses, 429 included, still raise right away. The first pause is still 0.1 s. - `tests/conftest.py`: a new stub mode, `sse_mode = "drop"`. Every connection gets 200 `text/event-stream` and closes without a terminal frame, by default with no frames at all. `sse_drop_snapshot = True` sends the on-connect `status: running` + `progress` snapshot before closing. `sse_hold_on_connect = N` / `sse_hold_seconds` hold connection N open first, using `threading.Event().wait` because tests fake `time.sleep`. - `tests/test_events.py`: five new tests: - sync and async backoff on empty drops: record pauses `[0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0]`; the sync one also checks that connects and polls are each at most pauses + 1 - sync and async backoff on snapshot-then-close: same pause schedule, even though every connection delivered frames - reset: 3 short snapshot-then-close connects, then one held past the threshold (patched to 0.3 s), then more short ones; the pauses come out as `0.1, 0.2, 0.4, 0.1, 0.2, 0.4` - `CHANGELOG.md`: new entry under Unreleased → Fixed. - `uv.lock`: resynced to `pyproject.toml`. Main's mypy specifier bump (#203) did not update the lock, so `uv run` rewrote it. This PR has no dependency changes of its own. ## Judgment calls - The async tests patch `asyncio.sleep` on the module, because `jobs.py` imports `asyncio` inside the function. The fake records every call and still yields to the loop with `real_sleep(0)`. The exact-list assertion would fail if anything else in that test path called `asyncio.sleep`. - The reset test covers only the sync path. The async body is a line-for-line mirror. - Health is judged by how long the connection stayed open, not by frame count, because the on-connect snapshot (status, latest progress, latest preview) has no fixed size, so "frames beyond the snapshot" can't be told apart reliably. - No jitter on the pause. This matches the SDK's other backoff (`wait()`/`result()` polling), and the 5 s cap already bounds a client at one connect plus one poll every 5 s. ## Provenance - **Authored by:** agent-work loop - **Verified:** - Tests: full suite green on 3.12, 1270 passed and 9 skipped (live-gateway integration tests skip when their env vars are not set) - Lint/format/mypy: `ruff check .`, `ruff format --check .` and `mypy src` all clean - Public-repo hygiene script: clean - Mutation-tested: forcing the reset to always fire fails all 5 new tests; forcing it to never fire fails the reset test - Not run locally: the 3.10/3.11/3.13 matrix legs, `build-check`, `codegen-drift` (no generated or vendored files touched) - **Deviations:** the reset rule changed from "the connection delivered a frame" to "the connection stayed open ≥ 5 s", after review showed the on-connect snapshot defeats the frame-based rule ## Residual - None for the behavior asked for. The TypeScript SDK has the same reconnect loop, and that change is tracked separately. It is not part of this repo. --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>

bsakaguchi added the

pull-request

label

2026-10-31 22:02:24 +00:00