diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd..2081cbc0 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 dbd0beeb..d25b5bfc 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 ddcc1890..5edd6917 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 033acef6..d146cb36 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