AsyncStream leaves its SSE event iterators to the garbage collector after [DONE] or close() #4174

Activity

1. Asutosa commented on Oct 24, 2026

   

   Asutosa

   on Oct 24, 2026

   More actions

   I investigated the async SSE iterator cleanup issue and pushed a proposed fix to my fork.

   Changes: Added explicit async iterator cleanup and preserved the direct iterator returned by AsyncStream.__aiter__().

   Validation: python -m pytest tests/test_streaming.py -q — 64 passed, 34 skipped.

   Commit: Asutosa@e0a065f

   I noticed the related fix already linked here and would appreciate maintainer feedback on whether this approach addresses the remaining cleanup cases.
2. hemv-857 commented on Oct 27, 2026

   

   hemv-857

   on Oct 27, 2026

   More actions

   Root-cause analysis and a complete patch, posted here per CONTRIBUTING (PRs are limited to collaborators).

   Where the generators leak

   src/openai/_streaming.py opens a nested chain of async generators and never closes it:

   1. AsyncStream._iter_events iterates self._decoder.aiter_bytes(self.response.aiter_bytes()). An interrupted async for does not close the iterator it was looping over, so both the decoder iterator and the response byte iterator stay suspended.
   2. __stream__ breaks at [DONE]; its finally closes only response, not iterator.
   3. close() closes only response, not self._iterator.
   4. After an early exit, the __aiter__ wrapper is also left suspended.

   asyncio's GC finalizer then starts a separate aclose() task per unreachable async generator, and a chain that becomes unreachable together is closed concurrently, which is where RuntimeError('aclose(): asynchronous generator is already running') comes from. The httpx-side half of the race is tracked in pydantic/httpx2#1280 / #1281; the openai-owned generators above are independent of that and still need explicit cleanup.

   Patch

   Four touchpoints in _streaming.py, closing inner-to-outer:

   # _iter_events: bind the nested generators and close them in the finally
   async def _iter_events(self) -> AsyncIterator[ServerSentEvent]:
       response_iter = self.response.aiter_bytes()
       decoded = self._decoder.aiter_bytes(response_iter)
       try:
           async for sse in decoded:
               yield sse
       except timeout_exceptions() as err:
           raise APITimeoutError(request=self.response.request) from err
       except request_exceptions() as err:
           raise APIConnectionError(request=self.response.request) from err
       finally:
           await decoded.aclose()
           await response_iter.aclose()
   
   # __aiter__: close the wrapper when the consumer stops iterating
   async def __aiter__(self) -> AsyncIterator[_T]:
       try:
           async for item in self._iterator:
               yield item
       finally:
           await self._iterator.aclose()
   
   # __stream__ finally: close the iterator, then the response
   finally:
       try:
           await iterator.aclose()
       finally:
           await response.aclose()
   
   # close(): close the iterator, then the response
   async def close(self) -> None:
       try:
           await self._iterator.aclose()
       finally:
           await self.response.aclose()

   Validation

   • New regression test in tests/test_streaming.py: consume a stream with an early exit, then assert every generator in the chain reports ag_frame == None after await stream.close() (i.e. all closed, none left for the GC finalizer).
   • python -m pytest tests/test_streaming.py -q (excluding the aiohttp lane, extra not installed): 60 passed.
   • ruff check / ruff format --check on both files: clean.

   Reference implementation with the full test: https://github.com/hemv-857/openai-python/tree/fix/4174-async-stream-close-nested-generators (commit fb9ea7c). Happy to reshape it to match maintainer preferences, or to fold in anything @Asutosa's approach covers that this one does not.