Skip to content

fix(examples): handle slow broadcasting clients - #209

Merged
sysid merged 2 commits into
sysid:mainfrom
varadfromeast:fix/broadcasting-backpressure
Sep 5, 2026
Merged

sysid merged 2 commits into
sysid:mainfrom
varadfromeast:fix/broadcasting-backpressure

Conversation

@varadfromeast

Copy link
Copy Markdown
Contributor

Summary

  • bound each broadcasting client's queue so slow consumers cannot grow memory without limit
  • disconnect slow clients when their queue fills and wake their stream with a sentinel
  • clean up registered clients when EventSourceResponse cancels the iterator on disconnect
  • add regression coverage for bounded queues, slow-client eviction, and cancellation cleanup

Root cause

The example used asyncio.Queue() with the default maxsize=0, so its QueueFull branch was unreachable. Simply bounding the queue was not sufficient: removing a full queue from the broadcaster did not terminate the corresponding stream, which could later wait forever for another event. In addition, asyncio.CancelledError is not caught by except Exception, so normal response cancellation could leave the client registered.

The example now uses a bounded queue with an explicit disconnect policy. When a queue fills, buffered events are discarded and a sentinel terminates the evicted stream immediately; other clients continue receiving events through put_nowait().

Verification

  • make test-unit — 81 passed, 2 deselected
  • make ty
  • uv run ruff check examples/02_broadcasting.py tests/test_broadcasting_example.py
  • uv run ruff format --check examples/02_broadcasting.py tests/test_broadcasting_example.py

@sysid

sysid commented Sep 3, 2026

Copy link
Copy Markdown
Owner

@varadfromeast, thanks for bringing this up. Please rebase and incorporate some minor improvements, which I have outlined in https://github.com/sysid/sse-starlette/tree/fix/pull209.

Could not force-push to your PR.

varadfromeast and others added 2 commits September 4, 2026 19:17
EventSourceResponse calls aclose() on the body iterator when a data send times out. BroadcastStream had no aclose(), so a timed-out client stayed registered until its bounded queue filled and it was evicted. Add aclose() so cleanup runs on that path too.

Merge the CancelledError and Exception handlers into one BaseException handler; cancellation is the normal disconnect path and must clean up like any other exit.

Reword the eviction rationale: both dropping and disconnecting lose events, disconnecting just makes the gap visible. Point to event ids and Last-Event-ID as the resync mechanism this example omits.

Tests: cover aclose(), add a healthy peer to the eviction test to prove other clients keep receiving, and follow the test_x_whenY_thenZ naming convention.
@varadfromeast
varadfromeast force-pushed the fix/broadcasting-backpressure branch from 94d5270 to cb2205e Compare September 4, 2026 13:48
@varadfromeast

varadfromeast commented Sep 4, 2026 •

Copy link
Copy Markdown
Contributor Author

@sysid Rebased onto the latest main and incorporated the improvements from fix/pull209. The update adds the send-timeout cleanup path and regression coverage; the unit suite, Ruff checks, and type checks pass. Could you please review when you have a chance?

@sysid
sysid merged commit faa2c1c into sysid:main Sep 5, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants