Add the jobqueue exactly-once demo

README.md

@@ -93,6 +93,16 @@ net.crash("node2")                              # no reset, just silence
 - **Replayable traces** — every scheduling and fault decision lands in
   an append-only trace whose hash proves a replay is exact.
 
+## Proving it: the jobqueue demo
+
+`examples/jobqueue/` is a complete distributed system — an exactly-once
+job scheduler (leases, fencing tokens, idempotency keys, backoff, and
+dead-lettering) written in plain asyncio — tested end to end with
+simloop. Its suite runs hundreds of seeds of partitions, crashes, and
+poison jobs, and shows that removing any load-bearing safeguard produces
+a violation the explorer finds and replays from a seed. The bug table
+lives in [examples/jobqueue/README.md](examples/jobqueue/README.md).
+
 ## Honest limits
 
 Code that goes through the event-loop API is supported; code that

examples/jobqueue/README.md

@@ -0,0 +1,62 @@
+# jobqueue — an exactly-once job scheduler, proven by simloop
+
+A demo distributed system written in plain asyncio (stdlib only, no simloop
+imports): one broker, stateless workers, submitting clients. Its test suite
+runs entirely under [simloop](../../README.md) — seeded scheduling, virtual
+time, simulated partitions and crashes — and every failure it can produce
+replays exactly from a seed.
+
+## The claim, stated honestly
+
+Attempts are at-least-once; **effects commit exactly once**. Workers may
+re-run a job after a crash or an expired lease, but the fenced, idempotent
+effect store accepts one commit per job. No acknowledged job is lost: it
+ends `done` (effect committed) or `dead` (dead-lettered after
+`max_attempts`, visible, never silent).
+
+Mechanisms: time-based leases with heartbeat renewal, per-job monotonic
+fencing tokens checked at both broker and store, idempotency-key submit
+dedupe, exponential-backoff requeue, dead-letter state. Time is the only
+failure detector — under simloop a crashed peer sends no reset and a
+partition stalls silently, exactly like production.
+
+## Invariants
+
+Checked after every simulated run (`tests/invariants.py`):
+
+1. **no-loss** — every acknowledged submit ends done or dead
+2. **exactly-once** — at most one accepted commit per job and per logical
+   submit; every done job has exactly one
+3. **no-zombie-writes** — no commit accepted from a superseded lease
+4. **convergence** — nothing left queued or leased at quiesce
+
+## The numbers
+
+- Scenario suite: 7 scenarios × 25–50 seeds each, all green.
+- Campaign: **300 seeds** of randomized partitions, a worker crash, and
+  poison jobs per seed — invariants held on every seed.
+- Ablations: remove any load-bearing safeguard and the explorer finds a
+  violating schedule. Found-during-development bugs are listed too.
+
+| # | Safeguard removed / bug | Invariant violated | Found at seed | Seeds searched | Reproduce |
+|---|---|---|---|---|---|
+| 1 | Store fencing off (`EffectStore(fenced=False)`) | no-zombie-writes | 0 | 1 | `uv run pytest examples/jobqueue/tests/test_mutations.py::test_unfenced_store_admits_a_zombie_write` |
+| 2 | Store idempotency off (`EffectStore(idempotent=False)`) | exactly-once | 0 | 1 | `... ::test_unidempotent_store_double_commits_after_a_crash` |
+| 3 | Broker fencing off + store ablated | exactly-once | 0 | 1 | `... ::test_broker_fencing_off_lets_a_zombie_finish_the_job` |
+| 4 | Client idempotency keys off | exactly-once | 0 | 1 | `... ::test_no_idempotency_key_duplicates_a_lost_ack_submit` |
+| 5 | Unbounded retries (`max_attempts=None`) | convergence | 0 | 1 | `... ::test_unbounded_attempts_never_converge_on_poison` |
+| 6 | Worker renewals off + store idempotency off | exactly-once | 0 | 1 | `... ::test_renew_off_paired_with_unidempotent_store_double_commits` |
+
+Rows 1–6 are labeled ablations — detection demonstrations, not bugs that
+were ever shipped. Renewals off *alone* stays safe across 75 seeds
+(defense in depth); so does broker fencing alone — the store is the
+last line, and the suite proves both lines independently.
+
+## Run it
+
+    uv run pytest examples/jobqueue/tests -q            # fast suite
+    uv run pytest examples/jobqueue/tests -q -m slow    # campaign + safety proofs
+
+Replay any sim-test failure exactly:
+
+    uv run pytest 'examples/jobqueue/tests/test_campaign.py::test_campaign_holds_invariants_under_chaos' -m slow --simloop-replay=0

