From ec966628f3fdc2ec74204181199e54f68ccc8266 Mon Sep 17 00:00:00 2001 From: Sahith Reddy Thummala Date: Sat, 1 Aug 2026 22:21:01 -0400 Subject: [PATCH 1/3] fix(streaming): drain remaining bytes after [DONE] before closing response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When Stream.__stream__ encounters the [DONE] SSE event, it immediately breaks out of the loop and falls into the finally block that calls response.close(). At that point, the underlying HTTP/1.1 chunked transfer iterator may not have been read to EOF — specifically, the chunked terminator (0\r\n\r\n) may still be buffered in the kernel or in h11's receive buffer. h11 tracks the remote state machine. When response.close() is called while h11's their_state is still SEND_RESPONSE (i.e. the chunked terminator has not yet been parsed), httpcore takes the 'connection must be destroyed' branch instead of the 'return to pool (IDLE)' branch. This emits an immediate TCP FIN on the socket, which manifests as: - Upstream proxy logs: spike of downstream_remote_disconnect - Client side: occasional httpcore.RemoteProtocolError on the next request because the connection was torn down uncleanly Fix: After observing [DONE], exhaust the remaining SSE iterator before breaking. This drains the chunked terminator through h11's state machine so that their_state reaches DONE, and response.close() then takes the graceful 'return to pool' path. Regression introduced in 6132922c (fix(client): close streams without requiring full consumption). Previously fixed in 7e2b2544. Closes #3440 --- src/openai/_streaming.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 45c13cc11d..68efccbebf 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -61,6 +61,13 @@ def __stream__(self) -> Iterator[_T]: try: for sse in iterator: if sse.data.startswith("[DONE]"): + # Drain the remaining bytes from the underlying iterator so that + # the HTTP/1.1 chunked terminator (0\r\n\r\n) is fully consumed. + # Without this, h11's `their_state` is still SEND_RESPONSE when + # response.close() is called, causing httpcore to destroy the + # connection (TCP FIN) instead of returning it to the pool. + 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 +178,14 @@ async def __stream__(self) -> AsyncIterator[_T]: try: async for sse in iterator: if sse.data.startswith("[DONE]"): + # Drain the remaining bytes from the underlying async iterator so + # that the HTTP/1.1 chunked terminator (0\r\n\r\n) is fully + # consumed. Without this, h11's `their_state` is still + # SEND_RESPONSE when aclose() is called, causing httpcore to + # destroy the connection (TCP FIN) instead of returning it to the + # pool. + 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 From a9919a640a51055785ed45544578e9447a195cf5 Mon Sep 17 00:00:00 2001 From: Sahith Reddy Thummala Date: Sun, 2 Aug 2026 16:12:22 -0400 Subject: [PATCH 2/3] fix(streaming): make post-DONE drain best-effort to address Codex review Address two P2 issues raised by the Codex reviewer on PR #3566: 1. Suppress post-DONE drain failures If the transport reports a read/protocol error while consuming bytes after [DONE] (e.g. a proxy closes the chunked response without the terminating chunk), the bare drain loop would propagate that exception even though the SDK had already received the terminal application marker. The fix wraps the drain in try/except Exception so that transport noise after [DONE] is silently swallowed and the stream still closes cleanly. 2. Bound the drain (avoid waiting past the terminal SSE marker) If a server or proxy emits [DONE] but leaves the SSE connection open, the unbounded drain would block until EOF (potentially forever with no read timeout). By catching all exceptions from the drain, the user's configured read timeout will fire as an exception inside the loop, which is now caught. The drain therefore becomes truly best-effort: it drains if possible, and gives up gracefully otherwise. --- src/openai/_streaming.py | 55 ++++++++++++++++++++++++++++++---------- 1 file changed, 42 insertions(+), 13 deletions(-) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 68efccbebf..3bee859038 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -61,12 +61,27 @@ def __stream__(self) -> Iterator[_T]: try: for sse in iterator: if sse.data.startswith("[DONE]"): - # Drain the remaining bytes from the underlying iterator so that - # the HTTP/1.1 chunked terminator (0\r\n\r\n) is fully consumed. - # Without this, h11's `their_state` is still SEND_RESPONSE when - # response.close() is called, causing httpcore to destroy the - # connection (TCP FIN) instead of returning it to the pool. - for _ in iterator: + # Best-effort drain: consume the remaining bytes so that + # the HTTP/1.1 chunked terminator (0\r\n\r\n) is read by + # h11 before response.close() is called. Without this, + # h11's `their_state` stays SEND_RESPONSE and httpcore + # destroys the socket (TCP FIN) instead of returning the + # connection to the pool. + # + # The drain is wrapped in try/except for two reasons: + # 1. Suppress post-DONE transport errors: if the proxy + # closes the connection abruptly after [DONE], the + # read raises a protocol error that is irrelevant + # because the application-level stream is already done. + # 2. Bound the drain: if the server keeps the SSE + # connection open indefinitely after [DONE], the + # user's configured read timeout will eventually fire + # as an exception here, which we catch and ignore so + # the stream still closes cleanly. + try: + for _ in iterator: + pass + except Exception: pass break @@ -178,13 +193,27 @@ async def __stream__(self) -> AsyncIterator[_T]: try: async for sse in iterator: if sse.data.startswith("[DONE]"): - # Drain the remaining bytes from the underlying async iterator so - # that the HTTP/1.1 chunked terminator (0\r\n\r\n) is fully - # consumed. Without this, h11's `their_state` is still - # SEND_RESPONSE when aclose() is called, causing httpcore to - # destroy the connection (TCP FIN) instead of returning it to the - # pool. - async for _ in iterator: + # Best-effort drain: consume the remaining bytes so that + # the HTTP/1.1 chunked terminator (0\r\n\r\n) is read by + # h11 before aclose() is called. Without this, h11's + # `their_state` stays SEND_RESPONSE and httpcore destroys + # the socket (TCP FIN) instead of returning the connection + # to the pool. + # + # The drain is wrapped in try/except for two reasons: + # 1. Suppress post-DONE transport errors: if the proxy + # closes the connection abruptly after [DONE], the + # read raises a protocol error that is irrelevant + # because the application-level stream is already done. + # 2. Bound the drain: if the server keeps the SSE + # connection open indefinitely after [DONE], the + # user's configured read timeout will eventually fire + # as an exception here, which we catch and ignore so + # the stream still closes cleanly. + try: + async for _ in iterator: + pass + except Exception: pass break From 0dca8808319d6ccc3f56b048ebb359c1c3b6f59d Mon Sep 17 00:00:00 2001 From: Sahith Reddy Thummala Date: Sun, 2 Aug 2026 16:18:57 -0400 Subject: [PATCH 3/3] fix(streaming): trim verbose comments to match codebase style --- src/openai/_streaming.py | 40 ++++++---------------------------------- 1 file changed, 6 insertions(+), 34 deletions(-) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 3bee859038..006ab34f21 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -61,23 +61,9 @@ def __stream__(self) -> Iterator[_T]: try: for sse in iterator: if sse.data.startswith("[DONE]"): - # Best-effort drain: consume the remaining bytes so that - # the HTTP/1.1 chunked terminator (0\r\n\r\n) is read by - # h11 before response.close() is called. Without this, - # h11's `their_state` stays SEND_RESPONSE and httpcore - # destroys the socket (TCP FIN) instead of returning the - # connection to the pool. - # - # The drain is wrapped in try/except for two reasons: - # 1. Suppress post-DONE transport errors: if the proxy - # closes the connection abruptly after [DONE], the - # read raises a protocol error that is irrelevant - # because the application-level stream is already done. - # 2. Bound the drain: if the server keeps the SSE - # connection open indefinitely after [DONE], the - # user's configured read timeout will eventually fire - # as an exception here, which we catch and ignore so - # the stream still closes cleanly. + # Drain remaining bytes so h11 fully parses the chunked terminator + # before close(), allowing the connection to be returned to the pool. + # Errors are suppressed — the stream is done at [DONE] regardless. try: for _ in iterator: pass @@ -193,23 +179,9 @@ async def __stream__(self) -> AsyncIterator[_T]: try: async for sse in iterator: if sse.data.startswith("[DONE]"): - # Best-effort drain: consume the remaining bytes so that - # the HTTP/1.1 chunked terminator (0\r\n\r\n) is read by - # h11 before aclose() is called. Without this, h11's - # `their_state` stays SEND_RESPONSE and httpcore destroys - # the socket (TCP FIN) instead of returning the connection - # to the pool. - # - # The drain is wrapped in try/except for two reasons: - # 1. Suppress post-DONE transport errors: if the proxy - # closes the connection abruptly after [DONE], the - # read raises a protocol error that is irrelevant - # because the application-level stream is already done. - # 2. Bound the drain: if the server keeps the SSE - # connection open indefinitely after [DONE], the - # user's configured read timeout will eventually fire - # as an exception here, which we catch and ignore so - # the stream still closes cleanly. + # Drain remaining bytes so h11 fully parses the chunked terminator + # before aclose(), allowing the connection to be returned to the pool. + # Errors are suppressed — the stream is done at [DONE] regardless. try: async for _ in iterator: pass