From bfaa5bbba6351f1131c238003416558629619421 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:08:43 +0000
Subject: [PATCH 1/6] Add a native asyncio network backend

Under asyncio, connections now use a backend built directly on the event loop's transports and protocols instead of anyio streams. Each read and write no longer sets up a cancel scope with a loop timer and an exception-mapping context manager: a read returns buffered data immediately when there is some and otherwise waits on a plain future with an optional timer, and a write hands the data to the transport and only waits when the transport applies backpressure. The protocol keeps reading while a connection is idle, bounded by a high-water mark, so a server close is known without probing the socket. anyio remains in use for trio and for the synchronization primitives, and AnyIOBackend stays available for explicit use.

The integration tests now run against both the automatic backend and the explicit anyio backend, and a dedicated test module covers the asyncio backend's timeouts, end-of-stream and connection-loss handling, backpressure, TLS and unix sockets.
---
 scripts/unasync.py                           |   1 +
 src/httpcore2/httpcore2/__init__.py          |   2 +
 src/httpcore2/httpcore2/_backends/asyncio.py | 257 +++++++++++++++
 src/httpcore2/httpcore2/_backends/auto.py    |   4 +-
 tests/httpcore2/_async/test_integration.py   |  17 +-
 tests/httpcore2/_sync/test_integration.py    |  17 +-
 tests/httpcore2/test_asyncio_backend.py      | 311 +++++++++++++++++++
 7 files changed, 595 insertions(+), 14 deletions(-)
 create mode 100644 src/httpcore2/httpcore2/_backends/asyncio.py
 create mode 100644 tests/httpcore2/test_asyncio_backend.py

diff --git a/scripts/unasync.py b/scripts/unasync.py
index 908662b0..eb6c9762 100644
--- a/scripts/unasync.py
+++ b/scripts/unasync.py
@@ -11,6 +11,7 @@
     ),
     ("import trio as concurrency", "from tests.httpcore2 import concurrency"),
     ("anyio.sleep", "concurrency.sleep"),
+    ("AnyIOBackend", "SyncBackend"),
     ("AsyncIterator", "Iterator"),
     ("Async([A-Z][A-Za-z0-9_]*)", r"\2"),
     ("async def", "def"),
diff --git a/src/httpcore2/httpcore2/__init__.py b/src/httpcore2/httpcore2/__init__.py
index 708eea9a..0f279f9b 100644
--- a/src/httpcore2/httpcore2/__init__.py
+++ b/src/httpcore2/httpcore2/__init__.py
@@ -10,6 +10,7 @@
     AsyncHTTPProxy,
     AsyncSOCKSProxy,
 )
