From 14d6dab24f87dd5785176d6c9c9b78ee7b64427f Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 16:18:23 -0400 Subject: [PATCH 1/5] Modal replay: at most two simulations across invocations Each Modal invocation runs one replay worker, and separate invocations shared no limit, so a third concurrent run would exceed the two the shared account allows. Before app.run(), an invocation now takes one of two slots in the Modal Dict scorecard-replay-simulation-slots: put(skip_if_exists=True) is an atomic create-if-absent on the server. A lease outlasts the function timeout; an expired one is reclaimed by whoever first creates its one-time reclaim token. The slot is released in a finally, interrupts included. Co-Authored-By: Claude Opus 5.5 --- docs/uk_replay/RECIPE.md | 8 ++- pipeline/modal_uk_event.py | 80 +++++++++++++++++++++++++- tests/test_modal_uk_event.py | 106 ++++++++++++++++++++++++++++++++++- 3 files changed, 188 insertions(+), 6 deletions(-) diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index 380aa252..c8e83b4f 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -598,9 +598,11 @@ fields and head effects. `--allow-changed-existing` permits replacement only after that report has been reviewed; original bytes remain in the backup. `--allow-partial` permits read-only inspection of a focused download, but partial or noncanonical outputs cannot be adopted. Keep raw receipts for -provenance. Each Modal invocation uses one replay worker. Count every -active invocation and local simulation toward the overall maximum of two; -separate invocations do not share a global concurrency gate. A full event +provenance. Each Modal invocation uses one replay worker and first takes one of two +simulation slots in the Modal Dict `scorecard-replay-simulation-slots` +(`modal_uk_event.SimulationSlots`), waiting while both are held, so Modal +invocations never run more than two simulations at once. Local simulations +are not in that lease: count them yourself toward the overall maximum of two. A full event plus one focused repeat uses both slots. Remote wall time is recorded separately in `MODAL_RECEIPT.json`; use the completed event's observed timings when planning the next lane. diff --git a/pipeline/modal_uk_event.py b/pipeline/modal_uk_event.py index d5f43612..b8d510ec 100644 --- a/pipeline/modal_uk_event.py +++ b/pipeline/modal_uk_event.py @@ -100,6 +100,74 @@ } +SLOTS_DICT = "scorecard-replay-simulation-slots" +SLOT_COUNT = 2 +# Longer than the function timeout (10,800 s), so a live run never loses its +# slot; a holder that died without releasing frees it after this. +LEASE_SECONDS = 4 * 60 * 60 + + +class SimulationSlots: + """At most two replay simulations at once across Modal invocations. + + Each invocation runs one worker, and separate invocations share no + Modal-side limit, so a third concurrent run would exceed the two the + shared account allows. Before starting, an invocation takes one of two + slots in a named Modal Dict: ``put(key, value, skip_if_exists=True)`` is + an atomic create-if-absent on the server. An expired slot is reclaimed by + whoever first creates its one-time reclaim token, so two reclaimers + cannot both take it. + """ + + def __init__( + self, store, *, holder, count=SLOT_COUNT, lease=LEASE_SECONDS, clock=time.time + ): + self.store, self.holder, self.count, self.lease, self.clock = ( + store, + holder, + count, + lease, + clock, + ) + + def _try(self, slot: int) -> bool: + key = f"slot-{slot}" + claim = {"holder": self.holder, "expires": self.clock() + self.lease} + if self.store.put(key, claim, skip_if_exists=True): + return True + current = self.store.get(key) + if current is None or current["expires"] > self.clock(): + return False + token = f"reclaim-{slot}-{current['holder']}-{current['expires']}" + if not self.store.put(token, self.holder, skip_if_exists=True): + return False + self.store.pop(key, None) + return bool(self.store.put(key, claim, skip_if_exists=True)) + + def acquire( + self, + *, + wait_seconds: float = 6 * 60 * 60, + poll_seconds: float = 30, + sleep=time.sleep, + ) -> int: + deadline = self.clock() + wait_seconds + while True: + for slot in range(self.count): + if self._try(slot): + return slot + if self.clock() >= deadline: + raise TimeoutError("no simulation slot became free") + print("[modal] both simulation slots are held; waiting", flush=True) + sleep(poll_seconds) + + def release(self, slot: int) -> None: + key = f"slot-{slot}" + current = self.store.get(key) + if current is not None and current["holder"] == self.holder: + self.store.pop(key, None) + + def canonical_bytes(value: object) -> bytes: return ( json.dumps(value, indent=1, sort_keys=True, allow_nan=False) + "\n" @@ -641,8 +709,16 @@ def main(argv: list[str] | None = None) -> int: raise ValueError("choose an empty download prefix before a remote run") modal, app, replay = modal_function(mounts) response = None - with modal.enable_output(), app.run(): - response = replay.remote(payload) + slots = SimulationSlots( + modal.Dict.from_name(SLOTS_DICT, create_if_missing=True), + holder=f"{platform.node()}:{os.getpid()}:{digest_bytes(canonical_bytes(request))[:12]}", + ) + slot = slots.acquire() + try: + with modal.enable_output(), app.run(): + response = replay.remote(payload) + finally: + slots.release(slot) if response is None: print("Modal run interrupted; no output downloaded", file=sys.stderr) return 130 diff --git a/tests/test_modal_uk_event.py b/tests/test_modal_uk_event.py index 4a75e473..0194adeb 100644 --- a/tests/test_modal_uk_event.py +++ b/tests/test_modal_uk_event.py @@ -8,6 +8,8 @@ from types import SimpleNamespace import pytest +from hypothesis import given, settings +from hypothesis import strategies as st from pipeline import modal_uk_event as runner @@ -270,11 +272,15 @@ def interrupted_remote(payload): runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) ) monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") + slots = FakeDict() monkeypatch.setattr( runner, "modal_function", lambda mounts: ( - SimpleNamespace(enable_output=nullcontext), + SimpleNamespace( + enable_output=nullcontext, + Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), + ), SimpleNamespace(run=SuppressInterrupt), SimpleNamespace(remote=interrupted_remote), ), @@ -286,6 +292,8 @@ def interrupted_remote(payload): ) assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 130 assert "interrupted; no output downloaded" in capsys.readouterr().err + # The interrupted run gave its simulation slot back. + assert not [key for key in slots.data if key.startswith("slot-")] def test_modal_factory_has_one_blocked_network_worker_and_explicit_uv_venv( @@ -359,3 +367,99 @@ def concurrent(**kwargs): for key in recorded["env"] ) assert len(recorded["mounts"]) == len(mounts) + 1 # build-time frozen requirements + + +class FakeDict: + """A Modal Dict's put/get/pop, with put(skip_if_exists=True) atomic.""" + + def __init__(self): + self.data = {} + + def put(self, key, value, *, skip_if_exists=False): + if skip_if_exists and key in self.data: + return False + self.data[key] = value + return True + + def get(self, key, default=None): + return self.data.get(key, default) + + def pop(self, key, default=None): + return self.data.pop(key, default) + + +class Clock: + def __init__(self): + self.now = 1_000.0 + + def __call__(self): + return self.now + + +def test_two_slots_then_a_third_run_waits(): + store, clock = FakeDict(), Clock() + a = runner.SimulationSlots(store, holder="a", clock=clock) + b = runner.SimulationSlots(store, holder="b", clock=clock) + c = runner.SimulationSlots(store, holder="c", clock=clock) + assert {a.acquire(), b.acquire()} == {0, 1} + sleeps = [] + + def sleep(seconds): + sleeps.append(seconds) + clock.now += seconds + + with pytest.raises(TimeoutError): + c.acquire(wait_seconds=60, poll_seconds=30, sleep=sleep) + assert sleeps == [30, 30] + c.release(0) # not its slot: no effect + assert store.get("slot-0")["holder"] in ("a", "b") + a.release(0 if store.get("slot-0")["holder"] == "a" else 1) + assert c.acquire(wait_seconds=0, sleep=sleep) in (0, 1) + + +def test_an_expired_slot_is_reclaimed_once(): + store, clock = FakeDict(), Clock() + dead = runner.SimulationSlots(store, holder="dead", clock=clock, count=1) + dead.acquire() + clock.now += runner.LEASE_SECONDS + 1 + first = runner.SimulationSlots(store, holder="first", clock=clock, count=1) + second = runner.SimulationSlots(store, holder="second", clock=clock, count=1) + assert first._try(0) is True + assert second._try(0) is False + assert store.get("slot-0")["holder"] == "first" + dead.release(0) # the dead holder cannot free the new lease + assert store.get("slot-0")["holder"] == "first" + + +def test_a_live_lease_outlasts_the_function_timeout(): + assert runner.LEASE_SECONDS > runner.RESOURCE_LIMITS["timeout_seconds"] + + +@settings(max_examples=300, deadline=None) +@given( + st.lists( + st.tuples(st.sampled_from("abcd"), st.sampled_from(["try", "release", "tick"])), + max_size=40, + ) +) +def test_never_more_than_two_live_holders(operations): + store, clock = FakeDict(), Clock() + slots = {h: runner.SimulationSlots(store, holder=h, clock=clock) for h in "abcd"} + held = {} + for holder, action in operations: + if action == "tick": + clock.now += 600 # well inside a lease + elif action == "try" and holder not in held: + for slot in range(runner.SLOT_COUNT): + if slots[holder]._try(slot): + held[holder] = slot + break + elif action == "release" and holder in held: + slots[holder].release(held.pop(holder)) + live = { + value["holder"] + for key, value in store.data.items() + if key.startswith("slot-") and value["expires"] > clock.now + } + assert len(live) <= runner.SLOT_COUNT + assert live == set(held) From dff81d0846ac71733a1471d93a67a86361b7f73e Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 17:30:05 -0400 Subject: [PATCH 2/5] Modal slots: generation claims, renewal before the run, a remote guard Review of 14d6dab found three ways the two-slot lease could fail: a reclaimer dying between creating its one-time token and taking the slot lost that slot for good; a holder past its lease could pop a newer claim in release, and a reclaimer's unconditional pop could delete a claim made after it looked, each allowing a third simulation; and the property test asserted something false once 24 ticks passed. Every change of holder, and every renewal, now creates the slot's next generation key with put(skip_if_exists=True), Modal's atomic create-if-absent; nothing is popped, and a holder writes only its own key, only to release or shorten it. A head key names the newest generation as a hint (readers walk forward), and a random chain id separates keys written after Modal expires idle entries. main() renews the lease after the image build, just before the remote call, and the remote function refuses to start unless its 3-hour timeout fits in the lease, so a run ends before its slot can pass on even if the client dies. A renewal after the lease lapsed is refused (exit 75). After ctrl-C, Modal's client asks the server to stop the app's tasks (modal/runner.py _disconnect), and the slot frees 10 minutes later instead of at once. Tests drive the protocol call by call: the review's three scenarios, a holder dying after each possible step, a stale renewal after Modal expired the chain, and a Hypothesis model check interleaving 2-4 holders' Dict calls with clock jumps, crashes, interrupts and 7-day entry expiry (no two runs overlap on a slot; both slots are free again once every lease lapses). 20,000 generated schedules hold; each protection's deterministic test fails when that protection is removed. Assumption, stated in the docstring and RECIPE: no process stalls for days inside one protocol step. Co-Authored-By: Claude Opus 5.5 --- docs/uk_replay/RECIPE.md | 28 ++- pipeline/modal_uk_event.py | 202 ++++++++++++++---- tests/test_modal_uk_event.py | 402 +++++++++++++++++++++++++++++------ 3 files changed, 526 insertions(+), 106 deletions(-) diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index c8e83b4f..43a24d14 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -601,9 +601,31 @@ partial or noncanonical outputs cannot be adopted. Keep raw receipts for provenance. Each Modal invocation uses one replay worker and first takes one of two simulation slots in the Modal Dict `scorecard-replay-simulation-slots` (`modal_uk_event.SimulationSlots`), waiting while both are held, so Modal -invocations never run more than two simulations at once. Local simulations -are not in that lease: count them yourself toward the overall maximum of two. A full event -plus one focused repeat uses both slots. Remote wall time is recorded +invocations never run more than two simulations at once. How the slots hold: + +* A slot changes hands only through `put(..., skip_if_exists=True)`, Modal's + atomic create-if-absent: each holder, and each renewal, creates the slot's + next generation key, and nothing is deleted. A holder writes only its own + key, and only to release or shorten it. +* The lease (4 hours) is renewed after the image build, just before the + remote call, and the remote function refuses to start unless its whole + 3-hour timeout fits inside the lease. A run therefore ends before its slot + can pass on, even if the client dies. +* A finished or never-started run frees its slot at once. After ctrl-C, + Modal's client tells the server to stop the app's tasks, and the slot frees + 10 minutes later. A client killed outright frees it when the lease lapses. +* A renewal after the lease lapsed is refused (exit 75): the slot may have + passed on. Waiting runs poll every 30 seconds in no set order. +* `tests/test_modal_uk_event.py` checks this call by call: Hypothesis + interleaves several holders' Dict calls with clock jumps, crashes, + interrupts and Modal's 7-day expiry of idle entries, and asserts that no two + runs overlap on a slot and that both slots can be taken once every lease + has lapsed. It assumes no process stalls for days inside one protocol step + (a few consecutive Dict calls). + +Local simulations are not in that lease: count them yourself toward the +overall maximum of two. A full event plus one focused repeat uses both +slots. Remote wall time is recorded separately in `MODAL_RECEIPT.json`; use the completed event's observed timings when planning the next lane. diff --git a/pipeline/modal_uk_event.py b/pipeline/modal_uk_event.py index b8d510ec..6a3a55ca 100644 --- a/pipeline/modal_uk_event.py +++ b/pipeline/modal_uk_event.py @@ -37,9 +37,11 @@ import os import platform import re +import secrets import sys import tempfile import time +from dataclasses import dataclass from pathlib import Path, PurePosixPath ROOT = Path(__file__).resolve().parent.parent @@ -102,9 +104,27 @@ SLOTS_DICT = "scorecard-replay-simulation-slots" SLOT_COUNT = 2 -# Longer than the function timeout (10,800 s), so a live run never loses its -# slot; a holder that died without releasing frees it after this. +# The function timeout (10,800 s) plus an hour for the container to be +# scheduled. The lease is renewed just before the remote call, and the remote +# function refuses to start unless its whole timeout fits inside the lease, +# so a run always ends before its slot can pass to anyone else. LEASE_SECONDS = 4 * 60 * 60 +# After an interrupt, Modal's client tells the server to stop the ephemeral +# app's tasks (modal/runner.py, _status_based_disconnect); the slot frees +# this long afterwards rather than at once. +INTERRUPT_GRACE_SECONDS = 10 * 60 + + +class LeaseLost(RuntimeError): + """The slot passed to another holder before this one renewed it.""" + + +@dataclass(frozen=True) +class Lease: + slot: int + chain: str + generation: int + expires: float class SimulationSlots: @@ -113,36 +133,116 @@ class SimulationSlots: Each invocation runs one worker, and separate invocations share no Modal-side limit, so a third concurrent run would exceed the two the shared account allows. Before starting, an invocation takes one of two - slots in a named Modal Dict: ``put(key, value, skip_if_exists=True)`` is - an atomic create-if-absent on the server. An expired slot is reclaimed by - whoever first creates its one-time reclaim token, so two reclaimers - cannot both take it. + slots in a named Modal Dict. + + The only operation that changes who holds a slot is + ``put(key, value, skip_if_exists=True)``, an atomic create-if-absent on + the server (modal/dict.py, ``DictUpdateRequest(if_not_exists=...)``). + Each change of holder, and each renewal, creates the slot's next + generation key, ``claim-{slot}-{chain}-{generation}``; exactly one caller + creates any generation, and nothing is ever deleted. A caller may take + the next generation only when the newest one is released or expired. + A holder writes only its own generation, and only to release it or to + shorten it, so no write can take a slot from anyone. + + ``head-{slot}`` is a hint naming the newest generation; readers walk + forward from it, so a stale hint costs reads, never safety. Its random + chain id separates generations written after Modal expires idle entries + (7 days without a read or write) from any older key that survives. + + Every Dict call is one step of a generator (``_try_steps``, + ``_renew_steps``, ``_release_steps``), so tests interleave several + holders call by call. The guarantee assumes no process stalls for days + inside one of these steps (a few consecutive Dict calls): a write made + after Modal expired the keys it read could recreate one. Stalls between + steps (an image build, a run, a closed laptop) are safe. """ def __init__( - self, store, *, holder, count=SLOT_COUNT, lease=LEASE_SECONDS, clock=time.time + self, + store, + *, + holder, + count=SLOT_COUNT, + lease=LEASE_SECONDS, + clock=time.time, + chain_id=lambda: secrets.token_hex(8), ): - self.store, self.holder, self.count, self.lease, self.clock = ( - store, - holder, - count, - lease, - clock, - ) - - def _try(self, slot: int) -> bool: - key = f"slot-{slot}" - claim = {"holder": self.holder, "expires": self.clock() + self.lease} - if self.store.put(key, claim, skip_if_exists=True): - return True - current = self.store.get(key) - if current is None or current["expires"] > self.clock(): - return False - token = f"reclaim-{slot}-{current['holder']}-{current['expires']}" - if not self.store.put(token, self.holder, skip_if_exists=True): - return False - self.store.pop(key, None) - return bool(self.store.put(key, claim, skip_if_exists=True)) + self.store, self.holder, self.count = store, holder, count + self.lease, self.clock, self.chain_id = lease, clock, chain_id + + @staticmethod + def claim_key(slot: int, chain: str, generation: int) -> str: + return f"claim-{slot}-{chain}-{generation}" + + def _claim(self, expires: float, *, released: bool = False) -> dict: + return {"holder": self.holder, "expires": expires, "released": released} + + def _try_steps(self, slot: int): + head_key = f"head-{slot}" + head = yield ("get", head_key) + if head is None: + fresh = {"chain": self.chain_id(), "generation": -1} + yield ("put", head_key, fresh, True) + head = yield ("get", head_key) + if head is None: + return None + chain, generation = head["chain"], head["generation"] + newest = None + if generation >= 0: + newest = yield ("get", self.claim_key(slot, chain, generation)) + while True: + following = yield ("get", self.claim_key(slot, chain, generation + 1)) + if following is None: + break + generation, newest = generation + 1, following + if ( + newest is not None + and not newest["released"] + and newest["expires"] > self.clock() + ): + return None + expires = self.clock() + self.lease + key = self.claim_key(slot, chain, generation + 1) + if not (yield ("put", key, self._claim(expires), True)): + return None + yield ("put", head_key, {"chain": chain, "generation": generation + 1}, False) + return Lease(slot, chain, generation + 1, expires) + + def _renew_steps(self, lease: Lease): + # While a lease is live no one else can create its next generation, + # so that key has never existed and cannot be one Modal expired; + # after it lapses the slot may have moved on, so renewal stops. + if self.clock() >= lease.expires: + raise LeaseLost(f"simulation slot {lease.slot} lease lapsed") + expires = self.clock() + self.lease + key = self.claim_key(lease.slot, lease.chain, lease.generation + 1) + if not (yield ("put", key, self._claim(expires), True)): + raise LeaseLost(f"simulation slot {lease.slot} passed to another holder") + head = {"chain": lease.chain, "generation": lease.generation + 1} + yield ("put", f"head-{lease.slot}", head, False) + return Lease(lease.slot, lease.chain, lease.generation + 1, expires) + + def _release_steps(self, lease: Lease, *, grace: float = 0.0): + key = self.claim_key(lease.slot, lease.chain, lease.generation) + if grace: + claim = self._claim(min(lease.expires, self.clock() + grace)) + else: + claim = self._claim(lease.expires, released=True) + yield ("put", key, claim, False) + + def _drive(self, steps): + """Run one protocol generator against the store.""" + try: + call = next(steps) + while True: + if call[0] == "get": + result = self.store.get(call[1]) + else: + result = self.store.put(call[1], call[2], skip_if_exists=call[3]) + call = steps.send(result) + except StopIteration as done: + return done.value def acquire( self, @@ -150,22 +250,26 @@ def acquire( wait_seconds: float = 6 * 60 * 60, poll_seconds: float = 30, sleep=time.sleep, - ) -> int: + ) -> Lease: deadline = self.clock() + wait_seconds while True: for slot in range(self.count): - if self._try(slot): - return slot + lease = self._drive(self._try_steps(slot)) + if lease is not None: + return lease if self.clock() >= deadline: raise TimeoutError("no simulation slot became free") print("[modal] both simulation slots are held; waiting", flush=True) sleep(poll_seconds) - def release(self, slot: int) -> None: - key = f"slot-{slot}" - current = self.store.get(key) - if current is not None and current["holder"] == self.holder: - self.store.pop(key, None) + def renew(self, lease: Lease) -> Lease: + """A fresh full lease, taken just before the remote call starts.""" + return self._drive(self._renew_steps(lease)) + + def release(self, lease: Lease, *, grace: float = 0.0) -> None: + """Free the slot now, or ``grace`` seconds from now when the remote + run may still be stopping.""" + self._drive(self._release_steps(lease, grace=grace)) def canonical_bytes(value: object) -> bytes: @@ -540,6 +644,7 @@ def modal_function(mounts: dict[str, Path]): for relative, path in sorted(mounts.items()): image = image.add_local_file(path, f"{REMOTE_ROOT}/{relative}", copy=False) app = modal.App("scorecard-uk-event-replay", include_source=False) + timeout_seconds = RESOURCE_LIMITS["timeout_seconds"] @app.function( image=image, @@ -547,19 +652,23 @@ def modal_function(mounts: dict[str, Path]): include_source=False, cpu=(2, 2), memory=(32768, 32768), - timeout=10800, + timeout=timeout_seconds, max_containers=1, retries=0, block_network=True, restrict_modal_access=True, ) @modal.concurrent(max_inputs=1) - def replay(payload: dict) -> dict: + def replay(payload: dict, lease_expires: float) -> dict: import hashlib import json import subprocess + import time from pathlib import Path + # A run that could outlive its simulation slot does not start. + if time.time() + timeout_seconds > lease_expires: + raise RuntimeError("the simulation slot lease does not cover the timeout") # Inherit stdout/stderr so Modal streams phase progress as it happens. subprocess.run( [ @@ -713,12 +822,23 @@ def main(argv: list[str] | None = None) -> int: modal.Dict.from_name(SLOTS_DICT, create_if_missing=True), holder=f"{platform.node()}:{os.getpid()}:{digest_bytes(canonical_bytes(request))[:12]}", ) - slot = slots.acquire() + lease = slots.acquire() + started = False try: with modal.enable_output(), app.run(): - response = replay.remote(payload) + # The image build can take a while: renew so the lease covers + # the remote run from its start. + lease = slots.renew(lease) + started = True + response = replay.remote(payload, lease.expires) + except LeaseLost: + print("Modal run not started: its simulation slot lapsed", file=sys.stderr) + return 75 finally: - slots.release(slot) + # A returned or never-started run frees its slot now; an interrupted + # one frees it after Modal has had time to stop the remote task. + finished = response is not None or not started + slots.release(lease, grace=0.0 if finished else INTERRUPT_GRACE_SECONDS) if response is None: print("Modal run interrupted; no output downloaded", file=sys.stderr) return 130 diff --git a/tests/test_modal_uk_event.py b/tests/test_modal_uk_event.py index 0194adeb..8b0dbf2d 100644 --- a/tests/test_modal_uk_event.py +++ b/tests/test_modal_uk_event.py @@ -1,8 +1,11 @@ """Offline Modal transport checks; no Modal installation or country imports.""" import hashlib +import itertools import json +import subprocess # noqa: F401 - patched in the factory test import sys +import time from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace @@ -264,7 +267,7 @@ def __enter__(self): def __exit__(self, kind, value, traceback): return kind is KeyboardInterrupt - def interrupted_remote(payload): + def interrupted_remote(payload, lease_expires): raise KeyboardInterrupt monkeypatch.setattr(runner, "validate_selection", lambda request: None) @@ -292,8 +295,10 @@ def interrupted_remote(payload): ) assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 130 assert "interrupted; no output downloaded" in capsys.readouterr().err - # The interrupted run gave its simulation slot back. - assert not [key for key in slots.data if key.startswith("slot-")] + # The interrupted run's slot frees once Modal has had time to stop it. + claims = [v for k, v in slots.data.items() if k.startswith("claim-")] + assert claims[-1]["released"] is False + assert claims[-1]["expires"] <= time.time() + runner.INTERRUPT_GRACE_SECONDS def test_modal_factory_has_one_blocked_network_worker_and_explicit_uv_venv( @@ -341,7 +346,16 @@ def concurrent(**kwargs): monkeypatch.setattr( runner.importlib.metadata, "version", lambda name: runner.MODAL_VERSION ) - runner.modal_function(mounts) + _, _, replay = runner.modal_function(mounts) + # The remote refuses to start unless its whole timeout fits in the lease. + monkeypatch.setattr( + sys.modules["subprocess"], + "run", + lambda *args, **kwargs: pytest.fail("a short lease must not start a run"), + ) + short = time.time() + runner.RESOURCE_LIMITS["timeout_seconds"] - 60 + with pytest.raises(RuntimeError, match="lease does not cover"): + replay({"request": {}}, short) configuration = recorded["function"] assert configuration["cpu"] == (2, 2) assert configuration["memory"] == (32768, 32768) @@ -370,23 +384,39 @@ def concurrent(**kwargs): class FakeDict: - """A Modal Dict's put/get/pop, with put(skip_if_exists=True) atomic.""" + """A Modal Dict's get/put, with put(skip_if_exists=True) atomic and, given + a clock, entries expiring after 7 days without a read or write (as + modal/dict.py documents).""" - def __init__(self): - self.data = {} + TTL = 7 * 24 * 60 * 60 + + def __init__(self, clock=None): + self.data, self.touched, self.clock = {}, {}, clock + + def _expire(self): + if self.clock is None: + return + for key in [k for k, t in self.touched.items() if self.clock() - t > self.TTL]: + del self.data[key], self.touched[key] + + def _touch(self, key): + if self.clock is not None and key in self.data: + self.touched[key] = self.clock() def put(self, key, value, *, skip_if_exists=False): + self._expire() if skip_if_exists and key in self.data: + self._touch(key) return False self.data[key] = value + self._touch(key) return True def get(self, key, default=None): + self._expire() + self._touch(key) return self.data.get(key, default) - def pop(self, key, default=None): - return self.data.pop(key, default) - class Clock: def __init__(self): @@ -396,12 +426,63 @@ def __call__(self): return self.now +CHAIN_IDS = itertools.count() + + +def slots_for(store, clock, holder, **kwargs): + return runner.SimulationSlots( + store, + holder=holder, + clock=clock, + chain_id=lambda: f"chain{next(CHAIN_IDS)}", + **kwargs, + ) + + +def run_steps(store, steps, *, limit=None): + """Drive a protocol generator for at most ``limit`` Dict calls; return + (finished, value).""" + calls = 0 + try: + call = next(steps) + while True: + if limit is not None and calls == limit: + return False, None + calls += 1 + if call[0] == "get": + result = store.get(call[1]) + else: + result = store.put(call[1], call[2], skip_if_exists=call[3]) + call = steps.send(result) + except StopIteration as done: + return True, done.value + + +def holders_of(store, clock): + """Each slot's newest generation, if it is held now.""" + held = {} + for key, head in store.data.items(): + if not key.startswith("head-"): + continue + slot = int(key.split("-")[1]) + generation, newest = head["generation"], None + while True: + claim = store.data.get(f"claim-{slot}-{head['chain']}-{generation + 1}") + if claim is None: + break + generation, newest = generation + 1, claim + newest = newest or store.data.get(f"claim-{slot}-{head['chain']}-{generation}") + if newest and not newest["released"] and newest["expires"] > clock(): + held[slot] = newest["holder"] + return held + + def test_two_slots_then_a_third_run_waits(): - store, clock = FakeDict(), Clock() - a = runner.SimulationSlots(store, holder="a", clock=clock) - b = runner.SimulationSlots(store, holder="b", clock=clock) - c = runner.SimulationSlots(store, holder="c", clock=clock) - assert {a.acquire(), b.acquire()} == {0, 1} + clock = Clock() + store = FakeDict(clock) + a, b, c = (slots_for(store, clock, h) for h in "abc") + leases = {"a": a.acquire(), "b": b.acquire()} + assert {lease.slot for lease in leases.values()} == {0, 1} sleeps = [] def sleep(seconds): @@ -411,55 +492,252 @@ def sleep(seconds): with pytest.raises(TimeoutError): c.acquire(wait_seconds=60, poll_seconds=30, sleep=sleep) assert sleeps == [30, 30] - c.release(0) # not its slot: no effect - assert store.get("slot-0")["holder"] in ("a", "b") - a.release(0 if store.get("slot-0")["holder"] == "a" else 1) - assert c.acquire(wait_seconds=0, sleep=sleep) in (0, 1) + a.release(leases["a"]) + assert c.acquire(wait_seconds=0, sleep=sleep).slot == leases["a"].slot + assert holders_of(store, clock) == { + leases["a"].slot: "c", + leases["b"].slot: "b", + } -def test_an_expired_slot_is_reclaimed_once(): - store, clock = FakeDict(), Clock() - dead = runner.SimulationSlots(store, holder="dead", clock=clock, count=1) - dead.acquire() +def test_a_stale_release_cannot_free_a_newer_holder(): + # Review of #166, 2(a): a holder past its lease releases after the slot + # passed on; the release writes only its own generation. + clock = Clock() + store = FakeDict(clock) + late = slots_for(store, clock, "late", count=1) + old = late.acquire() clock.now += runner.LEASE_SECONDS + 1 - first = runner.SimulationSlots(store, holder="first", clock=clock, count=1) - second = runner.SimulationSlots(store, holder="second", clock=clock, count=1) - assert first._try(0) is True - assert second._try(0) is False - assert store.get("slot-0")["holder"] == "first" - dead.release(0) # the dead holder cannot free the new lease - assert store.get("slot-0")["holder"] == "first" - - -def test_a_live_lease_outlasts_the_function_timeout(): - assert runner.LEASE_SECONDS > runner.RESOURCE_LIMITS["timeout_seconds"] - + successor = slots_for(store, clock, "successor", count=1).acquire() + late.release(old) + assert holders_of(store, clock) == {0: "successor"} + third = slots_for(store, clock, "third", count=1) + assert run_steps(store, third._try_steps(0)) == (True, None) + assert successor.generation == old.generation + 1 + + +def test_a_reclaimer_cannot_displace_a_claim_made_after_it_looked(): + # Review of #166, 2(b): A sees the slot free, then C takes it before A + # writes; A's create-if-absent of the same generation fails. + clock = Clock() + store = FakeDict(clock) + holder = slots_for(store, clock, "holder", count=1) + lease = holder.acquire() + a = slots_for(store, clock, "a", count=1) + c = slots_for(store, clock, "c", count=1) + clock.now += runner.LEASE_SECONDS + 1 + steps = a._try_steps(0) + call = next(steps) + while call[0] == "get": # A reads up to its create + call = steps.send(store.get(call[1])) + holder.release(lease) + assert run_steps(store, c._try_steps(0))[1] is not None + with pytest.raises(StopIteration) as done: # A's create now fails + steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + assert done.value.value is None + assert holders_of(store, clock) == {0: "c"} + + +@pytest.mark.parametrize("calls", range(8)) +def test_a_holder_that_dies_at_any_step_frees_its_slot_after_the_lease(calls): + # Review of #166, 1: a caller that dies mid-protocol never loses a slot + # for good. + clock = Clock() + store = FakeDict(clock) + dying = slots_for(store, clock, "dying", count=1) + run_steps(store, dying._try_steps(0), limit=calls) + clock.now += runner.LEASE_SECONDS + 1 + assert slots_for(store, clock, "next", count=1).acquire(wait_seconds=0).slot == 0 + + +def test_renewal_after_the_lease_lapsed_is_refused_even_untaken(): + # A laptop closed during the image build: the slot may have moved on and + # Modal may have expired the keys it used, so the run does not start. + clock = Clock() + store = FakeDict(clock) + slots = slots_for(store, clock, "a", count=1) + lease = slots.acquire() + clock.now += 8 * 24 * 60 * 60 + with pytest.raises(runner.LeaseLost): + slots.renew(lease) + + +def test_a_stale_renewal_cannot_recreate_an_expired_generation(): + # The slot changes hands across days and Modal expires the old keys; a + # client back from a week asleep must not renew into a key that no + # longer exists while someone else holds the slot. + day = 24 * 60 * 60 + clock = Clock() + store = FakeDict(clock) + sleeper = slots_for(store, clock, "sleeper", count=1) + stale = sleeper.acquire() + clock.now += runner.LEASE_SECONDS + 1 + for name in ("b", "c"): + slots = slots_for(store, clock, name, count=1) + slots.release(slots.acquire()) + clock.now += day + clock.now += 8 * day # every old key idles past Modal's 7 days + current = slots_for(store, clock, "current", count=1).acquire() + with pytest.raises(runner.LeaseLost): + sleeper.renew(stale) + assert holders_of(store, clock) == {0: "current"} + assert current.slot == 0 + + +def test_renewal_after_a_takeover_is_refused(): + clock = Clock() + store = FakeDict(clock) + slow = slots_for(store, clock, "slow", count=1) + lease = slow.acquire() + clock.now += runner.LEASE_SECONDS + 1 + slots_for(store, clock, "fast", count=1).acquire() + with pytest.raises(runner.LeaseLost): + slow.renew(lease) + assert holders_of(store, clock) == {0: "fast"} + + +def test_renewal_extends_from_now_and_keeps_the_slot(): + clock = Clock() + store = FakeDict(clock) + slots = slots_for(store, clock, "a", count=1) + lease = slots.acquire() + clock.now += 3_000 # image build + renewed = slots.renew(lease) + assert renewed.expires == clock.now + runner.LEASE_SECONDS + assert renewed.generation == lease.generation + 1 + clock.now += runner.LEASE_SECONDS - 1 + assert holders_of(store, clock) == {0: "a"} + + +def test_an_interrupt_frees_the_slot_after_the_grace_period(): + clock = Clock() + store = FakeDict(clock) + slots = slots_for(store, clock, "a", count=1) + lease = slots.acquire() + slots.release(lease, grace=runner.INTERRUPT_GRACE_SECONDS) + assert holders_of(store, clock) == {0: "a"} + clock.now += runner.INTERRUPT_GRACE_SECONDS + assert holders_of(store, clock) == {} + + +def test_the_lease_covers_the_timeout_and_scheduling(): + timeout = runner.RESOURCE_LIMITS["timeout_seconds"] + assert runner.LEASE_SECONDS >= timeout + 60 * 60 + assert runner.INTERRUPT_GRACE_SECONDS < runner.LEASE_SECONDS + + +TIMEOUT = runner.RESOURCE_LIMITS["timeout_seconds"] +# Modal's disconnect stops an ephemeral app's tasks "in a few seconds" +# (modal/runner.py); the model allows a minute, inside the grace period. +STOP_AFTER_DISCONNECT = 60 + + +def holder_program(slots): + """main()'s use of the slots, one Dict call or event at a time.""" + lease = None + while lease is None: + for slot in range(slots.count): + lease = yield from slots._try_steps(slot) + if lease is not None: + break + else: + yield ("pause",) + yield ("pause",) # image build + try: + lease = yield from slots._renew_steps(lease) + except runner.LeaseLost: + yield from slots._release_steps(lease) + return + outcome = yield ("start", lease) + if outcome == "running": + outcome = yield ("await",) + grace = 0.0 if outcome == "returned" else runner.INTERRUPT_GRACE_SECONDS + yield from slots._release_steps(lease, grace=grace) + + +@settings(max_examples=600, deadline=None) +@given(st.data()) +def test_interleaved_holders_never_overlap_on_a_slot_and_never_lose_one(data): + """Model check: holders interleaved call by call, with clock jumps, + crashes (no release; a started run continues to its end), interrupts and + Dict entry expiry. Safety: two runs never overlap on one slot, so at most + SLOT_COUNT run at once. Liveness: once every lease has lapsed, both slots + can be taken again.""" + clock = Clock() + store = FakeDict(clock) + count = data.draw(st.integers(2, 4), label="holders") + actors = [] + for index in range(count): + program = holder_program(slots_for(store, clock, f"h{index}")) + actors.append({"program": program, "call": next(program), "done": False}) + runs = [] + + def send(actor, value): + try: + actor["call"] = actor["program"].send(value) + except StopIteration: + actor["done"] = True + + for _ in range(data.draw(st.integers(0, 150), label="actions")): + action = data.draw( + st.sampled_from(["step", "step", "step", "tick", "crash", "interrupt"]) + ) + if action == "tick": + seconds = data.draw( + st.sampled_from([1, 60, 600, 3600, TIMEOUT, 4 * 3600, 8 * 86400]) + ) + # The stated assumption: no stall of days inside one protocol + # step (a pending Dict call); between steps, anything goes. + mid_step = any( + not a["done"] and a["call"][0] in ("get", "put") for a in actors + ) + clock.now += min(seconds, 6 * 86400) if mid_step else seconds + continue + live = [a for a in actors if not a["done"]] + if not live: + break + if action == "interrupt": # ctrl-C on a client whose run is going + awaiting = [a for a in live if a["call"][0] == "await"] + if awaiting: + actor = data.draw(st.sampled_from(awaiting)) + run = actor["run"] + run["end"] = min(run["end"], clock.now + STOP_AFTER_DISCONNECT) + send(actor, "interrupted") + continue + actor = data.draw(st.sampled_from(live)) + if action == "crash": + actor["done"] = True + continue + call = actor["call"] + if call[0] == "get": + send(actor, store.get(call[1])) + elif call[0] == "put": + send(actor, store.put(call[1], call[2], skip_if_exists=call[3])) + elif call[0] == "pause": + send(actor, None) + elif call[0] == "start": + lease = call[1] + if clock.now + TIMEOUT > lease.expires: # the remote guard refuses + send(actor, "refused") + continue + duration = data.draw(st.integers(1, TIMEOUT), label="run seconds") + run = {"slot": lease.slot, "start": clock.now, "end": clock.now + duration} + for other in runs: + if other["slot"] == run["slot"]: + assert other["end"] <= run["start"], (other, run) + runs.append(run) + actor["run"] = run + send(actor, "running") + elif call[0] == "await" and clock.now >= actor["run"]["end"]: + send(actor, "returned") + running = [r for r in runs if r["start"] <= clock.now < r["end"]] + assert len(running) <= runner.SLOT_COUNT -@settings(max_examples=300, deadline=None) -@given( - st.lists( - st.tuples(st.sampled_from("abcd"), st.sampled_from(["try", "release", "tick"])), - max_size=40, + clock.now += runner.LEASE_SECONDS + 1 + fresh = [slots_for(store, clock, f"fresh{n}") for n in range(runner.SLOT_COUNT)] + taken = [s.acquire(wait_seconds=0) for s in fresh] + assert {lease.slot for lease in taken} == set(range(runner.SLOT_COUNT)), ( + taken, + clock.now, + store.data, ) -) -def test_never_more_than_two_live_holders(operations): - store, clock = FakeDict(), Clock() - slots = {h: runner.SimulationSlots(store, holder=h, clock=clock) for h in "abcd"} - held = {} - for holder, action in operations: - if action == "tick": - clock.now += 600 # well inside a lease - elif action == "try" and holder not in held: - for slot in range(runner.SLOT_COUNT): - if slots[holder]._try(slot): - held[holder] = slot - break - elif action == "release" and holder in held: - slots[holder].release(held.pop(holder)) - live = { - value["holder"] - for key, value in store.data.items() - if key.startswith("slot-") and value["expires"] > clock.now - } - assert len(live) <= runner.SLOT_COUNT - assert live == set(held) From 4a5a27262bc4b7cb8712d295ba0847c68686f330 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 18:00:52 -0400 Subject: [PATCH 3/5] Modal slots: hold the claim when Modal was not told to stop the run Review of dff81d0: - Modal's runner re-raises a lost connection (modal.exception .ConnectionError) without sending its stop request (modal/runner.py, the ConnectionError branch of _run_app), yet main() freed the slot ten minutes later while the container could run for three hours. The grace now applies only when app.run() itself exits after a caught ctrl-C, where the runner awaited the stop request; any other exception after the run started leaves the claim until the lease lapses, which the remote guard puts after the run's end. A remote error now holds the slot until then too (conservative). - A writer stalled across Modal's 7-day expiry could overwrite head-{slot} with its dead chain and hide a live holder. A protocol step (acquire or renew) that took over a day now gives its claim back and writes no head. - The remote guard keeps a 10-minute margin inside the lease for clock differences and kill latency. - The model check now caps stalls per protocol step, not per tick, and models a lost connection (the run continues to its own end, nothing is written). New tests: a week-stalled acquire cannot replace a live chain; a lost connection keeps the claim at its full lease; a finished run frees the slot at once; the stale-renewal scenario now asserts that the key it would recreate is absent. 20,000 generated schedules hold; the head test fails when the stale-step bail is removed. Co-Authored-By: Claude Opus 5.5 --- docs/uk_replay/RECIPE.md | 18 ++-- pipeline/modal_uk_event.py | 46 +++++++--- tests/test_modal_uk_event.py | 169 ++++++++++++++++++++++++++++------- 3 files changed, 184 insertions(+), 49 deletions(-) diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index 43a24d14..816568d0 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -606,14 +606,20 @@ invocations never run more than two simulations at once. How the slots hold: * A slot changes hands only through `put(..., skip_if_exists=True)`, Modal's atomic create-if-absent: each holder, and each renewal, creates the slot's next generation key, and nothing is deleted. A holder writes only its own - key, and only to release or shorten it. + key, and only to release or shorten it. A head key names the newest + generation as a hint; a step that took over a day writes no head and + gives its claim back, so keys Modal expired (7 days idle) can never be + pointed back to. * The lease (4 hours) is renewed after the image build, just before the remote call, and the remote function refuses to start unless its whole - 3-hour timeout fits inside the lease. A run therefore ends before its slot - can pass on, even if the client dies. -* A finished or never-started run frees its slot at once. After ctrl-C, - Modal's client tells the server to stop the app's tasks, and the slot frees - 10 minutes later. A client killed outright frees it when the lease lapses. + 3-hour timeout plus a 10-minute margin fits inside the lease. A run + therefore ends before its slot can pass on, even if the client dies. +* A finished or never-started run frees its slot at once. When `app.run()` + exits after ctrl-C, Modal's client has already asked the server to stop + the app's tasks, and the slot frees 10 minutes later. Anything else after + the run started (a lost connection, which Modal's runner re-raises + without that request; a remote error; a failed stop request; a client + killed outright) leaves the claim until the lease lapses. * A renewal after the lease lapsed is refused (exit 75): the slot may have passed on. Waiting runs poll every 30 seconds in no set order. * `tests/test_modal_uk_event.py` checks this call by call: Hypothesis diff --git a/pipeline/modal_uk_event.py b/pipeline/modal_uk_event.py index 6a3a55ca..cf3cb1e0 100644 --- a/pipeline/modal_uk_event.py +++ b/pipeline/modal_uk_event.py @@ -109,10 +109,16 @@ # function refuses to start unless its whole timeout fits inside the lease, # so a run always ends before its slot can pass to anyone else. LEASE_SECONDS = 4 * 60 * 60 -# After an interrupt, Modal's client tells the server to stop the ephemeral -# app's tasks (modal/runner.py, _status_based_disconnect); the slot frees -# this long afterwards rather than at once. +# The remote guard also leaves this much of the lease unused, for clock +# differences between hosts and the time Modal takes to stop a timed-out run. +GUARD_MARGIN_SECONDS = 10 * 60 +# When app.run() exits after the runner caught ctrl-C, it has already asked +# the server to stop the ephemeral app's tasks (modal/runner.py, +# _status_based_disconnect); the slot frees this long afterwards. INTERRUPT_GRACE_SECONDS = 10 * 60 +# A protocol step (a few consecutive Dict calls) that took longer than this +# writes no head: Modal may have expired the keys it read (7 days idle). +STALE_STEP_SECONDS = 24 * 60 * 60 class LeaseLost(RuntimeError): @@ -187,6 +193,7 @@ def _try_steps(self, slot: int): head = yield ("get", head_key) if head is None: return None + looked = self.clock() chain, generation = head["chain"], head["generation"] newest = None if generation >= 0: @@ -206,6 +213,11 @@ def _try_steps(self, slot: int): key = self.claim_key(slot, chain, generation + 1) if not (yield ("put", key, self._claim(expires), True)): return None + if self.clock() - looked > STALE_STEP_SECONDS: + # What this step read may be gone: give the claim back and leave + # the head alone, so a stale chain never replaces a live one. + yield ("put", key, self._claim(expires, released=True), False) + return None yield ("put", head_key, {"chain": chain, "generation": generation + 1}, False) return Lease(slot, chain, generation + 1, expires) @@ -213,12 +225,16 @@ def _renew_steps(self, lease: Lease): # While a lease is live no one else can create its next generation, # so that key has never existed and cannot be one Modal expired; # after it lapses the slot may have moved on, so renewal stops. - if self.clock() >= lease.expires: + began = self.clock() + if began >= lease.expires: raise LeaseLost(f"simulation slot {lease.slot} lease lapsed") - expires = self.clock() + self.lease + expires = began + self.lease key = self.claim_key(lease.slot, lease.chain, lease.generation + 1) if not (yield ("put", key, self._claim(expires), True)): raise LeaseLost(f"simulation slot {lease.slot} passed to another holder") + if self.clock() - began > STALE_STEP_SECONDS: + yield ("put", key, self._claim(expires, released=True), False) + raise LeaseLost(f"simulation slot {lease.slot} renewal stalled") head = {"chain": lease.chain, "generation": lease.generation + 1} yield ("put", f"head-{lease.slot}", head, False) return Lease(lease.slot, lease.chain, lease.generation + 1, expires) @@ -645,6 +661,7 @@ def modal_function(mounts: dict[str, Path]): image = image.add_local_file(path, f"{REMOTE_ROOT}/{relative}", copy=False) app = modal.App("scorecard-uk-event-replay", include_source=False) timeout_seconds = RESOURCE_LIMITS["timeout_seconds"] + guard_seconds = timeout_seconds + GUARD_MARGIN_SECONDS @app.function( image=image, @@ -667,7 +684,7 @@ def replay(payload: dict, lease_expires: float) -> dict: from pathlib import Path # A run that could outlive its simulation slot does not start. - if time.time() + timeout_seconds > lease_expires: + if time.time() + guard_seconds > lease_expires: raise RuntimeError("the simulation slot lease does not cover the timeout") # Inherit stdout/stderr so Modal streams phase progress as it happens. subprocess.run( @@ -823,7 +840,7 @@ def main(argv: list[str] | None = None) -> int: holder=f"{platform.node()}:{os.getpid()}:{digest_bytes(canonical_bytes(request))[:12]}", ) lease = slots.acquire() - started = False + started = exited = False try: with modal.enable_output(), app.run(): # The image build can take a while: renew so the lease covers @@ -831,14 +848,21 @@ def main(argv: list[str] | None = None) -> int: lease = slots.renew(lease) started = True response = replay.remote(payload, lease.expires) + # app.run() returned: the run finished, or the runner caught ctrl-C + # after asking the server to stop the app's tasks. + exited = True except LeaseLost: print("Modal run not started: its simulation slot lapsed", file=sys.stderr) return 75 finally: - # A returned or never-started run frees its slot now; an interrupted - # one frees it after Modal has had time to stop the remote task. - finished = response is not None or not started - slots.release(lease, grace=0.0 if finished else INTERRUPT_GRACE_SECONDS) + if not started or response is not None: + slots.release(lease) + elif exited: + slots.release(lease, grace=INTERRUPT_GRACE_SECONDS) + # Otherwise (a lost connection, which the runner re-raises without a + # stop request; a remote error; an exception during the stop request) + # the remote run may still be going: the claim stays until the lease + # lapses, after the run's end by the remote guard. if response is None: print("Modal run interrupted; no output downloaded", file=sys.stderr) return 130 diff --git a/tests/test_modal_uk_event.py b/tests/test_modal_uk_event.py index 8b0dbf2d..4203165d 100644 --- a/tests/test_modal_uk_event.py +++ b/tests/test_modal_uk_event.py @@ -353,9 +353,10 @@ def concurrent(**kwargs): "run", lambda *args, **kwargs: pytest.fail("a short lease must not start a run"), ) - short = time.time() + runner.RESOURCE_LIMITS["timeout_seconds"] - 60 - with pytest.raises(RuntimeError, match="lease does not cover"): - replay({"request": {}}, short) + timeout = runner.RESOURCE_LIMITS["timeout_seconds"] + for slack in (-60, runner.GUARD_MARGIN_SECONDS - 60): + with pytest.raises(RuntimeError, match="lease does not cover"): + replay({"request": {}}, time.time() + timeout + slack) configuration = recorded["function"] assert configuration["cpu"] == (2, 2) assert configuration["memory"] == (32768, 32768) @@ -550,18 +551,6 @@ def test_a_holder_that_dies_at_any_step_frees_its_slot_after_the_lease(calls): assert slots_for(store, clock, "next", count=1).acquire(wait_seconds=0).slot == 0 -def test_renewal_after_the_lease_lapsed_is_refused_even_untaken(): - # A laptop closed during the image build: the slot may have moved on and - # Modal may have expired the keys it used, so the run does not start. - clock = Clock() - store = FakeDict(clock) - slots = slots_for(store, clock, "a", count=1) - lease = slots.acquire() - clock.now += 8 * 24 * 60 * 60 - with pytest.raises(runner.LeaseLost): - slots.renew(lease) - - def test_a_stale_renewal_cannot_recreate_an_expired_generation(): # The slot changes hands across days and Modal expires the old keys; a # client back from a week asleep must not renew into a key that no @@ -578,6 +567,8 @@ def test_a_stale_renewal_cannot_recreate_an_expired_generation(): clock.now += day clock.now += 8 * day # every old key idles past Modal's 7 days current = slots_for(store, clock, "current", count=1).acquire() + following = runner.SimulationSlots.claim_key(0, stale.chain, stale.generation + 1) + assert store.get(following) is None # a create here would succeed with pytest.raises(runner.LeaseLost): sleeper.renew(stale) assert holders_of(store, clock) == {0: "current"} @@ -620,9 +611,10 @@ def test_an_interrupt_frees_the_slot_after_the_grace_period(): assert holders_of(store, clock) == {} -def test_the_lease_covers_the_timeout_and_scheduling(): +def test_the_lease_covers_the_timeout_margin_and_scheduling(): timeout = runner.RESOURCE_LIMITS["timeout_seconds"] - assert runner.LEASE_SECONDS >= timeout + 60 * 60 + assert runner.LEASE_SECONDS >= timeout + runner.GUARD_MARGIN_SECONDS + 30 * 60 + assert runner.STALE_STEP_SECONDS < FakeDict.TTL assert runner.INTERRUPT_GRACE_SECONDS < runner.LEASE_SECONDS @@ -651,8 +643,13 @@ def holder_program(slots): outcome = yield ("start", lease) if outcome == "running": outcome = yield ("await",) - grace = 0.0 if outcome == "returned" else runner.INTERRUPT_GRACE_SECONDS - yield from slots._release_steps(lease, grace=grace) + # main(): a returned run frees the slot now; after a caught ctrl-C + # (app.run() exited, the stop request sent) it frees after the grace; + # after a lost connection or a remote error nothing is written. + if outcome == "returned": + yield from slots._release_steps(lease) + elif outcome == "interrupted": + yield from slots._release_steps(lease, grace=runner.INTERRUPT_GRACE_SECONDS) @settings(max_examples=600, deadline=None) @@ -669,40 +666,57 @@ def test_interleaved_holders_never_overlap_on_a_slot_and_never_lose_one(data): actors = [] for index in range(count): program = holder_program(slots_for(store, clock, f"h{index}")) - actors.append({"program": program, "call": next(program), "done": False}) + actors.append( + {"program": program, "call": next(program), "done": False, "since": 1e3} + ) runs = [] def send(actor, value): + before = actor["call"][0] try: actor["call"] = actor["program"].send(value) except StopIteration: actor["done"] = True + return + if actor["call"][0] in ("get", "put") and before not in ("get", "put"): + actor["since"] = clock.now # a protocol step begins for _ in range(data.draw(st.integers(0, 150), label="actions")): action = data.draw( - st.sampled_from(["step", "step", "step", "tick", "crash", "interrupt"]) + st.sampled_from( + ["step", "step", "step", "tick", "crash", "interrupt", "disconnect"] + ) ) if action == "tick": seconds = data.draw( st.sampled_from([1, 60, 600, 3600, TIMEOUT, 4 * 3600, 8 * 86400]) ) - # The stated assumption: no stall of days inside one protocol - # step (a pending Dict call); between steps, anything goes. - mid_step = any( - not a["done"] and a["call"][0] in ("get", "put") for a in actors + # The stated assumption: no protocol step (consecutive Dict + # calls) spans 7 days, so Modal never expires a key a step read + # before the step ends; between steps, anything goes. + limit = min( + ( + 6 * 86400 - (clock.now - a["since"]) + for a in actors + if not a["done"] and a["call"][0] in ("get", "put") + ), + default=seconds, ) - clock.now += min(seconds, 6 * 86400) if mid_step else seconds + clock.now += max(0, min(seconds, limit)) continue live = [a for a in actors if not a["done"]] if not live: break - if action == "interrupt": # ctrl-C on a client whose run is going + if action in ("interrupt", "disconnect"): awaiting = [a for a in live if a["call"][0] == "await"] if awaiting: actor = data.draw(st.sampled_from(awaiting)) - run = actor["run"] - run["end"] = min(run["end"], clock.now + STOP_AFTER_DISCONNECT) - send(actor, "interrupted") + if action == "interrupt": # ctrl-C: the stop request is sent + run = actor["run"] + run["end"] = min(run["end"], clock.now + STOP_AFTER_DISCONNECT) + send(actor, "interrupted") + else: # connection lost: the run goes on to its own end + send(actor, "connection_lost") continue actor = data.draw(st.sampled_from(live)) if action == "crash": @@ -717,8 +731,8 @@ def send(actor, value): send(actor, None) elif call[0] == "start": lease = call[1] - if clock.now + TIMEOUT > lease.expires: # the remote guard refuses - send(actor, "refused") + if clock.now + TIMEOUT + runner.GUARD_MARGIN_SECONDS > lease.expires: + send(actor, "refused") # the remote guard: a remote error continue duration = data.draw(st.integers(1, TIMEOUT), label="run seconds") run = {"slot": lease.slot, "start": clock.now, "end": clock.now + duration} @@ -741,3 +755,94 @@ def send(actor, value): clock.now, store.data, ) + + +def test_a_week_stalled_acquire_cannot_replace_a_live_chain(): + # Review of dff81d0, 2: A reads the head, stalls a week (Modal expires + # every key), B builds a new chain and holds the slot; A's late writes + # must not point the head back at A's dead chain. + day = 24 * 60 * 60 + clock = Clock() + store = FakeDict(clock) + old = slots_for(store, clock, "old", count=1) + old.release(old.acquire()) + a = slots_for(store, clock, "a", count=1) + steps = a._try_steps(0) + call = next(steps) + while call[0] == "get": + call = steps.send(store.get(call[1])) + clock.now += 8 * day + b = slots_for(store, clock, "b", count=1).acquire() + with pytest.raises(StopIteration) as done: + while True: + if call[0] == "get": + call = steps.send(store.get(call[1])) + else: + call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + assert done.value.value is None + assert store.get("head-0")["chain"] == b.chain + assert holders_of(store, clock) == {0: "b"} + third = slots_for(store, clock, "third", count=1) + assert run_steps(store, third._try_steps(0)) == (True, None) + + +def test_a_lost_connection_keeps_the_claim_until_the_lease_lapses( + monkeypatch, tmp_path +): + # Review of dff81d0, 1: the runner re-raises a lost connection without + # asking the server to stop the app, so the run may still be going. + class LostConnection(Exception): + pass + + def lost(payload, lease_expires): + raise LostConnection + + slots = FakeDict() + monkeypatch.setattr(runner, "validate_selection", lambda request: None) + monkeypatch.setattr( + runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) + ) + monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") + monkeypatch.setattr( + runner, + "modal_function", + lambda mounts: ( + SimpleNamespace( + enable_output=nullcontext, + Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), + ), + SimpleNamespace(run=nullcontext), + SimpleNamespace(remote=lost), + ), + ) + with pytest.raises(LostConnection): + runner.main(["--event", "autumn_budget_2024", "--execute"]) + claims = [v for k, v in slots.data.items() if k.startswith("claim-")] + newest = max(claims, key=lambda claim: claim["expires"]) + assert newest["released"] is False + assert newest["expires"] >= time.time() + runner.LEASE_SECONDS - 60 + + +def test_a_finished_run_frees_its_slot_at_once(monkeypatch, tmp_path): + slots = FakeDict() + monkeypatch.setattr(runner, "validate_selection", lambda request: None) + monkeypatch.setattr( + runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) + ) + monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") + monkeypatch.setattr(runner, "save_download", lambda *args: []) + monkeypatch.setattr( + runner, + "modal_function", + lambda mounts: ( + SimpleNamespace( + enable_output=nullcontext, + Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), + ), + SimpleNamespace(run=nullcontext), + SimpleNamespace(remote=lambda payload, lease_expires: {"files": {}}), + ), + ) + assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 0 + claims = [v for k, v in slots.data.items() if k.startswith("claim-")] + assert max(claims, key=lambda claim: claim["expires"])["released"] is True From bac8974411a0769eef9c5e2c5aa166c8fc378688 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 18:52:23 -0400 Subject: [PATCH 4/5] Modal slots: say exactly what holds under week-long stalls Review of 4a5a272: - The step's clock was read after the head response, so a response left unread through a week asleep looked fresh. It is now read before the first call. - A head write landing a week late, after its keys expired, could point readers at a dead chain and hide a live holder. A reader whose head names a claim that is gone now blocks the slot for one lease (any hidden holder's lease began before that write) instead of taking it. - A claim whose head write fails is given back at once, on acquire and on renewal, instead of holding the slot for a lease with no run. - A result that returned is downloaded even if leaving the app then fails. - The docstring and RECIPE now state the guarantee against Modal's 7-day entry expiry, the two detected long-stall cases, and the one case not covered (a head write 7+ days late naming a generation that still exists while later ones expired), instead of an unconditional claim. Tests: a head response read before a week asleep (with and without a claim), a head written a week late, a failed head write on acquire and renewal, a returned result surviving a failed app exit; the model's final phase allows one blocked lease. Each new protection's test fails when the protection is removed; 20,000 generated schedules hold. Co-Authored-By: Claude Opus 5.5 --- docs/uk_replay/RECIPE.md | 21 +++-- pipeline/modal_uk_event.py | 76 ++++++++++++---- tests/test_modal_uk_event.py | 169 +++++++++++++++++++++++++++++++++-- 3 files changed, 237 insertions(+), 29 deletions(-) diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index 816568d0..8c29aa4d 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -607,9 +607,16 @@ invocations never run more than two simulations at once. How the slots hold: atomic create-if-absent: each holder, and each renewal, creates the slot's next generation key, and nothing is deleted. A holder writes only its own key, and only to release or shorten it. A head key names the newest - generation as a hint; a step that took over a day writes no head and - gives its claim back, so keys Modal expired (7 days idle) can never be - pointed back to. + generation as a hint. +* Modal expires an entry after 7 days without a read or write, so the + guarantee is stated against that: with every protocol step (acquire or + renew, a few consecutive Dict calls) finishing within 7 days, two holders + never hold one slot, however long the gaps between steps. Longer stalls + are handled where a client can detect them: a step that took over a day + gives its claim back and writes no head, and a reader whose head names a + claim that is gone blocks the slot for one lease instead of taking it. + Not covered: a head write that arrives 7 or more days late and names a + generation that still exists while later ones have expired. * The lease (4 hours) is renewed after the image build, just before the remote call, and the remote function refuses to start unless its whole 3-hour timeout plus a 10-minute margin fits inside the lease. A run @@ -621,13 +628,15 @@ invocations never run more than two simulations at once. How the slots hold: without that request; a remote error; a failed stop request; a client killed outright) leaves the claim until the lease lapses. * A renewal after the lease lapsed is refused (exit 75): the slot may have - passed on. Waiting runs poll every 30 seconds in no set order. + passed on. Waiting runs poll every 30 seconds in no set order. A claim + whose head write fails is given back at once. * `tests/test_modal_uk_event.py` checks this call by call: Hypothesis interleaves several holders' Dict calls with clock jumps, crashes, interrupts and Modal's 7-day expiry of idle entries, and asserts that no two runs overlap on a slot and that both slots can be taken once every lease - has lapsed. It assumes no process stalls for days inside one protocol step - (a few consecutive Dict calls). + has lapsed (after one further lease where a reader blocked a slot). The + model keeps every step within 6 days; the longer stalls above have their + own deterministic tests. Local simulations are not in that lease: count them yourself toward the overall maximum of two. A full event plus one focused repeat uses both diff --git a/pipeline/modal_uk_event.py b/pipeline/modal_uk_event.py index cf3cb1e0..e0572f78 100644 --- a/pipeline/modal_uk_event.py +++ b/pipeline/modal_uk_event.py @@ -152,16 +152,25 @@ class SimulationSlots: shorten it, so no write can take a slot from anyone. ``head-{slot}`` is a hint naming the newest generation; readers walk - forward from it, so a stale hint costs reads, never safety. Its random - chain id separates generations written after Modal expires idle entries - (7 days without a read or write) from any older key that survives. + forward from it. Its random chain id separates generations written after + Modal expires idle entries (7 days without a read or write) from any + older key that survives. + + What is guaranteed, and when. With every protocol step (acquire or + renew: a few consecutive Dict calls) finishing within 7 days, two + holders never hold one slot. Stalls between steps (an image build, a + run, a closed laptop) are safe whatever their length. A step that stalls + longer is handled in the cases a client can detect: a step that took + over a day gives its claim back and writes no head, and a reader whose + head names a claim that is gone (a head written late, after its keys + expired) blocks the slot for one lease instead of taking it. One case + is not covered: a head write that reaches the server 7 or more days + late and names a generation that still exists while later ones have + expired. Every Dict call is one step of a generator (``_try_steps``, ``_renew_steps``, ``_release_steps``), so tests interleave several - holders call by call. The guarantee assumes no process stalls for days - inside one of these steps (a few consecutive Dict calls): a write made - after Modal expired the keys it read could recreate one. Stalls between - steps (an image build, a run, a closed laptop) are safe. + holders call by call. """ def __init__( @@ -186,6 +195,8 @@ def _claim(self, expires: float, *, released: bool = False) -> dict: def _try_steps(self, slot: int): head_key = f"head-{slot}" + # Before the first call: a response can sit unread through a stall. + looked = self.clock() head = yield ("get", head_key) if head is None: fresh = {"chain": self.chain_id(), "generation": -1} @@ -193,11 +204,23 @@ def _try_steps(self, slot: int): head = yield ("get", head_key) if head is None: return None - looked = self.clock() chain, generation = head["chain"], head["generation"] newest = None if generation >= 0: newest = yield ("get", self.claim_key(slot, chain, generation)) + if newest is None: + # A head is written after its claim, and readers touch both, + # so a head whose claim is gone was written late, after its + # keys expired, and may hide a live holder elsewhere. Any + # such holder's lease began before that write: block the + # slot for one lease instead of taking it. + blocker = { + "holder": f"blocked-by:{self.holder}", + "expires": self.clock() + self.lease, + "released": False, + } + yield ("put", self.claim_key(slot, chain, generation), blocker, True) + return None while True: following = yield ("get", self.claim_key(slot, chain, generation + 1)) if following is None: @@ -218,7 +241,13 @@ def _try_steps(self, slot: int): # the head alone, so a stale chain never replaces a live one. yield ("put", key, self._claim(expires, released=True), False) return None - yield ("put", head_key, {"chain": chain, "generation": generation + 1}, False) + try: + head = {"chain": chain, "generation": generation + 1} + yield ("put", head_key, head, False) + except Exception: + # The claim exists but the hint may not: do not hold the slot. + yield ("put", key, self._claim(expires, released=True), False) + raise return Lease(slot, chain, generation + 1, expires) def _renew_steps(self, lease: Lease): @@ -235,8 +264,12 @@ def _renew_steps(self, lease: Lease): if self.clock() - began > STALE_STEP_SECONDS: yield ("put", key, self._claim(expires, released=True), False) raise LeaseLost(f"simulation slot {lease.slot} renewal stalled") - head = {"chain": lease.chain, "generation": lease.generation + 1} - yield ("put", f"head-{lease.slot}", head, False) + try: + head = {"chain": lease.chain, "generation": lease.generation + 1} + yield ("put", f"head-{lease.slot}", head, False) + except Exception: + yield ("put", key, self._claim(expires, released=True), False) + raise return Lease(lease.slot, lease.chain, lease.generation + 1, expires) def _release_steps(self, lease: Lease, *, grace: float = 0.0): @@ -252,10 +285,18 @@ def _drive(self, steps): try: call = next(steps) while True: - if call[0] == "get": - result = self.store.get(call[1]) - else: - result = self.store.put(call[1], call[2], skip_if_exists=call[3]) + try: + if call[0] == "get": + result = self.store.get(call[1]) + else: + result = self.store.put( + call[1], call[2], skip_if_exists=call[3] + ) + except Exception as error: + # Let the step see a failed call, so it can give back a + # claim it just made. + call = steps.throw(error) + continue call = steps.send(result) except StopIteration as done: return done.value @@ -854,6 +895,11 @@ def main(argv: list[str] | None = None) -> int: except LeaseLost: print("Modal run not started: its simulation slot lapsed", file=sys.stderr) return 75 + except Exception as error: + if response is None: + raise + # The run returned its result; only leaving the app failed. + print(f"Modal app exit failed after the run returned: {error}", file=sys.stderr) finally: if not started or response is not None: slots.release(lease) diff --git a/tests/test_modal_uk_event.py b/tests/test_modal_uk_event.py index 4203165d..b2c25bc2 100644 --- a/tests/test_modal_uk_event.py +++ b/tests/test_modal_uk_event.py @@ -551,7 +551,7 @@ def test_a_holder_that_dies_at_any_step_frees_its_slot_after_the_lease(calls): assert slots_for(store, clock, "next", count=1).acquire(wait_seconds=0).slot == 0 -def test_a_stale_renewal_cannot_recreate_an_expired_generation(): +def test_a_lapsed_lease_is_not_renewed_even_when_its_next_key_is_free(): # The slot changes hands across days and Modal expires the old keys; a # client back from a week asleep must not renew into a key that no # longer exists while someone else holds the slot. @@ -747,14 +747,22 @@ def send(actor, value): running = [r for r in runs if r["start"] <= clock.now < r["end"]] assert len(running) <= runner.SLOT_COUNT + # Once every lease has lapsed both slots can be taken, after at most one + # further lease when a reader had to block a slot whose head outlived + # its claim. clock.now += runner.LEASE_SECONDS + 1 - fresh = [slots_for(store, clock, f"fresh{n}") for n in range(runner.SLOT_COUNT)] - taken = [s.acquire(wait_seconds=0) for s in fresh] - assert {lease.slot for lease in taken} == set(range(runner.SLOT_COUNT)), ( - taken, - clock.now, - store.data, - ) + taken = set() + for attempt in range(2): + for n in range(runner.SLOT_COUNT): + holder = slots_for(store, clock, f"fresh{attempt}-{n}") + for slot in range(runner.SLOT_COUNT): + if slot not in taken and holder._drive(holder._try_steps(slot)): + taken.add(slot) + break + if taken == set(range(runner.SLOT_COUNT)): + break + clock.now += runner.LEASE_SECONDS + 1 + assert taken == set(range(runner.SLOT_COUNT)), (taken, clock.now, store.data) def test_a_week_stalled_acquire_cannot_replace_a_live_chain(): @@ -846,3 +854,148 @@ def test_a_finished_run_frees_its_slot_at_once(monkeypatch, tmp_path): assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 0 claims = [v for k, v in slots.data.items() if k.startswith("claim-")] assert max(claims, key=lambda claim: claim["expires"])["released"] is True + + +def finish(store, steps, call): + """Drive a protocol generator from a pending call to its return value.""" + try: + while True: + if call[0] == "get": + call = steps.send(store.get(call[1])) + else: + call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + except StopIteration as done: + return done.value + + +def test_a_head_response_read_before_a_week_asleep_is_not_acted_on(): + # Review of 4a5a272, 1(a): the head was answered, then the machine slept + # a week before the step used it; the step must not look fresh. + day = 24 * 60 * 60 + clock = Clock() + store = FakeDict(clock) + old = slots_for(store, clock, "old", count=1) + old.release(old.acquire()) + a = slots_for(store, clock, "a", count=1) + steps = a._try_steps(0) + call = next(steps) + assert call == ("get", "head-0") + answered = store.get("head-0") # the server answers... + clock.now += 8 * day # ...and the response sits unread + b = slots_for(store, clock, "b", count=1).acquire() + assert finish(store, steps, steps.send(answered)) is None + assert store.get("head-0")["chain"] == b.chain + assert holders_of(store, clock) == {0: "b"} + + +def test_a_fresh_chain_head_read_before_a_week_asleep_is_not_acted_on(): + # The same stall on a chain with no claim yet (generation -1): only the + # step's own clock, read before its first call, shows it is stale. + day = 24 * 60 * 60 + clock = Clock() + store = FakeDict(clock) + store.put("head-0", {"chain": "abandoned", "generation": -1}) + a = slots_for(store, clock, "a", count=1) + steps = a._try_steps(0) + assert next(steps) == ("get", "head-0") + answered = store.get("head-0") + clock.now += 8 * day + b = slots_for(store, clock, "b", count=1).acquire() + assert finish(store, steps, steps.send(answered)) is None + assert store.get("head-0")["chain"] == b.chain + assert holders_of(store, clock) == {0: "b"} + + +def test_a_head_written_a_week_late_blocks_the_slot_instead_of_hiding_a_holder(): + # Review of 4a5a272, 1(b): A's head write lands after every key expired + # and B built a new chain. A reader must not take the slot from B. + day = 24 * 60 * 60 + clock = Clock() + store = FakeDict(clock) + a = slots_for(store, clock, "a", count=1) + steps = a._try_steps(0) + call = next(steps) + while not (call[0] == "put" and call[1] == "head-0" and not call[3]): + if call[0] == "get": + call = steps.send(store.get(call[1])) + else: + call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + late_head = call # A's claim exists; its head write is in flight + clock.now += 8 * day + b = slots_for(store, clock, "b", count=1) + b_lease = b.renew(b.acquire()) + store.put(late_head[1], late_head[2], skip_if_exists=False) # it lands + assert store.get("head-0")["chain"] != b_lease.chain + c = slots_for(store, clock, "c", count=1) + assert run_steps(store, c._try_steps(0)) == (True, None) + assert run_steps(store, c._try_steps(0)) == (True, None) # still blocked + # B's lease began before the late write, so it ends within the block. + clock.now += runner.LEASE_SECONDS + 1 + assert b_lease.expires <= clock.now + assert slots_for(store, clock, "d", count=1).acquire(wait_seconds=0).slot == 0 + + +class FailingHeadPuts(FakeDict): + """Dict whose unconditional head writes fail (a lost connection).""" + + def put(self, key, value, *, skip_if_exists=False): + if key.startswith("head-") and not skip_if_exists: + raise OSError("connection lost") + return super().put(key, value, skip_if_exists=skip_if_exists) + + +def test_a_failed_head_write_gives_the_claim_back(): + # A claim without its hint would hold the slot for a lease with no run. + clock = Clock() + store = FailingHeadPuts(clock) + a = slots_for(store, clock, "a", count=1) + with pytest.raises(OSError): + a.acquire(wait_seconds=0) + claims = [v for k, v in store.data.items() if k.startswith("claim-")] + assert claims and all(claim["released"] for claim in claims) + # The same on renewal: the new generation is released, not held. + store2 = FakeDict(clock) + b = slots_for(store2, clock, "b", count=1) + lease = b.acquire() + b.store = FailingHeadPuts(clock) + b.store.data, b.store.touched = store2.data, store2.touched + with pytest.raises(OSError): + b.renew(lease) + newest = store2.data[ + runner.SimulationSlots.claim_key(0, lease.chain, lease.generation + 1) + ] + assert newest["released"] is True + + +def test_a_returned_result_survives_a_failed_app_exit(monkeypatch, tmp_path): + class ExitFails: + def __enter__(self): + return self + + def __exit__(self, kind, value, traceback): + raise OSError("stop request failed") + + slots = FakeDict() + saved = [] + monkeypatch.setattr(runner, "validate_selection", lambda request: None) + monkeypatch.setattr( + runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) + ) + monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") + monkeypatch.setattr(runner, "save_download", lambda *args: saved.append(args) or []) + monkeypatch.setattr( + runner, + "modal_function", + lambda mounts: ( + SimpleNamespace( + enable_output=nullcontext, + Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), + ), + SimpleNamespace(run=ExitFails), + SimpleNamespace(remote=lambda payload, lease_expires: {"files": {}}), + ), + ) + assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 0 + assert len(saved) == 1 + claims = [v for k, v in slots.data.items() if k.startswith("claim-")] + assert max(claims, key=lambda claim: claim["expires"])["released"] is True From 3e058b58fb04634f743215ec361188cda5dfb13a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 19:57:15 -0400 Subject: [PATCH 5/5] Modal slots: day-scoped claim chains on Modal's clock Review of bac8974 reproduced three failures by running the code: 1. A head write delayed 6.5 days plus a reader stalled 0.95 days recreated expired generations and overlapped a live run, inside the guarantee as stated. Any pointer can be rolled back by a late write, so the pointer is gone. A slot now has one chain of claims per calendar day (claim-{slot}-{day}-{generation}); a caller creates a chain's next generation, with Modal's atomic create-if-absent, only when its newest one is released or lapsed. A lease that crosses midnight holds a claim in both days' chains, taken in order and given back together. A day's chain matters for at most a day plus a lease, far inside the 7 days after which Modal expires idle entries, so no claim that still matters can have expired and a late write lands in a chain that governs nothing. The guarantee no longer assumes anything about stalls. Head keys, chain ids, the blocker claim, the stale-step bail and renewal are removed. 2. A contender whose clock ran five hours fast took an occupied slot. Leases now run on Modal's clock: main() measures its offset with a small remote function, checks it again after waiting for a slot (drift over 30 s stops the run, exit 75), and a lapsed claim is taken only 2 minutes past its expiry. 3. A Dict failure at release discarded a completed result. A failed release is now reported and the result is still downloaded; the claim lapses. The slot is taken inside app.run(), after the image build, so a build no longer consumes lease time. Tests: the model check now has no stall cap and gives each holder its own clock within the enforced drift, with jumps of any size (including 6.5 and 8 days) at any point, midnight crossings, crashes, interrupts, lost connections and entry expiry; 20,000 generated schedules hold. New deterministic tests cover the clock margin, a lease across midnight, a refused second day, a step that outlasts its lease, the reviewer's late write and five-hour-fast-clock scenarios, the offset measurement, a failed release, and a clock that moved while waiting. Removing the takeover margin, the newest-claim check, the second day's claim or the time-left check each fails a deterministic test. Verified on Modal: a real --preflight-only invocation ran the clock function, took generation 0 in today's and tomorrow's chains (the lease crossed UTC midnight), passed the remote guard and released both claims. Co-Authored-By: Claude Opus 5.5 --- docs/uk_replay/RECIPE.md | 58 ++-- pipeline/modal_uk_event.py | 314 +++++++++-------- tests/test_modal_uk_event.py | 653 +++++++++++++++++------------------ 3 files changed, 512 insertions(+), 513 deletions(-) diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index 8c29aa4d..dcfe915a 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -603,40 +603,42 @@ simulation slots in the Modal Dict `scorecard-replay-simulation-slots` (`modal_uk_event.SimulationSlots`), waiting while both are held, so Modal invocations never run more than two simulations at once. How the slots hold: -* A slot changes hands only through `put(..., skip_if_exists=True)`, Modal's - atomic create-if-absent: each holder, and each renewal, creates the slot's - next generation key, and nothing is deleted. A holder writes only its own - key, and only to release or shorten it. A head key names the newest - generation as a hint. -* Modal expires an entry after 7 days without a read or write, so the - guarantee is stated against that: with every protocol step (acquire or - renew, a few consecutive Dict calls) finishing within 7 days, two holders - never hold one slot, however long the gaps between steps. Longer stalls - are handled where a client can detect them: a step that took over a day - gives its claim back and writes no head, and a reader whose head names a - claim that is gone blocks the slot for one lease instead of taking it. - Not covered: a head write that arrives 7 or more days late and names a - generation that still exists while later ones have expired. -* The lease (4 hours) is renewed after the image build, just before the - remote call, and the remote function refuses to start unless its whole - 3-hour timeout plus a 10-minute margin fits inside the lease. A run - therefore ends before its slot can pass on, even if the client dies. +* A slot has one chain of claims per calendar day, + `claim-{slot}-{day}-{generation}`. A slot changes hands only through + `put(..., skip_if_exists=True)`, Modal's atomic create-if-absent: a caller + creates a chain's next generation only when its newest one is released or + lapsed. Nothing is deleted, there is no pointer that a late write could + roll back, and a holder writes only its own claims, to release or shorten + them. A lease that crosses midnight holds a claim in both days' chains. +* A day's chain matters for at most a day plus a lease, far inside the 7 days + after which Modal expires an idle entry, so no claim that still matters + can have expired, and a write that arrives days late lands in a chain that + no longer governs anything. The guarantee needs no assumption about how + long a step stalls. +* Leases run on Modal's clock: each invocation measures its offset from + Modal with a small remote function, checks it again after waiting for a + slot (a drift over 30 seconds stops the run, exit 75), and takes over a + lapsed claim only 2 minutes after its expiry. +* The lease is 4 hours. The remote function refuses to start unless its + whole 3-hour timeout plus a 10-minute margin fits inside it, so a run + ends before its slot can pass on, even if the client dies. * A finished or never-started run frees its slot at once. When `app.run()` exits after ctrl-C, Modal's client has already asked the server to stop the app's tasks, and the slot frees 10 minutes later. Anything else after the run started (a lost connection, which Modal's runner re-raises without that request; a remote error; a failed stop request; a client - killed outright) leaves the claim until the lease lapses. -* A renewal after the lease lapsed is refused (exit 75): the slot may have - passed on. Waiting runs poll every 30 seconds in no set order. A claim - whose head write fails is given back at once. + killed outright) leaves the claim until the lease lapses. A failed + release never costs a returned result: it is downloaded and the claim + lapses. +* The slot is taken with the app already open, after the image build, so a + build never eats into a lease; an invocation waits for a slot (polling + every 30 seconds, in no set order) with its app open and idle. * `tests/test_modal_uk_event.py` checks this call by call: Hypothesis - interleaves several holders' Dict calls with clock jumps, crashes, - interrupts and Modal's 7-day expiry of idle entries, and asserts that no two - runs overlap on a slot and that both slots can be taken once every lease - has lapsed (after one further lease where a reader blocked a slot). The - model keeps every step within 6 days; the longer stalls above have their - own deterministic tests. + interleaves several holders' Dict calls, each on its own clock within the + enforced drift, with clock jumps of any size at any point, midnight + crossings, crashes, interrupts, lost connections and Modal's 7-day expiry, + and asserts that no two runs overlap on a slot and that both slots can be + taken once every lease has lapsed. Local simulations are not in that lease: count them yourself toward the overall maximum of two. A full event plus one focused repeat uses both diff --git a/pipeline/modal_uk_event.py b/pipeline/modal_uk_event.py index e0572f78..66f1d33b 100644 --- a/pipeline/modal_uk_event.py +++ b/pipeline/modal_uk_event.py @@ -37,7 +37,6 @@ import os import platform import re -import secrets import sys import tempfile import time @@ -104,33 +103,38 @@ SLOTS_DICT = "scorecard-replay-simulation-slots" SLOT_COUNT = 2 -# The function timeout (10,800 s) plus an hour for the container to be -# scheduled. The lease is renewed just before the remote call, and the remote -# function refuses to start unless its whole timeout fits inside the lease, -# so a run always ends before its slot can pass to anyone else. +# The function timeout (10,800 s) plus the guard margin and time for the +# container to be scheduled. The remote function refuses to start unless its +# whole timeout and the margin fit inside the lease, so a run always ends +# before its slot can pass to anyone else. LEASE_SECONDS = 4 * 60 * 60 -# The remote guard also leaves this much of the lease unused, for clock -# differences between hosts and the time Modal takes to stop a timed-out run. +# The remote guard leaves this much of the lease unused. GUARD_MARGIN_SECONDS = 10 * 60 +# Claims are grouped by calendar day (UTC, on Modal's clock). A day's claims +# matter for at most EPOCH_SECONDS + LEASE_SECONDS, far inside the 7 days +# after which Modal expires an idle Dict entry, so no claim that still +# matters can have expired. +EPOCH_SECONDS = 24 * 60 * 60 +# Every invocation measures its offset from Modal's clock and uses Modal +# time for leases. A claim is treated as lapsed only this long after its +# expiry, which covers the measurement error and drift of two invocations. +CLOCK_MARGIN_SECONDS = 2 * 60 +MAX_CLOCK_DRIFT_SECONDS = 30 # When app.run() exits after the runner caught ctrl-C, it has already asked # the server to stop the ephemeral app's tasks (modal/runner.py, # _status_based_disconnect); the slot frees this long afterwards. INTERRUPT_GRACE_SECONDS = 10 * 60 -# A protocol step (a few consecutive Dict calls) that took longer than this -# writes no head: Modal may have expired the keys it read (7 days idle). -STALE_STEP_SECONDS = 24 * 60 * 60 class LeaseLost(RuntimeError): - """The slot passed to another holder before this one renewed it.""" + """This invocation cannot safely hold a simulation slot.""" @dataclass(frozen=True) class Lease: slot: int - chain: str - generation: int expires: float + claims: tuple # ((epoch, generation), ...), ascending by epoch class SimulationSlots: @@ -141,36 +145,36 @@ class SimulationSlots: shared account allows. Before starting, an invocation takes one of two slots in a named Modal Dict. - The only operation that changes who holds a slot is - ``put(key, value, skip_if_exists=True)``, an atomic create-if-absent on - the server (modal/dict.py, ``DictUpdateRequest(if_not_exists=...)``). - Each change of holder, and each renewal, creates the slot's next - generation key, ``claim-{slot}-{chain}-{generation}``; exactly one caller - creates any generation, and nothing is ever deleted. A caller may take - the next generation only when the newest one is released or expired. - A holder writes only its own generation, and only to release it or to - shorten it, so no write can take a slot from anyone. - - ``head-{slot}`` is a hint naming the newest generation; readers walk - forward from it. Its random chain id separates generations written after - Modal expires idle entries (7 days without a read or write) from any - older key that survives. - - What is guaranteed, and when. With every protocol step (acquire or - renew: a few consecutive Dict calls) finishing within 7 days, two - holders never hold one slot. Stalls between steps (an image build, a - run, a closed laptop) are safe whatever their length. A step that stalls - longer is handled in the cases a client can detect: a step that took - over a day gives its claim back and writes no head, and a reader whose - head names a claim that is gone (a head written late, after its keys - expired) blocks the slot for one lease instead of taking it. One case - is not covered: a head write that reaches the server 7 or more days - late and names a generation that still exists while later ones have - expired. + A slot has one chain of claims per calendar day (epoch): + ``claim-{slot}-{epoch}-{generation}``. The only operation that takes a + slot is ``put(key, value, skip_if_exists=True)``, an atomic + create-if-absent on the server (modal/dict.py, + ``DictUpdateRequest(if_not_exists=...)``): a caller creates a chain's + next generation only when its newest one is released or lapsed, and + exactly one caller creates any generation. Nothing is deleted, there is + no pointer to go stale, and a holder writes only its own claims, to + release or shorten them. A lease that crosses midnight holds a claim in + both days' chains, taken in order and given back together. + + Why this is safe: + + * Within a chain, generations are contiguous and each newer one was + created only after the one before it was free, so at most one claim + is live at a time. A run at time t holds the live claim in t's chain. + * A chain governs only its own day, and its keys are at most + EPOCH_SECONDS + LEASE_SECONDS old while it does, so Modal's expiry of + entries idle for 7 days cannot remove one that still matters. A write + that arrives late lands in a chain that no longer governs anything. + * A step that stalls, however long, either finds its generation taken + (the create fails) or was unopposed all along; a lease is granted + only with enough time left for a run, on the clock read after the + last write. + * Clocks: ``clock`` must be Modal's time (main() measures the offset). + Takeover waits CLOCK_MARGIN_SECONDS past a claim's expiry, and the + remote guard ends every run GUARD_MARGIN_SECONDS before it. Every Dict call is one step of a generator (``_try_steps``, - ``_renew_steps``, ``_release_steps``), so tests interleave several - holders call by call. + ``_release_steps``), so tests interleave several holders call by call. """ def __init__( @@ -181,104 +185,77 @@ def __init__( count=SLOT_COUNT, lease=LEASE_SECONDS, clock=time.time, - chain_id=lambda: secrets.token_hex(8), + minimum=None, ): self.store, self.holder, self.count = store, holder, count - self.lease, self.clock, self.chain_id = lease, clock, chain_id + self.lease, self.clock = lease, clock + # A lease with less than this left is not worth holding: the remote + # guard would refuse the run. + self.minimum = ( + RESOURCE_LIMITS["timeout_seconds"] + GUARD_MARGIN_SECONDS + 60 + if minimum is None + else minimum + ) + + @staticmethod + def claim_key(slot: int, epoch: int, generation: int) -> str: + return f"claim-{slot}-{epoch}-{generation}" @staticmethod - def claim_key(slot: int, chain: str, generation: int) -> str: - return f"claim-{slot}-{chain}-{generation}" + def epochs(start: float, expires: float) -> range: + """The days a lease touches, padded by the clock margin.""" + first = int((start - CLOCK_MARGIN_SECONDS) // EPOCH_SECONDS) + last = int((expires + CLOCK_MARGIN_SECONDS) // EPOCH_SECONDS) + return range(first, last + 1) def _claim(self, expires: float, *, released: bool = False) -> dict: return {"holder": self.holder, "expires": expires, "released": released} + def _give_back(self, slot: int, claims, expires: float): + for epoch, generation in claims: + key = self.claim_key(slot, epoch, generation) + yield ("put", key, self._claim(expires, released=True), False) + def _try_steps(self, slot: int): - head_key = f"head-{slot}" - # Before the first call: a response can sit unread through a stall. - looked = self.clock() - head = yield ("get", head_key) - if head is None: - fresh = {"chain": self.chain_id(), "generation": -1} - yield ("put", head_key, fresh, True) - head = yield ("get", head_key) - if head is None: - return None - chain, generation = head["chain"], head["generation"] - newest = None - if generation >= 0: - newest = yield ("get", self.claim_key(slot, chain, generation)) - if newest is None: - # A head is written after its claim, and readers touch both, - # so a head whose claim is gone was written late, after its - # keys expired, and may hide a live holder elsewhere. Any - # such holder's lease began before that write: block the - # slot for one lease instead of taking it. - blocker = { - "holder": f"blocked-by:{self.holder}", - "expires": self.clock() + self.lease, - "released": False, - } - yield ("put", self.claim_key(slot, chain, generation), blocker, True) - return None - while True: - following = yield ("get", self.claim_key(slot, chain, generation + 1)) - if following is None: - break - generation, newest = generation + 1, following - if ( - newest is not None - and not newest["released"] - and newest["expires"] > self.clock() - ): - return None expires = self.clock() + self.lease - key = self.claim_key(slot, chain, generation + 1) - if not (yield ("put", key, self._claim(expires), True)): - return None - if self.clock() - looked > STALE_STEP_SECONDS: - # What this step read may be gone: give the claim back and leave - # the head alone, so a stale chain never replaces a live one. - yield ("put", key, self._claim(expires, released=True), False) - return None + made = [] try: - head = {"chain": chain, "generation": generation + 1} - yield ("put", head_key, head, False) + for epoch in self.epochs(expires - self.lease, expires): + generation, newest = -1, None + while True: + key = self.claim_key(slot, epoch, generation + 1) + following = yield ("get", key) + if following is None: + break + generation, newest = generation + 1, following + free = ( + newest is None + or newest["released"] + or newest["expires"] + CLOCK_MARGIN_SECONDS <= self.clock() + ) + key = self.claim_key(slot, epoch, generation + 1) + if not free or not (yield ("put", key, self._claim(expires), True)): + yield from self._give_back(slot, made, expires) + return None + made.append((epoch, generation + 1)) except Exception: - # The claim exists but the hint may not: do not hold the slot. - yield ("put", key, self._claim(expires, released=True), False) + # A failed call: hold nothing rather than a slot with no run. + yield from self._give_back(slot, made, expires) raise - return Lease(slot, chain, generation + 1, expires) - - def _renew_steps(self, lease: Lease): - # While a lease is live no one else can create its next generation, - # so that key has never existed and cannot be one Modal expired; - # after it lapses the slot may have moved on, so renewal stops. - began = self.clock() - if began >= lease.expires: - raise LeaseLost(f"simulation slot {lease.slot} lease lapsed") - expires = began + self.lease - key = self.claim_key(lease.slot, lease.chain, lease.generation + 1) - if not (yield ("put", key, self._claim(expires), True)): - raise LeaseLost(f"simulation slot {lease.slot} passed to another holder") - if self.clock() - began > STALE_STEP_SECONDS: - yield ("put", key, self._claim(expires, released=True), False) - raise LeaseLost(f"simulation slot {lease.slot} renewal stalled") - try: - head = {"chain": lease.chain, "generation": lease.generation + 1} - yield ("put", f"head-{lease.slot}", head, False) - except Exception: - yield ("put", key, self._claim(expires, released=True), False) - raise - return Lease(lease.slot, lease.chain, lease.generation + 1, expires) + if expires - self.clock() < self.minimum: + # The step itself took too long for a run to fit in the lease. + yield from self._give_back(slot, made, expires) + return None + return Lease(slot, expires, tuple(made)) def _release_steps(self, lease: Lease, *, grace: float = 0.0): - key = self.claim_key(lease.slot, lease.chain, lease.generation) - if grace: - claim = self._claim(min(lease.expires, self.clock() + grace)) - else: - claim = self._claim(lease.expires, released=True) - yield ("put", key, claim, False) + for epoch, generation in lease.claims: + key = self.claim_key(lease.slot, epoch, generation) + if grace: + claim = self._claim(min(lease.expires, self.clock() + grace)) + else: + claim = self._claim(lease.expires, released=True) + yield ("put", key, claim, False) def _drive(self, steps): """Run one protocol generator against the store.""" @@ -293,8 +270,8 @@ def _drive(self, steps): call[1], call[2], skip_if_exists=call[3] ) except Exception as error: - # Let the step see a failed call, so it can give back a - # claim it just made. + # Let the step see a failed call, so it can give back + # the claims it just made. call = steps.throw(error) continue call = steps.send(result) @@ -319,16 +296,25 @@ def acquire( print("[modal] both simulation slots are held; waiting", flush=True) sleep(poll_seconds) - def renew(self, lease: Lease) -> Lease: - """A fresh full lease, taken just before the remote call starts.""" - return self._drive(self._renew_steps(lease)) - def release(self, lease: Lease, *, grace: float = 0.0) -> None: """Free the slot now, or ``grace`` seconds from now when the remote run may still be stopping.""" self._drive(self._release_steps(lease, grace=grace)) +def clock_offset(remote_now, *, local=time.time, attempts: int = 4) -> float: + """Modal's clock minus this machine's, from a remote call bracketed by + two local readings; the first call may include a cold start, so a slow + round trip is retried.""" + for _ in range(attempts): + before = local() + remote = remote_now() + after = local() + if after - before <= 10: + return remote - (before + after) / 2 + raise LeaseLost("could not measure the clock offset from Modal") + + def canonical_bytes(value: object) -> bytes: return ( json.dumps(value, indent=1, sort_keys=True, allow_nan=False) + "\n" @@ -760,7 +746,20 @@ def replay(payload: dict, lease_expires: float) -> dict: } return {"files": files} - return modal, app, replay + @app.function( + image=modal.Image.debian_slim(python_version="3.12"), + serialized=True, + include_source=False, + timeout=60, + block_network=True, + restrict_modal_access=True, + ) + def modal_time() -> float: + import time + + return time.time() + + return modal, app, replay, modal_time def download_prefix(download_root: str, *, root: Path = ROOT) -> Path: @@ -874,26 +873,31 @@ def main(argv: list[str] | None = None) -> int: raise ValueError("download destination escapes its ignored prefix") if target.exists() and any(target.iterdir()): raise ValueError("choose an empty download prefix before a remote run") - modal, app, replay = modal_function(mounts) + modal, app, replay, modal_time = modal_function(mounts) response = None - slots = SimulationSlots( - modal.Dict.from_name(SLOTS_DICT, create_if_missing=True), - holder=f"{platform.node()}:{os.getpid()}:{digest_bytes(canonical_bytes(request))[:12]}", - ) - lease = slots.acquire() + slots = lease = None started = exited = False try: with modal.enable_output(), app.run(): - # The image build can take a while: renew so the lease covers - # the remote run from its start. - lease = slots.renew(lease) + # Leases run on Modal's clock, so two invocations on machines + # whose clocks disagree still agree on when a claim lapses. + offset = clock_offset(modal_time.remote) + slots = SimulationSlots( + modal.Dict.from_name(SLOTS_DICT, create_if_missing=True), + holder=f"{platform.node()}:{os.getpid()}:{digest_bytes(canonical_bytes(request))[:12]}", + clock=lambda: time.time() + offset, + ) + lease = slots.acquire() + # Waiting for a slot can take hours: check the clock again. + if abs(clock_offset(modal_time.remote) - offset) > MAX_CLOCK_DRIFT_SECONDS: + raise LeaseLost("this machine's clock moved against Modal's") started = True response = replay.remote(payload, lease.expires) # app.run() returned: the run finished, or the runner caught ctrl-C # after asking the server to stop the app's tasks. exited = True - except LeaseLost: - print("Modal run not started: its simulation slot lapsed", file=sys.stderr) + except LeaseLost as error: + print(f"Modal run not started: {error}", file=sys.stderr) return 75 except Exception as error: if response is None: @@ -901,14 +905,24 @@ def main(argv: list[str] | None = None) -> int: # The run returned its result; only leaving the app failed. print(f"Modal app exit failed after the run returned: {error}", file=sys.stderr) finally: - if not started or response is not None: - slots.release(lease) - elif exited: - slots.release(lease, grace=INTERRUPT_GRACE_SECONDS) - # Otherwise (a lost connection, which the runner re-raises without a - # stop request; a remote error; an exception during the stop request) - # the remote run may still be going: the claim stays until the lease - # lapses, after the run's end by the remote guard. + # A finished or never-started run frees its slot now; a caught + # ctrl-C frees it after the grace. Otherwise (a lost connection, + # which the runner re-raises without a stop request; a remote error; + # an exception during the stop request) the remote run may still be + # going, and the claim stays until the lease lapses, which the + # remote guard puts after the run's end. A failed release never + # costs a result: the claim simply lapses. + if lease is not None: + try: + if not started or response is not None: + slots.release(lease) + elif exited: + slots.release(lease, grace=INTERRUPT_GRACE_SECONDS) + except Exception as error: + print( + f"simulation slot not released ({error}); it will lapse", + file=sys.stderr, + ) if response is None: print("Modal run interrupted; no output downloaded", file=sys.stderr) return 130 diff --git a/tests/test_modal_uk_event.py b/tests/test_modal_uk_event.py index b2c25bc2..ed363e90 100644 --- a/tests/test_modal_uk_event.py +++ b/tests/test_modal_uk_event.py @@ -286,6 +286,7 @@ def interrupted_remote(payload, lease_expires): ), SimpleNamespace(run=SuppressInterrupt), SimpleNamespace(remote=interrupted_remote), + SimpleNamespace(remote=time.time), ), ) monkeypatch.setattr( @@ -297,8 +298,11 @@ def interrupted_remote(payload, lease_expires): assert "interrupted; no output downloaded" in capsys.readouterr().err # The interrupted run's slot frees once Modal has had time to stop it. claims = [v for k, v in slots.data.items() if k.startswith("claim-")] - assert claims[-1]["released"] is False - assert claims[-1]["expires"] <= time.time() + runner.INTERRUPT_GRACE_SECONDS + assert claims and all(claim["released"] is False for claim in claims) + assert all( + claim["expires"] <= time.time() + runner.INTERRUPT_GRACE_SECONDS + 1 + for claim in claims + ) def test_modal_factory_has_one_blocked_network_worker_and_explicit_uv_venv( @@ -334,7 +338,7 @@ def __init__(self, name, *, include_source): assert include_source is False def function(self, **kwargs): - recorded["function"] = kwargs + recorded.setdefault("functions", []).append(kwargs) return lambda fn: fn def concurrent(**kwargs): @@ -346,7 +350,8 @@ def concurrent(**kwargs): monkeypatch.setattr( runner.importlib.metadata, "version", lambda name: runner.MODAL_VERSION ) - _, _, replay = runner.modal_function(mounts) + _, _, replay, modal_time = runner.modal_function(mounts) + assert abs(modal_time() - time.time()) < 5 # The remote refuses to start unless its whole timeout fits in the lease. monkeypatch.setattr( sys.modules["subprocess"], @@ -357,7 +362,9 @@ def concurrent(**kwargs): for slack in (-60, runner.GUARD_MARGIN_SECONDS - 60): with pytest.raises(RuntimeError, match="lease does not cover"): replay({"request": {}}, time.time() + timeout + slack) - configuration = recorded["function"] + configuration, clock_configuration = recorded["functions"] + assert clock_configuration["block_network"] is True + assert clock_configuration["timeout"] == 60 assert configuration["cpu"] == (2, 2) assert configuration["memory"] == (32768, 32768) assert ( @@ -420,23 +427,24 @@ def get(self, key, default=None): class Clock: - def __init__(self): - self.now = 1_000.0 + """True (Modal) time. Starts mid-morning, well inside a day.""" + + def __init__(self, now=10 * 3600.0): + self.now = now def __call__(self): return self.now -CHAIN_IDS = itertools.count() +DAY = runner.EPOCH_SECONDS +TIMEOUT = runner.RESOURCE_LIMITS["timeout_seconds"] +MARGIN = runner.CLOCK_MARGIN_SECONDS -def slots_for(store, clock, holder, **kwargs): +def slots_for(store, clock, holder, *, offset=0.0, **kwargs): + """A holder whose own clock reads true time plus ``offset``.""" return runner.SimulationSlots( - store, - holder=holder, - clock=clock, - chain_id=lambda: f"chain{next(CHAIN_IDS)}", - **kwargs, + store, holder=holder, clock=lambda: clock() + offset, **kwargs ) @@ -460,24 +468,31 @@ def run_steps(store, steps, *, limit=None): def holders_of(store, clock): - """Each slot's newest generation, if it is held now.""" + """Each slot's live claim in today's chain, by true time.""" held = {} - for key, head in store.data.items(): - if not key.startswith("head-"): - continue - slot = int(key.split("-")[1]) - generation, newest = head["generation"], None + today = int(clock() // DAY) + for slot in range(runner.SLOT_COUNT): + generation, newest = -1, None while True: - claim = store.data.get(f"claim-{slot}-{head['chain']}-{generation + 1}") + claim = store.data.get(f"claim-{slot}-{today}-{generation + 1}") if claim is None: break generation, newest = generation + 1, claim - newest = newest or store.data.get(f"claim-{slot}-{head['chain']}-{generation}") if newest and not newest["released"] and newest["expires"] > clock(): held[slot] = newest["holder"] return held +def test_the_constants_leave_room_for_a_run_and_for_modal_expiry(): + assert runner.LEASE_SECONDS >= TIMEOUT + runner.GUARD_MARGIN_SECONDS + 30 * 60 + assert runner.INTERRUPT_GRACE_SECONDS < runner.LEASE_SECONDS + # A day's claims matter for a day plus a lease: far inside Modal's expiry. + assert runner.EPOCH_SECONDS + 2 * runner.LEASE_SECONDS < FakeDict.TTL / 2 + # Takeover waits out the clock error of two invocations. + assert runner.CLOCK_MARGIN_SECONDS >= 2 * runner.MAX_CLOCK_DRIFT_SECONDS + 20 + assert runner.GUARD_MARGIN_SECONDS > runner.CLOCK_MARGIN_SECONDS + + def test_two_slots_then_a_third_run_waits(): clock = Clock() store = FakeDict(clock) @@ -501,37 +516,42 @@ def sleep(seconds): } +def test_a_lapsed_claim_is_taken_only_after_the_clock_margin(): + clock = Clock() + store = FakeDict(clock) + dead = slots_for(store, clock, "dead", count=1).acquire() + nxt = slots_for(store, clock, "next", count=1) + clock.now = dead.expires + MARGIN - 1 + assert run_steps(store, nxt._try_steps(0)) == (True, None) + clock.now = dead.expires + MARGIN + assert run_steps(store, nxt._try_steps(0))[1] is not None + + def test_a_stale_release_cannot_free_a_newer_holder(): - # Review of #166, 2(a): a holder past its lease releases after the slot - # passed on; the release writes only its own generation. clock = Clock() store = FakeDict(clock) late = slots_for(store, clock, "late", count=1) old = late.acquire() - clock.now += runner.LEASE_SECONDS + 1 - successor = slots_for(store, clock, "successor", count=1).acquire() + clock.now = old.expires + MARGIN + slots_for(store, clock, "successor", count=1).acquire() late.release(old) assert holders_of(store, clock) == {0: "successor"} third = slots_for(store, clock, "third", count=1) assert run_steps(store, third._try_steps(0)) == (True, None) - assert successor.generation == old.generation + 1 -def test_a_reclaimer_cannot_displace_a_claim_made_after_it_looked(): - # Review of #166, 2(b): A sees the slot free, then C takes it before A - # writes; A's create-if-absent of the same generation fails. +def test_a_contender_cannot_displace_a_claim_made_after_it_looked(): clock = Clock() store = FakeDict(clock) holder = slots_for(store, clock, "holder", count=1) lease = holder.acquire() a = slots_for(store, clock, "a", count=1) c = slots_for(store, clock, "c", count=1) - clock.now += runner.LEASE_SECONDS + 1 + clock.now = lease.expires + MARGIN steps = a._try_steps(0) call = next(steps) while call[0] == "get": # A reads up to its create call = steps.send(store.get(call[1])) - holder.release(lease) assert run_steps(store, c._try_steps(0))[1] is not None with pytest.raises(StopIteration) as done: # A's create now fails steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) @@ -539,89 +559,185 @@ def test_a_reclaimer_cannot_displace_a_claim_made_after_it_looked(): assert holders_of(store, clock) == {0: "c"} -@pytest.mark.parametrize("calls", range(8)) +@pytest.mark.parametrize("calls", range(6)) def test_a_holder_that_dies_at_any_step_frees_its_slot_after_the_lease(calls): - # Review of #166, 1: a caller that dies mid-protocol never loses a slot - # for good. clock = Clock() store = FakeDict(clock) dying = slots_for(store, clock, "dying", count=1) run_steps(store, dying._try_steps(0), limit=calls) - clock.now += runner.LEASE_SECONDS + 1 + clock.now += runner.LEASE_SECONDS + MARGIN assert slots_for(store, clock, "next", count=1).acquire(wait_seconds=0).slot == 0 -def test_a_lapsed_lease_is_not_renewed_even_when_its_next_key_is_free(): - # The slot changes hands across days and Modal expires the old keys; a - # client back from a week asleep must not renew into a key that no - # longer exists while someone else holds the slot. - day = 24 * 60 * 60 +def test_a_lease_across_midnight_holds_both_days(): + clock = Clock(now=22 * 3600.0) # two hours before the day ends + store = FakeDict(clock) + night = slots_for(store, clock, "night", count=1) + lease = night.acquire() + assert [epoch for epoch, _ in lease.claims] == [0, 1] + clock.now = DAY + 600 # the next day, the run still going + assert holders_of(store, clock) == {0: "night"} + morning = slots_for(store, clock, "morning", count=1) + assert run_steps(store, morning._try_steps(0)) == (True, None) + night.release(lease) + assert run_steps(store, morning._try_steps(0))[1] is not None + + +def test_a_refused_second_day_gives_the_first_day_back(): + # Someone already holds tomorrow's chain: taking today's alone would + # hold a slot this run cannot keep past midnight. + clock = Clock(now=22 * 3600.0) + store = FakeDict(clock) + store.put( + "claim-0-1-0", + {"holder": "other", "expires": DAY + 3 * 3600, "released": False}, + ) + a = slots_for(store, clock, "a", count=1) + assert run_steps(store, a._try_steps(0)) == (True, None) + assert store.data["claim-0-0-0"]["released"] is True + + +def test_an_interrupt_frees_the_slot_after_the_grace_period(): clock = Clock() store = FakeDict(clock) - sleeper = slots_for(store, clock, "sleeper", count=1) - stale = sleeper.acquire() - clock.now += runner.LEASE_SECONDS + 1 - for name in ("b", "c"): - slots = slots_for(store, clock, name, count=1) - slots.release(slots.acquire()) - clock.now += day - clock.now += 8 * day # every old key idles past Modal's 7 days - current = slots_for(store, clock, "current", count=1).acquire() - following = runner.SimulationSlots.claim_key(0, stale.chain, stale.generation + 1) - assert store.get(following) is None # a create here would succeed - with pytest.raises(runner.LeaseLost): - sleeper.renew(stale) - assert holders_of(store, clock) == {0: "current"} - assert current.slot == 0 + slots = slots_for(store, clock, "a", count=1) + lease = slots.acquire() + slots.release(lease, grace=runner.INTERRUPT_GRACE_SECONDS) + assert holders_of(store, clock) == {0: "a"} + other = slots_for(store, clock, "b", count=1) + clock.now += runner.INTERRUPT_GRACE_SECONDS + MARGIN - 1 + assert run_steps(store, other._try_steps(0)) == (True, None) + clock.now += 1 + assert run_steps(store, other._try_steps(0))[1] is not None -def test_renewal_after_a_takeover_is_refused(): +def test_a_step_that_outlasts_its_lease_grants_nothing(): + # A machine asleep in the middle of taking a slot: whatever it wrote, + # it holds nothing afterwards, and nobody is displaced. clock = Clock() store = FakeDict(clock) - slow = slots_for(store, clock, "slow", count=1) - lease = slow.acquire() - clock.now += runner.LEASE_SECONDS + 1 - slots_for(store, clock, "fast", count=1).acquire() - with pytest.raises(runner.LeaseLost): - slow.renew(lease) - assert holders_of(store, clock) == {0: "fast"} + a = slots_for(store, clock, "a", count=1) + steps = a._try_steps(0) + call = next(steps) + while call[0] == "get": + call = steps.send(store.get(call[1])) + clock.now += 9 * DAY # asleep with the create in flight + b = slots_for(store, clock, "b", count=1).acquire() + try: + while True: + if call[0] == "get": + call = steps.send(store.get(call[1])) + else: + call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + except StopIteration as done: + assert done.value is None + assert holders_of(store, clock) == {0: "b"} + assert b.slot == 0 -def test_renewal_extends_from_now_and_keeps_the_slot(): +def test_a_week_late_write_lands_in_a_day_that_no_longer_matters(): + # Review of bac8974, 1: with no pointer to roll back, a write that + # arrives days late cannot change which claim is newest today. clock = Clock() store = FakeDict(clock) - slots = slots_for(store, clock, "a", count=1) - lease = slots.acquire() - clock.now += 3_000 # image build - renewed = slots.renew(lease) - assert renewed.expires == clock.now + runner.LEASE_SECONDS - assert renewed.generation == lease.generation + 1 - clock.now += runner.LEASE_SECONDS - 1 - assert holders_of(store, clock) == {0: "a"} + early = slots_for(store, clock, "early", count=1) + lease = early.acquire() + clock.now += 6.5 * DAY + current = slots_for(store, clock, "current", count=1).acquire() + early.release(lease) # the late write + stalled = slots_for(store, clock, "stalled", count=1) + steps = stalled._try_steps(0) + call = next(steps) + clock.now += 0.9 * DAY - runner.LEASE_SECONDS # stalls, then resumes + result = None + try: + while True: + if call[0] == "get": + call = steps.send(store.get(call[1])) + else: + call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) + except StopIteration as done: + result = done.value + live = current.expires > clock.now + assert result is None or not live -def test_an_interrupt_frees_the_slot_after_the_grace_period(): +def test_a_fast_clock_cannot_take_an_occupied_slot(): + # Review of bac8974, 2: main() runs every holder on Modal's clock; within + # the drift it enforces, a contender cannot see a live claim as lapsed + # before the run under it has ended. clock = Clock() store = FakeDict(clock) - slots = slots_for(store, clock, "a", count=1) - lease = slots.acquire() - slots.release(lease, grace=runner.INTERRUPT_GRACE_SECONDS) + drift = runner.MAX_CLOCK_DRIFT_SECONDS + holder = slots_for(store, clock, "holder", count=1, offset=-drift) + lease = holder.acquire() + run_must_end_by = lease.expires - runner.GUARD_MARGIN_SECONDS # remote guard + fast = slots_for(store, clock, "fast", count=1, offset=drift) + clock.now = run_must_end_by + assert run_steps(store, fast._try_steps(0)) == (True, None) + taken_at = None + while taken_at is None: + clock.now += 1 + if run_steps(store, fast._try_steps(0))[1] is not None: + taken_at = clock.now + assert taken_at >= run_must_end_by + runner.GUARD_MARGIN_SECONDS - 2 * drift + + +def test_a_machine_five_hours_fast_still_sees_modal_time(): + # Review of bac8974, 2, as reported: B's own clock is five hours ahead. + # main() gives the slots Modal's time (local plus the measured offset), + # so B sees A's claim as live. + clock = Clock() + store = FakeDict(clock) + slots_for(store, clock, "a", count=1).acquire() + + def b_local(): + return clock() + 5 * 3600 + + offset = runner.clock_offset(clock, local=b_local) + assert offset == -5 * 3600 + b = runner.SimulationSlots( + store, holder="b", count=1, clock=lambda: b_local() + offset + ) + assert run_steps(store, b._try_steps(0)) == (True, None) assert holders_of(store, clock) == {0: "a"} - clock.now += runner.INTERRUPT_GRACE_SECONDS - assert holders_of(store, clock) == {} -def test_the_lease_covers_the_timeout_margin_and_scheduling(): - timeout = runner.RESOURCE_LIMITS["timeout_seconds"] - assert runner.LEASE_SECONDS >= timeout + runner.GUARD_MARGIN_SECONDS + 30 * 60 - assert runner.STALE_STEP_SECONDS < FakeDict.TTL - assert runner.INTERRUPT_GRACE_SECONDS < runner.LEASE_SECONDS +def test_the_clock_offset_is_measured_against_modal(): + local = iter([100.0, 104.0]) + assert runner.clock_offset(lambda: 1_102.0, local=lambda: next(local)) == 1_000.0 + # A slow first round trip (a cold start) is retried. + local = iter([0.0, 50.0, 60.0, 61.0]) + assert runner.clock_offset(lambda: 5.5, local=lambda: next(local)) == -55.0 + slow = iter(range(0, 1000, 20)) + with pytest.raises(runner.LeaseLost): + runner.clock_offset(lambda: 0.0, local=lambda: float(next(slow))) -TIMEOUT = runner.RESOURCE_LIMITS["timeout_seconds"] -# Modal's disconnect stops an ephemeral app's tasks "in a few seconds" -# (modal/runner.py); the model allows a minute, inside the grace period. -STOP_AFTER_DISCONNECT = 60 +class FailingPuts(FakeDict): + """A Dict whose writes start failing after ``allowed`` of them.""" + + def __init__(self, clock, allowed): + super().__init__(clock) + self.allowed = allowed + + def put(self, key, value, *, skip_if_exists=False): + if self.allowed == 0: + raise OSError("connection lost") + self.allowed -= 1 + return super().put(key, value, skip_if_exists=skip_if_exists) + + +def test_a_failed_second_claim_gives_the_first_back_when_it_can(): + clock = Clock(now=22 * 3600.0) # the lease spans two days: two creates + store = FailingPuts(clock, allowed=1) + a = slots_for(store, clock, "a", count=1) + with pytest.raises(OSError): + a.acquire(wait_seconds=0) + store.allowed = 10 + # The first claim stays (its release failed too) but lapses with the lease. + clock.now += runner.LEASE_SECONDS + MARGIN + assert slots_for(store, clock, "b", count=1).acquire(wait_seconds=0).slot == 0 def holder_program(slots): @@ -634,12 +750,6 @@ def holder_program(slots): break else: yield ("pause",) - yield ("pause",) # image build - try: - lease = yield from slots._renew_steps(lease) - except runner.LeaseLost: - yield from slots._release_steps(lease) - return outcome = yield ("start", lease) if outcome == "running": outcome = yield ("await",) @@ -652,34 +762,37 @@ def holder_program(slots): yield from slots._release_steps(lease, grace=runner.INTERRUPT_GRACE_SECONDS) +# Modal's disconnect stops an ephemeral app's tasks "in a few seconds" +# (modal/runner.py); the model allows a minute, inside the grace period. +STOP_AFTER_DISCONNECT = 60 + + @settings(max_examples=600, deadline=None) @given(st.data()) def test_interleaved_holders_never_overlap_on_a_slot_and_never_lose_one(data): - """Model check: holders interleaved call by call, with clock jumps, - crashes (no release; a started run continues to its end), interrupts and - Dict entry expiry. Safety: two runs never overlap on one slot, so at most - SLOT_COUNT run at once. Liveness: once every lease has lapsed, both slots - can be taken again.""" - clock = Clock() + """Model check: holders interleaved call by call, each on its own clock + (within the drift main() enforces), with clock jumps of any size at any + point (a stalled step, a closed laptop, midnight), crashes (no release; + a started run continues to its end), interrupts, lost connections and + Dict entry expiry. Safety: two runs never overlap on one slot, so at + most SLOT_COUNT run at once. Liveness: once every lease has lapsed, both + slots can be taken again.""" + clock = Clock(now=float(data.draw(st.sampled_from([3600, 12 * 3600, 23 * 3600])))) store = FakeDict(clock) + drift = runner.MAX_CLOCK_DRIFT_SECONDS count = data.draw(st.integers(2, 4), label="holders") actors = [] for index in range(count): - program = holder_program(slots_for(store, clock, f"h{index}")) - actors.append( - {"program": program, "call": next(program), "done": False, "since": 1e3} - ) + offset = data.draw(st.sampled_from([-drift, 0, drift]), label="offset") + program = holder_program(slots_for(store, clock, f"h{index}", offset=offset)) + actors.append({"program": program, "call": next(program), "done": False}) runs = [] def send(actor, value): - before = actor["call"][0] try: actor["call"] = actor["program"].send(value) except StopIteration: actor["done"] = True - return - if actor["call"][0] in ("get", "put") and before not in ("get", "put"): - actor["since"] = clock.now # a protocol step begins for _ in range(data.draw(st.integers(0, 150), label="actions")): action = data.draw( @@ -688,21 +801,11 @@ def send(actor, value): ) ) if action == "tick": - seconds = data.draw( - st.sampled_from([1, 60, 600, 3600, TIMEOUT, 4 * 3600, 8 * 86400]) - ) - # The stated assumption: no protocol step (consecutive Dict - # calls) spans 7 days, so Modal never expires a key a step read - # before the step ends; between steps, anything goes. - limit = min( - ( - 6 * 86400 - (clock.now - a["since"]) - for a in actors - if not a["done"] and a["call"][0] in ("get", "put") - ), - default=seconds, + clock.now += data.draw( + st.sampled_from( + [1, 60, 600, 3600, TIMEOUT, 4 * 3600, 20 * 3600, 6.5 * DAY, 8 * DAY] + ) ) - clock.now += max(0, min(seconds, limit)) continue live = [a for a in actors if not a["done"]] if not live: @@ -731,8 +834,9 @@ def send(actor, value): send(actor, None) elif call[0] == "start": lease = call[1] + # The remote guard, on Modal's (true) clock. if clock.now + TIMEOUT + runner.GUARD_MARGIN_SECONDS > lease.expires: - send(actor, "refused") # the remote guard: a remote error + send(actor, "refused") continue duration = data.draw(st.integers(1, TIMEOUT), label="run seconds") run = {"slot": lease.slot, "start": clock.now, "end": clock.now + duration} @@ -747,227 +851,82 @@ def send(actor, value): running = [r for r in runs if r["start"] <= clock.now < r["end"]] assert len(running) <= runner.SLOT_COUNT - # Once every lease has lapsed both slots can be taken, after at most one - # further lease when a reader had to block a slot whose head outlived - # its claim. - clock.now += runner.LEASE_SECONDS + 1 - taken = set() - for attempt in range(2): - for n in range(runner.SLOT_COUNT): - holder = slots_for(store, clock, f"fresh{attempt}-{n}") - for slot in range(runner.SLOT_COUNT): - if slot not in taken and holder._drive(holder._try_steps(slot)): - taken.add(slot) - break - if taken == set(range(runner.SLOT_COUNT)): - break - clock.now += runner.LEASE_SECONDS + 1 + clock.now += runner.LEASE_SECONDS + MARGIN + drift + fresh = [slots_for(store, clock, f"fresh{n}") for n in range(runner.SLOT_COUNT)] + taken = {s.acquire(wait_seconds=0).slot for s in fresh} assert taken == set(range(runner.SLOT_COUNT)), (taken, clock.now, store.data) -def test_a_week_stalled_acquire_cannot_replace_a_live_chain(): - # Review of dff81d0, 2: A reads the head, stalls a week (Modal expires - # every key), B builds a new chain and holds the slot; A's late writes - # must not point the head back at A's dead chain. - day = 24 * 60 * 60 - clock = Clock() - store = FakeDict(clock) - old = slots_for(store, clock, "old", count=1) - old.release(old.acquire()) - a = slots_for(store, clock, "a", count=1) - steps = a._try_steps(0) - call = next(steps) - while call[0] == "get": - call = steps.send(store.get(call[1])) - clock.now += 8 * day - b = slots_for(store, clock, "b", count=1).acquire() - with pytest.raises(StopIteration) as done: - while True: - if call[0] == "get": - call = steps.send(store.get(call[1])) - else: - call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) - assert done.value.value is None - assert store.get("head-0")["chain"] == b.chain - assert holders_of(store, clock) == {0: "b"} - third = slots_for(store, clock, "third", count=1) - assert run_steps(store, third._try_steps(0)) == (True, None) - - -def test_a_lost_connection_keeps_the_claim_until_the_lease_lapses( - monkeypatch, tmp_path -): - # Review of dff81d0, 1: the runner re-raises a lost connection without - # asking the server to stop the app, so the run may still be going. - class LostConnection(Exception): - pass - - def lost(payload, lease_expires): - raise LostConnection - - slots = FakeDict() - monkeypatch.setattr(runner, "validate_selection", lambda request: None) - monkeypatch.setattr( - runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) - ) - monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") - monkeypatch.setattr( - runner, - "modal_function", - lambda mounts: ( - SimpleNamespace( - enable_output=nullcontext, - Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), - ), - SimpleNamespace(run=nullcontext), - SimpleNamespace(remote=lost), +def modal_fakes(slots, remote, run=nullcontext): + return lambda mounts: ( + SimpleNamespace( + enable_output=nullcontext, + Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), ), + SimpleNamespace(run=run), + SimpleNamespace(remote=remote), + SimpleNamespace(remote=time.time), ) - with pytest.raises(LostConnection): - runner.main(["--event", "autumn_budget_2024", "--execute"]) - claims = [v for k, v in slots.data.items() if k.startswith("claim-")] - newest = max(claims, key=lambda claim: claim["expires"]) - assert newest["released"] is False - assert newest["expires"] >= time.time() + runner.LEASE_SECONDS - 60 -def test_a_finished_run_frees_its_slot_at_once(monkeypatch, tmp_path): - slots = FakeDict() +@pytest.fixture +def stub_main(monkeypatch, tmp_path): monkeypatch.setattr(runner, "validate_selection", lambda request: None) monkeypatch.setattr( runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) ) monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") - monkeypatch.setattr(runner, "save_download", lambda *args: []) - monkeypatch.setattr( - runner, - "modal_function", - lambda mounts: ( - SimpleNamespace( - enable_output=nullcontext, - Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), - ), - SimpleNamespace(run=nullcontext), - SimpleNamespace(remote=lambda payload, lease_expires: {"files": {}}), - ), - ) - assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 0 - claims = [v for k, v in slots.data.items() if k.startswith("claim-")] - assert max(claims, key=lambda claim: claim["expires"])["released"] is True + saved = [] + monkeypatch.setattr(runner, "save_download", lambda *args: saved.append(args) or []) + return saved -def finish(store, steps, call): - """Drive a protocol generator from a pending call to its return value.""" - try: - while True: - if call[0] == "get": - call = steps.send(store.get(call[1])) - else: - call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) - except StopIteration as done: - return done.value +ARGS = ["--event", "autumn_budget_2024", "--execute"] -def test_a_head_response_read_before_a_week_asleep_is_not_acted_on(): - # Review of 4a5a272, 1(a): the head was answered, then the machine slept - # a week before the step used it; the step must not look fresh. - day = 24 * 60 * 60 - clock = Clock() - store = FakeDict(clock) - old = slots_for(store, clock, "old", count=1) - old.release(old.acquire()) - a = slots_for(store, clock, "a", count=1) - steps = a._try_steps(0) - call = next(steps) - assert call == ("get", "head-0") - answered = store.get("head-0") # the server answers... - clock.now += 8 * day # ...and the response sits unread - b = slots_for(store, clock, "b", count=1).acquire() - assert finish(store, steps, steps.send(answered)) is None - assert store.get("head-0")["chain"] == b.chain - assert holders_of(store, clock) == {0: "b"} - +def claims_in(slots): + return [v for k, v in slots.data.items() if k.startswith("claim-")] -def test_a_fresh_chain_head_read_before_a_week_asleep_is_not_acted_on(): - # The same stall on a chain with no claim yet (generation -1): only the - # step's own clock, read before its first call, shows it is stale. - day = 24 * 60 * 60 - clock = Clock() - store = FakeDict(clock) - store.put("head-0", {"chain": "abandoned", "generation": -1}) - a = slots_for(store, clock, "a", count=1) - steps = a._try_steps(0) - assert next(steps) == ("get", "head-0") - answered = store.get("head-0") - clock.now += 8 * day - b = slots_for(store, clock, "b", count=1).acquire() - assert finish(store, steps, steps.send(answered)) is None - assert store.get("head-0")["chain"] == b.chain - assert holders_of(store, clock) == {0: "b"} +def test_a_finished_run_frees_its_slot_at_once(monkeypatch, stub_main): + slots = FakeDict() + seen = {} -def test_a_head_written_a_week_late_blocks_the_slot_instead_of_hiding_a_holder(): - # Review of 4a5a272, 1(b): A's head write lands after every key expired - # and B built a new chain. A reader must not take the slot from B. - day = 24 * 60 * 60 - clock = Clock() - store = FakeDict(clock) - a = slots_for(store, clock, "a", count=1) - steps = a._try_steps(0) - call = next(steps) - while not (call[0] == "put" and call[1] == "head-0" and not call[3]): - if call[0] == "get": - call = steps.send(store.get(call[1])) - else: - call = steps.send(store.put(call[1], call[2], skip_if_exists=call[3])) - late_head = call # A's claim exists; its head write is in flight - clock.now += 8 * day - b = slots_for(store, clock, "b", count=1) - b_lease = b.renew(b.acquire()) - store.put(late_head[1], late_head[2], skip_if_exists=False) # it lands - assert store.get("head-0")["chain"] != b_lease.chain - c = slots_for(store, clock, "c", count=1) - assert run_steps(store, c._try_steps(0)) == (True, None) - assert run_steps(store, c._try_steps(0)) == (True, None) # still blocked - # B's lease began before the late write, so it ends within the block. - clock.now += runner.LEASE_SECONDS + 1 - assert b_lease.expires <= clock.now - assert slots_for(store, clock, "d", count=1).acquire(wait_seconds=0).slot == 0 + def remote(payload, lease_expires): + seen["expires"] = lease_expires + return {"files": {}} + monkeypatch.setattr(runner, "modal_function", modal_fakes(slots, remote)) + assert runner.main(ARGS) == 0 + assert len(stub_main) == 1 + assert all(claim["released"] is True for claim in claims_in(slots)) + # The run was told a lease long enough for its timeout and the margin. + assert seen["expires"] - time.time() > TIMEOUT + runner.GUARD_MARGIN_SECONDS -class FailingHeadPuts(FakeDict): - """Dict whose unconditional head writes fail (a lost connection).""" - def put(self, key, value, *, skip_if_exists=False): - if key.startswith("head-") and not skip_if_exists: - raise OSError("connection lost") - return super().put(key, value, skip_if_exists=skip_if_exists) +def test_a_lost_connection_keeps_the_claim_until_the_lease_lapses( + monkeypatch, stub_main +): + # The runner re-raises a lost connection without asking the server to + # stop the app, so the run may still be going. + class LostConnection(Exception): + pass + def lost(payload, lease_expires): + raise LostConnection -def test_a_failed_head_write_gives_the_claim_back(): - # A claim without its hint would hold the slot for a lease with no run. - clock = Clock() - store = FailingHeadPuts(clock) - a = slots_for(store, clock, "a", count=1) - with pytest.raises(OSError): - a.acquire(wait_seconds=0) - claims = [v for k, v in store.data.items() if k.startswith("claim-")] - assert claims and all(claim["released"] for claim in claims) - # The same on renewal: the new generation is released, not held. - store2 = FakeDict(clock) - b = slots_for(store2, clock, "b", count=1) - lease = b.acquire() - b.store = FailingHeadPuts(clock) - b.store.data, b.store.touched = store2.data, store2.touched - with pytest.raises(OSError): - b.renew(lease) - newest = store2.data[ - runner.SimulationSlots.claim_key(0, lease.chain, lease.generation + 1) - ] - assert newest["released"] is True + slots = FakeDict() + monkeypatch.setattr(runner, "modal_function", modal_fakes(slots, lost)) + with pytest.raises(LostConnection): + runner.main(ARGS) + assert all(claim["released"] is False for claim in claims_in(slots)) + assert all( + claim["expires"] >= time.time() + runner.LEASE_SECONDS - 60 + for claim in claims_in(slots) + ) -def test_a_returned_result_survives_a_failed_app_exit(monkeypatch, tmp_path): +def test_a_returned_result_survives_a_failed_app_exit(monkeypatch, stub_main): class ExitFails: def __enter__(self): return self @@ -976,26 +935,50 @@ def __exit__(self, kind, value, traceback): raise OSError("stop request failed") slots = FakeDict() - saved = [] - monkeypatch.setattr(runner, "validate_selection", lambda request: None) monkeypatch.setattr( - runner, "input_manifest", lambda event, hf_cache=None: ({"event": event}, {}) + runner, + "modal_function", + modal_fakes(slots, lambda payload, lease_expires: {"files": {}}, run=ExitFails), ) - monkeypatch.setattr(runner, "download_prefix", lambda value: tmp_path / "download") - monkeypatch.setattr(runner, "save_download", lambda *args: saved.append(args) or []) + assert runner.main(ARGS) == 0 + assert len(stub_main) == 1 + assert all(claim["released"] is True for claim in claims_in(slots)) + + +def test_a_returned_result_survives_a_failed_release(monkeypatch, stub_main, capsys): + # Review of bac8974, 3: the Dict failing at release must not cost a + # completed result; the claim simply lapses. + class ReleaseFails(FakeDict): + def put(self, key, value, *, skip_if_exists=False): + if not skip_if_exists: + raise OSError("connection lost") + return super().put(key, value, skip_if_exists=skip_if_exists) + + slots = ReleaseFails() monkeypatch.setattr( runner, "modal_function", - lambda mounts: ( - SimpleNamespace( - enable_output=nullcontext, - Dict=SimpleNamespace(from_name=lambda name, create_if_missing: slots), - ), - SimpleNamespace(run=ExitFails), - SimpleNamespace(remote=lambda payload, lease_expires: {"files": {}}), - ), + modal_fakes(slots, lambda payload, lease_expires: {"files": {}}), ) - assert runner.main(["--event", "autumn_budget_2024", "--execute"]) == 0 - assert len(saved) == 1 - claims = [v for k, v in slots.data.items() if k.startswith("claim-")] - assert max(claims, key=lambda claim: claim["expires"])["released"] is True + assert runner.main(ARGS) == 0 + assert len(stub_main) == 1 + assert "not released" in capsys.readouterr().err + + +def test_a_clock_that_moved_while_waiting_stops_the_run(monkeypatch, stub_main, capsys): + slots = FakeDict() + readings = iter([0.0, 120.0]) # Modal minus local: 0 at first, then 120 s + + def modal_now(): + return time.time() + next(readings) + + def fakes(mounts): + modal, app, replay, _ = modal_fakes( + slots, lambda payload, lease_expires: pytest.fail("the run must not start") + )(mounts) + return modal, app, replay, SimpleNamespace(remote=modal_now) + + monkeypatch.setattr(runner, "modal_function", fakes) + assert runner.main(ARGS) == 75 + assert "clock moved" in capsys.readouterr().err + assert all(claim["released"] is True for claim in claims_in(slots))