Fix retrying interrupted HTTP/1.1 response cleanup

httpcore/_async/http11.py

@@ -344,8 +344,13 @@ async def __aiter__(self) -> typing.AsyncIterator[bytes]:
     async def aclose(self) -> None:
         if not self._closed:
             self._closed = True
-            async with Trace("response_closed", logger, self._request):
-                await self._connection._response_closed()
+            try:
+                async with Trace("response_closed", logger, self._request):
+                    await self._connection._response_closed()
+            except BaseException:
+                # Allow the caller to retry cleanup if it was interrupted.
+                self._closed = False
+                raise
 
 
 class AsyncHTTP11UpgradeStream(AsyncNetworkStream):

httpcore/_sync/http11.py

@@ -344,8 +344,13 @@ def __iter__(self) -> typing.Iterator[bytes]:
     def close(self) -> None:
         if not self._closed:
             self._closed = True
-            with Trace("response_closed", logger, self._request):
-                self._connection._response_closed()
+            try:
+                with Trace("response_closed", logger, self._request):
+                    self._connection._response_closed()
+            except BaseException:
+                # Allow the caller to retry cleanup if it was interrupted.
+                self._closed = False
+                raise
 
 
 class HTTP11UpgradeStream(NetworkStream):

pyproject.toml

@@ -56,13 +56,17 @@ Source = "https://github.com/encode/httpcore"
 path = "httpcore/__init__.py"
 
 [tool.hatch.build.targets.sdist]
+core-metadata-version = "2.4"
 include = [
     "/httpcore",
     "/CHANGELOG.md",
     "/README.md",
     "/tests"
 ]
 
+[tool.hatch.build.targets.wheel]
+core-metadata-version = "2.4"
+
 [tool.hatch.metadata.hooks.fancy-pypi-readme]
 content-type = "text/markdown"
 

tests/_async/test_http11.py

@@ -378,3 +378,39 @@ async def test_http11_header_sub_100kb():
         response = await conn.request("GET", "https://example.com/")
         assert response.status == 200
         assert response.content == b""
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("read_body", [False, True])
+async def test_http11_retry_response_close(read_body):
+    """Failed cleanup can be retried; successful cleanup remains idempotent."""
+    close_attempts = 0
+
+    async def trace(name, info):
+        nonlocal close_attempts
+        if name == "http11.response_closed.started":
+            close_attempts += 1
+            if close_attempts == 1:
+                raise RuntimeError("Interrupted cleanup")
+
+    origin = httpcore.Origin(b"http", b"example.com", 80)
+    stream = httpcore.AsyncMockStream(
+        [b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nx"]
+    )
+    async with httpcore.AsyncHTTP11Connection(origin, stream) as connection:
+        request = httpcore.Request(
+            "GET",
+            "http://example.com",
+            headers={"Host": "example.com"},
+            extensions={"trace": trace},
+        )
+        response = await connection.handle_async_request(request)
+        if read_body:
+            assert await response.aread() == b"x"
+        with pytest.raises(RuntimeError, match="Interrupted cleanup"):
+            await response.aclose()
+        await response.aclose()
+        await response.aclose()
+        assert close_attempts == 2
+        assert connection.is_idle() == read_body
+        assert connection.is_closed() == (not read_body)

tests/_sync/test_http11.py

@@ -378,3 +378,39 @@ def test_http11_header_sub_100kb():
         response = conn.request("GET", "https://example.com/")
         assert response.status == 200
         assert response.content == b""
+
+
+
+@pytest.mark.parametrize("read_body", [False, True])
+def test_http11_retry_response_close(read_body):
+    """Failed cleanup can be retried; successful cleanup remains idempotent."""
+    close_attempts = 0
+
+    def trace(name, info):
+        nonlocal close_attempts
+        if name == "http11.response_closed.started":
+            close_attempts += 1
+            if close_attempts == 1:
+                raise RuntimeError("Interrupted cleanup")
+
+    origin = httpcore.Origin(b"http", b"example.com", 80)
+    stream = httpcore.MockStream(
+        [b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nx"]
+    )
+    with httpcore.HTTP11Connection(origin, stream) as connection:
+        request = httpcore.Request(
+            "GET",
+            "http://example.com",
+            headers={"Host": "example.com"},
+            extensions={"trace": trace},
+        )
+        response = connection.handle_request(request)
+        if read_body:
+            assert response.read() == b"x"
+        with pytest.raises(RuntimeError, match="Interrupted cleanup"):
+            response.close()
+        response.close()
+        response.close()
+        assert close_attempts == 2
+        assert connection.is_idle() == read_body
+        assert connection.is_closed() == (not read_body)

tests/test_cancellations.py

@@ -243,3 +243,89 @@ async def test_h2_timeout_during_response():
 
         assert not conn.is_closed()
         assert conn.is_idle()
+
+
+@pytest.mark.parametrize("cancel_during", ["state_lock", "network_close"])
+def test_http11_double_cancel_releases_connection(cancel_during):
+    """A second cancellation must not prevent the pool from retrying cleanup."""
+    import asyncio
+
+    async def run() -> None:
+        reading = asyncio.Event()
+        closing = asyncio.Event()
+        never = asyncio.Event()
+
+        class Stream(httpcore.AsyncMockStream):
+            async def read(self, max_bytes, timeout=None):
+                if not self._buffer:
+                    reading.set()
+                    await never.wait()
+                return await super().read(max_bytes, timeout)
+
+            async def aclose(self):
+                if cancel_during == "network_close" and not closing.is_set():
+                    closing.set()
+                    await never.wait()
+                await super().aclose()
+
+        streams = []
+
+        class Backend(httpcore.AsyncMockBackend):
+            async def connect_tcp(self, *args, **kwargs):
+                stream = Stream([b"HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nx"])
+                streams.append(stream)
+                return stream
+
+        async def trace(name, info):
+            if (
+                cancel_during == "state_lock"
+                and name == "http11.response_closed.started"
+            ):
+                closing.set()
+
+        async with httpcore.AsyncConnectionPool(
+            network_backend=Backend([]), max_connections=1
+        ) as pool:
+
+            async def consume() -> None:
+                async with pool.stream(
+                    "GET", "http://example.com", extensions={"trace": trace}
+                ) as response:
+                    async for _ in response.aiter_stream():
+                        pass
+
+            task = asyncio.create_task(consume())
+            await asyncio.wait_for(reading.wait(), 1)
+            pooled_connection = pool.connections[0]
+            assert isinstance(pooled_connection, httpcore.AsyncHTTPConnection)
+            connection = pooled_connection._connection
+            assert isinstance(connection, httpcore.AsyncHTTP11Connection)
+            lock = connection._state_lock
+            if cancel_during == "state_lock":
+                await lock.__aenter__()
+            try:
+                task.cancel()
+                await asyncio.wait_for(closing.wait(), 1)
+                task.cancel()
+                # Deliver cancellation while cleanup is still blocked.
+                await asyncio.sleep(0)
+            finally:
+                if cancel_during == "state_lock":
+                    await lock.__aexit__(None, None, None)
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, 1)
+
+            assert connection.is_closed()
+            assert streams[0]._closed
+            assert not pool.connections
+            assert not pool._requests
+
+            # A one-slot pool must still be able to service another request.
+            async with pool.stream(
+                "GET", "http://example.com", extensions={"timeout": {"pool": 0.1}}
+            ) as response:
+                assert response.status == 200
+            assert not pool.connections
+            assert streams[1]._closed
+
+    asyncio.run(run())