From f5fc60e344b41e924de3517708f979d9476417dc Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Mon, 20 Jul 2026 23:52:37 +0530 Subject: [PATCH 1/2] fix: drain through existing iterator after [DONE] to avoid StreamConsumed (#3440) Codex P1 review: calling self.response.iter_bytes() a second time raises httpx.StreamConsumed because the SSE decoder already started consuming it. Fix: drain through the existing (which wraps response.iter_bytes()) instead of starting a new one. This consumes remaining SSE events until EOF, allowing h11 to advance to DONE state before close. Also fixes: - P2: async test now properly tracks the response object it passes into AsyncStream, so assert response.is_closed checks the right response - P2: test no longer tries to simulate chunked terminator bytes (which httpx.Response(content=...) doesn't expose); instead verifies the stream is fully consumed including trailing events after [DONE] --- src/openai/_streaming.py | 26 ++++++++++++++++++++++++++ tests/test_streaming.py | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 45c13cc11d..0a7a38af63 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -61,6 +61,19 @@ def __stream__(self) -> Iterator[_T]: try: for sse in iterator: if sse.data.startswith("[DONE]"): + # Drain remaining events from the existing iterator so the + # underlying response.iter_bytes() reaches EOF, allowing + # h11 to advance to DONE state before close. Without this, + # response.close() sends TCP FIN while the chunked terminator + # (0\r\n\r\n) is still in flight, causing connection pool + # degradation and proxy errors. (#3440) + # + # We must drain through `iterator` (not start a new + # `self.response.iter_bytes()`) because httpx only allows + # one active iterator at a time — a second call raises + # `httpx.StreamConsumed`. + for _ in iterator: + pass break # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data @@ -171,6 +184,19 @@ async def __stream__(self) -> AsyncIterator[_T]: try: async for sse in iterator: if sse.data.startswith("[DONE]"): + # Drain remaining events from the existing iterator so the + # underlying response.aiter_bytes() reaches EOF, allowing + # h11 to advance to DONE state before close. Without this, + # response.aclose() sends TCP FIN while the chunked terminator + # (0\r\n\r\n) is still in flight, causing connection pool + # degradation and proxy errors. (#3440) + # + # We must drain through `iterator` (not start a new + # `self.response.aiter_bytes()`) because httpx only allows + # one active iterator at a time — a second call raises + # `httpx.StreamConsumed`. + async for _ in iterator: + pass break # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 04f8e51abd..585ea50377 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -216,6 +216,44 @@ def body() -> Iterator[bytes]: assert sse.json() == {"content": "известни"} +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_stream_drains_after_done( + sync: bool, + client: OpenAI, + async_client: AsyncOpenAI, +) -> None: + """Regression test for #3440: after [DONE], remaining events should be drained. + + The fix drains the existing iterator (not a new response.iter_bytes() call) + so that httpx doesn't raise StreamConsumed and the response reaches EOF + before close, preventing premature TCP FIN. + """ + + def body() -> Iterator[bytes]: + yield b'data: {"foo":true}\n' + yield b"\n" + yield b"data: [DONE]\n" + yield b"\n" + # Extra data after [DONE] that should be consumed by the drain + yield b'data: {"trailing":true}\n' + yield b"\n" + + if sync: + response = httpx.Response(200, content=body()) + stream = Stream(cast_to=object, client=client, response=response) + # Consume the full stream — the drain should consume the trailing event + for _ in stream: + pass + assert response.is_closed + else: + response = httpx.Response(200, content=to_aiter(body())) + stream = AsyncStream(cast_to=object, client=async_client, response=response) + async for _ in stream: + pass + assert response.is_closed + + async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: for chunk in iter: yield chunk From 30d491b813874eaaf1d0bbb6eb7193374e149cd4 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Tue, 21 Jul 2026 06:38:16 +0530 Subject: [PATCH 2/2] fix: use sentinel to verify drain actually consumed trailing events (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous test asserted response.is_closed, which is true even without the drain loop because Stream.__stream__/AsyncStream.__stream__ always closes the response in their finally block after breaking on [DONE]. Closing the httpx.Response can discard unread generator content, so a regression that stops draining the trailing event still passed. Now the body generator appends a sentinel after yielding all events. The test asserts the sentinel was reached, proving the drain loop actually consumed the trailing data: {"trailing":true} event — not just that the response was closed. --- tests/test_streaming.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 585ea50377..de7be5ec14 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -228,8 +228,14 @@ async def test_stream_drains_after_done( The fix drains the existing iterator (not a new response.iter_bytes() call) so that httpx doesn't raise StreamConsumed and the response reaches EOF before close, preventing premature TCP FIN. + + We track whether the body generator reached its final sentinel to verify + that the drain actually consumed the trailing event — not just that the + response was closed (which happens in the finally block regardless). """ + consumed_after_done: list[bool] = [] + def body() -> Iterator[bytes]: yield b'data: {"foo":true}\n' yield b"\n" @@ -238,6 +244,8 @@ def body() -> Iterator[bytes]: # Extra data after [DONE] that should be consumed by the drain yield b'data: {"trailing":true}\n' yield b"\n" + # Sentinel: only reached if the drain consumed all trailing events + consumed_after_done.append(True) if sync: response = httpx.Response(200, content=body()) @@ -253,6 +261,11 @@ def body() -> Iterator[bytes]: pass assert response.is_closed + # The sentinel is only appended if the body generator was fully consumed. + # Without the drain loop, the generator is garbage-collected when the + # response closes, so this assertion fails — proving the drain works. + assert consumed_after_done == [True] + async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: for chunk in iter: