Simulate write-side flow control on request

CHANGELOG.md

@@ -2,6 +2,33 @@
 
 ## 0.2.0 (unreleased)
 
+- `drain()` can finally block. `loop.net.set_flow_control()` gives stream
+  transports a write buffer that holds every byte written but not yet
+  received by the peer's protocol — still in flight, held by a partition,
+  queued behind an earlier sequence number, or parked because the peer called
+  `pause_reading()`. A slow reader, a cut link and a dead peer therefore all
+  push back, and a writer paused against a crashed peer keeps waiting until
+  its own timeout fires, because a crashed host sends no reset and a real
+  sender would wait too. Crossing the high mark calls `pause_writing()` and
+  falling back to the low one calls `resume_writing()`, both synchronously,
+  which is what turns a backpressure deadlock or an unhandled pause into
+  something a seed can find. Off unless you ask for it, and the watermarks
+  once armed are the standard library's own `(low=16 KiB, high=64 KiB)`,
+  overridable network-wide or per transport with `set_write_buffer_limits`.
+  The switch is what makes the defaults safe: `set_write_buffer_limits` on
+  its own records numbers without enforcing them, because libraries call it
+  uninvited — anyio sets limits on every stream, websockets on every
+  connection — and arming on their call would change, or deadlock, workloads
+  nobody touched. On hashes: the feature adds no packets and no scheduling
+  events of its own, so a run that never arms it is byte-identical to the run
+  it was before, checked against digests pinned from before the feature
+  existed; an armed run that actually crosses a mark hashes anew, because the
+  writer it wakes is a real scheduling decision. One honest divergence from
+  TCP: the buffer drains when the peer's *application* receives the bytes,
+  with no read-ahead, so simulated backpressure is strictly tighter than the
+  real thing — deliberately, since that is what makes a slow consumer
+  visibly slow. With the switch off `get_write_buffer_size()` reports `0`,
+  which is the truth: nothing is charged and writes leave immediately.
 - Traces now say *where*, and that changes every hash. Each scheduling event
   carries the host it belongs to — the machine that asked for a callback on
   `schedule`, the machine that owns it on `run` and `cancel`, and nothing at

README.md

@@ -296,7 +296,11 @@ simulated, so a client that connects a socket and hands it to
 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.
+Write-side flow control is simulated on request: `net.set_flow_control()`
+makes `drain()` really wait while the peer has not read, so backpressure
+deadlocks and missing pause/resume handling become findable. It is off by
+default, so a run that does not ask for it decides exactly what it decided
+before.
 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
 httpx actually do under simulation, measured rather than promised — is in

docs/design.md

@@ -222,6 +222,28 @@ Decisions inside that model, each doing real work:
   application can actually defend against — write, sync, and only then act
   on it. Skipping that sync is a bug the simulation can now find, which is
   the whole reason the model exists.
+- **A write buffer counts what the peer has not read.** A stream transport
+  owes every byte it wrote that the peer's protocol has not received — on the
+  wire, held by a partition, or parked behind the peer's `pause_reading()`.
+  That definition is the only one under which a slow reader, a cut link and a
+  dead peer all push back, and it needs no wire traffic: the simulation is one
+  process, so the receiving transport can credit the sending one directly. The
+  alternative — a real receive window with credit packets — was rejected on
+  price: every credit would be another packet, another uid, another latency
+  draw, moving the trace hash of every stream workload in the repository and
+  invalidating recorded seeds users already hold, all to buy fidelity a
+  single-process simulation does not need. Pause and resume fire
+  synchronously, from the write and from the peer's read, as the stdlib's own
+  transports do, so the feature adds no scheduling event of its own; the only
+  new event is the woken writer's, which is a real scheduling decision. It is
+  off unless asked for, for the same reason the disk is: a run that never
+  calls `set_flow_control` has to decide exactly what it decided before, and
+  libraries set write-buffer limits uninvited, so arming on their call would
+  change — or deadlock — workloads nobody touched. The honest divergence is
+  that the buffer drains on the peer *application's* read with no read-ahead,
+  making simulated backpressure strictly tighter than real backpressure. That
+  is deliberate: it is what makes a slow consumer visibly slow, and it is why
+  the stdlib's numbers ship inside an opt-in mode rather than by default.
 - **The accept is sequence 0.** The server builds its transport and sends
   `accept` before its protocol's `connection_made` can write; the client
   transport is built when the accept is *dispatched*, not when the connector
