Move SSE keep-alive to a transport-layer wrapper#47973
Open
antriksh30 wants to merge 5 commits into
Open
Conversation
antriksh30
requested review from
RaviPidaparthi,
ankitbko and
vangarp
as code owners
added 3 commits
July 9, 2026 18:05
…wrapper Keep-alive (periodic ': keep-alive' SSE comment frames) lived inside the event-pipeline orchestrator, which branched on the keep-alive interval. When enabled (hosted-platform default), background+stream ran through a merge path whose finally cancelled the handler on client disconnect, killing in-flight background runs so GET ?stream=true reconnect returned 200 with no new events. Keep-alive is a transport concern. Introduce with_keep_alive in streaming/_sse.py — a single-pump-task wrapper that interleaves heartbeats into any AsyncIterator[str] during idle gaps (passthrough when disabled), advancing the source in one task so its contextvars (request context, SSE sequence counter) are preserved. Apply it at the create-POST and reconnect-GET StreamingResponse sites in the endpoint handler. Remove all keep-alive knowledge from the orchestrator: _live_stream collapses from three keep-alive-aware paths to two lifecycle paths (non-bg simple iterator vs. bg+store shielded task). The shielded background path (which survives client disconnect, FR-012/FR-013) is now unconditional instead of gated on keep-alive being off. Reconnect GET streams now also get heartbeats. Add unit tests for with_keep_alive (passthrough, idle heartbeat, no drop/reorder, contextvar preservation, deterministic close, idle-from-start reconnect shape) and strengthen the sequence-number integrity assertion to catch contextvar regressions.
Move the with_keep_alive unit tests into test_sse_writer.py (the unit home for _sse.py) and drop the separate test file. Reword source and test comments to describe current behavior.
The multi-gap order test covered the same branch and assertions as the single idle-gap heartbeat test.
antriksh30
force-pushed
the
antrikshjain/keep-alive-fix
branch
from
17248db to
e25ea87
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
This PR relocates SSE keep-alive comment injection from the orchestrator into the transport layer by introducing a reusable with_keep_alive async wrapper. The orchestrator's _live_stream is simplified to emit only real events, and the endpoint handler now wraps its streaming responses with with_keep_alive.
Changes:
- Add
with_keep_alivetostreaming/_sse.py, which interleaves keep-alive frames during idle gaps using a background pump task and queue. - Remove the keep-alive/merge-queue logic from
_orchestrator._live_stream, leaving only real-event emission (background+store path preserved). - Wire
with_keep_aliveinto both streaming responses in_endpoint_handler.py, and add unit tests plus stronger contract-test assertions.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
streaming/_sse.py |
New with_keep_alive async wrapper emitting heartbeat frames on idle gaps. |
hosting/_orchestrator.py |
Removes in-orchestrator keep-alive path; _live_stream now emits only real events. |
hosting/_endpoint_handler.py |
Wraps both SSE streaming responses with with_keep_alive. |
tests/unit/test_sse_writer.py |
Adds unit tests for with_keep_alive behavior. |
tests/contract/test_keep_alive.py |
Strengthens sequence-number integrity assertions. |
Widen interval_seconds to float, surface source errors to the consumer instead of swallowing them, drop a redundant assertion, and rename a test variable flagged by spell-check.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Keep-alive (periodic
: keep-aliveSSE comment frames) lived inside the event-pipeline orchestrator, which branched on the keep-alive interval. When enabled (the hosted-platform default),background=true+stream=trueruns took a merge path whosefinallycancelled the handler on client disconnect. This killed the in-flight background run, so a laterGET /responses/{id}?stream=true&starting_after=Nreconnect returned200with no new events.Change
with_keep_aliveinstreaming/_sse.py: a single-pump-task wrapper that interleaves heartbeats into anyAsyncIterator[str]during idle gaps (passthrough when disabled). The source is advanced by one task, so its contextvars (request context, SSE sequence counter) are preserved.StreamingResponsesites in the endpoint handler._live_streamcollapses from three keep-alive-aware paths to two lifecycle paths (non-bg simple iterator vs. bg+store shielded task). The shielded background path that survives client disconnect is now unconditional instead of gated on keep-alive being off. Reconnect GET streams now also get heartbeats.Tests
Unit tests for
with_keep_alive(passthrough, idle heartbeat, contextvar preservation, deterministic close, idle-from-start reconnect, source error propagation) and a strengthened sequence-number integrity assertion.