diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index cbf5722e657..70d9c748526 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -33,6 +33,12 @@ ROOT = Path(__file__).resolve().parent.parent AZURE_PROVIDER = ROOT / "bin" / "fm-azure-worker-provider.py" STATE_SCHEMA = "fm.worker-lifecycle/v1" +# The scalar pending_action slot this schema carried is superseded by the +# per-slot pending_actions map. The sentinel is deliberately a string an OLD +# binary's verify_state refuses ("pending provider action is malformed"), so a +# rollback cannot read None, plan fresh work, and blind-overwrite a live claim: +# it refuses loudly instead, cured by rolling forward. +LEGACY_PENDING_SENTINEL = "superseded-by-pending-actions" REQUEST_SCHEMA = "fm.worker-request/v1" EXECUTION_SCHEMA = "fm.worker-execution/v1" EXECUTION_RESULT_SCHEMA = "fm.worker-execution-result/v1" @@ -90,6 +96,12 @@ # that actually ran. PROVIDER_GUEST_RUN_SLACK_SECONDS = 8400 MAX_PROVIDER_OUTPUT_BYTES = 2 * 1024 * 1024 +# Every provider mutation type the claim contract covers. admission-refused is +# a bare planning verdict with no idempotency key, never claimed, never sent. +ACTION_TYPES = frozenset({ + "create", "resume", "deallocate", "delete-compute", "reset", "execute", "steer", +}) + REQUIRED_RESOURCE_KINDS = ( "vm", "nic", "os-disk", "task-disk", "account-disk", "identity", "role-assignment", "state-container", "monitor-extension", "bootstrap-command", @@ -295,8 +307,25 @@ def environment(): } +_LOCK_STATE = {"held": False, "epoch": 0} + + +class FencedState(dict): + """The durable document, stamped with the lock epoch and disk revision it + was loaded under. A dict subclass serializes through json.dump unchanged; + the stamps live on attributes, never in the document.""" + + epoch = 0 + revision = 0 + + @contextlib.contextmanager def controller_lock(env): + # Re-entrant acquisition would deadlock on a second file description of the + # same lock file; refusing it loudly also structurally prevents replaying + # pending work from inside a hold. + if _LOCK_STATE["held"]: + raise LifecycleError("controller lock is already held by this process") env["state_dir"].mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(env["state_dir"], 0o700) with open(env["lock_path"], "a+", encoding="utf-8") as handle: @@ -305,10 +334,16 @@ def controller_lock(env): # crewmates serialize here. Callers WAIT rather than fail; making the # loser error out was tried and reverted, because status, reconcile and # release would then start failing under ordinary contention. Real - # concurrency needs per-action pending state so the provider call can - # run outside this lock, which is its own change. + # concurrency needs the provider call to run outside this lock, which + # is the next change in this series; the per-slot claim map, the load + # fence and the revision CAS below are its durable groundwork. fcntl.flock(handle.fileno(), fcntl.LOCK_EX) - yield + _LOCK_STATE["held"] = True + _LOCK_STATE["epoch"] += 1 + try: + yield + finally: + _LOCK_STATE["held"] = False def empty_state(env): @@ -326,7 +361,9 @@ def empty_state(env): "workers": {}, "capacity_reservations": {}, "completed_worker_seconds": 0.0, - "pending_action": None, + "pending_action": LEGACY_PENDING_SENTINEL, + "pending_actions": {}, + "revision": 0, "cleanup_refusals": [], "last_metrics": None, "executions": {}, @@ -379,8 +416,36 @@ def verify_state(env, state): require_binding("capacity reservation fence", reservation.get("fence_binding")) if "shape_id" in reservation: require_id("capacity shape id", reservation.get("shape_id")) - if state.get("pending_action") is not None and not isinstance(state["pending_action"], dict): + legacy = state.get("pending_action") + if legacy is not None and legacy != LEGACY_PENDING_SENTINEL: + # A dict here means load_state's migration did not run; anything else + # is corruption. Both refuse rather than guess. raise LifecycleError("pending provider action is malformed") + revision = state.get("revision") + if isinstance(revision, bool) or not isinstance(revision, int) or revision < 0: + raise LifecycleError("lifecycle state revision is malformed") + pending = state.get("pending_actions") + if not isinstance(pending, dict) or len(pending) > MAX_WORKERS: + raise LifecycleError("pending provider action inventory is malformed") + for slot_key, action in pending.items(): + # Bounded by MAX_WORKERS, never env max_workers: lowering + # FM_AZURE_WORKER_MAX must not make an existing state file unloadable. + # slot_key membership in workers is deliberately NOT required here: + # enforcing it converts a recoverable wedge into a file that refuses + # even `status`, and apply_action_result already raises on a missing + # worker in every branch. + if ( + not isinstance(action, dict) + or not slot_key.isdigit() + or not 1 <= int(slot_key) <= MAX_WORKERS + or str(action.get("slot")) != slot_key + or action.get("type") not in ACTION_TYPES + or action.get("deployment_generation") != expected["deployment_generation"] + or action.get("owner") != expected["owner"] + or action_id(action) != action.get("idempotency_key") + ): + raise LifecycleError("pending provider action is malformed") + require_binding("pending action idempotency key", action.get("idempotency_key")) def load_state(env): @@ -392,14 +457,60 @@ def load_state(env): state = empty_state(env) state.setdefault("capacity_reservations", {}) state.setdefault("executions", {}) + state.setdefault("pending_actions", {}) + state.setdefault("revision", 0) + legacy = state.get("pending_action") + if legacy is not None and legacy != LEGACY_PENDING_SENTINEL and not isinstance(legacy, dict): + # The old binary refused this shape loudly; paving it over with the + # sentinel would silently destroy whatever replay obligation the + # corrupted bytes used to be. Refuse rather than guess. + raise LifecycleError("pending provider action is malformed") + if isinstance(legacy, dict): + # apply_action_result has always addressed the worker by + # action["slot"], so the action already carries its own key; nothing + # is invented. Idempotent on every load; durable at the next save. + slot = str(legacy.get("slot", "")) + if not slot.isdigit(): + raise LifecycleError("legacy pending provider action carries no exact slot") + held = state["pending_actions"].get(slot) + if held is not None and held != legacy: + raise LifecycleError( + "legacy and per-slot pending actions disagree for slot {}".format(slot)) + state["pending_actions"][slot] = legacy + state["pending_action"] = LEGACY_PENDING_SENTINEL verify_state(env, state) - return state + fenced = FencedState(state) + fenced.epoch = _LOCK_STATE["epoch"] + fenced.revision = int(state["revision"]) + return fenced def save_state(env, state): + if not isinstance(state, FencedState): + raise LifecycleError("lifecycle state was not loaded through load_state") + if not _LOCK_STATE["held"] or state.epoch != _LOCK_STATE["epoch"]: + # The load fence: this object was loaded outside the lock hold that is + # trying to commit it, so anything read from it may already be stale. + raise LifecycleError("lifecycle state was loaded outside the committing lock hold") + on_disk = 0 + try: + current = read_json(env["state_path"], "lifecycle state") + on_disk = int(current.get("revision", 0)) + except LifecycleError as exc: + if "is absent" not in str(exc): + raise + if on_disk != state.revision: + # Last-writer-wins over this document would not corrupt the file; it + # would silently forget another writer's cloud resource identities and + # re-admit a VM that exists and is billing. Refuse, naming both. + raise LifecycleError( + "lifecycle state revision moved from {} to {} since this load; reload and retry".format( + state.revision, on_disk)) + state["revision"] = on_disk + 1 state["updated_at"] = iso_utc() verify_state(env, state) save_json_atomic(env["state_path"], state) + state.revision = state["revision"] def request_key(task, generation): @@ -1107,9 +1218,12 @@ def record_refusal(state, worker, note): # BOTH sides does not, because the apply writes only into the copy. # # It excludes the whole subtree rather than the aliased part, so an apply that -# rebound `state["pending_action"]` outright would go unseen. That is safe only -# because both call sites set it themselves immediately afterwards. -CALLER_OWNED_KEYS = ("pending_action",) +# rebound the pending claim state outright would go unseen. That is safe only +# because both call sites set it themselves immediately afterwards. The stored +# claims in pending_actions are deep copies and share nothing with the live +# document, but the exclusion also covers the caller popping its own slot's +# claim between apply and commit. +CALLER_OWNED_KEYS = ("pending_action", "pending_actions", "revision") def assert_scoped(before, after, *, slot, queue_key, request_digest): @@ -1183,14 +1297,22 @@ def apply_result_transactionally(env, state, action, result): def execute_action(env, state, action): - state["pending_action"] = action + slot = str(action.get("slot", "")) + if not slot.isdigit(): + raise LifecycleError("provider mutation carries no exact slot") + # deepcopy is load-bearing, not hygiene: make_action aliases live state + # into the action (action["request"] IS the queue entry; bindings and + # resources are the worker's own dicts), and apply_action_result mutates + # some of those in place. A stored claim that can change after it was + # hashed turns verify_state's self-hash check into a random wedge. + state["pending_actions"][slot] = copy.deepcopy(action) save_state(env, state) response = provider_call(env, "mutate", action) result = response.get("result") if not isinstance(result, dict) or result.get("idempotency_key") != action["idempotency_key"]: raise LifecycleError("provider mutation result is not bound to the exact idempotency key") apply_result_transactionally(env, state, action, result) - state["pending_action"] = None + state["pending_actions"].pop(slot, None) save_state(env, state) @@ -1297,17 +1419,18 @@ def apply_action_result(env, state, action, result): def replay_pending(env, state): - action = state.get("pending_action") - if action is None: - return False - response = provider_call(env, "mutate", action) - result = response.get("result") - if not isinstance(result, dict) or result.get("idempotency_key") != action.get("idempotency_key"): - raise LifecycleError("replayed provider mutation is not idempotently bound") - apply_result_transactionally(env, state, action, result) - state["pending_action"] = None - save_state(env, state) - return True + drained = False + for slot in sorted(state.get("pending_actions") or {}, key=int): + action = state["pending_actions"][slot] + response = provider_call(env, "mutate", action) + result = response.get("result") + if not isinstance(result, dict) or result.get("idempotency_key") != action.get("idempotency_key"): + raise LifecycleError("replayed provider mutation is not idempotently bound") + apply_result_transactionally(env, state, action, result) + state["pending_actions"].pop(slot, None) + save_state(env, state) + drained = True + return drained def refresh_classifications(state, inventory, now=None): @@ -1609,6 +1732,11 @@ def status_projection(env, state, inventory=None): "warm_idle_target": env["warm_idle"], "retained_disks": retained_disks, "cleanup_refusals": state["cleanup_refusals"][-10:], + "pending_mutations": [ + {"slot": int(slot), "type": action.get("type")} + for slot, action in sorted( + (state.get("pending_actions") or {}).items(), key=lambda p: int(p[0])) + ], } @@ -1640,6 +1768,9 @@ def print_status(status, json_output): print("specialized-queue: queued={} reserved={}".format( status["specialized_queued_reservations"], status["specialized_reserved_reservations"], )) + if status["pending_mutations"]: + print("pending-mutations: {}".format(json.dumps( + status["pending_mutations"], sort_keys=True, separators=(",", ":")))) print("idle: cooldown={}s warm={} retained-disks={} cleanup-refusals={}".format( status["idle_cooldown_seconds"], status["warm_idle_target"], status["retained_disks"], len(status["cleanup_refusals"]), @@ -2442,17 +2573,19 @@ def command_withdraw(env, args): # and raises, and because the pending action never clears, that repeats # forever. One stale entry would take the whole fleet's convergence # with it. - pending = state.get("pending_action") or {} - pending_request = pending.get("request") or {} - pending_bindings = pending.get("bindings") or {} - for candidate in (pending_request, pending_bindings): - if candidate.get("task") is None: - continue - if request_key(candidate["task"], candidate["task_generation"]) == key: - raise LifecycleError( - "withdraw refuses a task generation a pending {} action still names; " - "reconcile it first".format(pending.get("type", "provider")) - ) + for slot_key in sorted(state.get("pending_actions") or {}, key=int): + pending = state["pending_actions"][slot_key] + pending_request = pending.get("request") or {} + pending_bindings = pending.get("bindings") or {} + for candidate in (pending_request, pending_bindings): + if candidate.get("task") is None: + continue + if request_key(candidate["task"], candidate["task_generation"]) == key: + raise LifecycleError( + "withdraw refuses a task generation a pending {} action on slot {} " + "still names; reconcile it first".format( + pending.get("type", "provider"), slot_key) + ) del state["queue"][key] save_state(env, state) # A machine-readable receipt naming the exact entry that was deleted. The @@ -2531,7 +2664,7 @@ def command_surrender(env, args): raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") with controller_lock(env): state = load_state(env) - if state.get("pending_action") is not None: + if state.get("pending_actions"): raise LifecycleError("a pending provider action exists; reconcile first") key = request_key(args.task, args.task_generation) item = state["queue"].get(key) @@ -2657,7 +2790,7 @@ def command_resume(env, args): require_binding("repository binding", args.repository_binding) with controller_lock(env): state = load_state(env) - if state.get("pending_action") is not None: + if state.get("pending_actions"): replay_pending(env, state) key = request_key(require_id("task", args.task), require_id("task generation", args.task_generation)) item = state["queue"].get(key) diff --git a/docs/azure-requirements.md b/docs/azure-requirements.md index 69b4ee8effd..939e329f9e7 100644 --- a/docs/azure-requirements.md +++ b/docs/azure-requirements.md @@ -254,8 +254,9 @@ Acceptance: a measured review completes in 20 to 30 minutes, with the breakdown Status: NOT DONE. -The controller holds a single `pending_action`, so every provider mutation serializes. -That is the direct blocker on this requirement. +The durable state now holds per-slot `pending_actions` with a load fence and a revision CAS +(C2's second change), but every provider mutation still serializes behind the fleet lock. +That lock is the remaining direct blocker on this requirement. The lock is the other half, and the harder one: `controller_lock` is held across provider calls and for an execute's whole guest run, and the code's own note records that fixing only the lock was diff --git a/docs/azure-workers.md b/docs/azure-workers.md index cec7d01311f..1b3e2cda6e6 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -140,7 +140,7 @@ The general worker contract forbids a browser profile rather than allocating one The OS disk is disposable, while the account and task disks detach from VM deletion and remain encrypted by the guest contract. Provider-account credentials never enter ARM parameters, controller output, logs, tags, images, browser state, or the control home. -A submitted provider action has a canonical SHA-256 idempotency key and remains in `pending_action` until the exact provider result is durably applied. +A submitted provider action has a canonical SHA-256 idempotency key and remains in the per-slot `pending_actions` map until the exact provider result is durably applied; the map entry is a deep copy that re-derives its own key at every load, and the legacy scalar slot permanently holds a sentinel an old binary refuses rather than misreads. After a host restart, the same action and key are replayed. The Azure singleton deployment is incremental and receives the same task, home, assignment, and snapshot bindings, so replay converges one generation rather than creating a second assignment. A visible VM with another task or assignment binding refuses instead of being adopted. @@ -236,9 +236,9 @@ bin/fm-worker-lifecycle.sh reconcile \ --confirm-subscription "$FM_AZURE_SUBSCRIPTION_ID" ``` -One home lock serializes local state and one pending provider action. +One home lock serializes local state; unapplied provider actions are durable per slot in `pending_actions`, every load is fenced to the lock hold that commits it, and a save whose on-disk revision moved since its load refuses instead of overwriting another writer's document. Each reconcile refreshes Azure before selecting the next action and stops after 64 actions even if a provider never converges. -A provider error preserves the pending action and records a bounded cleanup refusal. +A provider error preserves the slot's pending action and records a bounded cleanup refusal. The next controller process replays that exact action before considering new work. `status` is local and bounded by default, while `status --live` refreshes Azure and cost evidence. diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index ed7fd97e893..78a06e0a11b 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -58,6 +58,8 @@ for marker in ( "REGIONAL_ADMISSION_CEILING_VCPUS = 128", "AUTHOR_PLAN_VCPUS = MAX_WORKERS * VCPUS_PER_WORKER", "SPECIALIZED_SHAPE_VCPUS = 40", "SHARED_HEADROOM_VCPUS = 22", "MAX_WORKERS = 16", 'FM_AZURE_WORKER_WARM_IDLE currently must remain zero', "pending_action", + "pending_actions", "LEGACY_PENDING_SENTINEL", "superseded-by-pending-actions", + "revision moved from", "FencedState", "capacity-reserve", "capacity-reserve-shape", "capacity-release", "merged_specialized_reservations", "command_withdraw", "command_surrender", "WORKER AUTHORITY REFUSED", "--confirm-discard-unlanded", @@ -1296,19 +1298,42 @@ assert "task-10@gen-10" not in after_withdraw["queue"], after_withdraw["queue"] # fleet's convergence with it. request(11) pending_state = controller_state() -pending_state["pending_action"] = { - "type": "create", "idempotency_key": "f" * 64, "slot": 9, - "bindings": {"task": "task-11", "task_generation": "gen-11"}, - "request": {"task": "task-11", "task_generation": "gen-11"}, +# The claims must be REAL minted actions: verify_state re-derives the +# idempotency key from the stored bytes, so a hand-typed "f"*64 claim now +# refuses at load, and a guard tested through an unloadable file tests +# nothing. Two entries prove the guard fans out over the whole map, not just +# its first entry. +import importlib.util as _ilu +_spec = _ilu.spec_from_file_location("controller_mod", controller_path) +_cmod = _ilu.module_from_spec(_spec) +_spec.loader.exec_module(_cmod) +_menv = {"deployment_generation": "dep-one", "owner": "owner"} +def _mint(slot, task): + worker = { + "slot": slot, "sku": "sku", "sku_family": "fam", "cloud_generation": 1, + "cloud_instance_id": None, "reservation_usd": 1.0, "resources": {}, + "bindings": {"task": task, "task_generation": "gen-11" if task == "task-11" else "gen-x"}, + } + return _cmod.make_action(_menv, "create", worker=worker, + item={"task": task, "task_generation": worker["bindings"]["task_generation"]}) +pending_state["pending_actions"] = { + "3": _mint(3, "task-unrelated"), + "9": _mint(9, "task-11"), } controller_file = Path(env["FM_HOME"]) / "state/azure-workers/controller.json" +pre_block = controller_file.read_text() controller_file.write_text(json.dumps(pending_state, sort_keys=True, separators=(",", ":"))) pending_refusal = run("withdraw", "--task", "task-11", "--task-generation", "gen-11", "--confirm-withdraw", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"], check=False) assert pending_refusal.returncode != 0, "withdraw dropped an entry a pending action names" -assert "pending" in pending_refusal.stderr, pending_refusal.stderr +assert "still names" in pending_refusal.stderr, pending_refusal.stderr +assert "slot 9" in pending_refusal.stderr, ( + "the guard did not fan out past the first map entry", pending_refusal.stderr) assert "task-11@gen-11" in controller_state()["queue"], "the refused entry was dropped anyway" +controller_file.write_text(pre_block) +run("withdraw", "--task", "task-11", "--task-generation", "gen-11", "--confirm-withdraw", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) # `assigning` is persisted before the provider call and, behind a slow create, # can hold for hours. It still counts as demand, so the refusal must not tell @@ -2099,11 +2124,19 @@ moved = { } for kind in controller.REQUIRED_RESOURCE_KINDS } -action = {"type": "create", "slot": 1, "idempotency_key": "k", "request_digest": "d"} +# A real minted action: execute_action now durably claims it, and the claim +# must self-hash at save, so a fake key would make this unit die at the save +# and never reach the moved-identity apply it exists to test. +MINT_WORKER = dict(copy.deepcopy(WORKER), sku="sku", sku_family="fam", + cloud_generation=1, cloud_instance_id=None, reservation_usd=1.0) +action = controller.make_action( + {"deployment_generation": "dep-one", "owner": "owner"}, "create", + worker=MINT_WORKER, item={"task": "one", "task_generation": "spawn:aaaaaaaaaaaaaaaa"}) # The provider is the only thing stubbed; execute_action itself is the real one. controller.provider_call = lambda environment, operation, payload: { - "result": {"idempotency_key": "k", "worker": {"slot": 1, "resources": moved}} + "result": {"idempotency_key": action["idempotency_key"], + "worker": {"slot": 1, "resources": moved}} } with controller.controller_lock(env): @@ -2117,6 +2150,39 @@ with controller.controller_lock(env): except controller.LifecycleError as error: raised = error assert raised is not None, "execute_action accepted a moved identity" +assert "identity" in str(raised) or "foreign" in str(raised) or "exact" in str(raised), ( + "the refusal was not the apply's; the unit degraded into testing something else", str(raised)) + +# The crash-during-provider window: the durable claim must already be on disk +# when the provider is called, or a crash there strands a cloud mutation with +# no replay obligation. The file, not the caller's object, is the proof. +durable = json.loads((env["state_path"]).read_text()) +assert durable["pending_actions"]["1"]["idempotency_key"] == action["idempotency_key"], ( + "execute_action called the provider without a durable claim on its slot") + +def _boom(environment, operation, payload): + raise controller.LifecycleError("provider process modeled as crashing") +controller.provider_call = _boom +with controller.controller_lock(env): + crash_state = controller.load_state(env) + crashed = None + try: + controller.execute_action(env, crash_state, json.loads(json.dumps(action))) + except controller.LifecycleError as error: + crashed = error +assert crashed is not None +durable = json.loads((env["state_path"]).read_text()) +assert durable["pending_actions"]["1"]["idempotency_key"] == action["idempotency_key"], ( + "a crash during the provider call left no durable claim to replay") +# And on the CALLER'S object after the earlier failed apply: a pop that moved +# ahead of the apply would erase the replay obligation from the very object a +# later refusal handler saves, silently wedging the slot instead of replaying. +assert state["pending_actions"]["1"]["idempotency_key"] == action["idempotency_key"], ( + "a failed apply left the caller's object without its slot's durable claim") +controller.provider_call = lambda environment, operation, payload: { + "result": {"idempotency_key": action["idempotency_key"], + "worker": {"slot": 1, "resources": moved}} +} # The object the caller still holds, not the file. execute_action does not save # after a failed apply, so nothing reaches disk on this path either way; what @@ -2138,7 +2204,16 @@ with controller.controller_lock(env): replayed = controller.load_state(env) replayed["workers"]["1"] = copy.deepcopy(WORKER) replayed["queue"][WORKER["queue_key"]] = {"status": "queued", "slot": None} - replayed["pending_action"] = action + # A real minted claim: verify_state re-derives the idempotency key from + # the stored bytes at save, so the fake used before this schema would + # make the document unsaveable and the unit vacuous. + mint_worker = dict(copy.deepcopy(WORKER), sku="sku", sku_family="fam", + cloud_generation=1, cloud_instance_id=None, reservation_usd=1.0) + replay_action = controller.make_action( + {"deployment_generation": env["deployment_generation"], "owner": env["owner"]}, + "create", worker=mint_worker, + item=replayed["queue"][WORKER["queue_key"]]) + replayed["pending_actions"]["1"] = replay_action controller.save_state(env, replayed) raised = None try: @@ -2233,29 +2308,46 @@ item = { "repository_binding": "4" * 64, "owner_kind": "primary", "role": "author", "eligible": True, "discretionary": True, "status": "queued", "enqueued_at": module.iso_utc(), } +item2 = dict(item, task="restart-task-two", task_generation="restart-gen-two", + home_binding="5" * 64, account_binding="6" * 64, + worktree_binding="7" * 64, repository_binding="8" * 64) +actions = [] with module.controller_lock(env): state = module.load_state(env) - state["queue"][module.request_key(item["task"], item["task_generation"])] = item + for entry in (item, item2): + state["queue"][module.request_key(entry["task"], entry["task_generation"])] = entry inventory = module.provider_call(env, "inventory")["inventory"] - action = module.next_reconcile_action(env, state, inventory) - state["pending_action"] = action - module.save_state(env, state) - # The provider completed, but the controller process is modeled as dying - # before it durably applied the response. - module.provider_call(env, "mutate", action) + revision_before = state["revision"] + for _ in range(2): + # Each planning pass claims the next free slot; parking the claim on + # the map (without applying) is exactly the crash-before-apply image, + # and two of them at once is the shape the scalar could never hold. + action = module.next_reconcile_action(env, state, inventory) + state["pending_actions"][str(action["slot"])] = action + module.save_state(env, state) + # The provider completed, but the controller process is modeled as + # dying before it durably applied the response. + module.provider_call(env, "mutate", action) + actions.append(action) +assert sorted(str(a["slot"]) for a in actions) == ["1", "2"] with module.controller_lock(env): restarted = module.load_state(env) - assert restarted["pending_action"]["idempotency_key"] == action["idempotency_key"] + assert restarted["pending_action"] == module.LEGACY_PENDING_SENTINEL + assert restarted["revision"] > revision_before + for action in actions: + held = restarted["pending_actions"][str(action["slot"])] + assert held["idempotency_key"] == action["idempotency_key"] assert module.replay_pending(env, restarted) is True - assert restarted["pending_action"] is None - assert len(restarted["workers"]) == 1 + assert restarted["pending_actions"] == {} + assert len(restarted["workers"]) == 2 fixture = json.loads(Path(os.environ["FIXTURE_STATE"]).read_text()) -matching = [call for call in fixture["calls"] if call["key"] == action["idempotency_key"]] -assert len(matching) == 2 -assert len(fixture["seen"]) == 1 and len(fixture["workers"]) == 1 +for action in actions: + matching = [call for call in fixture["calls"] if call["key"] == action["idempotency_key"]] + assert len(matching) == 2, (action["slot"], len(matching)) +assert len(fixture["seen"]) == 2 and len(fixture["workers"]) == 2 PY - pass "restart replays one exact idempotency key without duplicating assignment" + pass "restart replays each per-slot idempotency key exactly once without duplicating assignment" } @@ -2528,6 +2620,7 @@ def base_state(): "queue": {"task-1@gen-1": {"status": "assigned", "slot": 1, "task": "task-1", "task_generation": "gen-1"}}, "workers": {"1": worker_record()}, "pending_action": None, + "pending_actions": {}, "executions": {}, } @@ -2563,7 +2656,7 @@ expect_refusal(base_state(), args(task="../escape"), "bounded identifier charact # A pending provider action blocks the whole lane. state = base_state() -state["pending_action"] = {"type": "execute", "request": {"task": "other", "task_generation": "gen-9"}} +state["pending_actions"] = {"2": {"type": "execute", "request": {"task": "other", "task_generation": "gen-9"}}} expect_refusal(state, args(), "pending provider action") # A converged entry names its credential recovery instead of a generic refusal. @@ -2632,6 +2725,238 @@ PY } + +legacy_scalar_migration() { + local tmp provider fixture home envfile + fm_test_tmproot_into tmp fm-worker-legacy-migration + provider="$tmp/provider.py" + fixture="$tmp/provider-state.json" + home="$tmp/home" + mkdir -p "$home" + write_fixture_provider "$provider" + envfile="$tmp/env" + cat >"$envfile" <= 1 +PY + pass "a legacy scalar pending action migrates onto the map and poisons the scalar" +} + +state_fence_and_revision_cas() { + local tmp home + fm_test_tmproot_into tmp fm-worker-state-fence + home="$tmp/home" + mkdir -p "$home" + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID=$SUB \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + python3 - "$CONTROLLER" <<'PY' || fail "the load fence, revision CAS, or re-entrancy refusal is not enforced" +import importlib.util +import json +from pathlib import Path +import sys + +spec = importlib.util.spec_from_file_location("controller", sys.argv[1]) +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) +env = module.environment() + +# Re-entrant acquisition refuses instead of deadlocking. +with module.controller_lock(env): + try: + with module.controller_lock(env): + pass + except module.LifecycleError as exc: + assert "already held" in str(exc), exc + else: + raise AssertionError("re-entrant controller lock acquisition was allowed") + +# A state loaded outside the committing lock hold refuses to save. +with module.controller_lock(env): + stale = module.load_state(env) + module.save_state(env, stale) +with module.controller_lock(env): + try: + module.save_state(env, stale) + except module.LifecycleError as exc: + assert "outside the committing lock hold" in str(exc), exc + else: + raise AssertionError("a state loaded under an earlier hold was committed") + +# A plain dict never saves, whatever it claims to be. +with module.controller_lock(env): + fresh = module.load_state(env) + try: + module.save_state(env, dict(fresh)) + except module.LifecycleError as exc: + assert "not loaded through load_state" in str(exc), exc + else: + raise AssertionError("an unfenced document was committed") + +# A tampered claim refuses at load: the stored bytes no longer re-derive +# the stored idempotency key, so a hand-edited, truncated, or fabricated +# claim cannot ride the map into a replay. +with module.controller_lock(env): + doc = module.load_state(env) + worker = { + "slot": 1, "sku": "sku", "sku_family": "fam", "cloud_generation": 1, + "cloud_instance_id": None, "reservation_usd": 1.0, "resources": {}, + "bindings": {"task": "t", "task_generation": "g"}, + } + claim = module.make_action( + {"deployment_generation": env["deployment_generation"], "owner": env["owner"]}, + "deallocate", worker=worker) + doc["pending_actions"]["1"] = claim + module.save_state(env, doc) +raw = json.loads(Path(env["state_path"]).read_text()) +raw["pending_actions"]["1"]["cloud_generation"] = 7 +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) +with module.controller_lock(env): + try: + module.load_state(env) + except module.LifecycleError as exc: + assert "pending provider action is malformed" in str(exc), exc + else: + raise AssertionError("a tampered claim loaded cleanly") +# Restore an honest document for the CAS case below. +raw["pending_actions"] = {} +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) + +# A corrupt legacy scalar refuses at load instead of being paved over. +raw = json.loads(Path(env["state_path"]).read_text()) +raw["pending_action"] = "corrupted-garbage" +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) +with module.controller_lock(env): + try: + module.load_state(env) + except module.LifecycleError as exc: + assert "pending provider action is malformed" in str(exc), exc + else: + raise AssertionError("a corrupt legacy scalar was silently paved over") + +# A legacy scalar that DISAGREES with the map entry for its slot refuses. +raw["pending_action"] = claim +disagreeing = json.loads(json.dumps(claim)) +disagreeing["cloud_generation"] = 2 +disagreeing["idempotency_key"] = module.action_id(disagreeing) +raw["pending_actions"] = {"1": disagreeing} +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) +with module.controller_lock(env): + try: + module.load_state(env) + except module.LifecycleError as exc: + assert "disagree" in str(exc), exc + else: + raise AssertionError("disagreeing legacy and per-slot claims loaded cleanly") + +# A claim naming a slot outside 1..MAX_WORKERS refuses, even when it +# self-hashes: the planner can never produce one, so it is a hand edit. +out_of_range = { + "slot": 17, "sku": "sku", "sku_family": "fam", "cloud_generation": 1, + "cloud_instance_id": None, "reservation_usd": 1.0, "resources": {}, + "bindings": {"task": "t", "task_generation": "g"}, +} +range_claim = module.make_action( + {"deployment_generation": env["deployment_generation"], "owner": env["owner"]}, + "deallocate", worker=out_of_range) +raw["pending_action"] = None +raw["pending_actions"] = {"17": range_claim} +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) +with module.controller_lock(env): + try: + module.load_state(env) + except module.LifecycleError as exc: + assert "pending provider action is malformed" in str(exc), exc + else: + raise AssertionError("a claim outside the slot range loaded cleanly") +raw["pending_actions"] = {} +Path(env["state_path"]).write_text(json.dumps(raw, sort_keys=True, separators=(",", ":"))) + +# The revision CAS: a concurrent writer moved the file after this load. +with module.controller_lock(env): + mine = module.load_state(env) + on_disk = json.loads(Path(env["state_path"]).read_text()) + on_disk["revision"] = int(on_disk.get("revision", 0)) + 1 + Path(env["state_path"]).write_text(json.dumps(on_disk, sort_keys=True, separators=(",", ":"))) + try: + module.save_state(env, mine) + except module.LifecycleError as exc: + assert "revision moved from" in str(exc), exc + else: + raise AssertionError("a stale save overwrote a moved revision") +PY + pass "the load fence, revision CAS, and re-entrancy refusal each fail loudly" +} + + static_contract classification_and_admission_matrix azure_provider_refusal_matrix @@ -2646,5 +2971,7 @@ partial_apply_never_persists surrender_lane surrender_refuses_when_ordinary_authority_passes surrender_refusal_matrix +legacy_scalar_migration +state_fence_and_revision_cas echo "# fm-worker-lifecycle.test.sh: all assertions passed"