examples/jobqueue/conftest.py

@@ -0,0 +1,8 @@
+"""Make ``import jobqueue`` work when pytest runs from the repository root."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))

examples/jobqueue/jobqueue/__init__.py

@@ -0,0 +1,9 @@
+"""An exactly-once job scheduler in plain asyncio, tested with simloop.
+
+Plain-stdlib demo application: a single broker leases jobs to stateless
+workers under time-based leases with fencing tokens; clients submit with
+idempotency keys. Nothing in this package imports simloop — the harness
+lives entirely in the test suite.
+"""
+
+from __future__ import annotations

examples/jobqueue/jobqueue/broker.py

@@ -0,0 +1,182 @@
+"""The broker: single authority for the job state machine.
+
+States per job: queued -> leased -> done, with leased -> queued on lease
+expiry and queued -> dead once the attempt budget is spent. Time is the
+only failure detector — a worker that goes silent is indistinguishable
+from a partitioned one, so the lease deadline decides, never the
+connection.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from typing import Any
+
+from jobqueue import wire
+
+
+@dataclass
+class _Job:
+    job_id: str
+    key: str
+    payload: dict[str, Any]
+    state: str = "queued"
+    attempts: int = 0
+    token: int = 0
+    ready_at: float = 0.0
+    expiry: asyncio.TimerHandle | None = field(default=None, repr=False)
+
+
+class Broker:
+    def __init__(
+        self,
+        *,
+        lease_s: float = 2.0,
+        max_attempts: int | None = 3,
+        backoff_base_s: float = 0.5,
+        fencing: bool = True,
+    ) -> None:
+        self._lease_s = lease_s
+        self._max_attempts = max_attempts
+        self._backoff_base_s = backoff_base_s
+        self._fencing = fencing
+        self._jobs: dict[str, _Job] = {}
+        self._by_key: dict[str, str] = {}
+        self._order: list[str] = []
+        self._ready = asyncio.Event()
+        self._next_id = 0
+
+    async def serve(self, port: int = 7000) -> None:
+        server = await asyncio.start_server(self._connection, "0.0.0.0", port)
+        async with server:
+            await server.serve_forever()
+
+    async def _connection(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        try:
+            while True:
+                message = await wire.read_message(reader)
+                response = await self.handle(message)
+                wire.write_message(writer, response)
+                await writer.drain()
+        except (asyncio.IncompleteReadError, ConnectionResetError, wire.FrameError):
+            pass
+        finally:
+            writer.close()
+
+    async def handle(self, message: dict[str, Any]) -> dict[str, Any]:
+        op = message["op"]
+        if op == "submit":
+            return self._submit(message["key"], message["payload"])
+        if op == "acquire":
+            return await self._acquire(message["worker_id"], message["wait_s"])
+        if op == "renew":
+            return self._renew(message["job_id"], message["token"])
+        if op == "complete":
+            return self._complete(message["job_id"], message["token"])
+        if op == "status":
+            return self._status(message["job_id"])
+        return {"error": f"unknown op {op!r}"}
+
+    def snapshot(self) -> dict[str, tuple[str, int]]:
+        return {j.job_id: (j.state, j.attempts) for j in self._jobs.values()}
+
+    def _submit(self, key: str, payload: dict[str, Any]) -> dict[str, Any]:
+        job_id = self._by_key.get(key)
+        if job_id is None:
+            self._next_id += 1
+            job_id = f"j{self._next_id}"
+            self._jobs[job_id] = _Job(job_id=job_id, key=key, payload=payload)
+            self._by_key[key] = job_id
+            self._order.append(job_id)
+            self._ready.set()
+        return {"job_id": job_id}
+
+    async def _acquire(self, worker_id: str, wait_s: float) -> dict[str, Any]:
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + wait_s
+        while True:
+            job = self._next_ready(loop.time())
+            if job is not None:
+                return self._lease(job)
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                return {"job_id": None}
+            self._ready.clear()
+            try:
+                async with asyncio.timeout(remaining):
+                    await self._ready.wait()
+            except TimeoutError:
+                return {"job_id": None}
+
+    def _next_ready(self, now: float) -> _Job | None:
+        for job_id in self._order:
+            job = self._jobs[job_id]
+            if job.state == "queued" and job.ready_at <= now:
+                return job
+        return None
+
+    def _lease(self, job: _Job) -> dict[str, Any]:
+        loop = asyncio.get_running_loop()
+        job.state = "leased"
+        job.token += 1
+        job.attempts += 1
+        self._order.remove(job.job_id)
+        job.expiry = loop.call_later(
+            self._lease_s, self._expire, job.job_id, job.token
+        )
+        return {
+            "job_id": job.job_id,
+            "token": job.token,
+            "payload": job.payload,
+            "lease_s": self._lease_s,
+        }
+
+    def _expire(self, job_id: str, token: int) -> None:
+        job = self._jobs.get(job_id)
+        if job is None or job.state != "leased" or job.token != token:
+            return
+        if self._max_attempts is not None and job.attempts >= self._max_attempts:
+            job.state = "dead"
+            return
+        loop = asyncio.get_running_loop()
+        delay = self._backoff_base_s * 2 ** (job.attempts - 1)
+        job.state = "queued"
+        job.ready_at = loop.time() + delay
+        self._order.append(job_id)
+        loop.call_later(delay, self._ready.set)
+
+    def _renew(self, job_id: str, token: int) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        if job is None or job.state != "leased":
+            return {"ok": False}
+        if self._fencing and token != job.token:
+            return {"ok": False}
+        if job.expiry is not None:
+            job.expiry.cancel()
+        loop = asyncio.get_running_loop()
+        job.expiry = loop.call_later(
+            self._lease_s, self._expire, job.job_id, job.token
+        )
+        return {"ok": True}
+
+    def _complete(self, job_id: str, token: int) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        if job is None or job.state == "dead":
+            return {"ok": False}
+        if job.state == "done":
+            return {"ok": True}
+        if self._fencing and (job.state != "leased" or token != job.token):
+            return {"ok": False}
+        if job.expiry is not None:
+            job.expiry.cancel()
+        if job.state == "queued":
+            self._order.remove(job.job_id)
+        job.state = "done"
+        return {"ok": True}
+
+    def _status(self, job_id: str) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        return {"state": None if job is None else job.state}

examples/jobqueue/jobqueue/client.py

@@ -0,0 +1,74 @@
+"""A submitting client: idempotency keys, timeouts, retries with backoff.
+
+A timed-out submit is ambiguous — the broker may or may not have created
+the job. The client retries with the *same* idempotency key, so the broker
+can answer "that one again" instead of minting a duplicate.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from jobqueue import wire
+
+
+class Client:
+    def __init__(
+        self,
+        client_id: str,
+        *,
+        broker_host: str = "broker",
+        broker_port: int = 7000,
+        idempotency: bool = True,
+        rpc_timeout_s: float = 1.0,
+        retries: int = 6,
+        retry_base_s: float = 0.25,
+        poll_s: float = 0.5,
+    ) -> None:
+        self._client_id = client_id
+        self._broker_host = broker_host
+        self._broker_port = broker_port
+        self._idempotency = idempotency
+        self._rpc_timeout_s = rpc_timeout_s
+        self._retries = retries
+        self._retry_base_s = retry_base_s
+        self._poll_s = poll_s
+        self._counter = 0
+        self.acknowledged: dict[str, str] = {}
+
+    async def submit(
+        self, value: str, *, duration: float = 0.1, poison: bool = False
+    ) -> str | None:
+        """Submit one logical job; None means never acknowledged."""
+        self._counter += 1
+        key = f"{self._client_id}.{self._counter}"
+        payload = {"value": value, "duration": duration, "poison": poison}
+        for attempt in range(self._retries):
+            wire_key = key if self._idempotency else f"{key}.try{attempt}"
+            response = await wire.call(
+                self._broker_host,
+                self._broker_port,
+                {"op": "submit", "key": wire_key, "payload": payload},
+                timeout_s=self._rpc_timeout_s,
+            )
+            if response is not None:
+                job_id: str = response["job_id"]
+                self.acknowledged[key] = job_id
+                return job_id
+            await asyncio.sleep(self._retry_base_s * 2**attempt)
+        return None
+
+    async def wait(self, job_id: str) -> str:
+        """Poll until the job reaches a terminal state; returns it."""
+        while True:
+            response = await wire.call(
+                self._broker_host,
+                self._broker_port,
+                {"op": "status", "job_id": job_id},
+                timeout_s=self._rpc_timeout_s,
+            )
+            if response is not None and response["state"] in ("done", "dead"):
+                state: str = response["state"]
+                return state
+            await asyncio.sleep(self._poll_s)