From 41421b102f5939713d071b00b3390be674b75874 Mon Sep 17 00:00:00 2001 From: wwmanidumaneesha Date: Sat, 25 Jul 2026 17:59:35 +0530 Subject: [PATCH] Reset the HTTP/2 stream when a request is cancelled `_response_closed()` releases our stream semaphore, but never told `h2` that the stream was finished. When a request is cancelled before its END_STREAM frame reaches the network, `h2` continues to count the stream as open while we count it as closed. The two therefore drift apart by one on every cancelled request, until `h2` rejects new streams with "Max outbound streams is 100, 100 open" even though the pool still believes it has spare capacity. The stream is now reset locally so that both sides agree. The RST_STREAM frame this queues is not flushed here, since this runs inside a cancellation shield during teardown and taking the write lock at that point risks deadlocking against the request holding it. The frame is written out by the next call to `_write_outgoing_data()`. Closes #1022. --- CHANGELOG.md | 1 + httpcore/_async/http2.py | 22 +++++++++++++ httpcore/_sync/http2.py | 22 +++++++++++++ tests/test_cancellations.py | 66 +++++++++++++++++++++++++++++++++++++ 4 files changed, 111 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd0..2081cbc0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## Unreleased - Fix `max_keepalive_connections` not being properly handled. (#1000) +- Reset the HTTP/2 stream when a request is cancelled, keeping `h2`'s count of open streams in step with our own. (#1022) ## Version 1.0.9 (April 24th, 2025) diff --git a/httpcore/_async/http2.py b/httpcore/_async/http2.py index dbd0beeb4..d25b5bfc2 100644 --- a/httpcore/_async/http2.py +++ b/httpcore/_async/http2.py @@ -407,6 +407,28 @@ async def _receive_remote_settings_change( self._max_streams -= 1 async def _response_closed(self, stream_id: int) -> None: + # If the stream is still open as far as `h2` is concerned then we are + # tearing it down early -- typically because the task was cancelled + # before the END_STREAM frame could be written. Reset it so that `h2` + # also drops the stream from its own count of open streams. + # + # Without this, our semaphore and h2's accounting drift apart on every + # cancelled request, until h2 rejects new streams with "Max outbound + # streams" while the pool still believes there is spare capacity. + # + # The RST_STREAM frame that this queues is deliberately not flushed + # here. This runs inside a cancellation shield during teardown, and + # taking the write lock at this point risks deadlocking against the + # request that currently holds it. The frame is written out by the + # next call to `_write_outgoing_data()` instead. + h2_stream = self._h2_state.streams.get(stream_id) + if h2_stream is not None and not h2_stream.closed: + try: + self._h2_state.reset_stream(stream_id) + except h2.exceptions.ProtocolError: # pragma: nocover + # The stream may already have been reset by the remote end. + pass + await self._max_streams_semaphore.release() del self._events[stream_id] async with self._state_lock: diff --git a/httpcore/_sync/http2.py b/httpcore/_sync/http2.py index ddcc18900..5edd6917d 100644 --- a/httpcore/_sync/http2.py +++ b/httpcore/_sync/http2.py @@ -407,6 +407,28 @@ def _receive_remote_settings_change( self._max_streams -= 1 def _response_closed(self, stream_id: int) -> None: + # If the stream is still open as far as `h2` is concerned then we are + # tearing it down early -- typically because the task was cancelled + # before the END_STREAM frame could be written. Reset it so that `h2` + # also drops the stream from its own count of open streams. + # + # Without this, our semaphore and h2's accounting drift apart on every + # cancelled request, until h2 rejects new streams with "Max outbound + # streams" while the pool still believes there is spare capacity. + # + # The RST_STREAM frame that this queues is deliberately not flushed + # here. This runs inside a cancellation shield during teardown, and + # taking the write lock at this point risks deadlocking against the + # request that currently holds it. The frame is written out by the + # next call to `_write_outgoing_data()` instead. + h2_stream = self._h2_state.streams.get(stream_id) + if h2_stream is not None and not h2_stream.closed: + try: + self._h2_state.reset_stream(stream_id) + except h2.exceptions.ProtocolError: # pragma: nocover + # The stream may already have been reset by the remote end. + pass + self._max_streams_semaphore.release() del self._events[stream_id] with self._state_lock: diff --git a/tests/test_cancellations.py b/tests/test_cancellations.py index 033acef60..d146cb36d 100644 --- a/tests/test_cancellations.py +++ b/tests/test_cancellations.py @@ -243,3 +243,69 @@ async def test_h2_timeout_during_response(): assert not conn.is_closed() assert conn.is_idle() + + +class HandshakeThenBlockedWriteStream(httpcore.AsyncNetworkStream): + """ + A stream that allows the HTTP/2 handshake to complete, then blocks on + every subsequent write, signalling when it has done so. + + This lets a request be cancelled at a precise point, once its headers have + been handed to `h2` but before they have reached the network. + """ + + def __init__(self) -> None: + self._handshake_complete = False + self.blocked = anyio.Event() + + async def write( + self, buffer: bytes, timeout: typing.Optional[float] = None + ) -> None: + if not self._handshake_complete: + self._handshake_complete = True + return + self.blocked.set() + await anyio.sleep(999) + + async def read( + self, max_bytes: int, timeout: typing.Optional[float] = None + ) -> bytes: + await anyio.sleep(999) # pragma: nocover + return b"" # pragma: nocover + + async def aclose(self) -> None: + await anyio.sleep(0) + + +@pytest.mark.anyio +async def test_h2_stream_count_stays_in_sync_after_cancellation(): + """ + Cancelling an HTTP/2 request releases our own stream semaphore, so `h2` + must be told that the stream is finished as well. + + If the two drift apart then the connection eventually rejects new streams + with "Max outbound streams is 100, 100 open", while we still believe there + is spare capacity available. + + See https://github.com/encode/httpcore/issues/1022 + """ + origin = httpcore.Origin(b"http", b"example.com", 80) + stream = HandshakeThenBlockedWriteStream() + async with httpcore.AsyncHTTP2Connection(origin, stream) as conn: + # An overall ceiling, so that a regression here fails rather than + # hanging. The cancellation itself is triggered deterministically, + # once the request headers are known to have reached the stream. + with anyio.move_on_after(5.0): + async with anyio.create_task_group() as task_group: + + async def send_request() -> None: + await conn.request("GET", "http://example.com") + + task_group.start_soon(send_request) + await stream.blocked.wait() + task_group.cancel_scope.cancel() + + # We have released our own stream slot, so h2's count of open streams + # has to agree, or the two drift apart by one on every cancellation + # until no further streams can be opened at all. + assert conn._h2_state.open_outbound_streams == 0