Restart, disks and skewed clocks

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 ## 0.2.0 (unreleased)
 
+- Crashed hosts can come back: `loop.net.restart(name)` (or
+  `host.restart()`) revives a machine as a fresh incarnation. It restores
+  liveness and nothing else — the old tasks stay cancelled and the
+  listeners are gone, so the caller boots the machine again the way it
+  booted it the first time. Traffic due while the host was dead is lost,
+  leaving peers to notice the outage from their own timeouts. Crashes are
+  now recorded too: `crash()` writes a trace event and consumes a uid, so
+  a workload that crashes a host hashes differently than it did in 0.1.0
+  (workloads that never crash stay byte-identical to 0.1.0).
+- Every host now has `host.disk`, a mapping that survives its crashes:
+  where state a real process would fsync belongs. Writes are atomic at
+  assignment; there is no partial-write model.
+- Clocks can lie per host: `loop.net.set_clock(name, offset=...)` skews
+  what that host's tasks read from `loop.time()`, and the deadlines they
+  hand to `call_at` with it, while durations (`sleep`, `timeout`,
+  `wait_for`, `call_later`) cost the same everywhere — which is what a
+  wrong wall clock does to a real machine. Traces stay on the true clock,
+  and runs that configure no offset are byte-identical to 0.1.0.
 - A second flagship demo: `examples/raft/` is a teaching-sized Raft (leader
   election + log replication, plain asyncio on streams) tested only under
   simulation — four safety invariants checked over 50,000 chaos seeds, five

README.md

@@ -171,6 +171,14 @@ loop.call_later(5.0, net.heal)                  # heals in virtual time
 net.crash("node2")                              # no reset, just silence
 ```
 
+- **Crashes with a way back** — `net.crash` kills a host's tasks and
+  binds, `net.restart` brings it back as a fresh incarnation, and
+  `host.disk` is a mapping that outlives both: machines die, reboot, and
+  remember what they wrote down.
+- **Clocks that lie** — `net.set_clock(name, offset=...)` skews what one
+  host reads from the clock without changing how long anything takes, for
+  testing lease and timeout code against machines that disagree about the
+  time.
 - **Replayable traces** — every scheduling and fault decision lands in
   an append-only trace whose hash proves a replay is exact.
 - **Seeded stand-ins** — `sim.random`, `sim.uuid4()` and `sim.time()`

docs/supported-api.md

@@ -11,7 +11,7 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of
 | API | Behavior under simulation |
 |---|---|
 | `loop.call_soon` / `call_later` / `call_at` | Seeded ready-queue ordering; `(deadline, seq)` timer tie-break |
-| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0; never waits on wall time |
+| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0; never waits on wall time. What `loop.time()` *reads* is shifted by whatever offset the calling task's host is configured with; durations such as `asyncio.sleep` cost the same everywhere |
 | `loop.create_task` / `asyncio.create_task` | Real stdlib `Task`s, including custom task factories |
 | `loop.create_future` | Real stdlib `Future`s |
 | `run_until_complete` / `run_forever` / `stop` / `close` | Deadlock detection: raises `SimulationDeadlockError` when nothing can run |
@@ -40,6 +40,9 @@ a host belong to an implicit `driver` host.
 | `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 |