@@ -234,8 +256,7 @@ Decisions inside that model, each doing real work:
   `datagram_received` belongs to the receiving machine, and `crash` knows
   exactly which tasks to kill.
 
-What was cut, deliberately: write-side flow control (`drain()` never blocks
-— buffers are unbounded), retransmission and congestion modeling, IP
+What was cut, deliberately: retransmission and congestion modeling, IP
 addresses, TLS. Each would deepen the simulation without widening the class
 of bugs it can catch; the supported-subset contract beats chasing 100% of
 the asyncio surface.

docs/supported-api.md

@@ -46,6 +46,7 @@ a host belong to an implicit `driver` host.
 | `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. By default a write is durable the moment it is made. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk. `disk.sync()` exists on every disk and does nothing on one that does not buffer, so the code under test is written the same way either way |
 | `loop.net.set_disk` | Makes a host's disk lie about when a write lands. `buffered=True` queues writes and deletes in order and only makes them durable on `sync()`; reads on that host see the queue merged over what is durable, in a fixed order — durable keys where the durable state has them, then the keys the queue invented, in write order, with a queued delete hiding a key. A crash throws the queue away and the reboot finds what was synced; with `torn=True` a seeded prefix of the queue survives instead, which is the state a machine that lost power part-way through a batch comes back with. A prefix is the whole model: writes never land out of order, and no value is ever half-written. Tearing needs a buffer to tear (`torn=True` alone is a `ValueError`), and reconfiguring a disk flushes whatever it was holding. The prefix is drawn from a seed-derived stream of its own, so a torn run makes exactly the network draws it would have made untorn, and a run that never calls `set_disk` draws nothing and records nothing — storage is not a scheduling event and has no trace events at all |
 | `loop.net.set_clock` / `clock_offset` | Per-host clock skew, in seconds. The offset changes what that host's tasks *read*: `loop.time()` (and `sim.time()` with it) returns true time plus the offset, and a deadline handed to `call_at` is interpreted on the calling task's clock. Durations are immune — `asyncio.sleep`, `asyncio.timeout`, `wait_for` and `call_later` cost the same everywhere, which is exactly what a wrong wall clock does to a real machine. By default the driver and unconfigured hosts read true time; the driver can be given an offset too. Trace timestamps stay on the true clock, so skew never perturbs scheduling and traces from skewed runs stay comparable |
+| `loop.net.set_flow_control` / `transport.set_write_buffer_limits` | Makes writes push back when the peer is not keeping up. A stream transport's write buffer holds every byte it has written that the peer's protocol has not received yet — still in flight, held by a partition, waiting behind an earlier sequence number, or parked because the peer called `pause_reading()`. So a slow reader, a cut link and a dead peer all apply backpressure. Crossing `high` calls `pause_writing()` on the protocol and dropping back to `low` calls `resume_writing()`; both happen synchronously, from the write and from the peer's read, so `drain()` really waits and no scheduling event of its own is added. Marks default to the standard library's `(low=16 KiB, high=64 KiB)`, settable network-wide here and per transport with `set_write_buffer_limits(high, low)`, which derives and validates them exactly as the stdlib does (a `high` below `low` is a `ValueError`). **Off until `loop.net.set_flow_control()` says otherwise**: a transport's own `set_write_buffer_limits` records numbers without enforcing them until then, because libraries set them uninvited — anyio sets limits on every stream, websockets on every connection — and upgrading should not deadlock a workload nobody changed. With the switch off `get_write_buffer_size()` reports `0`, which is honest: nothing is charged and writes leave immediately. Honest divergence from TCP: the buffer drains when the peer's *application* receives the bytes, with no read-ahead, so simulated backpressure is strictly tighter than real backpressure — which is what makes a slow consumer visibly slow. A crashed peer sends no reset, so a writer paused against one stays paused until its own timeout fires, exactly as a real sender does. Datagram endpoints have no write buffer at all |
 | `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
 | `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 |
@@ -71,11 +72,10 @@ by construction — in `examples/jobqueue/` only the broker reads a clock,
 so skewing a worker changes nothing the cluster decides. Clock faults
 reach only code that compares timestamps taken on different machines.
 
-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; and addressing is IPv4-only and entirely synthetic
-— there are no routes, no netmasks, and no service-name database.
+Limitations, stated honestly: there is no retransmission or congestion model
+— streams are reliable by construction; and addressing is IPv4-only and
+entirely synthetic — there are no routes, no netmasks, and no service-name
+database.
 
 ## Fenced
 
@@ -121,7 +121,9 @@ host)`:
 A `schedule` event names the host that *asked* for the callback, while `run`
 and `cancel` name the host the callback belongs to. The difference is the
 point: a wakeup that crosses machines is a `schedule` on one host and a `run`
-on another. An empty host means the event belongs to the simulation rather
+on another. Flow control is that shape exactly: it adds no packets and no
+scheduling events of its own, and what does appear is the wakeup of a writer
+that was waiting in `drain()`, scheduled by the host whose read released it. An empty host means the event belongs to the simulation rather
 than to any machine — a clock advance, which is global; the network's own
 delivery step, which happens on the wire between two machines rather than on
 either of them; and every `net` event, whose label already says which

src/simloop/_net.py

@@ -18,7 +18,11 @@
 
 import asyncio
 
-from simloop._transports import _SimDatagramTransport, _SimStreamTransport
+from simloop._transports import (
+    _check_limits,
+    _SimDatagramTransport,
+    _SimStreamTransport,
+)
 
 if TYPE_CHECKING:
     from simloop._loop import SimLoop
@@ -408,6 +412,8 @@ def __init__(self, loop: SimLoop) -> None:
         self._default_latency: tuple[float, float] = (0.0, 0.0)
         self._default_drop = 0.0
         self._default_duplicate = 0.0
+        self._flow_control = False
+        self._flow_defaults: tuple[int, int] = (16 * 1024, 64 * 1024)
         self._links: dict[tuple[str, str], _Link] = {}
         self._cuts: set[frozenset[str]] = set()
         self._held: list[_Packet] = []
@@ -598,6 +604,36 @@ def set_disk(
             )
         self.host(name).disk._configure(buffered=buffered, torn=torn)
 
+    def set_flow_control(
+        self, *, enabled: bool = True, high: int | None = None, low: int | None = None
+    ) -> None:
+        """Make stream writes push back when the peer is not keeping up.
+
+        Armed, a transport's write buffer counts every byte it has written
+        that the peer's protocol has not received — still on the wire, held
+        by a partition, or parked because the peer paused reading. Crossing
+        ``high`` calls ``pause_writing`` on the protocol, so ``drain()``
+        really waits; dropping back to ``low`` calls ``resume_writing``. Both
+        happen at once, without a scheduling step of their own. Watermarks
+        default to the standard library's 64 KiB and 16 KiB and can be set
+        here for the whole network or per transport with
+        ``set_write_buffer_limits``.
+
+        Off unless asked for, and a transport's own limits are recorded but
+        inert until then, because libraries set them uninvited: a run that
+        never calls this decides exactly what it decided without it.
+        Turning it off releases whatever is paused.
+
+        The buffer drains on the peer *application's* read, with no
+        read-ahead, so this pushes back sooner than a real socket does.
+        """
+        if high is not None or low is not None:
+            self._flow_defaults = _check_limits(high, low)
+        self._flow_control = enabled
+        if not enabled:
+            for transport in list(self._streams.values()):
+                transport._release_flow_control()
+
     def clock_offset(self, name: str) -> float:
         self._require_host(name)
         return self._clock_offsets.get(name, 0.0)