Reset the HTTP/2 stream when a request is cancelled

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)
 

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:

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:

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