diff --git a/httpcore/_async/http11.py b/httpcore/_async/http11.py index e6d6d7098..4c2d03517 100644 --- a/httpcore/_async/http11.py +++ b/httpcore/_async/http11.py @@ -154,9 +154,20 @@ async def _send_request_body(self, request: Request) -> None: timeout = timeouts.get("write", None) assert isinstance(request.stream, typing.AsyncIterable) - async for chunk in request.stream: - event = h11.Data(data=chunk) - await self._send_event(event, timeout=timeout) + body = request.stream.__aiter__() + try: + async for chunk in body: + event = h11.Data(data=chunk) + await self._send_event(event, timeout=timeout) + finally: + # If sending the body fails part-way through, for example with + # a `WriteError`, the async iterator would otherwise be abandoned + # mid-iteration, and then garbage collected without ever being + # exhausted, triggering `ResourceWarning`. + # See https://github.com/encode/httpx/issues/3597 + aclose = getattr(body, "aclose", None) + if aclose is not None: + await aclose() await self._send_event(h11.EndOfMessage(), timeout=timeout) diff --git a/httpcore/_sync/http11.py b/httpcore/_sync/http11.py index ebd3a9748..b8758fa65 100644 --- a/httpcore/_sync/http11.py +++ b/httpcore/_sync/http11.py @@ -154,9 +154,20 @@ def _send_request_body(self, request: Request) -> None: timeout = timeouts.get("write", None) assert isinstance(request.stream, typing.Iterable) - for chunk in request.stream: - event = h11.Data(data=chunk) - self._send_event(event, timeout=timeout) + body = request.stream.__iter__() + try: + for chunk in body: + event = h11.Data(data=chunk) + self._send_event(event, timeout=timeout) + finally: + # If sending the body fails part-way through, for example with + # a `WriteError`, the async iterator would otherwise be abandoned + # mid-iteration, and then garbage collected without ever being + # exhausted, triggering `ResourceWarning`. + # See https://github.com/encode/httpx/issues/3597 + close = getattr(body, "close", None) + if close is not None: + close() self._send_event(h11.EndOfMessage(), timeout=timeout) diff --git a/tests/_async/test_http11.py b/tests/_async/test_http11.py index 94f2febf0..15d565532 100644 --- a/tests/_async/test_http11.py +++ b/tests/_async/test_http11.py @@ -1,3 +1,5 @@ +import typing + import pytest import httpcore @@ -378,3 +380,60 @@ 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 +async def test_http11_write_error_closes_request_body(): + """ + If a `WriteError` occurs part-way through sending the request body, + the request body's async iterator must be closed rather than abandoned + mid-iteration. Otherwise it is garbage collected without ever being + exhausted, triggering `ResourceWarning`. + See https://github.com/encode/httpx/issues/3597. + """ + + class WriteErrorStream(httpcore.AsyncMockStream): + def __init__(self, buffer: typing.List[bytes], fail_after_writes: int) -> None: + super().__init__(buffer) + self._writes = 0 + self._fail_after_writes = fail_after_writes + + async def write( + self, buffer: bytes, timeout: typing.Optional[float] = None + ) -> None: + self._writes += 1 + if self._writes > self._fail_after_writes: + raise httpcore.WriteError("Simulated write failure") + await super().write(buffer, timeout) + + body_closed = False + + async def streaming_body() -> typing.AsyncIterator[bytes]: + nonlocal body_closed + try: + for index in range(10): + yield b"chunk-%d" % index + except GeneratorExit: + body_closed = True + raise + + origin = httpcore.Origin(b"https", b"example.com", 443) + # Two writes succeed: the request headers, then the first body chunk. + # The write of the second body chunk raises `WriteError`. + stream = WriteErrorStream( + [ + b"HTTP/1.1 200 OK\r\n", + b"Content-Length: 0\r\n", + b"\r\n", + ], + fail_after_writes=2, + ) + async with httpcore.AsyncHTTP11Connection(origin=origin, stream=stream) as conn: + response = await conn.request( + "POST", "https://example.com/", content=streaming_body() + ) + # The `WriteError` is suppressed, and the response is still readable. + assert response.status == 200 + assert response.content == b"" + + assert body_closed diff --git a/tests/_sync/test_http11.py b/tests/_sync/test_http11.py index f2fa28f4c..1f3546347 100644 --- a/tests/_sync/test_http11.py +++ b/tests/_sync/test_http11.py @@ -1,3 +1,5 @@ +import typing + import pytest import httpcore @@ -378,3 +380,60 @@ def test_http11_header_sub_100kb(): response = conn.request("GET", "https://example.com/") assert response.status == 200 assert response.content == b"" + + + +def test_http11_write_error_closes_request_body(): + """ + If a `WriteError` occurs part-way through sending the request body, + the request body's async iterator must be closed rather than abandoned + mid-iteration. Otherwise it is garbage collected without ever being + exhausted, triggering `ResourceWarning`. + See https://github.com/encode/httpx/issues/3597. + """ + + class WriteErrorStream(httpcore.MockStream): + def __init__(self, buffer: typing.List[bytes], fail_after_writes: int) -> None: + super().__init__(buffer) + self._writes = 0 + self._fail_after_writes = fail_after_writes + + def write( + self, buffer: bytes, timeout: typing.Optional[float] = None + ) -> None: + self._writes += 1 + if self._writes > self._fail_after_writes: + raise httpcore.WriteError("Simulated write failure") + super().write(buffer, timeout) + + body_closed = False + + def streaming_body() -> typing.Iterator[bytes]: + nonlocal body_closed + try: + for index in range(10): + yield b"chunk-%d" % index + except GeneratorExit: + body_closed = True + raise + + origin = httpcore.Origin(b"https", b"example.com", 443) + # Two writes succeed: the request headers, then the first body chunk. + # The write of the second body chunk raises `WriteError`. + stream = WriteErrorStream( + [ + b"HTTP/1.1 200 OK\r\n", + b"Content-Length: 0\r\n", + b"\r\n", + ], + fail_after_writes=2, + ) + with httpcore.HTTP11Connection(origin=origin, stream=stream) as conn: + response = conn.request( + "POST", "https://example.com/", content=streaming_body() + ) + # The `WriteError` is suppressed, and the response is still readable. + assert response.status == 200 + assert response.content == b"" + + assert body_closed