Add a simulated network

README.md

@@ -1,7 +1,7 @@
 # simloop
 
 Deterministic simulation testing for Python asyncio — seeded task scheduling,
-virtual time, and (planned) a simulated network with fault injection. Any failure
+virtual time, and a simulated network with fault injection. Any failure
 it finds replays exactly from a seed.
 
 Rust has [madsim](https://github.com/madsim-rs/madsim) and
@@ -17,16 +17,15 @@ a test that deadlocks *every* time at seed 4217, in milliseconds, under a debugg
 virtual time, replay-proving trace hashes, deadlock detection, and seeded
 `sim.random` / `sim.uuid4` / `sim.time` shims, with the stdlib coordination
 primitives (`Queue`, `Event`, `Lock`, `gather`, `TaskGroup`, `timeout`)
-running unchanged on top. No simulated network yet. See
+running unchanged on top. The simulated network is in: `asyncio.open_connection`,
+`start_server`, and datagram endpoints run unchanged over in-memory transports
+with seeded latency, drops, duplication, partitions, and host crashes. See
 [the supported subset](docs/supported-api.md) for the exact contract.
 
 ## Planned
 
 - **SimLoop** — a deterministic `asyncio` event loop: virtual clock, seeded
   ready-queue ordering, append-only scheduling trace with replay-proving hashes.
-- **SimNetwork** — in-memory transports behind the standard asyncio streams and
-  protocol APIs, with seeded fault injection: latency, drops, duplication,
-  reordering, and network partitions.
 - **pytest plugin** — `@sim_test(seeds=1000)` to explore schedules, and an exact
   replay flag for any failing seed.
 

docs/supported-api.md

@@ -26,11 +26,27 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of
 — everything built purely on futures, tasks and timers. Timeouts fire in
 virtual time. Each of these claims is exercised by the test suite.
 
-## Planned: simulated network
+## Simulated network
 
-`create_connection`, `create_server`, `create_datagram_endpoint` and the
-streams API, backed by in-memory transports with seeded fault injection
-(latency, drops, duplication, reordering, partitions).
+Networking runs over an in-memory packet layer with seeded fault injection.
+Hosts are named machines; a task started via `loop.net.host("name").create_task(...)`
+— and every task it spawns — belongs to that host. Tasks never started under
+a host belong to an implicit `driver` host.
+
+| API | Behavior under simulation |
+|---|---|
+| `loop.create_connection` / `create_server`, `asyncio.open_connection` / `start_server` | Real transports and protocols over reliable, ordered in-memory streams; connecting costs one round trip of virtual latency; connecting to a closed port raises `ConnectionRefusedError` |
+| `loop.create_datagram_endpoint` | Unreliable messaging: per-link drop, duplication, and latency apply per datagram |
+| `loop.net.set_defaults` / `set_link` | Per-direction latency ranges, drop and duplication probabilities, drawn from a seed-derived stream |
+| `loop.net.partition` / `heal` | Silent blackhole: datagrams are lost, stream traffic is held and resumes intact after healing; nothing errors — only your own timeouts fire |
+| `loop.net.crash` | A host's tasks are cancelled and it goes silent; no reset is sent — peers cannot tell a crash from a partition |
+| `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
+
+Limitations, stated honestly: write-side flow control is not simulated
+(`drain()` never blocks, write buffers are unbounded, the peer cannot pause
+your writes); there is no retransmission or congestion model — streams are
+reliable by construction; addresses are host names, not IPs, and
+`getaddrinfo` stays fenced.
 
 ## Fenced
 

src/simloop/__init__.py

@@ -1,14 +1,17 @@
 """simloop — deterministic simulation testing for Python asyncio."""
 
 from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError
+from simloop._net import Host, SimNetwork
 from simloop._sim import Sim, sim
 from simloop._trace import TraceEvent
 
 __version__ = "0.0.1.dev0"
 
 __all__ = [
+    "Host",
     "Sim",
     "SimLoop",
+    "SimNetwork",
     "SimulationDeadlockError",
     "SimulationFenceError",
     "TraceEvent",

src/simloop/_loop.py

@@ -15,6 +15,7 @@
 if TYPE_CHECKING:
     from asyncio.events import _TaskFactory
 
+from simloop._net import SimNetwork
 from simloop._trace import TraceEvent, TraceRecorder
 
 _Ts = TypeVarTuple("_Ts")
@@ -45,6 +46,14 @@ def _fence(api: str) -> NoReturn:
     )
 
 
+def _reject_kwargs(api: str, kwargs: dict[str, Any]) -> None:
+    # Optional stdlib arguments (ssl, sock, interface selectors, ...) reach
+    # outside the simulation; anything actually requested must fail loudly.
+    for name, value in kwargs.items():
+        if value:
+            _fence(f"{api}({name}=...)")
+
+
 def _label(callback: Callable[..., object]) -> str:
     # Labels feed the trace hash, so they must be stable across processes:
     # qualified names only, never repr() (which can embed memory addresses).
@@ -97,6 +106,7 @@ def __init__(self, seed: int = 0) -> None:
         self._unhandled: list[BaseException] = []
         self._exception_handler: _ExceptionHandler | None = None
         self._task_factory: _TaskFactory | None = None
+        self._net = SimNetwork(self)
 
     # ------------------------------------------------------------------
     # Introspection
@@ -113,6 +123,10 @@ def trace(self) -> tuple[TraceEvent, ...]:
     def trace_hash(self) -> str:
         return self._recorder.hash()
 
+    @property
+    def net(self) -> SimNetwork:
+        return self._net
+
     # ------------------------------------------------------------------
     # Clock and scheduling
     # ------------------------------------------------------------------
@@ -285,19 +299,56 @@ def create_task(
     ) -> asyncio.Task[Any]:
         self._check_closed()
         if self._task_factory is None:
-            return asyncio.Task(coro, loop=self, name=name, context=context)
-        factory: Any = self._task_factory
-        task: asyncio.Task[Any] = (
-            factory(self, coro)
-            if context is None
-            else factory(self, coro, context=context)
-        )
-        if name is not None:
-            set_name = getattr(task, "set_name", None)
-            if set_name is not None:
-                set_name(name)
+            task: asyncio.Task[Any] = asyncio.Task(
+                coro, loop=self, name=name, context=context
+            )
+        else:
+            factory: Any = self._task_factory
+            task = (
+                factory(self, coro)
+                if context is None
+                else factory(self, coro, context=context)
+            )
+            if name is not None:
+                set_name = getattr(task, "set_name", None)
+                if set_name is not None:
+                    set_name(name)
+        self._net._register_task(task)
         return task
 
+    async def create_datagram_endpoint(
+        self,
+        protocol_factory: Any,
+        local_addr: Any = None,
+        remote_addr: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        _reject_kwargs("create_datagram_endpoint", kwargs)
+        return await self._net._open_datagram_endpoint(
+            protocol_factory, local_addr, remote_addr
+        )
+
+    async def create_connection(
+        self,
+        protocol_factory: Any,
+        host: Any = None,
+        port: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        _reject_kwargs("create_connection", kwargs)
+        return await self._net._open_connection(protocol_factory, host, port)
+
+    async def create_server(
+        self,
+        protocol_factory: Any,
+        host: Any = None,
+        port: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        kwargs.pop("backlog", None)  # accepted and irrelevant: no accept queue
+        _reject_kwargs("create_server", kwargs)
+        return await self._net._start_server(protocol_factory, host, port)
+
     def set_task_factory(self, factory: _TaskFactory | None) -> None:
         if factory is not None and not callable(factory):
             raise TypeError("task factory must be a callable or None")
@@ -429,12 +480,6 @@ def getaddrinfo(self, *args: Any, **kwargs: Any) -> Any:
     def getnameinfo(self, *args: Any, **kwargs: Any) -> Any:
         _fence("getnameinfo")
 
-    def create_connection(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_connection")
-
-    def create_server(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_server")
-
     def start_tls(self, *args: Any, **kwargs: Any) -> Any:
         _fence("start_tls")
 
@@ -444,9 +489,6 @@ def sendfile(self, *args: Any, **kwargs: Any) -> Any:
     def sock_sendfile(self, *args: Any, **kwargs: Any) -> Any:
         _fence("sock_sendfile")
 
-    def create_datagram_endpoint(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_datagram_endpoint")
-
     def connect_read_pipe(self, *args: Any, **kwargs: Any) -> Any:
         _fence("connect_read_pipe")