+from ._backends.asyncio import AsyncioBackend
 from ._backends.base import (
     SOCKET_OPTION,
     AsyncNetworkBackend,
@@ -99,6 +100,7 @@ def __init__(self, *args, **kwargs):  # type: ignore
     # network backends, implementations
     "SyncBackend",
     "AnyIOBackend",
+    "AsyncioBackend",
     "TrioBackend",
     # network backends, mock implementations
     "AsyncMockBackend",
diff --git a/src/httpcore2/httpcore2/_backends/asyncio.py b/src/httpcore2/httpcore2/_backends/asyncio.py
new file mode 100644
index 00000000..7e55be92
--- /dev/null
+++ b/src/httpcore2/httpcore2/_backends/asyncio.py
@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+import asyncio
+import collections
+import ssl
+import typing
+
+from .._exceptions import (
+    ConnectError,
+    ConnectTimeout,
+    ReadError,
+    ReadTimeout,
+    WriteError,
+    WriteTimeout,
+)
+from .base import SOCKET_OPTION, AsyncNetworkBackend, AsyncNetworkStream
+
+# Stop reading from the socket once this much data is buffered but unread,
+# and start again once the buffer drains below the low-water mark.
+RECEIVE_HIGH_WATER = 256 * 1024
+RECEIVE_LOW_WATER = 64 * 1024
+
+
+class _Timeout(Exception):
+    """
+    Raised on a waiter future when its deadline passes.
+    """
+
+
+def _timeout(waiter: asyncio.Future[None]) -> None:
+    if not waiter.done():
+        waiter.set_exception(_Timeout())
+
+
+def _wake(waiter: asyncio.Future[None] | None) -> None:
+    if waiter is not None and not waiter.done():
+        waiter.set_result(None)
+
+
+class AsyncioStreamProtocol(asyncio.Protocol):
+    """
+    Buffers received data for `AsyncioStream`, applying backpressure to the
+    transport once too much is buffered, and wakes up pending reads and writes.
+    """
+
+    def __init__(self) -> None:
+        self.transport: asyncio.Transport | None = None
+        self.chunks: collections.deque[bytes] = collections.deque()
+        self.buffered = 0
+        self.reading_paused = False
+        self.writing_paused = False
+        self.eof = False
+        self.closed = False
+        self.exception: Exception | None = None
+        self.read_waiter: asyncio.Future[None] | None = None
+        self.write_waiter: asyncio.Future[None] | None = None
+
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
+        # The transport implements the interface without necessarily subclassing it.
+        self.transport = typing.cast(asyncio.Transport, transport)
+
+    def data_received(self, data: bytes) -> None:
+        self.chunks.append(data)
+        self.buffered += len(data)
+        if self.buffered >= RECEIVE_HIGH_WATER and not self.reading_paused:
+            assert self.transport is not None
+            self.transport.pause_reading()
+            self.reading_paused = True
+        _wake(self.read_waiter)
+
+    def eof_received(self) -> bool:
+        self.eof = True
+        _wake(self.read_waiter)
+        # Let the transport close: an HTTP peer that has sent a FIN is done.
+        return False
+
+    def connection_lost(self, exc: Exception | None) -> None:
+        self.closed = True
+        self.exception = exc
+        _wake(self.read_waiter)
+        _wake(self.write_waiter)
+
+    def pause_writing(self) -> None:
+        self.writing_paused = True
+
+    def resume_writing(self) -> None:
+        self.writing_paused = False
+        _wake(self.write_waiter)
+
+
+class AsyncioStream(AsyncNetworkStream):
+    def __init__(self, transport: asyncio.Transport, protocol: AsyncioStreamProtocol) -> None:
+        self._transport = transport
+        self._protocol = protocol
+
+    async def read(self, max_bytes: int, timeout: float | None = None) -> bytes:
+        protocol = self._protocol
+        if not protocol.chunks:
+            if protocol.eof or protocol.closed:
+                return self._read_at_end()
+            try:
+                await self._wait("read", timeout)
+            except _Timeout:
+                raise ReadTimeout("timed out") from None
+            if not protocol.chunks:
+                return self._read_at_end()
+
+        chunk = protocol.chunks[0]
+        if len(chunk) <= max_bytes:
+            protocol.chunks.popleft()
+        else:
+            protocol.chunks[0] = chunk[max_bytes:]
+            chunk = chunk[:max_bytes]
+        protocol.buffered -= len(chunk)
+        if protocol.reading_paused and protocol.buffered <= RECEIVE_LOW_WATER:
+            protocol.reading_paused = False
+            self._transport.resume_reading()
+        return chunk
+
+    def _read_at_end(self) -> bytes:
+        # No buffered data and no more coming: a clean EOF reads as empty,
+        # a connection dropped by an error is a read error.
+        if self._protocol.exception is not None:
+            raise ReadError(str(self._protocol.exception)) from self._protocol.exception
+        return b""
+
+    async def write(self, buffer: bytes, timeout: float | None = None) -> None:
+        if not buffer:
+            return
+        protocol = self._protocol
+        if protocol.closed or self._transport.is_closing():
+            raise WriteError("Connection closed")
+        self._transport.write(buffer)
+        if protocol.writing_paused:
+            # The transport's send buffer is full; wait for it to drain.
+            try:
+                await self._wait("write", timeout)
+            except _Timeout:
+                raise WriteTimeout("timed out") from None
+            if protocol.closed:
+                raise WriteError(str(protocol.exception or "Connection closed"))
+
+    async def _wait(self, kind: str, timeout: float | None) -> None:
+        loop = asyncio.get_running_loop()
+        waiter: asyncio.Future[None] = loop.create_future()
+        protocol = self._protocol
+        if kind == "read":
+            protocol.read_waiter = waiter
+        else:
+            protocol.write_waiter = waiter
+        handle = None if timeout is None else loop.call_later(timeout, _timeout, waiter)
+        try:
+            await waiter
+        finally:
+            if handle is not None:
+                handle.cancel()
+            if kind == "read":
+                protocol.read_waiter = None
+            else:
+                protocol.write_waiter = None
+
+    async def aclose(self) -> None:
+        if self._protocol.closed:
+            return
+        self._transport.close()
+        # Closing only schedules the socket close on the event loop. Yield once
+        # so it runs now, then force it if unsent data is still holding it up.
+        await asyncio.sleep(0)
+        if not self._protocol.closed:
+            self._transport.abort()
+
+    async def start_tls(
+        self,
+        ssl_context: ssl.SSLContext,
+        server_hostname: str | None = None,
+        timeout: float | None = None,
+    ) -> AsyncNetworkStream:
+        loop = asyncio.get_running_loop()
+        # The loop reports its own handshake timeout as a connection error,
+        # so the deadline is applied here instead to raise a timeout.
+        handshake = loop.start_tls(self._transport, self._protocol, ssl_context, server_hostname=server_hostname)
+        try:
+            transport = await asyncio.wait_for(handshake, timeout)
+        except (TimeoutError, asyncio.TimeoutError):
+            self._transport.close()
+            raise ConnectTimeout("timed out") from None
+        except (OSError, ssl.SSLError) as exc:
+            self._transport.close()
+            raise ConnectError(str(exc)) from exc
+        if transport is None:  # pragma: no cover
+            raise ConnectError("TLS handshake failed")
+        self._protocol.transport = transport
+        return AsyncioStream(transport, self._protocol)
+
+    def get_extra_info(self, info: str) -> typing.Any:
+        if info == "ssl_object":
+            return self._transport.get_extra_info("ssl_object")
+        if info == "client_addr":
+            return self._transport.get_extra_info("sockname")
+        if info == "server_addr":
+            return self._transport.get_extra_info("peername")
+        if info == "socket":
+            return self._transport.get_extra_info("socket")
+        if info == "is_readable":
+            # The event loop keeps reading while the connection is idle, so a
+            # FIN or stray data from the server is already known here without
+            # touching the socket.
+            protocol = self._protocol
+            return bool(protocol.chunks) or protocol.eof or protocol.closed
+        return None
+
+
+class AsyncioBackend(AsyncNetworkBackend):
+    async def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+    ) -> AsyncNetworkStream:
+        loop = asyncio.get_running_loop()
+        local_addr = None if local_address is None else (local_address, 0)
+        # By default TCP sockets opened in `asyncio` include TCP_NODELAY.
+        connect = loop.create_connection(AsyncioStreamProtocol, host, port, local_addr=local_addr)
+        return await self._connect(connect, timeout, socket_options)
+
+    async def connect_unix_socket(
+        self,
+        path: str,
+        timeout: float | None = None,
+        socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
+    ) -> AsyncNetworkStream:
+        loop = asyncio.get_running_loop()
+        connect = loop.create_unix_connection(AsyncioStreamProtocol, path)
+        return await self._connect(connect, timeout, socket_options)
+
+    async def _connect(
+        self,
+        connect: typing.Coroutine[typing.Any, typing.Any, tuple[asyncio.BaseTransport, AsyncioStreamProtocol]],
+        timeout: float | None,
+        socket_options: typing.Iterable[SOCKET_OPTION] | None,
+    ) -> AsyncioStream:
+        try:
+            transport, protocol = await asyncio.wait_for(connect, timeout)
+        except (TimeoutError, asyncio.TimeoutError):
+            raise ConnectTimeout("timed out") from None
+        except OSError as exc:
+            raise ConnectError(str(exc)) from exc
+        if socket_options:
+            sock = transport.get_extra_info("socket")
+            for option in socket_options:
+                sock.setsockopt(*option)
+        return AsyncioStream(typing.cast(asyncio.Transport, transport), protocol)
+
+    async def sleep(self, seconds: float) -> None:
+        await asyncio.sleep(seconds)
diff --git a/src/httpcore2/httpcore2/_backends/auto.py b/src/httpcore2/httpcore2/_backends/auto.py
index c946c9e6..ea92817a 100644
--- a/src/httpcore2/httpcore2/_backends/auto.py
+++ b/src/httpcore2/httpcore2/_backends/auto.py
@@ -15,9 +15,9 @@ async def _init_backend(self) -> None:
 
                 self._backend: AsyncNetworkBackend = TrioBackend()
             else:
-                from .anyio import AnyIOBackend
+                from .asyncio import AsyncioBackend
 
-                self._backend = AnyIOBackend()
+                self._backend = AsyncioBackend()
 
     async def connect_tcp(
         self,
diff --git a/tests/httpcore2/_async/test_integration.py b/tests/httpcore2/_async/test_integration.py
index 1325c734..527d6c0e 100644
--- a/tests/httpcore2/_async/test_integration.py
+++ b/tests/httpcore2/_async/test_integration.py
@@ -5,30 +5,35 @@
 
 import httpcore2
 
+BACKENDS = [None, httpcore2.AnyIOBackend()]
+
 
 @pytest.mark.anyio
-async def test_request(httpbin: Server) -> None:
-    async with httpcore2.AsyncConnectionPool() as pool:
+@pytest.mark.parametrize("network_backend", BACKENDS)
+async def test_request(httpbin: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None:
+    async with httpcore2.AsyncConnectionPool(network_backend=network_backend) as pool:
         response = await pool.request("GET", httpbin.url)
         assert response.status == 200
 
 
 @pytest.mark.anyio
-async def test_ssl_request(httpbin_secure: Server) -> None:
+@pytest.mark.parametrize("network_backend", BACKENDS)
+async def test_ssl_request(httpbin_secure: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None:
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
-    async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context) as pool:
+    async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool:
         response = await pool.request("GET", httpbin_secure.url)
         assert response.status == 200
 
 
 @pytest.mark.anyio
-async def test_extra_info(httpbin_secure: Server) -> None:
+@pytest.mark.parametrize("network_backend", BACKENDS)
+async def test_extra_info(httpbin_secure: Server, network_backend: httpcore2.AsyncNetworkBackend | None) -> None:
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
-    async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context) as pool:
+    async with httpcore2.AsyncConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool:
         async with pool.stream("GET", httpbin_secure.url) as response:
             assert response.status == 200
             stream = response.extensions["network_stream"]
diff --git a/tests/httpcore2/_sync/test_integration.py b/tests/httpcore2/_sync/test_integration.py
index 06f1aafb..5bac9e2b 100644
--- a/tests/httpcore2/_sync/test_integration.py
+++ b/tests/httpcore2/_sync/test_integration.py
@@ -5,30 +5,35 @@
 
 import httpcore2
 
+BACKENDS = [None, httpcore2.SyncBackend()]
 
 
-def test_request(httpbin: Server) -> None:
-    with httpcore2.ConnectionPool() as pool:
+
+@pytest.mark.parametrize("network_backend", BACKENDS)
+def test_request(httpbin: Server, network_backend: httpcore2.NetworkBackend | None) -> None:
+    with httpcore2.ConnectionPool(network_backend=network_backend) as pool:
         response = pool.request("GET", httpbin.url)
         assert response.status == 200
 
 
 
-def test_ssl_request(httpbin_secure: Server) -> None:
+@pytest.mark.parametrize("network_backend", BACKENDS)
+def test_ssl_request(httpbin_secure: Server, network_backend: httpcore2.NetworkBackend | None) -> None:
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
-    with httpcore2.ConnectionPool(ssl_context=ssl_context) as pool:
+    with httpcore2.ConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool:
         response = pool.request("GET", httpbin_secure.url)
         assert response.status == 200
 
 
 
-def test_extra_info(httpbin_secure: Server) -> None:
+@pytest.mark.parametrize("network_backend", BACKENDS)
+def test_extra_info(httpbin_secure: Server, network_backend: httpcore2.NetworkBackend | None) -> None:
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
-    with httpcore2.ConnectionPool(ssl_context=ssl_context) as pool:
+    with httpcore2.ConnectionPool(ssl_context=ssl_context, network_backend=network_backend) as pool:
         with pool.stream("GET", httpbin_secure.url) as response:
             assert response.status == 200
             stream = response.extensions["network_stream"]
diff --git a/tests/httpcore2/test_asyncio_backend.py b/tests/httpcore2/test_asyncio_backend.py
new file mode 100644
index 00000000..14ba2bde
--- /dev/null
+++ b/tests/httpcore2/test_asyncio_backend.py
@@ -0,0 +1,311 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import socket
+import ssl
+import sys
+import typing
+from collections.abc import AsyncIterator, Awaitable, Callable
+
+import pytest
+import trustme
+
+import httpcore2
+from httpcore2._backends.asyncio import (
+    RECEIVE_HIGH_WATER,
+    RECEIVE_LOW_WATER,
+    AsyncioStream,
+    AsyncioStreamProtocol,
+)
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+Handler = Callable[[asyncio.StreamReader, asyncio.StreamWriter], Awaitable[None]]
+
+
+async def echo(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    # The client may drop the connection mid-way, for example when a TLS handshake fails.
+    with contextlib.suppress(ConnectionError, ssl.SSLError):
+        while data := await reader.read(65536):
+            writer.write(data)
+            await writer.drain()
+    writer.close()
+
+
+async def silent(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    # Hold the connection open, without answering, until the client goes away.
+    await reader.read()
+    writer.close()
+
+
+@pytest.fixture
+async def serve() -> AsyncIterator[Callable[..., Awaitable[int]]]:
+    servers: list[asyncio.base_events.Server] = []
+
+    async def start(handler: Handler = echo, **kwargs: typing.Any) -> int:
+        server = await asyncio.start_server(handler, "127.0.0.1", 0, **kwargs)
+        servers.append(server)
+        port: int = server.sockets[0].getsockname()[1]
+        return port
+
+    yield start
+    for server in servers:
+        server.close()
+        await server.wait_closed()
+
+
+async def test_tcp_roundtrip_and_extra_info(serve: Callable[..., Awaitable[int]]) -> None:
+    port = await serve()
+    backend = httpcore2.AsyncioBackend()
+    stream = await backend.connect_tcp("127.0.0.1", port, socket_options=[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)])
+    assert stream.get_extra_info("socket").getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) == 1
+    assert stream.get_extra_info("client_addr")[0] == "127.0.0.1"
+    assert stream.get_extra_info("server_addr") == ("127.0.0.1", port)
+    assert stream.get_extra_info("ssl_object") is None
+    assert stream.get_extra_info("invalid") is None
+    assert not stream.get_extra_info("is_readable")
+
+    await stream.write(b"hello")
+    assert await stream.read(1024) == b"hello"
+    assert not stream.get_extra_info("is_readable")
+
+    await stream.aclose()
+    assert stream.get_extra_info("is_readable")
+
+
+async def test_read_at_eof(serve: Callable[..., Awaitable[int]]) -> None:
+    async def close_after_hello(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        writer.write(b"hello")
+        await writer.drain()
+        writer.close()
+        await writer.wait_closed()
+
+    port = await serve(close_after_hello)
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    assert await stream.read(1024) == b"hello"
+    assert await stream.read(1024) == b""
+    assert await stream.read(1024) == b""
+    await stream.aclose()
+
+
+async def test_read_timeout(serve: Callable[..., Awaitable[int]]) -> None:
+    port = await serve(silent)
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    with pytest.raises(httpcore2.ReadTimeout):
+        await stream.read(1024, timeout=0.05)
+    await stream.aclose()
+
+
+async def test_connect_refused() -> None:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+    with pytest.raises(httpcore2.ConnectError):
+        await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+
+
+async def test_connect_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def never_connects(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
+        await asyncio.sleep(10)
+
+    monkeypatch.setattr(asyncio.get_running_loop(), "create_connection", never_connects)
+    with pytest.raises(httpcore2.ConnectTimeout):
+        await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", 1, timeout=0.05)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="Unix sockets are not available on Windows.")
+async def test_unix_socket(tmp_path: typing.Any) -> None:
+    path = str(tmp_path / "socket")
+    server = await asyncio.start_unix_server(echo, path)
+    try:
+        stream = await httpcore2.AsyncioBackend().connect_unix_socket(path)
+        await stream.write(b"hello")
+        assert await stream.read(1024) == b"hello"
+        await stream.aclose()
+    finally:
+        server.close()
+        await server.wait_closed()
+
+    with pytest.raises(httpcore2.ConnectError):
+        await httpcore2.AsyncioBackend().connect_unix_socket(str(tmp_path / "missing"))
+
+
+async def test_start_tls(serve: Callable[..., Awaitable[int]]) -> None:
+    ca = trustme.CA()
+    server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    ca.issue_cert("localhost").configure_cert(server_context)
+    client_context = ssl.create_default_context()
+    ca.configure_trust(client_context)
+
+    port = await serve(ssl=server_context)
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    tls_stream = await stream.start_tls(client_context, server_hostname="localhost", timeout=5.0)
+    assert tls_stream.get_extra_info("ssl_object").version() == "TLSv1.3"
+    await tls_stream.write(b"hello")
+    assert await tls_stream.read(1024) == b"hello"
+    await tls_stream.aclose()
+
+
+async def test_start_tls_failure(serve: Callable[..., Awaitable[int]]) -> None:
+    # A plain echo server answers the ClientHello with the ClientHello.
+    port = await serve()
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    with pytest.raises(httpcore2.ConnectError):
+        await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=5.0)
+
+
+async def test_start_tls_timeout(serve: Callable[..., Awaitable[int]]) -> None:
+    port = await serve(silent)
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    with pytest.raises(httpcore2.ConnectTimeout):
+        await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=0.05)
+
+
+async def test_sleep() -> None:
+    await httpcore2.AsyncioBackend().sleep(0)
+
+
+# The remaining behaviour is driven through a fake transport, so the timing
+# of the real network is not involved.
+
+
+class FakeTransport(asyncio.Transport):
+    def __init__(self) -> None:
+        super().__init__()
+        self.written: list[bytes] = []
+        self.reading_paused = False
+        self.closed = False
+
+    def write(self, data: typing.Any) -> None:
+        self.written.append(bytes(data))
+
+    def is_closing(self) -> bool:
+        return self.closed
+
+    def close(self) -> None:
+        self.closed = True
+
+    def abort(self) -> None:
+        self.closed = True
+
+    def pause_reading(self) -> None:
+        self.reading_paused = True
+
+    def resume_reading(self) -> None:
+        self.reading_paused = False
+
+
+def fake_stream() -> tuple[AsyncioStream, AsyncioStreamProtocol, FakeTransport]:
+    transport = FakeTransport()
+    protocol = AsyncioStreamProtocol()
+    protocol.connection_made(transport)
+    return AsyncioStream(transport, protocol), protocol, transport
+
+
+async def test_read_splits_large_chunks() -> None:
+    stream, protocol, _ = fake_stream()
+    protocol.data_received(b"abcdef")
+    assert await stream.read(4) == b"abcd"
+    assert await stream.read(4) == b"ef"
+    assert protocol.buffered == 0
+
+
+async def test_backpressure_pauses_and_resumes_reading() -> None:
+    stream, protocol, transport = fake_stream()
+    protocol.data_received(b"x" * RECEIVE_HIGH_WATER)
+    assert transport.reading_paused
+    protocol.data_received(b"y")
+    assert transport.reading_paused
+
+    await stream.read(RECEIVE_HIGH_WATER - RECEIVE_LOW_WATER - 1)
+    assert protocol.buffered == RECEIVE_LOW_WATER + 2
+    assert transport.reading_paused
+    await stream.read(2)
+    assert not transport.reading_paused
+
+
+async def test_pending_read_wakes_on_eof() -> None:
+    stream, protocol, _ = fake_stream()
+    read = asyncio.ensure_future(stream.read(1024))
+    await asyncio.sleep(0)
+    assert protocol.eof_received() is False
+    assert await read == b""
+
+
+async def test_pending_read_wakes_on_connection_lost_with_error() -> None:
+    stream, protocol, _ = fake_stream()
+    read = asyncio.ensure_future(stream.read(1024))
+    await asyncio.sleep(0)
+    protocol.connection_lost(OSError("reset"))
+    with pytest.raises(httpcore2.ReadError):
+        await read
+    with pytest.raises(httpcore2.ReadError):
+        await stream.read(1024)
+
+
+async def test_connection_lost_without_error_reads_as_eof() -> None:
+    stream, protocol, _ = fake_stream()
+    protocol.data_received(b"tail")
+    protocol.connection_lost(None)
+    assert await stream.read(1024) == b"tail"
+    assert await stream.read(1024) == b""
+    assert stream.get_extra_info("is_readable")
+
+
+async def test_write() -> None:
+    stream, protocol, transport = fake_stream()
+    await stream.write(b"")
+    await stream.write(b"hello")
+    assert transport.written == [b"hello"]
+
+    transport.close()
+    with pytest.raises(httpcore2.WriteError):
+        await stream.write(b"more")
+
+    protocol.connection_lost(None)
+    with pytest.raises(httpcore2.WriteError):
+        await stream.write(b"more")
+
+
+async def test_aclose_forces_a_stuck_close() -> None:
+    stream, protocol, transport = fake_stream()
+    # The fake transport never reports the connection as lost by itself.
+    await stream.aclose()
+    assert transport.closed
+    protocol.connection_lost(None)
+    await stream.aclose()
+
+
+async def test_write_waits_for_the_transport_to_drain() -> None:
+    stream, protocol, transport = fake_stream()
+    protocol.pause_writing()
+    write = asyncio.ensure_future(stream.write(b"hello"))
+    await asyncio.sleep(0)
+    assert not write.done()
+    protocol.resume_writing()
+    await write
+    assert transport.written == [b"hello"]
+
+
+async def test_write_timeout_while_draining() -> None:
+    stream, protocol, _ = fake_stream()
+    protocol.pause_writing()
+    with pytest.raises(httpcore2.WriteTimeout):
+        await stream.write(b"hello", timeout=0.01)
+
+
+async def test_write_fails_when_the_connection_is_lost_while_draining() -> None:
+    stream, protocol, _ = fake_stream()
+    protocol.pause_writing()
+    write = asyncio.ensure_future(stream.write(b"hello"))
+    await asyncio.sleep(0)
+    protocol.connection_lost(OSError("reset"))
+    with pytest.raises(httpcore2.WriteError):
+        await write

From 883aefdf91b03ef0ba801a119780452f840893d5 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:12:15 +0000
Subject: [PATCH 2/6] Report socket option failures as connection errors

Setting a socket option after the connection is established now closes the connection and raises ConnectError, like any other failure to establish a usable connection, instead of leaking the transport and raising a raw OSError.
---
 src/httpcore2/httpcore2/_backends/asyncio.py | 11 ++++++++---
 tests/httpcore2/test_asyncio_backend.py      |  6 ++++++
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/src/httpcore2/httpcore2/_backends/asyncio.py b/src/httpcore2/httpcore2/_backends/asyncio.py
index 7e55be92..52bd466a 100644
--- a/src/httpcore2/httpcore2/_backends/asyncio.py
+++ b/src/httpcore2/httpcore2/_backends/asyncio.py
@@ -247,11 +247,16 @@ async def _connect(
             raise ConnectTimeout("timed out") from None
         except OSError as exc:
             raise ConnectError(str(exc)) from exc
+        stream = AsyncioStream(typing.cast(asyncio.Transport, transport), protocol)
         if socket_options:
             sock = transport.get_extra_info("socket")
-            for option in socket_options:
-                sock.setsockopt(*option)
-        return AsyncioStream(typing.cast(asyncio.Transport, transport), protocol)
+            try:
+                for option in socket_options:
+                    sock.setsockopt(*option)
+            except OSError as exc:
+                await stream.aclose()
+                raise ConnectError(str(exc)) from exc
+        return stream
 
     async def sleep(self, seconds: float) -> None:
         await asyncio.sleep(seconds)
diff --git a/tests/httpcore2/test_asyncio_backend.py b/tests/httpcore2/test_asyncio_backend.py
index 14ba2bde..c5471122 100644
--- a/tests/httpcore2/test_asyncio_backend.py
+++ b/tests/httpcore2/test_asyncio_backend.py
@@ -103,6 +103,12 @@ async def test_read_timeout(serve: Callable[..., Awaitable[int]]) -> None:
     await stream.aclose()
 
 
+async def test_invalid_socket_option(serve: Callable[..., Awaitable[int]]) -> None:
+    port = await serve()
+    with pytest.raises(httpcore2.ConnectError):
+        await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port, socket_options=[(socket.SOL_SOCKET, -1, 1)])
+
+
 async def test_connect_refused() -> None:
     with socket.socket() as sock:
         sock.bind(("127.0.0.1", 0))

From 6d7231dbb1664656418aef7e2775f07c7f8abdb5 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:21:36 +0000
Subject: [PATCH 3/6] Harden TLS upgrades and connection attempts in the
 asyncio backend

start_tls now refuses to proceed if any data was received before the handshake, since it would be plaintext that must not be mistaken for data received over TLS, and it sets the event loop's own handshake timeout from the requested timeout so the requested deadline is the one that fires. connect_tcp staggers attempts across resolved addresses (Happy Eyeballs) on event loops that support it, as the anyio backend did.
---
 src/httpcore2/httpcore2/_backends/asyncio.py | 45 +++++++++++++++++---
 tests/httpcore2/test_asyncio_backend.py      | 28 ++++++++++++
 2 files changed, 67 insertions(+), 6 deletions(-)

diff --git a/src/httpcore2/httpcore2/_backends/asyncio.py b/src/httpcore2/httpcore2/_backends/asyncio.py
index 52bd466a..e430a069 100644
--- a/src/httpcore2/httpcore2/_backends/asyncio.py
+++ b/src/httpcore2/httpcore2/_backends/asyncio.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import collections
+import inspect
 import ssl
 import typing
 
@@ -20,6 +21,22 @@
 RECEIVE_HIGH_WATER = 256 * 1024
 RECEIVE_LOW_WATER = 64 * 1024
 
+# Stagger connection attempts across resolved addresses, as in RFC 8305.
+HAPPY_EYEBALLS_DELAY = 0.25
+# The event loop insists on a TLS handshake timeout; this stands in for "none".
+NO_HANDSHAKE_TIMEOUT = 365 * 24 * 60 * 60.0
+
+_happy_eyeballs_support: dict[type[asyncio.AbstractEventLoop], bool] = {}
+
+
+def _connection_kwargs(loop: asyncio.AbstractEventLoop) -> dict[str, typing.Any]:
+    # Not every event loop implements Happy Eyeballs; use it where available.
+    supported = _happy_eyeballs_support.get(type(loop))
+    if supported is None:
+        supported = "happy_eyeballs_delay" in inspect.signature(loop.create_connection).parameters
+        _happy_eyeballs_support[type(loop)] = supported
+    return {"happy_eyeballs_delay": HAPPY_EYEBALLS_DELAY} if supported else {}
+
 
 class _Timeout(Exception):
     """
@@ -175,10 +192,24 @@ async def start_tls(
         server_hostname: str | None = None,
         timeout: float | None = None,
     ) -> AsyncNetworkStream:
+        protocol = self._protocol
+        if protocol.chunks or protocol.eof or protocol.closed:
+            # Nothing may arrive before the handshake: anything already buffered
+            # is plaintext that must not be mistaken for data received over TLS.
+            await self.aclose()
+            raise ConnectError("Received unexpected data before the TLS handshake")
+
         loop = asyncio.get_running_loop()
-        # The loop reports its own handshake timeout as a connection error,
-        # so the deadline is applied here instead to raise a timeout.
-        handshake = loop.start_tls(self._transport, self._protocol, ssl_context, server_hostname=server_hostname)
+        # The loop reports its own handshake timeout as a connection error, so
+        # the deadline is applied here to raise a timeout, with the loop's own
+        # deadline kept out of the way.
+        handshake = loop.start_tls(
+            self._transport,
+            protocol,
+            ssl_context,
+            server_hostname=server_hostname,
+            ssl_handshake_timeout=NO_HANDSHAKE_TIMEOUT if timeout is None else timeout + 1.0,
+        )
         try:
             transport = await asyncio.wait_for(handshake, timeout)
         except (TimeoutError, asyncio.TimeoutError):
@@ -189,8 +220,8 @@ async def start_tls(
             raise ConnectError(str(exc)) from exc
         if transport is None:  # pragma: no cover
             raise ConnectError("TLS handshake failed")
-        self._protocol.transport = transport
-        return AsyncioStream(transport, self._protocol)
+        protocol.transport = transport
+        return AsyncioStream(transport, protocol)
 
     def get_extra_info(self, info: str) -> typing.Any:
         if info == "ssl_object":
@@ -222,7 +253,9 @@ async def connect_tcp(
         loop = asyncio.get_running_loop()
         local_addr = None if local_address is None else (local_address, 0)
         # By default TCP sockets opened in `asyncio` include TCP_NODELAY.
-        connect = loop.create_connection(AsyncioStreamProtocol, host, port, local_addr=local_addr)
+        connect = loop.create_connection(
+            AsyncioStreamProtocol, host, port, local_addr=local_addr, **_connection_kwargs(loop)
+        )
         return await self._connect(connect, timeout, socket_options)
 
     async def connect_unix_socket(
diff --git a/tests/httpcore2/test_asyncio_backend.py b/tests/httpcore2/test_asyncio_backend.py
index c5471122..68f4e792 100644
--- a/tests/httpcore2/test_asyncio_backend.py
+++ b/tests/httpcore2/test_asyncio_backend.py
@@ -159,6 +159,34 @@ async def test_start_tls(serve: Callable[..., Awaitable[int]]) -> None:
     await tls_stream.aclose()
 
 
+async def test_start_tls_rejects_data_received_before_the_handshake(serve: Callable[..., Awaitable[int]]) -> None:
+    async def banner(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        writer.write(b"220 hello\r\n")
+        await writer.drain()
+        await reader.read()
+        writer.close()
+
+    port = await serve(banner)
+    stream = await httpcore2.AsyncioBackend().connect_tcp("127.0.0.1", port)
+    # Give the banner time to arrive and be buffered.
+    await asyncio.sleep(0.05)
+    with pytest.raises(httpcore2.ConnectError):
+        await stream.start_tls(ssl.create_default_context(), server_hostname="localhost", timeout=5.0)
+
+
+async def test_happy_eyeballs_is_used_where_supported() -> None:
+    from httpcore2._backends.asyncio import HAPPY_EYEBALLS_DELAY, _connection_kwargs
+
+    assert _connection_kwargs(asyncio.get_running_loop()) == {"happy_eyeballs_delay": HAPPY_EYEBALLS_DELAY}
+
+    class MinimalLoop:
+        async def create_connection(self, protocol_factory: typing.Any, host: str, port: int) -> None: ...
+
+    loop = typing.cast(asyncio.AbstractEventLoop, MinimalLoop())
+    assert _connection_kwargs(loop) == {}
+    assert _connection_kwargs(loop) == {}
+
+
 async def test_start_tls_failure(serve: Callable[..., Awaitable[int]]) -> None:
     # A plain echo server answers the ClientHello with the ClientHello.
     port = await serve()

From 9871eef10f6169e80fb1802f03c53b45ba2d9332 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:38:40 +0000
Subject: [PATCH 4/6] Tolerate event loops whose connection method has no
 signature

Detecting Happy Eyeballs support inspects the event loop's create_connection signature; extension-implemented loops may expose none, in which case the feature is treated as unsupported rather than failing every connection.
---
 src/httpcore2/httpcore2/_backends/asyncio.py |  6 +++++-
 tests/httpcore2/test_asyncio_backend.py      | 15 +++++++++++++++
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/src/httpcore2/httpcore2/_backends/asyncio.py b/src/httpcore2/httpcore2/_backends/asyncio.py
index e430a069..0d2b60a8 100644
--- a/src/httpcore2/httpcore2/_backends/asyncio.py
+++ b/src/httpcore2/httpcore2/_backends/asyncio.py
@@ -33,7 +33,11 @@ def _connection_kwargs(loop: asyncio.AbstractEventLoop) -> dict[str, typing.Any]
     # Not every event loop implements Happy Eyeballs; use it where available.
     supported = _happy_eyeballs_support.get(type(loop))
     if supported is None:
-        supported = "happy_eyeballs_delay" in inspect.signature(loop.create_connection).parameters
+        try:
+            supported = "happy_eyeballs_delay" in inspect.signature(loop.create_connection).parameters
+        except (TypeError, ValueError):
+            # Some extension-implemented loops expose no signature to inspect.
+            supported = False
         _happy_eyeballs_support[type(loop)] = supported
     return {"happy_eyeballs_delay": HAPPY_EYEBALLS_DELAY} if supported else {}
 
diff --git a/tests/httpcore2/test_asyncio_backend.py b/tests/httpcore2/test_asyncio_backend.py
index 68f4e792..d74aeec0 100644
--- a/tests/httpcore2/test_asyncio_backend.py
+++ b/tests/httpcore2/test_asyncio_backend.py
@@ -187,6 +187,21 @@ async def create_connection(self, protocol_factory: typing.Any, host: str, port:
     assert _connection_kwargs(loop) == {}
 
 
+async def test_happy_eyeballs_detection_tolerates_loops_without_signatures(monkeypatch: pytest.MonkeyPatch) -> None:
+    import inspect
+
+    from httpcore2._backends.asyncio import _connection_kwargs
+
+    class OpaqueLoop:
+        def create_connection(self, *args: typing.Any, **kwargs: typing.Any) -> None: ...
+
+    def no_signature(callable: typing.Any) -> typing.Any:
+        raise ValueError("no signature found")
+
+    monkeypatch.setattr(inspect, "signature", no_signature)
+    assert _connection_kwargs(typing.cast(asyncio.AbstractEventLoop, OpaqueLoop())) == {}
+
+
 async def test_start_tls_failure(serve: Callable[..., Awaitable[int]]) -> None:
     # A plain echo server answers the ClientHello with the ClientHello.
     port = await serve()

From 777cd8647e8ffbf79cc537c8cc26b7bddc2a2910 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:38:40 +0000
Subject: [PATCH 5/6] Add the original httpx to the benchmark harness

The client worker can drive httpx as well as httpx2, since the two share an API, giving the comparison a reference point for the package httpx2 forked from.
---
 benchmark/client.py | 25 ++++++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/benchmark/client.py b/benchmark/client.py
index 5e18a13c..dda7e40d 100644
--- a/benchmark/client.py
+++ b/benchmark/client.py
@@ -65,6 +65,18 @@ def check_size(received: int, expected: int) -> None:
 def build_httpx2(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]:
     import httpx2
 
+    return _build_httpx_like(httpx2, scenario, payload)
+
+
+def build_httpx(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]:
+    # The original httpx, for reference; it shares the httpx2 API.
+    import httpx
+
+    return _build_httpx_like(httpx, scenario, payload)
+
+
+def _build_httpx_like(httpx2: Any, scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]:
+
     client = httpx2.AsyncClient(
         limits=httpx2.Limits(
             max_connections=scenario.max_connections,
@@ -79,13 +91,15 @@ def build_httpx2(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn
     if scenario.post:
         headers.append(("content-length", str(len(payload))))
 
-    class RequestBody(httpx2.AsyncByteStream):
-        async def __aiter__(self) -> AsyncIterator[bytes]:
-            if scenario.post:
-                yield payload
+    async def request_body(self: object) -> AsyncIterator[bytes]:
+        if scenario.post:
+            yield payload
+
+    # Built dynamically so the helper can serve both httpx2 and httpx.
+    request_body_stream = type("RequestBody", (httpx2.AsyncByteStream,), {"__aiter__": request_body})
 
     async def stream_one() -> None:
-        request = httpx2.Request(scenario.method, scenario.url, headers=headers, stream=RequestBody())
+        request = httpx2.Request(scenario.method, scenario.url, headers=headers, stream=request_body_stream())
         response = await client.send(request, stream=True, follow_redirects=False)
         received = 0
         async for chunk in response.aiter_raw(scenario.chunk_size):
@@ -198,6 +212,7 @@ async def read_one() -> None:
 
 BUILDERS: dict[str, Builder] = {
     "httpx2": build_httpx2,
+    "httpx": build_httpx,
     "httpcore2": build_httpcore2,
     "aiohttp": build_aiohttp,
     "punkreq": build_punkreq,

From c6d2d00605e38349c250abd46b86de31b715975b Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Thu, 27 Aug 2026 08:54:07 +0000
Subject: [PATCH 6/6] Run the sync integration tests once and document the
 benchmark's reference packages

The sync connection pool has a single backend, so the generated sync integration tests no longer parametrize over a duplicate of it. The benchmark README explains how to provision httpx and punkreq in a separate interpreter, since neither belongs in the project environment.
---
 benchmark/README.md                        | 9 ++++++++-
 benchmark/client.py                        | 3 ++-
 scripts/unasync.py                         | 2 +-
 tests/httpcore2/_async/test_integration.py | 1 +
 tests/httpcore2/_sync/test_integration.py  | 3 ++-
 5 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index d1678966..94ffe87b 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -28,7 +28,14 @@ git worktree add ../httpx2-main main && (cd ../httpx2-main && uv sync)
 scripts/benchmark --python main=../httpx2-main/.venv/bin/python --python work=.venv/bin/python --lib httpx2
 ```
 
-`punkreq` is supported as an extra reference point (`--lib punkreq`) if it is installed.
+The original `httpx` and `punkreq` are supported as reference points (`--lib httpx`, `--lib punkreq`).
+Neither is part of the project's environment (installing `httpx` next to `httpx2` interferes with the
+alias tests), so provision them in a separate interpreter and pass it with `--python`:
+
+```
+uv venv --python 3.14 /tmp/refs && uv pip install --python /tmp/refs/bin/python httpx punkreq zuvloop
+scripts/benchmark --python refs=/tmp/refs/bin/python --lib httpx --lib punkreq
+```
 
 ## Reading the results
 
diff --git a/benchmark/client.py b/benchmark/client.py
index dda7e40d..8399da7c 100644
--- a/benchmark/client.py
+++ b/benchmark/client.py
@@ -69,7 +69,8 @@ def build_httpx2(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn
 
 
 def build_httpx(scenario: Scenario, payload: bytes) -> tuple[RequestFn, CloseFn]:
-    # The original httpx, for reference; it shares the httpx2 API.
+    # The original httpx, for reference; it shares the httpx2 API. Not part of the
+    # project environment: see the README for how to provision it.
     import httpx
 
     return _build_httpx_like(httpx, scenario, payload)
diff --git a/scripts/unasync.py b/scripts/unasync.py
index eb6c9762..43e28ac3 100644
--- a/scripts/unasync.py
+++ b/scripts/unasync.py
@@ -11,7 +11,7 @@
     ),
     ("import trio as concurrency", "from tests.httpcore2 import concurrency"),
     ("anyio.sleep", "concurrency.sleep"),
-    ("AnyIOBackend", "SyncBackend"),
+    ("BACKENDS = \\[None, httpcore2.AnyIOBackend\\(\\)\\]", "BACKENDS = [None]"),
     ("AsyncIterator", "Iterator"),
     ("Async([A-Z][A-Za-z0-9_]*)", r"\2"),
     ("async def", "def"),
diff --git a/tests/httpcore2/_async/test_integration.py b/tests/httpcore2/_async/test_integration.py
index 527d6c0e..8a7f12a9 100644
--- a/tests/httpcore2/_async/test_integration.py
+++ b/tests/httpcore2/_async/test_integration.py
@@ -5,6 +5,7 @@
 
 import httpcore2
 
+# The automatic backend, and the explicit anyio backend it no longer selects under asyncio.
 BACKENDS = [None, httpcore2.AnyIOBackend()]
 
 
diff --git a/tests/httpcore2/_sync/test_integration.py b/tests/httpcore2/_sync/test_integration.py
index 5bac9e2b..2cf9d9b0 100644
--- a/tests/httpcore2/_sync/test_integration.py
+++ b/tests/httpcore2/_sync/test_integration.py
@@ -5,7 +5,8 @@
 
 import httpcore2
 
-BACKENDS = [None, httpcore2.SyncBackend()]
+# The automatic backend, and the explicit anyio backend it no longer selects under asyncio.
+BACKENDS = [None]