+| `loop.net.restart` / `host.restart()` | The counterpart to a crash: the host comes back as a fresh incarnation. Liveness is all that is revived — the old tasks stay cancelled, its listeners and binds are gone, and the caller boots whatever should run on the machine again, the same way it booted it the first time. Cancellation is requested at crash and lands on the next scheduler step, so a restart in the same step can briefly coexist with a dying task that swallows `CancelledError`. A packet is checked against liveness when it arrives, so traffic due during the dead window is lost; a packet that was already in flight and lands after the machine is back is delivered, and finds a host that no longer holds the old incarnation's connections. Peers still learn about the outage only from their own timeouts |
+| `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. Writes are atomic at assignment; there is no partial-write model. 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 |
+| `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 |
 | `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 |
@@ -54,6 +57,17 @@ address, an `AF_INET6` or `SOCK_RAW` request, a service name — raises
 `socket.gaierror(EAI_NONAME)`. Resolution is a pure lookup, not a scheduling
 decision: it never blocks and records no trace event.
 
+Partitions, crashes and reboots are all silent, so time is the only
+failure detector the code under test has — and `set_clock` lets that
+detector be wrong. A lease holder whose clock runs fast and an issuer
+whose clock runs slow disagree about when the lease expired, which is the
+disagreement leases exist to survive, and a test can produce it on
+purpose. The converse is worth knowing before reaching for it: a protocol
+that puts durations on the wire rather than timestamps is immune to skew
+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

examples/jobqueue/tests/test_lease_skew.py

@@ -0,0 +1,126 @@
+"""Leases held by a worker whose clock runs two leases fast.
+
+The broker is the only clock that decides expiry, and everything the
+protocol puts on the wire is a duration, never a timestamp. These runs
+put a badly skewed worker in the middle of the lease machinery and check
+that the exactly-once story is unchanged.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from simloop import SimLoop, sim_test
+
+import helpers
+
+LEASE_S = 2.0  # the broker's default lease
+SKEW_S = 2 * LEASE_S  # a clock two full leases ahead of the rest of the cluster
+
+
+async def _clock_reading() -> float:
+    return helpers.sim_loop().time()
+
+
+@sim_test(seeds=25)
+async def test_a_fast_clock_does_not_cost_the_worker_its_lease() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=1, clients=1)
+    loop.net.set_clock("w1", offset=SKEW_S)
+    # The skew is real from inside the worker: its tasks read a clock two
+    # leases ahead of the broker's.
+    worker_now = await loop.net.host("w1").create_task(_clock_reading())
+    assert worker_now - loop.time() == pytest.approx(SKEW_S)
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("fast-clock", duration=1.5 * LEASE_S)
+    )
+    assert job_id is not None
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    # Renewals are heartbeats on a timer, not deadline arithmetic, so the
+    # skewed worker keeps the lease it was granted: one attempt, one commit,
+    # no re-run to fence.
+    assert cluster.broker.snapshot()[job_id] == ("done", 1)
+    assert [(c.token, c.value) for c in cluster.store.commits] == [(1, "fast-clock")]
+    assert cluster.store.rejected == []
+
+
+@sim_test(seeds=50)
+async def test_a_fast_worker_takes_over_a_lapsed_lease() -> None:
+    loop = helpers.sim_loop()
+    loop.net.host("w2")  # register it early so it can be skewed before it runs
+    loop.net.set_clock("w2", offset=SKEW_S)
+    # w1 is partitioned mid-job and turns into a zombie; the skewed w2 is the
+    # worker that picks the job back up once the broker expires the lease.
+    cluster = await helpers.zombie_run(helpers.EffectStore())
+    helpers.verify(cluster)
+    assert any(reason == "stale" for _, _, reason in cluster.store.rejected)
+    assert len(cluster.store.commits) == 1
+    assert cluster.store.commits[0].token >= 2  # the re-run's lease, not the zombie's
+
+
+@sim_test(seeds=25)
+async def test_skewed_and_honest_workers_share_a_queue() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    loop.net.set_clock("w2", offset=SKEW_S)
+    client = cluster.clients[0]
+    host = loop.net.host("c1")
+    job_ids = [await host.create_task(client.submit(f"m{i}")) for i in range(4)]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    assert len(cluster.store.commits) == 4
+
+
+async def _skew_workload() -> str:
+    """Run a small cluster to quiesce; return the trace hash of that run."""
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    client = cluster.clients[0]
+    host = loop.net.host("c1")
+    job_ids = [await host.create_task(client.submit(f"t{i}")) for i in range(2)]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    trace = loop.trace_hash()  # taken at quiesce: the run, not the teardown
+    # Then shut everything down — servers, workers, open connection handlers —
+    # so the loop closes with nothing suspended mid-write. `all_tasks()` is a
+    # set, so sort it by name: the teardown cannot smuggle in id ordering.
+    # (The hash above is taken before any of this runs, deliberately.)
+    here = asyncio.current_task()
+    running = sorted(
+        (task for task in asyncio.all_tasks() if task is not here),
+        key=lambda task: task.get_name(),
+    )
+    for task in running:
+        task.cancel()
+    await asyncio.gather(*running, return_exceptions=True)
+    await asyncio.sleep(0.1)  # let cancelled renew heartbeats finish unwinding
+    return trace
+
+
+def _skew_hash(seed: int, offset: float) -> str:
+    loop = SimLoop(seed=seed)
+    # Every run registers the hosts in the same order; only the offset differs.
+    for name in ("broker", "w1", "w2", "c1"):
+        loop.net.host(name)
+    loop.net.set_clock("w2", offset=offset)
+    try:
+        trace: str = loop.run_until_complete(_skew_workload())
+    finally:
+        loop.close()
+    return trace
+
+
+def test_skew_alone_does_not_change_what_happens() -> None:
+    # Why the skew is absorbed rather than merely survived: no timestamp ever
+    # crosses a host boundary here. The broker sends `lease_s`, a duration;
+    # the worker sleeps and heartbeats on durations; only the broker compares
+    # readings, all of them its own. A worker's wrong clock is therefore
+    # unobservable — the packet trace is byte-identical either way, which is
+    # also why no ablation of *this* app can turn skew into a violation.
+    for seed in (0, 7):
+        hashes = {_skew_hash(seed, offset) for offset in (0.0, SKEW_S, -SKEW_S)}
+        assert len(hashes) == 1, f"seed {seed}: the skew moved the trace"

src/simloop/__init__.py

@@ -2,7 +2,7 @@
 
 from simloop._explore import SeedReport, explore, sim_test
 from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError
-from simloop._net import Host, SimNetwork
+from simloop._net import Host, SimDisk, SimNetwork
 from simloop._sim import Sim, sim
 from simloop._trace import TraceEvent
 
@@ -12,6 +12,7 @@
     "Host",
     "SeedReport",
     "Sim",
+    "SimDisk",
     "SimLoop",
     "SimNetwork",
     "SimulationDeadlockError",

src/simloop/_loop.py

@@ -185,6 +185,18 @@ def net(self) -> SimNetwork:
     # ------------------------------------------------------------------
 
     def time(self) -> float:
+        # What the calling task's machine believes the time is. Everything
+        # internal — timer ordering, deadline advance, trace timestamps —
+        # uses the true clock (self._now) directly, so skew never perturbs
+        # scheduling and traces from skewed runs stay comparable.
+        return self._now + self._net._offset_now()
+
+    def _true_time(self) -> float:
+        """The shared virtual clock, the same reading for whoever asks.
+
+        The clock the simulation itself runs on, for the internals that must
+        not see a host's skew — the network's trace timestamps above all.
+        """
         return self._now
 
     def call_soon(
@@ -209,7 +221,14 @@ def call_later(
         *args: Unpack[_Ts],
         context: Context | None = None,
     ) -> asyncio.TimerHandle:
-        return self.call_at(self._now + delay, callback, *args, context=context)
+        # A delay is a duration, so it is measured on the true clock and no
+        # caller's offset touches it: a wrong wall clock does not make a
+        # second take longer. Going through call_at instead would subtract
+        # the caller's offset from a deadline that is already true.
+        when = self._now + delay
+        return self._call_at_true(
+            when, when + self._net._offset_now(), callback, args, context
+        )
 
     def call_at(
         self,
@@ -218,8 +237,34 @@ def call_at(
         *args: Unpack[_Ts],
         context: Context | None = None,
     ) -> asyncio.TimerHandle:
+        # ``when`` is a reading of the *scheduling* task's clock, so it is
+        # converted to true time once, here, with that task's offset. The
+        # callback runs later under the context asyncio copied into the
+        # handle — the scheduling task's — so it reads the same clock the
+        # scheduler did, which is coherent for the cases that matter:
+        # asyncio.timeout, sleep and protocol timers all schedule from the
+        # task they serve. Nothing re-converts per callback.
+        return self._call_at_true(
+            when - self._net._offset_now(), when, callback, args, context
+        )
+
+    def _call_at_true(
+        self,
+        when: float,
+        shown: float,
+        callback: Callable[[Unpack[_Ts]], object],
+        args: tuple[Unpack[_Ts]],
+        context: Context | None,
+    ) -> asyncio.TimerHandle:
+        """Push a timer whose ``when`` is already true time.
+
+        The heap orders on true time — the one clock the loop itself runs on
+        — while the handle advertises ``shown``, the same deadline on the
+        clock the caller reads, so ``timer.when()`` stays comparable with
+        that task's ``loop.time()``.
+        """
         self._check_closed()
-        timer = asyncio.TimerHandle(when, callback, args, self, context)
+        timer = asyncio.TimerHandle(shown, callback, args, self, context)
         seq = self._next_seq
         self._next_seq += 1
         label = _label(callback)