Skip to content

fix: close async stream when an error event arrives mid-stream - #3356

Open
rcleveng wants to merge 3 commits into
googleapis:mainfrom
rcleveng:fix/close-async-stream-on-error
Open

rcleveng wants to merge 3 commits into
googleapis:mainfrom
rcleveng:fix/close-async-stream-on-error

Conversation

@rcleveng

Copy link
Copy Markdown

Problem

When the server sends an {"error": ...} event partway through a streamed response, BaseApiClient.async_request_streamed raises APIError from inside its async for over the response. Leaving an async for by exception does not close the iterator, so two generators stay suspended:

  • HttpResponse.async_segments
  • HttpResponse._aiter_response_stream, which async_segments reads

The HTTP response is closed only when the garbage collector finalizes them. asyncio runs that finalization later, in a task of its own, outside the Context of the request that failed. In production we saw this after mid-stream 504 DEADLINE_EXCEEDED errors from Gemini: the connection was released late, and context-dependent cleanup (OpenTelemetry spans, ContextVar resets) ran in the wrong Context.

Fix

Wrap both loops in contextlib.aclosing, so the stream is closed before the error reaches the caller. Each wrapper is needed: closing async_segments alone leaves _aiter_response_stream suspended, because a GeneratorExit leaving an async for does not close that iterator either.

async_segments and _aiter_response_stream are now annotated AsyncGenerator rather than AsyncIterator, so mypy accepts them as closeable.

Testing

  • New test test_error_event_in_streamed_responses_async_closes_stream. The mock server sends an error event and keeps the connection open, and the test asserts the stream's aclose has already run when ServerError surfaces. It fails without the change, and it fails with either aclosing removed.
  • google/genai/tests in --mode tap: no new failures. The tests that fail do so identically with and without the change, because the replay files are not in the public repo.
  • mypy google/genai/: no new errors.

async_request_streamed raises APIError from inside its `async for` over the
response when the server sends an `{"error": ...}` event. Leaving an
`async for` by exception does not close the iterator, so
HttpResponse.async_segments and the _aiter_response_stream it reads stay
suspended. The connection is released only when the garbage collector
finalizes them, which asyncio does later in a task of its own, outside the
Context of the request that failed.

Wrap both loops in contextlib.aclosing, so the stream is closed before the
error reaches the caller. Annotate the two generators as AsyncGenerator so
the type checker accepts them as closeable.
@Venkaiahbabuneelam Venkaiahbabuneelam self-assigned this Nov 12, 2026
@Venkaiahbabuneelam Venkaiahbabuneelam added the size:L Code changes between 40-100 lines label Nov 12, 2026
@Venkaiahbabuneelam

Copy link
Copy Markdown

Hi @rcleveng, Thanks for reaching out us!
The branch is looks like out of date, kindly please update the branch.
thanks

@rcleveng

Copy link
Copy Markdown
Author

Hey @Venkaiahbabuneelam I updated it yesterday and then again today - PTAL

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

size:L Code changes between 40-100 lines

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants