Let a host connect to its own listener

CHANGELOG.md

@@ -113,6 +113,13 @@
   threads are real ones spawned through no loop API — a real thread racing
   a virtual clock ends in the cross-thread fence, a hang, or the caller's
   own timeout, whichever the race picks.
+- A host can connect to its own listener. The stream registry used to key
+  each connection end by host alone, so the two ends of a self-connection
+  collapsed onto one entry and the run deadlocked; the key now carries the
+  end's own port, the connect handshake answers to the connector's port
+  rather than the listener's, and a loopback connect to a closed port is
+  refused instead of hanging. Packets between distinct hosts are keyed,
+  ordered and traced exactly as before, so existing hashes do not move.
 - `server.sockets` on a simulated server answers with an empty tuple
   instead of not existing, which is all aiohttp's `web.TCPSite` and
   websockets' `serve()` need to start; both now run their documented

src/simloop/_net.py

@@ -413,8 +413,11 @@ def __init__(self, loop: SimLoop) -> None:
         self._held: list[_Packet] = []
         self._datagrams: dict[tuple[str, int], _SimDatagramTransport] = {}
         self._listeners: dict[tuple[str, int], _Listener] = {}
-        self._streams: dict[tuple[int, str], _SimStreamTransport] = {}
-        self._inbound: dict[tuple[int, str], _InOrder] = {}
+        # Keyed by (conn, host, local port): a host can connect to its own
+        # listener, and without the port the two ends of that connection
+        # would collapse onto one key and overwrite each other.
+        self._streams: dict[tuple[int, str, int], _SimStreamTransport] = {}
+        self._inbound: dict[tuple[int, str, int], _InOrder] = {}
         self._pending: dict[int, _Connect] = {}
         self._next_conn = 0
         self._next_uid = 0
@@ -730,7 +733,7 @@ def _deliver(self, packet: _Packet) -> None:
             _current_host.reset(token)
 
     def _dispatch_stream(self, packet: _Packet) -> None:
-        key = (packet.conn, packet.dst)
+        key = (packet.conn, packet.dst, packet.dst_port)
         queue = self._inbound.get(key)
         if queue is None:
             queue = self._inbound[key] = _InOrder(self)
@@ -753,18 +756,18 @@ def _dispatch_ready(self, packet: _Packet) -> None:
                 client = _SimStreamTransport(
                     self, packet.conn, local=connect.local, remote=connect.remote
                 )
-                self._streams[(packet.conn, connect.local[0])] = client
+                self._streams[
+                    (packet.conn, connect.local[0], connect.local[1])
+                ] = client
                 protocol = connect.factory()
                 client._begin(protocol)
                 connect.fut.set_result((client, protocol))
             else:
                 connect.fut.set_exception(
-                    ConnectionRefusedError(
-                        f"connect to ({packet.src!r}, {packet.dst_port}) refused"
-                    )
+                    ConnectionRefusedError(f"connect to {connect.remote!r} refused")
                 )
             return
-        transport = self._streams.get((packet.conn, packet.dst))
+        transport = self._streams.get((packet.conn, packet.dst, packet.dst_port))
         if transport is None:
             return  # connection already torn down locally
         if packet.kind == "data":
@@ -783,7 +786,8 @@ def _handle_syn(self, packet: _Packet) -> None:
                 dst=packet.src,
                 conn=packet.conn,
                 seq=0,
-                dst_port=packet.dst_port,
+                src_port=packet.dst_port,
+                dst_port=packet.src_port,
             )
             return
         transport = _SimStreamTransport(
@@ -792,12 +796,21 @@ def _handle_syn(self, packet: _Packet) -> None:
             local=(packet.dst, packet.dst_port),
             remote=(packet.src, packet.src_port),
         )
-        self._streams[(packet.conn, packet.dst)] = transport
+        self._streams[(packet.conn, packet.dst, packet.dst_port)] = transport
         # The accept is seq 0 of the server-to-client direction, so any data
         # the protocol writes from connection_made (seq 1+) can never arrive
         # ahead of the accept, whatever the latency draws say.
+        # The accept (and the refusal above) answers to the connector's own
+        # port: the answer belongs to the client end's inbound direction, and
+        # on a self-connection that direction is told apart by port alone.
         self._send_stream(
-            kind="accept", src=packet.dst, dst=packet.src, conn=packet.conn, seq=0
+            kind="accept",
+            src=packet.dst,
+            dst=packet.src,
+            conn=packet.conn,
+            seq=0,
+            src_port=packet.dst_port,
+            dst_port=packet.src_port,
         )
         protocol = listener.factory()
         transport._begin(protocol)
@@ -828,8 +841,8 @@ def _send_stream(
             )
         )
 
-    def _drop_stream(self, conn: int, host: str) -> None:
-        self._streams.pop((conn, host), None)
+    def _drop_stream(self, conn: int, host: str, port: int) -> None:
+        self._streams.pop((conn, host, port), None)
 
     async def _open_connection(
         self, protocol_factory: Any, host: Any, port: Any
@@ -842,6 +855,10 @@ async def _open_connection(
         conn = self._next_conn
         self._next_conn += 1
         src_port = self._ephemeral()
+        if host == src and src_port == port:
+            # A self-connection whose ends share a port would collapse onto
+            # one registry key; the next ephemeral number cannot collide.
+            src_port = self._ephemeral()
         fut: asyncio.Future[tuple[_SimStreamTransport, Any]] = (
             self._loop.create_future()
         )

src/simloop/_transports.py

@@ -211,6 +211,11 @@ def _send(self, kind: str, payload: bytes = b"") -> None:
             conn=self._conn,
             seq=seq,
             payload=payload,
+            # The ports address the peer's end of this connection: on a
+            # self-connection the two ends share a host and only the port
+            # tells the registry which one a packet is for.
+            src_port=self._local[1],
+            dst_port=self._remote[1],
         )
 
     def write(self, data: Any) -> None:
@@ -254,7 +259,7 @@ def _finish(self, exc: Exception | None) -> None:
             return
         self._closed = True
         self._closing = True
-        self._net._drop_stream(self._conn, self._local[0])
+        self._net._drop_stream(self._conn, self._local[0], self._local[1])
         if self._extra_socket is not None:
             self._extra_socket._dispose()
         protocol, self._protocol = self._protocol, None

tests/test_streams.py

@@ -127,6 +127,57 @@ async def main() -> float:
     assert elapsed == pytest.approx(0.1)  # syn there + refusal back
 
 
+def test_a_host_can_connect_to_its_own_listener() -> None:
+    # Both ends of this connection live on one machine, so the stream
+    # registry must tell them apart by more than the host name — the
+    # client's ephemeral port against the listener's port is what does it.
+    loop = SimLoop(seed=0)
+    loop.net.host("solo")
+
+    async def handle(
+        reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        await reader.readline()
+        writer.write(b"pong\n")
+        await writer.drain()
+        writer.close()
+
+    async def main() -> bytes:
+        server = await asyncio.start_server(handle, "0.0.0.0", 9000)
+        reader, writer = await asyncio.open_connection("solo", 9000)
+        writer.write(b"ping\n")
+        await writer.drain()
+        reply = await reader.readline()
+        writer.close()
+        await writer.wait_closed()
+        server.close()
+        return reply
+
+    try:
+        reply = loop.run_until_complete(loop.net.host("solo").create_task(main()))
+    finally:
+        loop.close()
+    assert reply == b"pong\n"
+    assert not loop.net._streams  # both ends existed, and both were torn down
+
+
+def test_a_loopback_connect_to_a_closed_port_is_refused() -> None:
+    # The refusal answers to the connector's own port. Addressed to the
+    # listener's port instead, it would land on the queue the syn already
+    # advanced and wait there forever — a deadlock instead of an error.
+    loop = SimLoop(seed=0)
+    loop.net.host("solo")
+
+    async def main() -> None:
+        with pytest.raises(ConnectionRefusedError):
+            await asyncio.open_connection("solo", 9999)
+
+    try:
+        loop.run_until_complete(loop.net.host("solo").create_task(main()))
+    finally:
+        loop.close()
+
+
 def test_bytes_arrive_complete_and_in_order_under_latency_chaos() -> None:
     loop = _network(seed=5)
     loop.net.set_defaults(latency=(0.001, 0.2))