From da5b37a9599ee8d10621063fbaa927df738b06a9 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:10:07 +0530 Subject: [PATCH 01/14] Expose a socket object on stream transports --- src/simloop/_transports.py | 45 ++++++++++++++ tests/test_client_sockets.py | 110 +++++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 tests/test_client_sockets.py diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index 0845114..1d295b5 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import socket from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -89,6 +90,45 @@ def get_protocol(self) -> Any: return self._protocol +class _SimSocket: + """Stand-in for the OS socket a simulated stream connection does not have. + + Client stacks introspect the transport's socket: anyio eagerly calls + getpeername()/.family on every typed-attribute lookup, aiohttp sets + TCP_NODELAY through it, and httpcore polls fileno() to decide whether a + pooled connection died. Addresses are reported as synthetic-IP tuples so + callers that feed them to ipaddress.ip_address() get a plausible + AF_INET sockaddr; option and teardown calls are accepted and do nothing. + Anything not listed here raises AttributeError, keeping the simulation + loud about surface it does not fake. + """ + + def __init__(self, transport: _SimStreamTransport) -> None: + self._transport = transport + self.family = socket.AF_INET + self.type = socket.SOCK_STREAM + self.proto = socket.IPPROTO_TCP + + def _address(self, endpoint: _Addr) -> tuple[str, int]: + name, port = endpoint + return (self._transport._net.address(name), port) + + def getsockname(self) -> tuple[str, int]: + return self._address(self._transport._local) + + def getpeername(self) -> tuple[str, int]: + return self._address(self._transport._remote) + + def setsockopt(self, *args: Any) -> None: + return None + + def shutdown(self, how: int) -> None: + return None + + def close(self) -> None: + return None + + class _SimStreamTransport(asyncio.Transport): """One end of a reliable, ordered byte-stream connection. @@ -116,6 +156,7 @@ def __init__( self._backlog: list[bytes] = [] self._eof_pending = False self._limits = (16 * 1024, 64 * 1024) # (low, high): recorded, inert + self._extra_socket: _SimSocket | None = None def _begin(self, protocol: Any) -> None: self._protocol = protocol @@ -244,6 +285,10 @@ def get_extra_info(self, name: str, default: Any = None) -> Any: return self._local if name == "peername": return self._remote + if name == "socket": + if self._extra_socket is None: + self._extra_socket = _SimSocket(self) + return self._extra_socket return default def set_protocol(self, protocol: Any) -> None: diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py new file mode 100644 index 0000000..93fe044 --- /dev/null +++ b/tests/test_client_sockets.py @@ -0,0 +1,110 @@ +"""The fake socket stream transports expose for client-stack introspection. + +anyio reads get_extra_info("socket") and eagerly calls getpeername()/.family +on it (anyio/abc/_sockets.py); httpcore polls fileno() to decide whether a +pooled connection died. These tests pin the contract both stacks rely on. +""" + +from __future__ import annotations + +import asyncio +import ipaddress +import socket +from typing import Any + +import pytest + +from simloop import SimLoop + + +def _network(seed: int = 0) -> SimLoop: + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.host("client") + return loop + + +async def _hold_open(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await reader.read() # stay open until the client closes + writer.close() + + +def _connected_pair(loop: SimLoop, handler: Any = _hold_open) -> tuple[Any, Any]: + """One established client connection; returns (client_writer, server).""" + + async def main() -> tuple[Any, Any]: + server = await loop.net.host("server").create_task( + asyncio.start_server(handler, "0.0.0.0", 9000) + ) + writer_box: list[Any] = [] + + async def connect() -> None: + _, writer = await asyncio.open_connection("server", 9000) + writer_box.append(writer) + + await loop.net.host("client").create_task(connect()) + return writer_box[0], server + + return loop.run_until_complete(main()) + + +def test_stream_transport_exposes_a_socket_object() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock is not None + assert sock is writer.transport.get_extra_info("socket") # stable singleton + assert sock.family == socket.AF_INET + assert sock.type == socket.SOCK_STREAM + assert sock.proto == socket.IPPROTO_TCP + server.close() + loop.close() + + +def test_socket_addresses_are_synthetic_ip_tuples() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + local_ip, local_port = sock.getsockname() + peer_ip, peer_port = sock.getpeername() + # Plausible AF_INET sockaddrs: ip_address() must accept them. + ipaddress.ip_address(local_ip) + ipaddress.ip_address(peer_ip) + assert local_ip == loop.net.address("client") + assert peer_ip == loop.net.address("server") + assert peer_port == 9000 + # Same endpoints the transport itself reports, by name. + assert writer.transport.get_extra_info("peername") == ("server", 9000) + assert writer.transport.get_extra_info("sockname")[1] == local_port + server.close() + loop.close() + + +def test_socket_options_and_teardown_calls_are_inert() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) is None + assert sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) is None + assert sock.shutdown(socket.SHUT_RDWR) is None + assert sock.close() is None + with pytest.raises(AttributeError): + sock.recv(1) # no pretending to be readable + server.close() + loop.close() + + +def test_datagram_transports_still_answer_none() -> None: + loop = _network() + + async def main() -> Any: + transport, _ = await loop.create_datagram_endpoint( + asyncio.DatagramProtocol, local_addr=("0.0.0.0", 5000) + ) + try: + return transport.get_extra_info("socket") + finally: + transport.close() + + assert loop.run_until_complete(main()) is None + loop.close() From 54547cdf9b5442093e7f9c7d04667bde7834fe24 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:20:45 +0530 Subject: [PATCH 02/14] Track peer liveness through a parked descriptor --- src/simloop/_transports.py | 43 +++++++++++++++++ tests/test_client_sockets.py | 94 +++++++++++++++++++++++++++++++++++- 2 files changed, 136 insertions(+), 1 deletion(-) diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index 1d295b5..1fb1dcc 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -108,6 +108,8 @@ def __init__(self, transport: _SimStreamTransport) -> None: self.family = socket.AF_INET self.type = socket.SOCK_STREAM self.proto = socket.IPPROTO_TCP + self._park: tuple[socket.socket, socket.socket] | None = None + self._disposed = False def _address(self, endpoint: _Addr) -> tuple[str, int]: name, port = endpoint @@ -128,6 +130,38 @@ def shutdown(self, how: int) -> None: def close(self) -> None: return None + def fileno(self) -> int: + # httpcore's is_socket_readable() polls this fd to decide whether a + # pooled connection died. A parked socketpair end reads not-readable + # until its peer end is written to or closed, so the poll tracks the + # simulated connection: alive while the transport lives, readable + # (= discard me) once the peer is gone. -1 after teardown is what a + # closed real socket reports. Lazy: connections nobody introspects + # cost no descriptors, which matters at campaign scale. + if self._disposed or self._transport._closed: + return -1 + if self._park is None: + self._park = socket.socketpair() + # The peer's FIN/RST can land before anyone asks for the socket, + # so a freshly parked pair inherits whatever the transport + # already saw rather than claiming a dead connection is live. + if self._transport._peer_closed: + self._park[0].close() + return self._park[1].fileno() + + def _peer_gone(self) -> None: + # EOF or reset arrived: closing the held end makes the exposed end + # poll readable, exactly when a real kernel would report it. + if self._park is not None: + self._park[0].close() + + def _dispose(self) -> None: + self._disposed = True + if self._park is not None: + self._park[0].close() + self._park[1].close() + self._park = None + class _SimStreamTransport(asyncio.Transport): """One end of a reliable, ordered byte-stream connection. @@ -157,6 +191,7 @@ def __init__( self._eof_pending = False self._limits = (16 * 1024, 64 * 1024) # (low, high): recorded, inert self._extra_socket: _SimSocket | None = None + self._peer_closed = False # the peer's FIN or RST has arrived def _begin(self, protocol: Any) -> None: self._protocol = protocol @@ -220,6 +255,8 @@ def _finish(self, exc: Exception | None) -> None: self._closed = True self._closing = True self._net._drop_stream(self._conn, self._local[0]) + if self._extra_socket is not None: + self._extra_socket._dispose() protocol, self._protocol = self._protocol, None if protocol is not None: protocol.connection_lost(exc) @@ -237,6 +274,9 @@ def _data_arrived(self, data: bytes) -> None: self._protocol.data_received(data) def _eof_arrived(self) -> None: + self._peer_closed = True + if self._extra_socket is not None: + self._extra_socket._peer_gone() if self._closed: return if self._read_paused: @@ -247,6 +287,9 @@ def _eof_arrived(self) -> None: self.close() def _reset_arrived(self) -> None: + self._peer_closed = True + if self._extra_socket is not None: + self._extra_socket._peer_gone() if self._closed: return self._finish(ConnectionResetError("Connection reset by peer")) diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py index 93fe044..cefaa5d 100644 --- a/tests/test_client_sockets.py +++ b/tests/test_client_sockets.py @@ -9,6 +9,7 @@ import asyncio import ipaddress +import select import socket from typing import Any @@ -45,7 +46,8 @@ async def connect() -> None: await loop.net.host("client").create_task(connect()) return writer_box[0], server - return loop.run_until_complete(main()) + pair: tuple[Any, Any] = loop.run_until_complete(main()) + return pair def test_stream_transport_exposes_a_socket_object() -> None: @@ -108,3 +110,93 @@ async def main() -> Any: assert loop.run_until_complete(main()) is None loop.close() + + +def _readable(fd: int) -> bool: + """The exact check httpcore's is_socket_readable performs.""" + if fd < 0: + return True + rready, _, _ = select.select([fd], [], [], 0) + return bool(rready) + + +def test_fileno_is_lazy_and_not_readable_while_live() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock._park is None # no kernel object until someone asks + fd = sock.fileno() + assert fd >= 0 + assert fd == sock.fileno() # stable + assert not _readable(fd) # live connection: httpcore keeps pooling it + server.close() + writer.close() + loop.run_until_complete(asyncio.sleep(1.0)) + loop.close() + + +def test_fd_turns_readable_when_the_peer_closes() -> None: + loop = _network() + + async def close_after_a_beat( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + await asyncio.sleep(0.5) # after the client has taken its fd + writer.close() + + writer, server = _connected_pair(loop, close_after_a_beat) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + # Let the FIN cross the simulated network. + loop.run_until_complete(asyncio.sleep(1.0)) + assert _readable(fd) # httpcore now sees the connection as expired + server.close() + writer.close() + loop.run_until_complete(asyncio.sleep(1.0)) + loop.close() + + +def test_fd_born_readable_when_the_peer_left_first() -> None: + """A FIN can land before anyone asks for the socket.""" + loop = _network() + + async def close_immediately( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + writer.close() + + writer, server = _connected_pair(loop, close_immediately) + loop.run_until_complete(asyncio.sleep(1.0)) # FIN already delivered + sock = writer.transport.get_extra_info("socket") + assert _readable(sock.fileno()) # never claim a dead connection is live + server.close() + writer.close() + loop.run_until_complete(asyncio.sleep(1.0)) + loop.close() + + +def test_fd_lifecycle_ends_with_the_transport() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + writer.close() + loop.run_until_complete(asyncio.sleep(1.0)) + assert sock._park is None # both ends closed, nothing leaked + assert sock.fileno() == -1 # closed-socket semantics + server.close() + loop.close() + + +def test_fileno_after_teardown_never_creates_a_descriptor() -> None: + loop = _network() + writer, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + writer.close() + loop.run_until_complete(asyncio.sleep(1.0)) + assert sock.fileno() == -1 + assert sock._park is None + server.close() + loop.close() From e44d53c35bb1677e4a0e4b99cb8b08f302a3f4ba Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:31:38 +0530 Subject: [PATCH 03/14] Accept TCP sock_connect and park the target address --- src/simloop/_loop.py | 23 +++++++++++-- tests/test_client_sockets.py | 64 +++++++++++++++++++++++++++++++++++- 2 files changed, 84 insertions(+), 3 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 8b20d35..bf5dd18 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -6,6 +6,7 @@ import gc import heapq import random +import socket import sys from array import array from asyncio import events @@ -117,6 +118,10 @@ def __init__(self, seed: int = 0) -> None: self._unhandled: list[BaseException] = [] self._exception_handler: _ExceptionHandler | None = None self._task_factory: _TaskFactory | None = None + # sock_connect(sock, addr) -> create_connection(sock=sock) is how + # aiohttp reaches the network; the address is only visible in the + # first call, so it is parked here until the upgrade claims it. + self._sock_targets: dict[Any, tuple[Any, int]] = {} self._net = SimNetwork(self) @classmethod @@ -328,6 +333,7 @@ def is_closed(self) -> bool: def close(self) -> None: if self._running: raise RuntimeError("cannot close a running event loop") + self._sock_targets.clear() self._closed = True def _check_closed(self) -> None: @@ -585,8 +591,21 @@ def sock_recv_into(self, *args: Any, **kwargs: Any) -> Any: def sock_sendall(self, *args: Any, **kwargs: Any) -> Any: _fence("sock_sendall") - def sock_connect(self, *args: Any, **kwargs: Any) -> Any: - _fence("sock_connect") + async def sock_connect(self, sock: Any, address: Any) -> None: + # aiohttp's connector creates a real TCP socket and connects it here + # before handing it to create_connection(sock=...). The simulation + # accepts exactly that shape; every other socket kind still fences. + if ( + getattr(sock, "family", None) != socket.AF_INET + or getattr(sock, "type", None) != socket.SOCK_STREAM + ): + _fence("sock_connect (only AF_INET stream sockets are simulated)") + host, port = address[0], address[1] + self._net._resolve(host) # unknown targets fail here, loudly + # No packet moves and no time passes yet: the connection handshake + # (and its one-RTT cost) happens when create_connection claims the + # socket, keeping the total cost identical to a direct connect. + self._sock_targets[sock] = (host, port) def sock_accept(self, *args: Any, **kwargs: Any) -> Any: _fence("sock_accept") diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py index cefaa5d..56d7fb3 100644 --- a/tests/test_client_sockets.py +++ b/tests/test_client_sockets.py @@ -15,7 +15,7 @@ import pytest -from simloop import SimLoop +from simloop import SimLoop, SimulationFenceError def _network(seed: int = 0) -> SimLoop: @@ -200,3 +200,65 @@ def test_fileno_after_teardown_never_creates_a_descriptor() -> None: assert sock._park is None server.close() loop.close() + + +def test_sock_connect_records_a_stream_socket_target() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setblocking(False) # what aiohappyeyeballs does + try: + await loop.sock_connect(sock, (loop.net.address("server"), 9000)) + assert loop._sock_targets[sock] == (loop.net.address("server"), 9000) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_rejects_unknown_addresses() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="10.9.9.9"): + await loop.sock_connect(sock, ("10.9.9.9", 9000)) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_still_fences_datagram_sockets() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + with pytest.raises(SimulationFenceError): + await loop.sock_connect(sock, (loop.net.address("server"), 9000)) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_consumes_no_virtual_time() -> None: + loop = _network() + + async def main() -> float: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + before = loop.time() + await loop.sock_connect(sock, ("server", 9000)) + return loop.time() - before + finally: + sock.close() + + assert loop.run_until_complete(main()) == 0.0 + loop.close() From 814bd24f64b67875c717cf2b8200f9409dd17211 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:39:57 +0530 Subject: [PATCH 04/14] Upgrade parked sockets into simulated connections --- src/simloop/_loop.py | 17 +++++++ tests/test_client_sockets.py | 92 ++++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index bf5dd18..083f7f2 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -392,7 +392,24 @@ async def create_connection( port: Any = None, **kwargs: Any, ) -> Any: + sock = kwargs.pop("sock", None) _reject_kwargs("create_connection", kwargs) + if sock is not None: + # The stdlib treats a passed-in socket as already connected and + # takes ownership of it. Here "connected" means sock_connect + # parked a target for it; the real descriptor is closed at once + # because the simulation only needed the address it carried. + if host is not None or port is not None: + raise ValueError( + "host/port and sock can not be specified at the same time" + ) + target = self._sock_targets.pop(sock, None) + if target is None: + raise OSError( + "the given socket was not connected via sock_connect on this loop" + ) + sock.close() + host, port = target return await self._net._open_connection(protocol_factory, host, port) async def create_server( diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py index 56d7fb3..9f66001 100644 --- a/tests/test_client_sockets.py +++ b/tests/test_client_sockets.py @@ -11,6 +11,7 @@ import ipaddress import select import socket +from collections.abc import Callable, Coroutine from typing import Any import pytest @@ -262,3 +263,94 @@ async def main() -> float: assert loop.run_until_complete(main()) == 0.0 loop.close() + + +def _aiohttp_style_connect( + loop: SimLoop, port: int = 9000 +) -> Callable[[], Coroutine[Any, Any, tuple[Any, Any, Any]]]: + """The exact two-call sequence aiohttp + aiohappyeyeballs performs.""" + + async def connect() -> tuple[Any, Any, Any]: + infos = await loop.getaddrinfo("server", port, type=socket.SOCK_STREAM) + family, kind, proto, _, address = infos[0] + sock = socket.socket(family=family, type=kind, proto=proto) + sock.setblocking(False) + await loop.sock_connect(sock, address) + transport, protocol = await loop.create_connection( + asyncio.Protocol, ssl=None, server_hostname=None, sock=sock + ) + return sock, transport, protocol + + return connect + + +def test_parked_socket_upgrades_to_a_sim_connection() -> None: + loop = _network() + loop.net.set_defaults(latency=(0.001, 0.001)) # so the round trip is visible + + async def main() -> tuple[Any, Any]: + server = await loop.net.host("server").create_task( + asyncio.start_server(_hold_open, "0.0.0.0", 9000) + ) + started = loop.time() + sock, transport, _ = await loop.net.host("client").create_task( + _aiohttp_style_connect(loop)() + ) + elapsed = loop.time() - started + transport.close() + server.close() + return sock, elapsed + + sock, elapsed = loop.run_until_complete(main()) + loop.run_until_complete(asyncio.sleep(1.0)) # let both ends finish closing + assert sock.fileno() == -1 # the loop took ownership and closed it + assert elapsed > 0.0 # the handshake cost its round trip + assert not loop._sock_targets # the parked entry was claimed + loop.close() + + +def test_refused_port_raises_from_the_upgrade() -> None: + loop = _network() + loop.net.set_defaults(latency=(0.001, 0.001)) + + async def main() -> None: + with pytest.raises(ConnectionRefusedError): + await loop.net.host("client").create_task( + _aiohttp_style_connect(loop, port=9999)() + ) + + loop.run_until_complete(main()) + loop.close() + + +def test_unparked_socket_is_rejected() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="sock_connect"): + await loop.create_connection(asyncio.Protocol, sock=sock) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_and_host_together_follow_stdlib_rules() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + await loop.sock_connect(sock, ("server", 9000)) + with pytest.raises(ValueError): + await loop.create_connection( + asyncio.Protocol, "server", 9000, sock=sock + ) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() From d7eef8ba680d8bde8085f1209f1c15aaca7fa561 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:46:22 +0530 Subject: [PATCH 05/14] Pin one resolver row per socket kind --- tests/test_resolution.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/test_resolution.py b/tests/test_resolution.py index 0c97c89..603e401 100644 --- a/tests/test_resolution.py +++ b/tests/test_resolution.py @@ -164,6 +164,17 @@ def test_getaddrinfo_honors_type_family_and_proto_filters() -> None: assert len(_resolve("server", 80, family=socket.AF_INET)) == 2 +def test_getaddrinfo_returns_exactly_one_row_per_socket_kind() -> None: + # aiohappyeyeballs (aiohttp) and anyio (httpx) both stagger connection + # attempts across candidate addresses when more than one comes back, + # arming extra timers and racing tasks. Both stay on their simple path + # because this resolver returns exactly one row per socket kind. Anyone + # making this return more rows must first decide what happy-eyeballs + # should mean under simulation. + assert len(_resolve("server", 80, type=socket.SOCK_STREAM)) == 1 + assert len(_resolve("server", 80, type=socket.SOCK_DGRAM)) == 1 + + def test_getaddrinfo_accepts_a_bytes_host_name() -> None: # Resolver stacks encode names before resolving them — anyio hands the # stdlib resolver ASCII bytes — so bytes must resolve like their text. From bfe0696fe062e78ef0328f7832887d090c2196f3 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 11:58:21 +0530 Subject: [PATCH 06/14] Refresh the compatibility table after the client unlocks --- docs/compatibility.md | 70 ++++++++++++++++++++++++++----------------- docs/supported-api.md | 15 ++++++++-- 2 files changed, 56 insertions(+), 29 deletions(-) diff --git a/docs/compatibility.md b/docs/compatibility.md index e423514..02d3105 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -46,11 +46,12 @@ carries a date instead. | anyio | 4.14.2 | works: task group, memory object stream (one, two, three), anyio.sleep and move_on_after; virtual clock reached 1.75s | Asyncio backend only; nothing here touches a socket. | | redis (RESP wire protocol) | n/a | works: PING, SET and GET round trips over one connection: ['+PONG', '+OK', '0'] | Hand-rolled RESP over sim streams; no client library, no real server. | | websockets | 17.0.1 | works: handshake, one echoed frame ('HELLO') and close over ws:// | asyncio server and client on two sim hosts, ws:// only. | -| aiohttp (client) | 3.14.3 | fenced: simloop does not simulate 'sock_connect'; see docs/supported-api.md for the supported asyncio subset | ClientSession GET at a sim host answered by a raw stream server. | -| httpx | 0.28.1 | fails: AttributeError: 'NoneType' object has no attribute 'getpeername' | AsyncClient GET at a sim host answered by a raw stream server. | +| aiohttp (client) | 3.14.3 | works: ClientSession GET returned 'hello from the simulation' | ClientSession GET at a sim host answered by a raw stream server. | +| httpx | 0.28.1 | works: AsyncClient GET returned 'hello from the simulation' | AsyncClient GET at a sim host answered by a raw stream server. | -Rows are grouped: the libraries expected to run on the loop first, then the -client stacks expected to leave it. +Rows are grouped: the libraries that need nothing but the loop and its +streams first, then the client stacks that expect a socket object +underneath them. ## Reading the rows @@ -78,29 +79,44 @@ clients are built on — a length-prefixed request/response protocol on one long-lived connection — so the probe speaks RESP by hand against a small server on a second sim host, and the row claims no more than that. -**aiohttp's client** leaves the simulation at its first connection attempt. -The fence, verbatim: - -``` -simloop does not simulate 'sock_connect'; see docs/supported-api.md for the supported asyncio subset -``` - -Its connector resolves the name through `loop.getaddrinfo` (which the -simulation answers), then hands the addresses to `aiohappyeyeballs`, which -opens a real socket and calls `loop.sock_connect` on it. Raw sockets are -fenced, so the simulation stops there rather than letting a real connection -out. - -**httpx** never reaches a fence, and gets further than its verdict looks. -It goes through httpcore and anyio; anyio resolves the name (the simulated -resolver accepts the ASCII-encoded form anyio sends), connects through -`loop.create_connection`, and the connection *succeeds*. The failure is -introspection after the fact: httpcore asks the new stream for its local -and remote addresses, anyio answers by reading the raw socket object out of -`transport.get_extra_info("socket")` (`anyio/abc/_sockets.py`, -`extra_attributes`), and the simulation's honest answer for a transport -with no operating-system socket is `None`. Whether a request would complete -past that line is untested here. +**aiohttp's client** issues its GET and reads the body back. Its connector +resolves the name through `loop.getaddrinfo`, then hands the addresses to +`aiohappyeyeballs`, which creates a real `AF_INET` stream socket, calls +`loop.sock_connect` on it, and passes that socket to +`loop.create_connection(sock=...)`. The simulation answers the sequence +without letting the socket reach a network: `sock_connect` resolves the +target against the host table and records it, moving no packet and no +clock, and the `create_connection` call closes the real descriptor and +opens a simulated connection to the recorded address, paying the same +single round trip a direct `create_connection` would. The connector's +`setsockopt(TCP_NODELAY)` lands on the stand-in object +`get_extra_info("socket")` returns, which accepts option calls and does +nothing with them. + +**httpx** completes the same request through httpcore and anyio. anyio +resolves the name (the simulated resolver accepts the ASCII-encoded form +anyio sends) and connects through `loop.create_connection`, with no socket +of its own. The step that used to end this row comes next: httpcore asks +the new stream who it is connected to, and anyio answers by reading the +socket object out of `transport.get_extra_info("socket")` +(`anyio/abc/_sockets.py`, `extra_attributes`) and calling `getpeername()` +on it. A transport with no operating-system socket now answers with a +stand-in that reports the peer's synthetic address and port, so the +introspection succeeds and the response body comes back. + +Both client probes make one request against a responder that sends +`Connection: close`, so neither row says anything about connection reuse. +The piece a pool depends on is the descriptor `fileno()` returns: httpcore +polls it to decide whether a pooled connection has died, and the +simulation backs it with a parked descriptor the transport owns, which +stays unreadable while the peer is alive and becomes readable once the +peer's EOF or reset arrives. That contract is pinned by the test suite, +not by these rows. + +TLS is where both clients still stop. The two rows are `http://` only: a +request to an `https://` URL fences before the handshake starts, whether +the stack asks for it through `start_tls` or through +`create_connection(ssl=...)`. ## Not tested diff --git a/docs/supported-api.md b/docs/supported-api.md index f5c8360..32f845e 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -44,6 +44,8 @@ a host belong to an implicit `driver` host. | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary | | `loop.getnameinfo` | Reverse lookup: a synthetic address maps back to its host name, and a host name (what `get_extra_info("peername")` reports) maps to itself. `NI_NUMERICHOST` returns the address instead; services are always numeric | | `loop.net.address` / `hostname` | The mapping itself: every registered host owns one synthetic IPv4 address from `10.7.0.0/16` — `10.7.0.1`, `10.7.0.2`, ... handed out in registration order, starting with the implicit `driver` host | +| `loop.sock_connect` + `create_connection(sock=...)` | The two-call connect sequence aiohttp's connector performs. On an `AF_INET` stream socket, `sock_connect` places the target in the host table and records it against the socket — no packet moves and no virtual time passes, and a target the table cannot place raises `OSError` there. `create_connection(sock=...)` then claims that recorded address: it closes the real descriptor (the loop takes ownership, as the stdlib does) and opens a simulated connection, paying the same one round trip a direct connect pays, so a closed port raises `ConnectionRefusedError` from this call rather than the first. Passing a socket that no `sock_connect` on this loop parked raises `OSError`; passing `host`/`port` alongside `sock` raises `ValueError`. Binding a source address first — `TCPConnector(local_addr=...)` — is not supported: the connector binds the real socket before the simulation is consulted, and a synthetic address belongs to no real interface, so the bind fails outside the loop | +| `transport.get_extra_info("socket")` (streams) | A stand-in object, not a network socket: `family` / `type` / `proto` report `AF_INET` / `SOCK_STREAM` / `IPPROTO_TCP`, `getsockname()` and `getpeername()` return `(synthetic address, port)` tuples for the two ends, and `setsockopt`, `shutdown` and `close` are accepted and do nothing. `fileno()` returns a parked descriptor the transport owns, created on first call, that polls unreadable while the peer is alive and readable once the peer's EOF or reset arrives — which is how a pool that checks readability sees a dead connection. It is closed with the transport, and reports `-1` from then on. No bytes ever cross it: `recv`, `send` and anything else not listed above raise `AttributeError` rather than pretend. Datagram transports still report `None` | Names and their synthetic addresses are interchangeable wherever an endpoint is accepted, so a client can resolve a name and connect to what it got back. @@ -62,5 +64,14 @@ reliable by construction; and addressing is IPv4-only and entirely synthetic Anything that reaches outside the simulation raises `SimulationFenceError`: executors and threads (`run_in_executor`, `call_soon_threadsafe`), signal -handlers, subprocesses, raw sockets (`sock_*`), file-descriptor callbacks -(`add_reader` / `add_writer`), TLS upgrades, `sendfile`, and pipes. +handlers, subprocesses, file-descriptor callbacks (`add_reader` / +`add_writer`), TLS upgrades, `sendfile`, and pipes. + +The socket calls are fenced with one exception. `sock_connect` on an +`AF_INET` stream socket is simulated — it is how client stacks reach the +network, and the table above says what it does. Every other socket kind +fences there — datagram, raw and IPv6 alike — and the rest of the family +stays fenced outright: `sock_recv`, `sock_recv_into`, `sock_sendall`, +`sock_sendto`, `sock_recvfrom`, `sock_recvfrom_into`, `sock_accept` and +`sock_sendfile`. Client stacks do not need them: the socket they connect +is upgraded into a transport instead of being read and written directly. From 0f2c568ffbf474805d9f3376335f158027d6e537 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 12:01:21 +0530 Subject: [PATCH 07/14] Correct the socket fence in the honest limits --- README.md | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 34f8d84..0fc8aab 100644 --- a/README.md +++ b/README.md @@ -213,11 +213,15 @@ and the campaign results: ## Honest limits Code that goes through the event-loop API is supported; code that -bypasses it is fenced: threads and executors, raw sockets, subprocesses, -signals, and TLS raise `SimulationFenceError` rather than silently -breaking determinism. Name resolution stays inside the simulation: -`getaddrinfo` resolves sim host names to stable synthetic addresses and -raises `socket.gaierror` for anything else — no real DNS, ever. +bypasses it is fenced: threads and executors, raw socket reads and +writes, subprocesses, signals, and TLS raise `SimulationFenceError` +rather than silently breaking determinism. `sock_connect` on an +`AF_INET` stream socket is the exception — it is simulated, so a client +that connects a socket and hands it to `create_connection` runs, while +the datagram and raw variants still fence. Name resolution stays inside +the simulation: `getaddrinfo` resolves sim host names to stable synthetic +addresses and raises `socket.gaierror` for anything else — no real DNS, +ever. Write-side flow control is not simulated. The full contract is in [docs/supported-api.md](https://github.com/dhruvl/simloop/blob/main/docs/supported-api.md). What that contract costs real libraries — what aiohttp, anyio, websockets and From 604c633065be5bee9b40af9d69feaacf587dd42a Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 12:10:35 +0530 Subject: [PATCH 08/14] Say what really happens when a client asks for TLS --- docs/compatibility.md | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/docs/compatibility.md b/docs/compatibility.md index 02d3105..3257a84 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -113,18 +113,32 @@ stays unreadable while the peer is alive and becomes readable once the peer's EOF or reset arrives. That contract is pinned by the test suite, not by these rows. -TLS is where both clients still stop. The two rows are `http://` only: a -request to an `https://` URL fences before the handshake starts, whether -the stack asks for it through `start_tls` or through -`create_connection(ssl=...)`. +Both client rows are `http://` only, and the two stacks stop differently +on `https://`. aiohttp asks for TLS through `create_connection(ssl=...)`, +which fences: + +``` +simloop does not simulate 'create_connection(ssl=...)'; see docs/supported-api.md for the supported asyncio subset +``` + +httpx reaches no fence. httpcore wraps the byte stream with anyio's +`TLSStream`, which drives an `ssl` memory BIO inside the process and +sends the handshake as ordinary bytes over the simulated connection, so +`loop.start_tls` is never called and nothing stops the attempt. Where it +ends is up to whatever is listening: aimed at the plaintext responder +these probes use, the handshake goes unanswered and the request dies of +httpx's own `ConnectTimeout`. That is a one-off measurement rather than a +row — no probe on this page requests `https://`. ## Not tested - **asyncpg**: reaching its first fence needs a live PostgreSQL server to connect to, which no probe can provide; it is untested rather than fenced-or-not. -- **TLS anywhere**: `start_tls` is fenced, so `https://` and `wss://` are out - of scope for every probe on this page. +- **TLS anywhere**: no probe on this page requests `https://` or `wss://`. + simloop fences `start_tls` and `create_connection(ssl=...)`, but a stack + that runs its handshake in memory reaches neither — it reaches a simulated + network with nothing on it that speaks TLS. - Anything that reaches outside the loop by design — threads, executors, subprocesses, signals, real DNS. Those are fences, listed in [docs/supported-api.md](supported-api.md), not compatibility questions. From 3b0d0fa077fb73dff06212dc2b6f63d324ba2b55 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 12:14:51 +0530 Subject: [PATCH 09/14] Scope the TLS fence to loop-level upgrades --- README.md | 11 ++++++----- docs/compatibility.md | 6 +++--- docs/supported-api.md | 5 ++++- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 0fc8aab..3093ad9 100644 --- a/README.md +++ b/README.md @@ -214,11 +214,12 @@ and the campaign results: Code that goes through the event-loop API is supported; code that bypasses it is fenced: threads and executors, raw socket reads and -writes, subprocesses, signals, and TLS raise `SimulationFenceError` -rather than silently breaking determinism. `sock_connect` on an -`AF_INET` stream socket is the exception — it is simulated, so a client -that connects a socket and hands it to `create_connection` runs, while -the datagram and raw variants still fence. Name resolution stays inside +writes, subprocesses, signals, and loop-level TLS upgrades raise +`SimulationFenceError` rather than silently breaking determinism. +`sock_connect` on an `AF_INET` stream socket is the exception — it is +simulated, so a client that connects a socket and hands it to +`create_connection` runs, while the datagram and raw variants still +fence. Name resolution stays inside the simulation: `getaddrinfo` resolves sim host names to stable synthetic addresses and raises `socket.gaierror` for anything else — no real DNS, ever. diff --git a/docs/compatibility.md b/docs/compatibility.md index 3257a84..28db7d6 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -127,8 +127,8 @@ sends the handshake as ordinary bytes over the simulated connection, so `loop.start_tls` is never called and nothing stops the attempt. Where it ends is up to whatever is listening: aimed at the plaintext responder these probes use, the handshake goes unanswered and the request dies of -httpx's own `ConnectTimeout`. That is a one-off measurement rather than a -row — no probe on this page requests `https://`. +httpx's own `ConnectTimeout`. That timeout is a one-off measurement +rather than a row — no probe on this page requests `https://`. ## Not tested @@ -138,7 +138,7 @@ row — no probe on this page requests `https://`. - **TLS anywhere**: no probe on this page requests `https://` or `wss://`. simloop fences `start_tls` and `create_connection(ssl=...)`, but a stack that runs its handshake in memory reaches neither — it reaches a simulated - network with nothing on it that speaks TLS. + network with nothing on it that speaks TLS unless the test puts it there. - Anything that reaches outside the loop by design — threads, executors, subprocesses, signals, real DNS. Those are fences, listed in [docs/supported-api.md](supported-api.md), not compatibility questions. diff --git a/docs/supported-api.md b/docs/supported-api.md index 32f845e..39ba38d 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -65,7 +65,10 @@ reliable by construction; and addressing is IPv4-only and entirely synthetic Anything that reaches outside the simulation raises `SimulationFenceError`: executors and threads (`run_in_executor`, `call_soon_threadsafe`), signal handlers, subprocesses, file-descriptor callbacks (`add_reader` / -`add_writer`), TLS upgrades, `sendfile`, and pipes. +`add_writer`), loop-level TLS upgrades (`start_tls`, +`create_connection(ssl=...)`), `sendfile`, and pipes. TLS a library +performs in memory reaches no loop API and so reaches no fence; what that +means in practice is in [docs/compatibility.md](compatibility.md). The socket calls are fenced with one exception. `sock_connect` on an `AF_INET` stream socket is simulated — it is how client stacks reach the From 4dcde0c5f4c24cbe38e531f60055180f0fbb51d5 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 13:01:54 +0530 Subject: [PATCH 10/14] Close the parked descriptor outright when a reset arrives --- src/simloop/_transports.py | 25 ++++++++++++++++--------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index 1fb1dcc..8d13bef 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -133,12 +133,12 @@ def close(self) -> None: def fileno(self) -> int: # httpcore's is_socket_readable() polls this fd to decide whether a # pooled connection died. A parked socketpair end reads not-readable - # until its peer end is written to or closed, so the poll tracks the - # simulated connection: alive while the transport lives, readable - # (= discard me) once the peer is gone. -1 after teardown is what a - # closed real socket reports. Lazy: connections nobody introspects - # cost no descriptors, which matters at campaign scale. - if self._disposed or self._transport._closed: + # until its peer end is closed, so the poll answers exactly one + # question: has the peer's EOF arrived? A reset or a teardown closes + # both ends instead and reports -1, which the same poll reads as dead + # just as a closed real socket would. Lazy: connections nobody + # introspects cost no descriptors, which matters at campaign scale. + if self._disposed: return -1 if self._park is None: self._park = socket.socketpair() @@ -150,7 +150,7 @@ def fileno(self) -> int: return self._park[1].fileno() def _peer_gone(self) -> None: - # EOF or reset arrived: closing the held end makes the exposed end + # The peer's EOF arrived: closing the held end makes the exposed end # poll readable, exactly when a real kernel would report it. if self._park is not None: self._park[0].close() @@ -287,9 +287,11 @@ def _eof_arrived(self) -> None: self.close() def _reset_arrived(self) -> None: + # No _peer_gone() here: a reset tears the connection down on the spot, + # and _finish closes both parked ends. The descriptor goes to -1 + # rather than turning readable, which a liveness poll reads the same + # way — as a connection to discard. self._peer_closed = True - if self._extra_socket is not None: - self._extra_socket._peer_gone() if self._closed: return self._finish(ConnectionResetError("Connection reset by peer")) @@ -331,6 +333,11 @@ def get_extra_info(self, name: str, default: Any = None) -> Any: if name == "socket": if self._extra_socket is None: self._extra_socket = _SimSocket(self) + if self._closed: + # Asked for after teardown: born closed, like the socket a + # finished connection leaves behind. Nothing here may hand + # out a live descriptor for a connection that is gone. + self._extra_socket._dispose() return self._extra_socket return default From 04526ae4077469826ff872425afdf380a39074f9 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 13:01:54 +0530 Subject: [PATCH 11/14] Take descriptors back at close and vet the connect address --- src/simloop/_loop.py | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 083f7f2..f2ba3e2 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -10,7 +10,7 @@ import sys from array import array from asyncio import events -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Sequence from contextvars import Context from typing import TYPE_CHECKING, Any, NoReturn, TypeVarTuple, Unpack @@ -334,6 +334,14 @@ def close(self) -> None: if self._running: raise RuntimeError("cannot close a running event loop") self._sock_targets.clear() + # Connections still open at the end of a run — what a pooling client + # leaves behind — hold their liveness descriptors inside the socket + # object's reference cycle with the transport, so the descriptors + # would only come back when the cycle collector ran. Nothing can poll + # them once the loop is closed, so release them here instead. + for transport in self._net._streams.values(): + if transport._extra_socket is not None: + transport._extra_socket._dispose() self._closed = True def _check_closed(self) -> None: @@ -616,7 +624,25 @@ async def sock_connect(self, sock: Any, address: Any) -> None: getattr(sock, "family", None) != socket.AF_INET or getattr(sock, "type", None) != socket.SOCK_STREAM ): - _fence("sock_connect (only AF_INET stream sockets are simulated)") + raise SimulationFenceError( + "simloop does not simulate 'sock_connect' for anything but " + "AF_INET stream sockets; see docs/supported-api.md for the " + "supported asyncio subset" + ) + if ( + isinstance(address, (str, bytes, bytearray)) + or not isinstance(address, Sequence) + or len(address) != 2 + or not isinstance(address[0], (str, bytes)) + or isinstance(address[1], bool) + or not isinstance(address[1], int) + ): + # Caught here rather than deeper: a bare host string would be read + # character by character, and a string port would surface much + # later as a confusing error about something else entirely. + raise OSError( + f"sock_connect needs an AF_INET (host, port) address, got {address!r}" + ) host, port = address[0], address[1] self._net._resolve(host) # unknown targets fail here, loudly # No packet moves and no time passes yet: the connection handshake From c2b3d4c98760d5f627b15e75018b03842eebf96f Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 13:32:00 +0530 Subject: [PATCH 12/14] Tighten the socket tests around teardown and determinism Every connection the tests open now gets closed and settled before the loop goes down, so the file runs clean under -W error::ResourceWarning. Adds the reset path, the descriptors a run leaves behind, the malformed connect addresses, and a trace-hash pair over the whole client sequence. --- tests/test_client_sockets.py | 405 ++++++++++++++++++++++++++--------- 1 file changed, 306 insertions(+), 99 deletions(-) diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py index 9f66001..af5965c 100644 --- a/tests/test_client_sockets.py +++ b/tests/test_client_sockets.py @@ -8,7 +8,9 @@ from __future__ import annotations import asyncio +import gc import ipaddress +import os import select import socket from collections.abc import Callable, Coroutine @@ -31,12 +33,24 @@ async def _hold_open(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) writer.close() -def _connected_pair(loop: SimLoop, handler: Any = _hold_open) -> tuple[Any, Any]: - """One established client connection; returns (client_writer, server).""" +def _connected_pair(loop: SimLoop, handler: Any = _hold_open) -> tuple[Any, Any, Any]: + """One established client connection. - async def main() -> tuple[Any, Any]: + Returns (client_writer, accepted_writer, server): the two ends of the same + connection plus the listening server. + """ + + accepted: list[Any] = [] + + async def accept( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + accepted.append(writer) + await handler(reader, writer) + + async def main() -> tuple[Any, Any, Any]: server = await loop.net.host("server").create_task( - asyncio.start_server(handler, "0.0.0.0", 9000) + asyncio.start_server(accept, "0.0.0.0", 9000) ) writer_box: list[Any] = [] @@ -45,56 +59,78 @@ async def connect() -> None: writer_box.append(writer) await loop.net.host("client").create_task(connect()) - return writer_box[0], server + return writer_box[0], accepted[0], server + + trio: tuple[Any, Any, Any] = loop.run_until_complete(main()) + return trio - pair: tuple[Any, Any] = loop.run_until_complete(main()) - return pair + +def _settle(loop: SimLoop) -> None: + """Let queued closes cross the simulated network before teardown.""" + loop.run_until_complete(asyncio.sleep(1.0)) def test_stream_transport_exposes_a_socket_object() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - assert sock is not None - assert sock is writer.transport.get_extra_info("socket") # stable singleton - assert sock.family == socket.AF_INET - assert sock.type == socket.SOCK_STREAM - assert sock.proto == socket.IPPROTO_TCP - server.close() - loop.close() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock is not None + assert sock is writer.transport.get_extra_info("socket") # stable singleton + assert sock.family == socket.AF_INET + assert sock.type == socket.SOCK_STREAM + assert sock.proto == socket.IPPROTO_TCP + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_socket_addresses_are_synthetic_ip_tuples() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - local_ip, local_port = sock.getsockname() - peer_ip, peer_port = sock.getpeername() - # Plausible AF_INET sockaddrs: ip_address() must accept them. - ipaddress.ip_address(local_ip) - ipaddress.ip_address(peer_ip) - assert local_ip == loop.net.address("client") - assert peer_ip == loop.net.address("server") - assert peer_port == 9000 - # Same endpoints the transport itself reports, by name. - assert writer.transport.get_extra_info("peername") == ("server", 9000) - assert writer.transport.get_extra_info("sockname")[1] == local_port - server.close() - loop.close() + try: + writer, accepted, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + local_ip, local_port = sock.getsockname() + peer_ip, peer_port = sock.getpeername() + # Plausible AF_INET sockaddrs: ip_address() must accept them. + ipaddress.ip_address(local_ip) + ipaddress.ip_address(peer_ip) + assert local_ip == loop.net.address("client") + assert peer_ip == loop.net.address("server") + assert peer_port == 9000 + # Same endpoints the transport itself reports, by name. + assert writer.transport.get_extra_info("peername") == ("server", 9000) + assert writer.transport.get_extra_info("sockname")[1] == local_port + # The accepting side sees the client from the other direction: the two + # fake sockets describe one connection, not two unrelated ones. + accepted_sock = accepted.transport.get_extra_info("socket") + assert accepted_sock.getpeername() == (local_ip, local_port) + assert accepted_sock.getsockname() == (peer_ip, 9000) + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_socket_options_and_teardown_calls_are_inert() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - assert sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) is None - assert sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) is None - assert sock.shutdown(socket.SHUT_RDWR) is None - assert sock.close() is None - with pytest.raises(AttributeError): - sock.recv(1) # no pretending to be readable - server.close() - loop.close() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) is None + assert sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) is None + assert sock.shutdown(socket.SHUT_RDWR) is None + assert sock.close() is None + with pytest.raises(AttributeError): + sock.recv(1) # no pretending to be readable + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_datagram_transports_still_answer_none() -> None: @@ -109,8 +145,10 @@ async def main() -> Any: finally: transport.close() - assert loop.run_until_complete(main()) is None - loop.close() + try: + assert loop.run_until_complete(main()) is None + finally: + loop.close() def _readable(fd: int) -> bool: @@ -123,17 +161,19 @@ def _readable(fd: int) -> bool: def test_fileno_is_lazy_and_not_readable_while_live() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - assert sock._park is None # no kernel object until someone asks - fd = sock.fileno() - assert fd >= 0 - assert fd == sock.fileno() # stable - assert not _readable(fd) # live connection: httpcore keeps pooling it - server.close() - writer.close() - loop.run_until_complete(asyncio.sleep(1.0)) - loop.close() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock._park is None # no kernel object until someone asks + fd = sock.fileno() + assert fd >= 0 + assert fd == sock.fileno() # stable + assert not _readable(fd) # live connection: httpcore keeps pooling it + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_fd_turns_readable_when_the_peer_closes() -> None: @@ -145,17 +185,19 @@ async def close_after_a_beat( await asyncio.sleep(0.5) # after the client has taken its fd writer.close() - writer, server = _connected_pair(loop, close_after_a_beat) - sock = writer.transport.get_extra_info("socket") - fd = sock.fileno() - assert not _readable(fd) - # Let the FIN cross the simulated network. - loop.run_until_complete(asyncio.sleep(1.0)) - assert _readable(fd) # httpcore now sees the connection as expired - server.close() - writer.close() - loop.run_until_complete(asyncio.sleep(1.0)) - loop.close() + try: + writer, _, server = _connected_pair(loop, close_after_a_beat) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + # Let the FIN cross the simulated network. + _settle(loop) + assert _readable(fd) # httpcore now sees the connection as expired + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_fd_born_readable_when_the_peer_left_first() -> None: @@ -167,40 +209,72 @@ async def close_immediately( ) -> None: writer.close() - writer, server = _connected_pair(loop, close_immediately) - loop.run_until_complete(asyncio.sleep(1.0)) # FIN already delivered - sock = writer.transport.get_extra_info("socket") - assert _readable(sock.fileno()) # never claim a dead connection is live - server.close() - writer.close() - loop.run_until_complete(asyncio.sleep(1.0)) - loop.close() + try: + writer, _, server = _connected_pair(loop, close_immediately) + _settle(loop) # FIN already delivered + sock = writer.transport.get_extra_info("socket") + assert sock.fileno() >= 0 # the local end is still open: a real fd + assert _readable(sock.fileno()) # never claim a dead connection is live + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_fd_lifecycle_ends_with_the_transport() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - fd = sock.fileno() - assert not _readable(fd) - writer.close() - loop.run_until_complete(asyncio.sleep(1.0)) - assert sock._park is None # both ends closed, nothing leaked - assert sock.fileno() == -1 # closed-socket semantics - server.close() - loop.close() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + writer.close() + _settle(loop) + assert sock._park is None # both ends closed, nothing leaked + assert sock.fileno() == -1 # closed-socket semantics + server.close() + finally: + loop.close() + + +def test_reset_closes_the_descriptor_instead_of_arming_it() -> None: + """An aborting peer tears the connection down; the fd goes, not readable.""" + loop = _network() + + async def abort_after_a_beat( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + await asyncio.sleep(0.5) # after the client has taken its fd + writer.transport.abort() + + try: + writer, _, server = _connected_pair(loop, abort_after_a_beat) + sock = writer.transport.get_extra_info("socket") + assert sock.fileno() >= 0 + _settle(loop) # let the RST cross the network + assert sock.fileno() == -1 # the reset finished the connection outright + assert sock._park is None + assert _readable(sock.fileno()) # a poll still reads it as dead + server.close() + writer.close() + _settle(loop) + finally: + loop.close() def test_fileno_after_teardown_never_creates_a_descriptor() -> None: loop = _network() - writer, server = _connected_pair(loop) - sock = writer.transport.get_extra_info("socket") - writer.close() - loop.run_until_complete(asyncio.sleep(1.0)) - assert sock.fileno() == -1 - assert sock._park is None - server.close() - loop.close() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + writer.close() + _settle(loop) + assert sock.fileno() == -1 + assert sock._park is None + server.close() + finally: + loop.close() def test_sock_connect_records_a_stream_socket_target() -> None: @@ -234,13 +308,36 @@ async def main() -> None: loop.close() +@pytest.mark.parametrize( + "address", + [ + "server", # a bare string is a sequence of characters, not an endpoint + ("server",), # one item short + ("server", "9000"), # a port that only looks like a number + ], +) +def test_sock_connect_rejects_malformed_addresses(address: Any) -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="host, port"): + await loop.sock_connect(sock, address) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + def test_sock_connect_still_fences_datagram_sockets() -> None: loop = _network() async def main() -> None: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: - with pytest.raises(SimulationFenceError): + with pytest.raises(SimulationFenceError, match="sock_connect"): await loop.sock_connect(sock, (loop.net.address("server"), 9000)) finally: sock.close() @@ -266,7 +363,7 @@ async def main() -> float: def _aiohttp_style_connect( - loop: SimLoop, port: int = 9000 + loop: SimLoop, port: int = 9000, sockets: list[Any] | None = None ) -> Callable[[], Coroutine[Any, Any, tuple[Any, Any, Any]]]: """The exact two-call sequence aiohttp + aiohappyeyeballs performs.""" @@ -275,6 +372,8 @@ async def connect() -> tuple[Any, Any, Any]: family, kind, proto, _, address = infos[0] sock = socket.socket(family=family, type=kind, proto=proto) sock.setblocking(False) + if sockets is not None: + sockets.append(sock) # so a refused connect can be inspected too await loop.sock_connect(sock, address) transport, protocol = await loop.create_connection( asyncio.Protocol, ssl=None, server_hostname=None, sock=sock @@ -288,7 +387,7 @@ def test_parked_socket_upgrades_to_a_sim_connection() -> None: loop = _network() loop.net.set_defaults(latency=(0.001, 0.001)) # so the round trip is visible - async def main() -> tuple[Any, Any]: + async def main() -> tuple[Any, float]: server = await loop.net.host("server").create_task( asyncio.start_server(_hold_open, "0.0.0.0", 9000) ) @@ -302,9 +401,9 @@ async def main() -> tuple[Any, Any]: return sock, elapsed sock, elapsed = loop.run_until_complete(main()) - loop.run_until_complete(asyncio.sleep(1.0)) # let both ends finish closing + _settle(loop) # let both ends finish closing assert sock.fileno() == -1 # the loop took ownership and closed it - assert elapsed > 0.0 # the handshake cost its round trip + assert elapsed == pytest.approx(0.002) # SYN out, accept back: one round trip assert not loop._sock_targets # the parked entry was claimed loop.close() @@ -312,14 +411,20 @@ async def main() -> tuple[Any, Any]: def test_refused_port_raises_from_the_upgrade() -> None: loop = _network() loop.net.set_defaults(latency=(0.001, 0.001)) + sockets: list[Any] = [] - async def main() -> None: + async def main() -> float: + started = loop.time() with pytest.raises(ConnectionRefusedError): await loop.net.host("client").create_task( - _aiohttp_style_connect(loop, port=9999)() + _aiohttp_style_connect(loop, port=9999, sockets=sockets)() ) + return loop.time() - started - loop.run_until_complete(main()) + elapsed = loop.run_until_complete(main()) + # Refusal costs what a real one does: SYN out, refusal back. + assert elapsed == pytest.approx(0.002) + assert sockets[0].fileno() == -1 # the loop owned it and closed it anyway loop.close() @@ -354,3 +459,105 @@ async def main() -> None: loop.run_until_complete(main()) loop.close() + + +def test_closing_the_loop_releases_parked_descriptors() -> None: + """A pooling client leaves its connections open; the run still ends clean.""" + loop = _network() + + async def main() -> tuple[Any, Any, Any]: + # A bare protocol server, not start_server: the connection has to + # survive to the end of the run, and a stream handler parked on the + # other side of it would be a leak of its own. + server = await loop.net.host("server").create_task( + loop.create_server(asyncio.Protocol, "0.0.0.0", 9000) + ) + _, transport, _ = await loop.net.host("client").create_task( + _aiohttp_style_connect(loop)() + ) + return server, transport, transport.get_extra_info("socket") + + server, transport, sock = loop.run_until_complete(main()) + fd = sock.fileno() + assert fd >= 0 + server.close() + loop.close() # the connection is still open, exactly as a pool leaves it + assert sock._park is None + with pytest.raises(OSError): + os.fstat(fd) # the descriptor went back to the OS, not just its owner + gc.collect() # under -W error::ResourceWarning: nothing left to complain + assert transport.get_extra_info("socket") is sock + + +class _Collect(asyncio.Protocol): + """A protocol that keeps what it was sent, so replies are observable.""" + + def __init__(self) -> None: + self.data = bytearray() + + def data_received(self, data: bytes) -> None: + self.data += data + + +async def _echo_once( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter +) -> None: + writer.write(await reader.read(64)) + await writer.drain() + writer.close() + + +async def _client_stack_probe(loop: SimLoop, name: str) -> None: + """One aiohttp-shaped request: resolve, connect, introspect, write, close.""" + infos = await loop.getaddrinfo("server", 9000, type=socket.SOCK_STREAM) + family, kind, proto, _, address = infos[0] + sock = socket.socket(family=family, type=kind, proto=proto) + sock.setblocking(False) + await loop.sock_connect(sock, address) + transport, _ = await loop.create_connection(_Collect, sock=sock) + fake = transport.get_extra_info("socket") + fake.getpeername() + fake.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) + fake.fileno() # the descriptor a pool would poll + transport.write(name.encode()) + await asyncio.sleep(0.05) + transport.close() + + +def _run_client_stack(seed: int) -> str: + """Three concurrent clients driving the whole new surface; trace hash out.""" + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.set_defaults(latency=(0.001, 0.005)) + + async def main() -> None: + server = await loop.net.host("server").create_task( + asyncio.start_server(_echo_once, "0.0.0.0", 9000) + ) + clients = [ + loop.net.host(name).create_task(_client_stack_probe(loop, name)) + for name in ("alice", "bob", "carol") + ] + for client in clients: + await client + server.close() + + try: + loop.run_until_complete(main()) + loop.run_until_complete(asyncio.sleep(1.0)) + return loop.trace_hash() + finally: + loop.close() + + +def test_client_stack_path_is_deterministic() -> None: + for seed in range(3): + hashes = {_run_client_stack(seed) for _ in range(2)} + assert len(hashes) == 1, f"seed {seed} produced diverging traces" + + +def test_client_stack_path_still_varies_with_the_seed() -> None: + # The fake socket's descriptors are OS-assigned and deliberately absent + # from the trace, so only the schedule and the latency draws move. + hashes = {_run_client_stack(seed) for seed in range(10)} + assert len(hashes) >= 8 From 9192a40da499d321df52a61139ffdccf772c782f Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 13:32:09 +0530 Subject: [PATCH 13/14] Describe both ways the liveness descriptor reads dead The descriptor only turns readable on the peer's EOF; a reset, a local close or the end of the run closes it and fileno() reports -1, which a readability poll reads as dead just the same. Also notes that the number is OS-assigned and outside the determinism guarantee. --- docs/compatibility.md | 5 +++-- docs/supported-api.md | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/compatibility.md b/docs/compatibility.md index 28db7d6..0ec7b74 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -110,8 +110,9 @@ The piece a pool depends on is the descriptor `fileno()` returns: httpcore polls it to decide whether a pooled connection has died, and the simulation backs it with a parked descriptor the transport owns, which stays unreadable while the peer is alive and becomes readable once the -peer's EOF or reset arrives. That contract is pinned by the test suite, -not by these rows. +peer's EOF arrives; a reset or a teardown closes it and `fileno()` returns +`-1`, which the same poll reads as dead just as well. That contract is +pinned by the test suite, not by these rows. Both client rows are `http://` only, and the two stacks stop differently on `https://`. aiohttp asks for TLS through `create_connection(ssl=...)`, diff --git a/docs/supported-api.md b/docs/supported-api.md index 39ba38d..5583a04 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -44,8 +44,8 @@ a host belong to an implicit `driver` host. | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary | | `loop.getnameinfo` | Reverse lookup: a synthetic address maps back to its host name, and a host name (what `get_extra_info("peername")` reports) maps to itself. `NI_NUMERICHOST` returns the address instead; services are always numeric | | `loop.net.address` / `hostname` | The mapping itself: every registered host owns one synthetic IPv4 address from `10.7.0.0/16` — `10.7.0.1`, `10.7.0.2`, ... handed out in registration order, starting with the implicit `driver` host | -| `loop.sock_connect` + `create_connection(sock=...)` | The two-call connect sequence aiohttp's connector performs. On an `AF_INET` stream socket, `sock_connect` places the target in the host table and records it against the socket — no packet moves and no virtual time passes, and a target the table cannot place raises `OSError` there. `create_connection(sock=...)` then claims that recorded address: it closes the real descriptor (the loop takes ownership, as the stdlib does) and opens a simulated connection, paying the same one round trip a direct connect pays, so a closed port raises `ConnectionRefusedError` from this call rather than the first. Passing a socket that no `sock_connect` on this loop parked raises `OSError`; passing `host`/`port` alongside `sock` raises `ValueError`. Binding a source address first — `TCPConnector(local_addr=...)` — is not supported: the connector binds the real socket before the simulation is consulted, and a synthetic address belongs to no real interface, so the bind fails outside the loop | -| `transport.get_extra_info("socket")` (streams) | A stand-in object, not a network socket: `family` / `type` / `proto` report `AF_INET` / `SOCK_STREAM` / `IPPROTO_TCP`, `getsockname()` and `getpeername()` return `(synthetic address, port)` tuples for the two ends, and `setsockopt`, `shutdown` and `close` are accepted and do nothing. `fileno()` returns a parked descriptor the transport owns, created on first call, that polls unreadable while the peer is alive and readable once the peer's EOF or reset arrives — which is how a pool that checks readability sees a dead connection. It is closed with the transport, and reports `-1` from then on. No bytes ever cross it: `recv`, `send` and anything else not listed above raise `AttributeError` rather than pretend. Datagram transports still report `None` | +| `loop.sock_connect` + `create_connection(sock=...)` | The two-call connect sequence aiohttp's connector performs. On an `AF_INET` stream socket, `sock_connect` places the target in the host table and records it against the socket — no packet moves and no virtual time passes, and an address that is not a `(host, port)` pair, or a target the host table cannot place, raises `OSError` there. `create_connection(sock=...)` then claims that recorded address: it closes the real descriptor (the loop takes ownership, as the stdlib does) and opens a simulated connection, paying the same one round trip a direct connect pays, so a closed port raises `ConnectionRefusedError` from this call rather than the first. Passing a socket that no `sock_connect` on this loop parked raises `OSError`; passing `host`/`port` alongside `sock` raises `ValueError`. Binding a source address first — `TCPConnector(local_addr=...)` — is not supported: the connector binds the real socket before the simulation is consulted, and a synthetic address belongs to no real interface, so the bind fails outside the loop | +| `transport.get_extra_info("socket")` (streams) | A stand-in object, not a network socket: `family` / `type` / `proto` report `AF_INET` / `SOCK_STREAM` / `IPPROTO_TCP`, `getsockname()` and `getpeername()` return `(synthetic address, port)` tuples for the two ends, and `setsockopt`, `shutdown` and `close` are accepted and do nothing. `fileno()` returns a parked descriptor the transport owns, created on first call, that polls unreadable while the peer is alive and readable once the peer's EOF arrives — which is how a pool that checks readability sees a dead connection. A reset, a local close or the end of the run closes the descriptor instead, and `fileno()` reports `-1` from then on; a readability poll reads that as dead too. The number itself is assigned by the operating system, so it is not reproducible across runs and is outside the determinism guarantee — nothing in a trace depends on it. No bytes ever cross it: `recv`, `send` and anything else not listed above raise `AttributeError` rather than pretend. Datagram transports still report `None` | Names and their synthetic addresses are interchangeable wherever an endpoint is accepted, so a client can resolve a name and connect to what it got back. From 11bcba2462c45dab14cb0eafbce382b20d7df8d0 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sat, 1 Aug 2026 13:32:09 +0530 Subject: [PATCH 14/14] Re-wrap the honest-limits paragraph --- README.md | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 3093ad9..b458766 100644 --- a/README.md +++ b/README.md @@ -218,11 +218,10 @@ writes, subprocesses, signals, and loop-level TLS upgrades raise `SimulationFenceError` rather than silently breaking determinism. `sock_connect` on an `AF_INET` stream socket is the exception — it is simulated, so a client that connects a socket and hands it to -`create_connection` runs, while the datagram and raw variants still -fence. Name resolution stays inside -the simulation: `getaddrinfo` resolves sim host names to stable synthetic -addresses and raises `socket.gaierror` for anything else — no real DNS, -ever. +`create_connection` runs, while the datagram and raw variants still fence. +Name resolution stays inside the simulation: `getaddrinfo` resolves sim +host names to stable synthetic addresses and raises `socket.gaierror` for +anything else — no real DNS, ever. Write-side flow control is not simulated. The full contract is in [docs/supported-api.md](https://github.com/dhruvl/simloop/blob/main/docs/supported-api.md). What that contract costs real libraries — what aiohttp, anyio, websockets and