From 42fdf2376c653aa649d40bec70de1270cae1e9c6 Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Wed, 19 Aug 2026 14:36:38 -0400 Subject: [PATCH 1/3] feat(worker): run provider mutations outside the fleet lock under per-slot leases (C2 3 of 3) The fleet lock now covers only short read-validate-claim and apply sections. Every provider mutation goes through provider_mutate under a non-blocking per-slot flock lease (provider_call refuses 'mutate' outright), claimed durably by claim_pending before the provider is touched and applied by apply_pending against a FRESH load, never the caller's pre-call object. Mutations for different slots run concurrently; an execute no longer blocks status, dry reconcile, requests, or withdrawals for its whole guest run. Reconcile fetches inventory unlocked, plans and claims in one short hold, mutates unlocked, applies in a fresh hold, and records refusals on a CLEAN load. The drain of stranded claims runs AFTER convergence with strict=False, skipping any slot whose lease a live process holds, so a wedged or hours-long replay cannot stop the fleet converging; the planner and the classification refresh skip claimed slots, whose durable records are deliberately not yet the truth. resume drains only its own slot, strictly. claim_pending refuses a different key on a claimed slot, closing the blind overwrite that silently discarded the first claim's replay obligation and ran the guest twice. The sanctioned exit that overwrite used to provide is now explicit: abandon-claim takes the lease, replays the mutation itself, requires the provider result to bind the exact idempotency key (the mutation is provably complete and its result final under key-idempotency), records the refusal verbatim with the result digest in cleanup_refusals, and only then clears the claim. The capacity commands keep their single holds deliberately: merged_specialized_reservations ignores non-reserved locals, so a split would let two concurrent reserves each admit against a budget that fits one. The two-writer proof parks two creates INSIDE the provider on a real FIFO rendezvous and asserts a two-entry pending_actions map with two distinct assignment generations - a shape structurally unsatisfiable under the old fleet-lock discipline, with a one-child positive control that must fail the two-arrival wait. The fixture provider gains an flock around its read-modify-write so a lost update cannot masquerade as a controller bug. --- bin/fm-worker-lifecycle.py | 394 ++++++++++++++++++---- bin/fm-worker-lifecycle.sh | 3 +- docs/azure-requirements.md | 9 +- docs/azure-workers.md | 3 +- tests/behavior-test-durations.tsv | 2 +- tests/fm-worker-lifecycle.test.sh | 523 +++++++++++++++++++++++++----- 6 files changed, 773 insertions(+), 161 deletions(-) diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index 70d9c748526..16ba24a3cc9 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -295,6 +295,7 @@ def environment(): "state_dir": state_dir, "state_path": state_dir / "controller.json", "lock_path": state_dir / ".lock", + "slot_lock_dir": state_dir / "slots", "max_workers": max_workers, "cooldown_seconds": cooldown, "warm_idle": warm_idle, @@ -346,6 +347,70 @@ def controller_lock(env): _LOCK_STATE["held"] = False +class SlotBusy(LifecycleError): + pass + + +class SlotLease: + __slots__ = ("slot", "handle") + + def __init__(self, slot, handle): + self.slot = slot + self.handle = handle + + +@contextlib.contextmanager +def slot_lease(env, slot): + """Exclusive claim on ONE slot's provider mutation, for the call's duration. + + LOCK_NB is hardcoded here so no call site can choose otherwise: exactly one + lock in the system is ever waited on (the fleet lock), which makes deadlock + impossible even though the apply phase takes the fleet lock while holding + this one. The kernel drops it on process death, which is what makes a + crashed owner's claim drainable at once; a durable lease would wedge the + slot until manual repair, and a timed lease would have to exceed the + longest provider deadline, which is not a lease. + """ + slot = str(slot) + if not slot.isdigit(): + raise LifecycleError("slot lease requires one exact decimal slot") + env["slot_lock_dir"].mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(env["slot_lock_dir"], 0o700) + path = env["slot_lock_dir"] / "slot-{}.lock".format(slot) + with open(path, "a+", encoding="utf-8") as handle: + os.chmod(path, 0o600) + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + raise SlotBusy("slot {} provider mutation is owned by a live process".format(slot)) + yield SlotLease(slot, handle) + + +def slot_lease_held(env, slot): + """Liveness display only: whether some live process holds this slot's lease.""" + path = env["slot_lock_dir"] / "slot-{}.lock".format(str(slot)) + if not path.exists(): + return False + with open(path, "a+", encoding="utf-8") as handle: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + return True + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + return False + + +def provider_mutate(env, action, lease): + """The only path to a provider mutation: a live lease on the exact slot.""" + if not isinstance(lease, SlotLease) or lease.slot != str(action.get("slot")): + raise LifecycleError("provider mutation is not covered by its slot's lease") + response = _provider_call_raw(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") + return result + + def empty_state(env): return { "schema": STATE_SCHEMA, @@ -563,6 +628,14 @@ def provider_action_timeout(action): def provider_call(env, operation, action=None): + if operation == "mutate": + # The only mutate path is provider_mutate, which requires a live slot + # lease; a bare mutate here is a call site the lock discipline missed. + raise LifecycleError("provider mutations go through provider_mutate with a slot lease") + return _provider_call_raw(env, operation, action) + + +def _provider_call_raw(env, operation, action=None): request = { "schema": PROVIDER_REQUEST_SCHEMA, "operation": operation, @@ -1128,6 +1201,8 @@ def action_id(action): def make_action(env, action_type, worker=None, item=None, **fields): + if action_type in ACTION_TYPES and worker is None: + raise LifecycleError("a provider mutation cannot be minted without its exact worker") action = { "type": action_type, "deployment_generation": env["deployment_generation"], @@ -1296,10 +1371,21 @@ def apply_result_transactionally(env, state, action, result): state.update(working) -def execute_action(env, state, action): +def claim_pending(env, state, action): + """Durably claim one slot's provider mutation. Caller holds the fleet lock + AND the slot's lease; the claim's save must land before the provider is + called, or a crash there strands a cloud mutation with no replay owner.""" slot = str(action.get("slot", "")) if not slot.isdigit(): raise LifecycleError("provider mutation carries no exact slot") + existing = state["pending_actions"].get(slot) + if existing is not None and existing.get("idempotency_key") != action["idempotency_key"]: + # Overwriting a live claim silently discards its replay obligation: + # the first execution never lands in executions, its dedupe + # short-circuit never fires, and the guest command runs twice. + raise LifecycleError( + "slot {} still has an unapplied {} action; reconcile it first".format( + slot, existing.get("type", "provider"))) # 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 @@ -1307,13 +1393,69 @@ def execute_action(env, state, action): # 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") + + +def apply_pending(env, action, result): + """Apply one mutation's result. Caller holds the fleet lock AND the lease. + + The state is ALWAYS re-loaded here, never the caller's pre-call object: + the provider ran outside the fleet lock, and anything read before it may + already be stale. + """ + state = load_state(env) + slot = str(action["slot"]) + claimed = state["pending_actions"].get(slot) + if not isinstance(claimed, dict) or claimed.get("idempotency_key") != action["idempotency_key"]: + raise LifecycleError("durable claim for slot {} is no longer this action".format(slot)) apply_result_transactionally(env, state, action, result) state["pending_actions"].pop(slot, None) save_state(env, state) + return state + + +def drain_pending(env, slot=None, strict=True): + """Replay unapplied claims. Returns (drained_slots, refusals). + + A claim whose owning process is still ALIVE is skipped, never replayed: + re-sending a key that is still in flight is the one thing the single + fleet lock used to make impossible, and the only new way this design + could create two cloud assignments for one key. Must not be called under + the fleet lock (controller_lock refuses re-entry). + """ + with controller_lock(env): + snapshot = sorted( + (load_state(env).get("pending_actions") or {}).items(), key=lambda p: int(p[0])) + drained = [] + refusals = [] + for slot_key, action in snapshot: + if slot is not None and slot_key != str(slot): + continue + try: + with slot_lease(env, slot_key) as lease: + # The snapshot may be stale: another process can have applied + # this claim between the snapshot and this lease. Re-read + # before re-sending, or an already-applied key is mutated a + # second time for nothing and its absence then reads as a + # refusal. + with controller_lock(env): + current = load_state(env).get("pending_actions", {}).get(slot_key) + if not isinstance(current, dict) or current.get("idempotency_key") != action.get("idempotency_key"): + continue + result = provider_mutate(env, action, lease) + with controller_lock(env): + apply_pending(env, action, result) + drained.append(slot_key) + except SlotBusy: + continue + except LifecycleError as exc: + if strict: + raise + with controller_lock(env): + clean = load_state(env) + record_refusal(clean, clean["workers"].get(slot_key), exc) + save_state(env, clean) + refusals.append({"type": "replay-refused", "slot": int(slot_key), "reason": str(exc)[:500]}) + return drained, refusals def apply_action_result(env, state, action, result): @@ -1418,24 +1560,14 @@ def apply_action_result(env, state, action, result): raise LifecycleError("unsupported provider mutation result: {}".format(action_type)) -def replay_pending(env, state): - 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): cloud = inventory_by_slot(inventory) + claimed = state.get("pending_actions") or {} for worker in state["workers"].values(): + if str(worker["slot"]) in claimed: + # An unapplied mutation owns this slot; a display value derived + # from a record whose durable phase is not yet true would lie. + continue classification, note = classify_worker(worker, cloud.get(worker["slot"]), now=now) worker["last_classification"] = classification worker["classification_note"] = note @@ -1460,7 +1592,16 @@ def next_reconcile_action(env, state, inventory, now=None): # Released work is the only path to ordinary destruction. With queued work, # reset immediately; otherwise deallocate first and honor the short cooldown. waiting = queued_items(state) + claimed = state.get("pending_actions") or {} for slot_key in sorted(state["workers"], key=int): + if slot_key in claimed: + # An unapplied mutation owns this slot, so its durable record is + # deliberately not yet the truth: delete-compute writes phase + # before it can raise, and reset marks the queue complete before + # parse_time can raise. The fleet-wide pre-drain used to make + # planning on such a record unreachable; this skip replaces that + # shield, and the post-convergence drain owns the replay. + continue worker = state["workers"][slot_key] current = cloud.get(worker["slot"]) classification, note = classify_worker(worker, current, now=now) @@ -1513,41 +1654,58 @@ def next_reconcile_action(env, state, inventory, now=None): return action -def reconcile(env, state, apply, confirm_subscription): +def reconcile(env, apply, confirm_subscription): + """One bounded convergence pass. Provider calls run OUTSIDE the fleet lock; + each iteration is (short hold: plan+claim) -> (no hold: mutate) -> + (short hold: re-read+apply). The drain of previously stranded claims runs + AFTER convergence, in command_reconcile: planning past a claimed slot is + safe (the planner skips it), so a wedged or hours-long replay can no + longer stop the fleet from converging, and the convergence work is + already durable when a drain blocks.""" if apply and confirm_subscription != env["subscription"]: raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") - if apply and replay_pending(env, state): - pass actions = [] + inventory = None for _ in range(64): - response = provider_call(env, "inventory") - inventory = response["inventory"] - state["last_metrics"] = metrics_from_inventory(inventory) - refresh_classifications(state, inventory) - action = next_reconcile_action(env, state, inventory) - if action is None: - save_state(env, state) - return actions, inventory - if action.get("type") == "admission-refused": - actions.append(action) - save_state(env, state) - return actions, inventory - actions.append(action) - if not apply: - # Dry planning may have allocated an in-memory worker record. Do not - # persist or continue beyond the first mutation boundary. - if action["type"] == "create": - state["workers"].pop(str(action.get("slot")), None) - key = request_key(action["request"]["task"], action["request"]["task_generation"]) - state["queue"][key]["status"] = "queued" - return actions, inventory - try: - execute_action(env, state, action) - except LifecycleError as exc: - worker = state["workers"].get(str(action.get("slot"))) - record_refusal(state, worker, exc) - save_state(env, state) - raise + inventory = provider_call(env, "inventory")["inventory"] + with contextlib.ExitStack() as stack: + action = None + with controller_lock(env): + state = load_state(env) + state["last_metrics"] = metrics_from_inventory(inventory) + refresh_classifications(state, inventory) + action = next_reconcile_action(env, state, inventory) + if action is None: + save_state(env, state) + return actions, inventory + if action.get("type") == "admission-refused": + actions.append(action) + save_state(env, state) + return actions, inventory + actions.append(action) + if not apply: + # Dry planning may have allocated an in-memory worker + # record. Do not persist or continue beyond the first + # mutation boundary. + if action["type"] == "create": + state["workers"].pop(str(action.get("slot")), None) + key = request_key(action["request"]["task"], action["request"]["task_generation"]) + state["queue"][key]["status"] = "queued" + return actions, inventory + lease = stack.enter_context(slot_lease(env, action["slot"])) + claim_pending(env, state, action) + try: + result = provider_mutate(env, action, lease) + with controller_lock(env): + apply_pending(env, action, result) + except LifecycleError as exc: + # NEVER the half-applied object: the refusal is recorded on a + # clean load, and the durable claim stays for the drain. + with controller_lock(env): + clean = load_state(env) + record_refusal(clean, clean["workers"].get(str(action.get("slot"))), exc) + save_state(env, clean) + raise raise LifecycleError("reconcile exceeded its bounded 64-action convergence allowance") @@ -1733,7 +1891,8 @@ def status_projection(env, state, inventory=None): "retained_disks": retained_disks, "cleanup_refusals": state["cleanup_refusals"][-10:], "pending_mutations": [ - {"slot": int(slot), "type": action.get("type")} + {"slot": int(slot), "type": action.get("type"), + "lease_held": slot_lease_held(env, slot)} for slot, action in sorted( (state.get("pending_actions") or {}).items(), key=lambda p: int(p[0])) ], @@ -1813,6 +1972,15 @@ def parser(): help="acknowledge that recorded execute outcomes never proven landed are discarded", ) surrender_parser.add_argument("--confirm-subscription", required=True) + abandon_parser = sub.add_parser( + "abandon-claim", + help="retire one slot's unapplied claim after proving its mutation is complete", + ) + abandon_parser.add_argument("--slot", required=True) + abandon_parser.add_argument("--idempotency-key", required=True) + abandon_parser.add_argument("--confirm-abandon", action="store_true") + abandon_parser.add_argument("--confirm-subscription", required=True) + reconcile_parser = sub.add_parser("reconcile", help="plan or apply bounded convergence") reconcile_parser.add_argument("--apply", action="store_true") reconcile_parser.add_argument("--confirm-subscription") @@ -2018,9 +2186,12 @@ def public_action(action): def command_reconcile(env, args): + actions, inventory = reconcile(env, args.apply, args.confirm_subscription) + if args.apply: + drained, refusals = drain_pending(env, strict=False) + actions.extend(refusals) with controller_lock(env): state = load_state(env) - actions, inventory = reconcile(env, state, args.apply, args.confirm_subscription) status = status_projection(env, state, inventory) safe_actions = [public_action(action) for action in actions] output = {"actions": safe_actions, "status": status} @@ -2030,6 +2201,8 @@ def command_reconcile(env, args): for action in safe_actions: if action["type"] == "admission-refused": print("admission refused: {}".format(action["reason"])) + elif action["type"] == "replay-refused": + print("replay refused: slot={} {}".format(action.get("slot"), action["reason"])) else: print("{}: slot={} generation={}".format( action["type"], action.get("slot"), @@ -2403,7 +2576,9 @@ def command_execute(env, args): account_manifest = staged_directory_manifest( "account", args.account_dir, total_bound=ACCOUNT_TOTAL_BOUND, ) - with controller_lock(env): + inventory = provider_call(env, "inventory")["inventory"] + with contextlib.ExitStack() as stack: + with controller_lock(env): state = load_state(env) key = request_key(require_id("task", args.task), require_id("task generation", args.task_generation)) item = state["queue"].get(key) @@ -2414,7 +2589,6 @@ def command_execute(env, args): raise LifecycleError("execute assignment generation is not exact") if worker.get("release_proof") is not None: raise LifecycleError("released work cannot execute") - inventory = provider_call(env, "inventory")["inventory"] cloud = inventory_by_slot(inventory).get(worker["slot"]) classification, reason = classify_worker(worker, cloud) if classification != "assigned": @@ -2457,8 +2631,12 @@ def command_execute(env, args): env, "execute", worker=worker, request=request, request_digest=request["request_digest"], **staged, ) - execute_action(env, state, action) - result = state["executions"][request["request_digest"]] + lease = stack.enter_context(slot_lease(env, worker["slot"])) + claim_pending(env, state, action) + mutation = provider_mutate(env, action, lease) + with controller_lock(env): + applied = apply_pending(env, action, mutation) + result = applied["executions"][request["request_digest"]] print(json.dumps(result, sort_keys=True, separators=(",", ":"))) @@ -2782,17 +2960,86 @@ def write_surrender_output(path, proof): ) +def command_abandon_claim(env, args): + """Retire one slot's unapplied claim after proving its mutation is complete. + + The lock discipline removed the old, unsafe exit from a wedged claim: a + second execute used to blind-overwrite the first, silently discarding its + replay obligation and running the guest twice. The sanctioned exit must + exist, because an apply can refuse a provider result deterministically + (a version-skewed supervisor answering an outcome-expected execute with no + outcome disposition is the recorded case), and the planner deliberately + skips a claimed slot, so release, reset, and every other lane stays + blocked behind the claim forever. + + This command takes the slot lease (no live owner), REPLAYS the claim + itself and requires the provider result to bind the exact idempotency key + - proving the mutation is complete at the provider and its result is + final under key-idempotency - then attempts the ordinary apply. An apply + that succeeds ends the claim normally. An apply that refuses is recorded + verbatim, with the result digest, in cleanup_refusals before the claim is + cleared, so the abandonment preserves the evidence it retires. + """ + if not args.confirm_abandon: + raise LifecycleError("--confirm-abandon is required") + if args.confirm_subscription != env["subscription"]: + raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") + slot = str(args.slot) + if not slot.isdigit(): + raise LifecycleError("abandon-claim requires one exact decimal slot") + with controller_lock(env): + state = load_state(env) + action = (state.get("pending_actions") or {}).get(slot) + if not isinstance(action, dict): + raise LifecycleError("slot {} holds no unapplied claim".format(slot)) + if action.get("idempotency_key") != args.idempotency_key: + raise LifecycleError( + "slot {} holds a different claim; pass its exact idempotency key".format(slot)) + with slot_lease(env, slot) as lease: + result = provider_mutate(env, action, lease) + with controller_lock(env): + try: + apply_pending(env, action, result) + print("claim applied cleanly; nothing was abandoned") + return + except LifecycleError as exc: + refusal = exc + clean = load_state(env) + current = (clean.get("pending_actions") or {}).get(slot) + if not isinstance(current, dict) or current.get("idempotency_key") != action["idempotency_key"]: + raise LifecycleError("slot {} claim changed while abandoning; retry".format(slot)) + worker = clean["workers"].get(slot) + record_refusal(clean, worker, LifecycleError( + "claim abandoned by operator: {} (result digest {})".format( + str(refusal)[:400], result.get("result_digest") or digest_value(result)))) + clean["pending_actions"].pop(slot, None) + save_state(env, clean) + print("FM-ABANDONED-CLAIM {} {}".format(slot, action["idempotency_key"])) + print("abandoned claim recorded in cleanup refusals; the slot plans normally again") + + def command_resume(env, args): if not args.confirm_resume: raise LifecycleError("--confirm-resume is required") if args.confirm_subscription != env["subscription"]: raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") require_binding("repository binding", args.repository_binding) + key = request_key(require_id("task", args.task), require_id("task generation", args.task_generation)) + # A stranded claim on THIS slot must replay before resume can judge the + # worker, and it must replay strictly: resume's preconditions are only + # meaningful against fully applied state for this worker. Slot-scoping + # keeps an unrelated slot's wedged or hours-long replay from blocking or + # failing this resume; apply only touches the owning slot's compartment. with controller_lock(env): + peek = load_state(env) + peek_item = peek["queue"].get(key) + resume_slot = str((peek_item or {}).get("slot", "")) + if resume_slot.isdigit(): + drain_pending(env, slot=resume_slot, strict=True) + inventory = provider_call(env, "inventory")["inventory"] + with contextlib.ExitStack() as stack: + with controller_lock(env): state = load_state(env) - 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) if item is None or item.get("status") != "assigned": raise LifecycleError("resume requires one exact assigned task generation") @@ -2803,7 +3050,6 @@ def command_resume(env, args): raise LifecycleError("released work cannot use dirty-task resume") if worker.get("bindings", {}).get("repository_binding") != args.repository_binding: raise LifecycleError("retained repository/task generation proof is not exact") - inventory = provider_call(env, "inventory")["inventory"] cloud = inventory_by_slot(inventory).get(worker["slot"]) classification, reason = classify_worker(worker, cloud) if classification != "retained-for-investigation" or cloud is None: @@ -2829,7 +3075,11 @@ def command_resume(env, args): "reservation_usd": worker["reservation_usd"], }), ) - execute_action(env, state, action) + lease = stack.enter_context(slot_lease(env, worker["slot"])) + claim_pending(env, state, action) + mutation = provider_mutate(env, action, lease) + with controller_lock(env): + apply_pending(env, action, mutation) print("replacement generation attached the exact retained task and account disks") @@ -2839,7 +3089,9 @@ def command_steer(env, args): if args.confirm_subscription != env["subscription"]: raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") request_digest = require_binding("steer request digest", args.request_digest) - with controller_lock(env): + inventory = provider_call(env, "inventory")["inventory"] + with contextlib.ExitStack() as stack: + with controller_lock(env): state = load_state(env) key = request_key(require_id("task", args.task), require_id("task generation", args.task_generation)) item = state["queue"].get(key) @@ -2848,22 +3100,26 @@ def command_steer(env, args): worker = state["workers"].get(str(item.get("slot"))) if worker is None or worker["assignment_generation"] != args.assignment_generation: raise LifecycleError("steer assignment generation is not exact") - inventory = provider_call(env, "inventory")["inventory"] cloud = inventory_by_slot(inventory).get(worker["slot"]) classification, reason = classify_worker(worker, cloud) if classification != "assigned": raise LifecycleError("steer refuses a non-assigned or ambiguous worker: {}".format(reason)) action = make_action(env, "steer", worker=worker, request_digest=request_digest) - execute_action(env, state, action) + lease = stack.enter_context(slot_lease(env, worker["slot"])) + claim_pending(env, state, action) + mutation = provider_mutate(env, action, lease) + with controller_lock(env): + apply_pending(env, action, mutation) print("steer request digest delivered to the exact worker generation") def command_status(env, args): + inventory = None + if args.live: + inventory = provider_call(env, "inventory")["inventory"] with controller_lock(env): state = load_state(env) - inventory = None - if args.live: - inventory = provider_call(env, "inventory")["inventory"] + if inventory is not None: state["last_metrics"] = metrics_from_inventory(inventory) refresh_classifications(state, inventory) save_state(env, state) @@ -2921,6 +3177,8 @@ def main(argv=None): command_withdraw(env, args) elif args.command == "surrender": command_surrender(env, args) + elif args.command == "abandon-claim": + command_abandon_claim(env, args) elif args.command == "resume": command_resume(env, args) elif args.command == "steer": diff --git a/bin/fm-worker-lifecycle.sh b/bin/fm-worker-lifecycle.sh index 4ddbc3b2a0e..883c4b78292 100755 --- a/bin/fm-worker-lifecycle.sh +++ b/bin/fm-worker-lifecycle.sh @@ -35,6 +35,7 @@ # fm-worker-lifecycle.sh proof-template --task --task-generation # fm-worker-lifecycle.sh release --task --task-generation --proof-file # fm-worker-lifecycle.sh withdraw --task --task-generation --confirm-withdraw --confirm-subscription +# fm-worker-lifecycle.sh abandon-claim --slot --idempotency-key --confirm-abandon --confirm-subscription # fm-worker-lifecycle.sh surrender --task --task-generation --reason --output --confirm-surrender --confirm-subscription # fm-worker-lifecycle.sh resume # fm-worker-lifecycle.sh steer @@ -49,7 +50,7 @@ SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P) . "$SCRIPT_DIR/fm-cloud-state-lib.sh" case "${1:-}" in - request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release) + request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|abandon-claim) fm_refuse_if_gate_agent exec python3 "$SCRIPT_DIR/fm-worker-lifecycle.py" "$@" ;; diff --git a/docs/azure-requirements.md b/docs/azure-requirements.md index 20f7b1aaff0..ba80525fc6f 100644 --- a/docs/azure-requirements.md +++ b/docs/azure-requirements.md @@ -324,9 +324,12 @@ Acceptance: a measured review completes in 20 to 30 minutes, with the breakdown Status: NOT DONE. -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. +All three C2 changes are landed: the transactional apply, the per-slot `pending_actions` map +with its load fence and revision CAS, and the lock discipline that runs every provider mutation +outside the fleet lock under a non-blocking per-slot lease, with the drain after convergence and +`abandon-claim` as the evidence-preserving exit from a deterministically refused claim. +What remains for DONE is the acceptance itself: many crewmates, no-mistakes runs, and +crosschecks demonstrated running in parallel against live capacity without contention. 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 1b3e2cda6e6..07f9bd680ed 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -236,7 +236,8 @@ bin/fm-worker-lifecycle.sh reconcile \ --confirm-subscription "$FM_AZURE_SUBSCRIPTION_ID" ``` -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. +The home lock now covers only short read-validate-claim and apply sections; every provider mutation runs outside it under a non-blocking per-slot lease, so mutations for different slots run concurrently while readers and unrelated mutations proceed. 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. +Reconcile drains stranded claims AFTER convergence, skipping any slot whose claim a live process still owns, so a wedged or hours-long replay cannot stop the fleet; a claim whose provider result is final but whose apply deterministically refuses is retired only through `abandon-claim`, which replays the mutation itself under the lease, proves the result binds the exact idempotency key, and records the refusal verbatim before clearing the claim. 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 slot's pending action and records a bounded cleanup refusal. The next controller process replays that exact action before considering new work. diff --git a/tests/behavior-test-durations.tsv b/tests/behavior-test-durations.tsv index 87e2eeb8a1e..4ce5c5edc19 100644 --- a/tests/behavior-test-durations.tsv +++ b/tests/behavior-test-durations.tsv @@ -112,7 +112,7 @@ 180000 tests/fm-watch-pause-absorb.test.sh 95645 tests/fm-watch-triage.test.sh 20435 tests/fm-watcher-lock.test.sh -8000 tests/fm-worker-lifecycle.test.sh +12000 tests/fm-worker-lifecycle.test.sh 1000 tests/fm-worker-outcome-transport.test.sh 1000 tests/fm-worker-supervisor.test.sh 23744 tests/fm-x-mode.test.sh diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index 78a06e0a11b..d185b838bc5 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -59,7 +59,8 @@ for marker in ( "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", + "revision moved from", "FencedState", "slot_lease", "LOCK_NB", "provider_mutate", + "drain_pending", "claim_pending", "apply_pending", "command_abandon_claim", "capacity-reserve", "capacity-reserve-shape", "capacity-release", "merged_specialized_reservations", "command_withdraw", "command_surrender", "WORKER AUTHORITY REFUSED", "--confirm-discard-unlanded", @@ -792,9 +793,38 @@ import os from pathlib import Path import sys +import fcntl + path = Path(os.environ["FIXTURE_STATE"]) request = json.load(sys.stdin) controller = request["controller"] + +def barrier(action): + # A real kernel rendezvous, not a sleep: arrival is an O_EXCL file named + # by the idempotency key; release is a blocking FIFO read. It must not + # perturb the action payload, which the key check below re-derives, and it + # must run BEFORE the fixture state lock, or the first parked mutate would + # hold the lock and the second could never arrive. + barrier_dir = os.environ.get("FIXTURE_BARRIER_DIR") + if not barrier_dir: + return + kinds = os.environ.get("FIXTURE_BARRIER_TYPES", "").split(",") + if action.get("type") not in kinds: + return + arrived = Path(barrier_dir) / "arrived" + arrived.mkdir(parents=True, exist_ok=True) + fd = os.open(str(arrived / action["idempotency_key"]), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.close(fd) + with open(Path(barrier_dir) / "release", "r") as gate: + gate.read(1) + +if request["operation"] == "mutate": + barrier(request["action"]) +# The stub is not concurrency-safe without this: two concurrent mutates would +# lose one another's read-modify-write and fire the create assertion below +# spuriously, indistinguishable from a controller bug. +lock_handle = open(str(path) + ".lock", "a+") +fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) if path.exists(): state = json.loads(path.read_text()) else: @@ -1201,6 +1231,32 @@ skewed = armed_execute( command="/bin/echo", ) assert skewed.returncode != 0 and "no outcome disposition" in skewed.stderr, skewed.stderr + +# The skewed execute left a durable claim, and its apply refuses +# deterministically (the provider result is final under key-idempotency), so +# the slot is deliberately wedged: a different execute refuses instead of +# blind-overwriting the claim, which is the double-run defect this discipline +# closes. The sanctioned exit is abandon-claim, which replays the mutation +# itself, proves the result binds the exact key, records the refusal with the +# result digest, and only then clears the claim. +skew_state = controller_state() +skew_slot = str(skew_state["queue"]["task-2@gen-2"]["slot"]) +skew_claim = skew_state["pending_actions"][skew_slot] +blocked = armed_execute( + "--payload-dir", str(payload_dir), "--account-dir", str(account_dir), + "--outcome-dir", str(outcome_dir), +) +assert blocked.returncode != 0 and "still has an unapplied execute action" in blocked.stderr, blocked.stderr +refused_abandon = run("abandon-claim", "--slot", skew_slot, "--idempotency-key", "f" * 64, + "--confirm-abandon", "--confirm-subscription", + env["FM_AZURE_SUBSCRIPTION_ID"], check=False) +assert refused_abandon.returncode != 0 and "different claim" in refused_abandon.stderr, refused_abandon.stderr +run("abandon-claim", "--slot", skew_slot, "--idempotency-key", skew_claim["idempotency_key"], + "--confirm-abandon", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) +after_abandon = controller_state() +assert skew_slot not in after_abandon["pending_actions"], after_abandon["pending_actions"] +assert any("claim abandoned by operator" in str(entry.get("note", "")) + for entry in after_abandon["cleanup_refusals"]), after_abandon["cleanup_refusals"] armed = armed_execute( "--payload-dir", str(payload_dir), "--account-dir", str(account_dir), "--outcome-dir", str(outcome_dir), @@ -2133,97 +2189,120 @@ 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": action["idempotency_key"], - "worker": {"slot": 1, "resources": moved}} +# The provider is the only thing stubbed, at the RAW boundary below the +# mutate ban; claim_pending, provider_mutate, apply_pending and drain_pending +# are the real ones, driven in the exact call-site sequence. +import contextlib as _ctx + +def _stub(resources): + def raw(environment, operation, payload): + return {"result": {"idempotency_key": payload["idempotency_key"], + "worker": {"slot": 1, "resources": copy.deepcopy(resources)}}} + return raw + +good = { + kind: { + "id": "/subscriptions/s/resourceGroups/g/providers/x/{}".format(kind), + "immutable_id": "imm-{}".format(kind), + "tags": dict(TAGS), + } + for kind in controller.REQUIRED_RESOURCE_KINDS } -with controller.controller_lock(env): - state = controller.load_state(env) - state["workers"]["1"] = copy.deepcopy(WORKER) - state["queue"][WORKER["queue_key"]] = {"status": "queued", "slot": None} - controller.save_state(env, state) - raised = None +# The mutate ban: no call site can reach the provider without a lease. +banned = None +try: + controller.provider_call(env, "mutate", action) +except controller.LifecycleError as error: + banned = error +assert banned is not None and "slot lease" in str(banned), banned + +controller._provider_call_raw = _stub(moved) +with _ctx.ExitStack() as stack: + with controller.controller_lock(env): + state = controller.load_state(env) + state["workers"]["1"] = copy.deepcopy(WORKER) + state["queue"][WORKER["queue_key"]] = {"status": "queued", "slot": None} + lease = stack.enter_context(controller.slot_lease(env, 1)) + controller.claim_pending(env, state, action) + + # The claim is durable BEFORE any provider call: a crash in the provider + # window leaves a replay obligation, proven from the file. + durable = json.loads(env["state_path"].read_text()) + assert durable["pending_actions"]["1"]["idempotency_key"] == action["idempotency_key"], ( + "claim_pending returned without a durable claim on its slot") + + # A DIFFERENT key on a claimed slot refuses: the blind overwrite that used + # to discard the first claim and run the guest twice is closed. + other = controller.make_action( + {"deployment_generation": "dep-one", "owner": "owner"}, "deallocate", + worker=MINT_WORKER) + with controller.controller_lock(env): + second = controller.load_state(env) + blocked = None + try: + controller.claim_pending(env, second, other) + except controller.LifecycleError as error: + blocked = error + assert blocked is not None and "still has an unapplied" in str(blocked), blocked + + # A lease on the wrong slot never reaches the provider. + wrong = None try: - controller.execute_action(env, state, action) + controller.provider_mutate(env, dict(other, slot=2), lease) except controller.LifecycleError as error: - raised = error -assert raised is not None, "execute_action accepted a moved identity" + wrong = error + assert wrong is not None and "slot's lease" in str(wrong), wrong + + result = controller.provider_mutate(env, action, lease) + with controller.controller_lock(env): + raised = None + try: + controller.apply_pending(env, action, result) + except controller.LifecycleError as error: + raised = error +assert raised is not None, "apply_pending 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()) +# The failed apply changed NOTHING durable: the point-3 image (claim present, +# worker untouched) is still exactly what the file holds. +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()) + "a failed apply erased the durable replay obligation") +assert durable["workers"]["1"]["phase"] == "creating" +assert not durable["workers"]["1"]["resources"], ( + "a failed apply left adopted resources on the durable record") + +# drain_pending with strict=False records the refusal and RETAINS the claim; +# the wedge stays visible instead of being silently discarded. +drained, refusals = controller.drain_pending(env, strict=False) +assert drained == [] and len(refusals) == 1 and refusals[0]["slot"] == 1, (drained, refusals) +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 -# differs is that the in-place apply leaves the caller holding a worker record -# carrying resources its create never got, and the very next thing a caller -# does with that object may be to save it. -assert state["workers"]["1"]["phase"] == "creating", state["workers"]["1"]["phase"] -assert not state["workers"]["1"]["resources"], ( - "execute_action left the caller holding the adopted resource set of a refused create" -) -assert state["queue"][WORKER["queue_key"]]["status"] == "queued", ( - "execute_action left the caller holding a queue entry a refused create had assigned" -) - -# The other call site. A restart replays whatever claim survived, and the -# existing restart coverage only exercises the path where the replay succeeds, -# where an in-place apply is indistinguishable from a transactional one. + "a refused replay dropped the claim") +assert any("provider mutation result" in str(entry.get("note", "")) or entry + for entry in durable["cleanup_refusals"]) + +# With an honest provider result, the drain applies and clears the claim. +controller._provider_call_raw = _stub(good) +drained, refusals = controller.drain_pending(env, strict=False) +assert drained == ["1"] and refusals == [], (drained, refusals) +durable = json.loads(env["state_path"].read_text()) +assert durable["pending_actions"] == {}, "a successful apply left its claim behind" +assert durable["workers"]["1"]["resources"], "the applied create adopted nothing" + +# Applying an action whose durable claim is GONE refuses on the fresh load: +# without this, a caller that lost a race with the drain (or with an operator +# abandon) would apply the same mutation's effects a second time. 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} - # 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 + stale_apply = None try: - controller.replay_pending(env, replayed) + controller.apply_pending(env, action, {"idempotency_key": action["idempotency_key"], + "worker": {"slot": 1, "resources": good}}) except controller.LifecycleError as error: - raised = error -assert raised is not None, "replay_pending accepted a moved identity" -assert not replayed["workers"]["1"]["resources"], ( - "replay_pending left the caller holding the adopted resource set of a refused create" -) + stale_apply = error +assert stale_apply is not None and "no longer this action" in str(stale_apply), stale_apply PY # An apply whose effects reach outside the slot it names is refused rather @@ -2325,11 +2404,14 @@ with module.controller_lock(env): 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"] +for action in actions: + # The provider completed, but the controller process is modeled as dying + # before it durably applied the response. Submission goes through the only + # legal mutate path: a live lease on the exact slot. + with module.slot_lease(env, action["slot"]) as lease: + module.provider_mutate(env, action, lease) with module.controller_lock(env): restarted = module.load_state(env) @@ -2338,9 +2420,12 @@ with module.controller_lock(env): 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_actions"] == {} - assert len(restarted["workers"]) == 2 +drained, refusals = module.drain_pending(env) +assert sorted(drained) == ["1", "2"] and refusals == [], (drained, refusals) +with module.controller_lock(env): + drained_state = module.load_state(env) + assert drained_state["pending_actions"] == {} + assert len(drained_state["workers"]) == 2 fixture = json.loads(Path(os.environ["FIXTURE_STATE"]).read_text()) for action in actions: matching = [call for call in fixture["calls"] if call["key"] == action["idempotency_key"]] @@ -2790,7 +2875,7 @@ result = subprocess.run([wrapper, "status", "--json"], env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) assert result.returncode == 0, result.stderr status = json.loads(result.stdout) -assert status["pending_mutations"] == [{"slot": 1, "type": "create"}], status["pending_mutations"] +assert status["pending_mutations"] == [{"slot": 1, "type": "create", "lease_held": False}], status["pending_mutations"] # A saving command makes the migration durable. result = subprocess.run([ @@ -2957,6 +3042,268 @@ PY } + +concurrent_mutations_do_not_serialize() { + local tmp provider fixture home envfile + fm_test_tmproot_into tmp fm-worker-concurrent + provider="$tmp/provider.py" + fixture="$tmp/provider-state.json" + home="$tmp/home" + mkdir -p "$home" "$tmp/barrier" + mkfifo "$tmp/barrier/release" + write_fixture_provider "$provider" + envfile="$tmp/env" + cat >"$envfile" <= count: + return True + time.sleep(0.05) + return False + +request(1) +request(2) + +# Two reconciles, each due to claim one create and park inside the provider. +children = [ + subprocess.Popen( + [wrapper, "reconcile", "--apply", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]], + env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + for _ in range(2) +] +try: + # Load-bearing: under the old fleet-lock discipline the second caller + # blocks at LOCK_EX before it ever reaches the provider, so a second + # arrival is structurally unsatisfiable there. + assert wait_for_arrivals(2, 60), "second mutation never reached the provider while the first was parked" + + state = controller_state() + pending = state["pending_actions"] + assert sorted(pending) == ["1", "2"], pending + generations = {entry["bindings"]["assignment_generation"] for entry in pending.values()} + assert len(generations) == 2, generations + assert sorted(state["workers"]) == ["1", "2"], sorted(state["workers"]) + assert state["pending_action"] == "superseded-by-pending-actions" + + # Readers and unrelated mutations proceed while both are parked. + status = json.loads(run("status", "--json").stdout) + held = {entry["slot"]: entry["lease_held"] for entry in status["pending_mutations"]} + assert held == {1: True, 2: True}, held + dry = run("reconcile", "--json") + assert json.loads(dry.stdout)["status"]["queue_depth"] >= 2 + request(9) + run("withdraw", "--task", "task-9", "--task-generation", "gen-9", "--confirm-withdraw", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) + + # A steer at a PARKED slot refuses without a third provider call: the + # queue entry is honestly still `assigning` while the create is in + # flight, so the earliest gate fires. (The claimed-slot refusal itself, + # "still has an unapplied ... action", is pinned at command level by the + # end-to-end skew scenario.) + arrivals_before = len(list((Path(barrier) / "arrived").iterdir())) + blocked = run("steer", "--task", "task-1", "--task-generation", "gen-1", + "--assignment-generation", state["workers"]["1"]["assignment_generation"], + "--request-digest", "a" * 64, "--confirm-steer", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"], check=False) + assert blocked.returncode != 0 and "one exact assigned task generation" in blocked.stderr, blocked.stderr + assert len(list((Path(barrier) / "arrived").iterdir())) == arrivals_before +finally: + with open(Path(barrier) / "release", "w") as gate: + gate.write("xx") + outcomes = [child.wait(timeout=120) for child in children] + +assert outcomes == [0, 0], [child.stderr.read() for child in children] +state = controller_state() +fixture = json.loads(Path(fixture_path).read_text()) +assert state["pending_actions"] == {} +assert len(fixture["workers"]) == 2 and len(fixture["seen"]) == 2 +for key, count in {}.items(): + pass +create_calls = [entry for entry in fixture["calls"] if entry["type"] == "create"] +per_key = {} +for entry in create_calls: + per_key[entry["key"]] = per_key.get(entry["key"], 0) + 1 +assert sorted(per_key.values()) == [1, 1], per_key +assert state["revision"] > 2 + +# Positive control: the same harness with ONE child must FAIL the two-arrival +# wait, or the wait proves nothing about concurrency. +for stale in (Path(barrier) / "arrived").iterdir(): + stale.unlink() +request(3) +lone = subprocess.Popen( + [wrapper, "reconcile", "--apply", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]], + env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) +try: + assert not wait_for_arrivals(2, 5), "one child produced two arrivals; the detector is broken" +finally: + with open(Path(barrier) / "release", "w") as gate: + gate.write("x") + assert lone.wait(timeout=120) == 0, lone.stderr.read() +PY + pass "two slots' provider mutations run concurrently while readers and unrelated mutations proceed" +} + +wedged_slot_does_not_stop_the_fleet() { + local tmp provider fixture home envfile + fm_test_tmproot_into tmp fm-worker-wedged + provider="$tmp/provider.py" + fixture="$tmp/provider-state.json" + home="$tmp/home" + mkdir -p "$home" + write_fixture_provider "$provider" + envfile="$tmp/env" + cat >"$envfile" < Date: Wed, 19 Aug 2026 15:06:58 -0400 Subject: [PATCH 2/3] harden(worker): absorb the lock-discipline review - pin the live lease refusal, close abandon's race The review proved the LOCK_NB discipline had no behavioral pin (a blocking lease survived the suite; the static marker was satisfied by a docstring): the concurrent unit now probes slot 1's lease from a bounded subprocess while the parked child owns it, requiring an immediate SlotBusy. abandon- claim gains the drain's under-lease re-read before re-sending, so a claim a concurrent drain applied is never mutated again without a durable claim naming it. The strict drain mode is pinned against the wedged claim (resume depends on it). The abandon docstring and help now say the proof is obtained BY submitting and what a refused create leaves behind; dead loop and stale execute_action comments removed. --- bin/fm-worker-lifecycle.py | 18 +++++++- tests/fm-worker-lifecycle.test.sh | 75 +++++++++++++++++++++++++++++-- 2 files changed, 88 insertions(+), 5 deletions(-) diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index 16ba24a3cc9..3c9a7b9b738 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -1974,7 +1974,7 @@ def parser(): surrender_parser.add_argument("--confirm-subscription", required=True) abandon_parser = sub.add_parser( "abandon-claim", - help="retire one slot's unapplied claim after proving its mutation is complete", + help="retire one slot's unapplied claim by replaying its mutation to completion first", ) abandon_parser.add_argument("--slot", required=True) abandon_parser.add_argument("--idempotency-key", required=True) @@ -2975,7 +2975,12 @@ def command_abandon_claim(env, args): This command takes the slot lease (no live owner), REPLAYS the claim itself and requires the provider result to bind the exact idempotency key - proving the mutation is complete at the provider and its result is - final under key-idempotency - then attempts the ordinary apply. An apply + final under key-idempotency - then attempts the ordinary apply. The proof + is obtained BY submitting: for a claim that never reached the provider, + the replay IS the first submission (abandoning a create builds the VM), + because dropping a claim of unknown provider-side status could strand a + resource that exists and is billing. An abandoned refused create leaves + its cloud resources to the ordinary planner, which now sees the slot. An apply that succeeds ends the claim normally. An apply that refuses is recorded verbatim, with the result digest, in cleanup_refusals before the claim is cleared, so the abandonment preserves the evidence it retires. @@ -2996,6 +3001,15 @@ def command_abandon_claim(env, args): raise LifecycleError( "slot {} holds a different claim; pass its exact idempotency key".format(slot)) with slot_lease(env, slot) as lease: + # The pre-lease read may be stale: a drain can have applied and + # cleared this claim between the check above and this lease. Re-read + # under the lease before re-sending, exactly as the drain does, so an + # already-applied key is never mutated again without a durable claim + # naming it. + with controller_lock(env): + current = (load_state(env).get("pending_actions") or {}).get(slot) + if not isinstance(current, dict) or current.get("idempotency_key") != action["idempotency_key"]: + raise LifecycleError("slot {} claim changed while abandoning; retry".format(slot)) result = provider_mutate(env, action, lease) with controller_lock(env): try: diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index d185b838bc5..d0f452b8c19 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -2180,7 +2180,7 @@ moved = { } for kind in controller.REQUIRED_RESOURCE_KINDS } -# A real minted action: execute_action now durably claims it, and the claim +# A real minted action: the call-site sequence 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", @@ -2303,6 +2303,43 @@ with controller.controller_lock(env): except controller.LifecycleError as error: stale_apply = error assert stale_apply is not None and "no longer this action" in str(stale_apply), stale_apply + +# abandon-claim's under-lease re-read: a claim a concurrent drain applied +# between abandon's pre-check and its lease must refuse WITHOUT re-sending +# the mutation. The load sequence is substituted (present at the pre-check, +# gone under the lease); the raw provider boundary asserts no call happens. +import types as _types +real_load = controller.load_state +with controller.controller_lock(env): + seeded = real_load(env) + seeded["workers"]["1"] = copy.deepcopy(WORKER) + seeded["pending_actions"]["1"] = copy.deepcopy(action) + controller.save_state(env, seeded) +loads = {"count": 0} +def sequenced_load(environment): + loads["count"] += 1 + state = real_load(environment) + if loads["count"] >= 2: + state["pending_actions"].pop("1", None) + return state +def never_mutate(environment, operation, payload): + raise AssertionError("abandon re-sent a mutation after its claim was applied elsewhere") +controller.load_state = sequenced_load +controller._provider_call_raw = never_mutate +raced = None +try: + controller.command_abandon_claim(env, _types.SimpleNamespace( + slot="1", idempotency_key=action["idempotency_key"], + confirm_abandon=True, confirm_subscription=env["subscription"])) +except controller.LifecycleError as error: + raced = error +finally: + controller.load_state = real_load +assert raced is not None and "changed while abandoning" in str(raced), raced +with controller.controller_lock(env): + cleanup = real_load(env) + cleanup["pending_actions"].pop("1", None) + controller.save_state(env, cleanup) PY # An apply whose effects reach outside the slot it names is refused rather @@ -3067,6 +3104,7 @@ FIXTURE_BARRIER_DIR=$tmp/barrier FIXTURE_BARRIER_TYPES=create EOF + FM_LIFECYCLE_CONTROLLER="$CONTROLLER" \ python3 - "$WRAPPER" "$envfile" "$fixture" "$tmp/barrier" <<'PY' || fail "two provider mutations did not run concurrently" import json import os @@ -3077,6 +3115,7 @@ import time wrapper, envfile, fixture_path, barrier = sys.argv[1:] env = os.environ.copy() +env["FM_LIFECYCLE_CONTROLLER"] = os.environ.get("FM_LIFECYCLE_CONTROLLER", "") for line in Path(envfile).read_text().splitlines(): key, value = line.split("=", 1) env[key] = value @@ -3148,6 +3187,29 @@ try: run("withdraw", "--task", "task-9", "--task-generation", "gen-9", "--confirm-withdraw", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) + # The lease itself, against a LIVE cross-process holder: the parked child + # owns slot 1's lease, so a second taker must refuse with SlotBusy at + # once. This is the only behavioral pin the LOCK_NB discipline has; a + # blocking lease here deadlocks the fleet-lock/lease cycle the design + # rules out, so the probe runs in a bounded subprocess. + probe = subprocess.run( + [sys.executable, "-c", ( + "import importlib.util, sys\n" + "spec = importlib.util.spec_from_file_location('c', sys.argv[1])\n" + "m = importlib.util.module_from_spec(spec)\n" + "spec.loader.exec_module(m)\n" + "env = m.environment()\n" + "try:\n" + " with m.slot_lease(env, 1):\n" + " print('ACQUIRED')\n" + "except m.SlotBusy as exc:\n" + " print('BUSY:', exc)\n" + ), os.environ.get("FM_LIFECYCLE_CONTROLLER", "")], + env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20) + assert probe.returncode == 0, probe.stderr + assert "BUSY:" in probe.stdout and "owned by a live process" in probe.stdout, ( + "a live holder did not refuse a second lease taker immediately", probe.stdout, probe.stderr) + # A steer at a PARKED slot refuses without a third provider call: the # queue entry is honestly still `assigning` while the create is in # flight, so the earliest gate fires. (The claimed-slot refusal itself, @@ -3170,8 +3232,6 @@ state = controller_state() fixture = json.loads(Path(fixture_path).read_text()) assert state["pending_actions"] == {} assert len(fixture["workers"]) == 2 and len(fixture["seen"]) == 2 -for key, count in {}.items(): - pass create_calls = [entry for entry in fixture["calls"] if entry["type"] == "create"] per_key = {} for entry in create_calls: @@ -3299,6 +3359,15 @@ plan = json.loads(run("reconcile", "--json").stdout) planned_slots = [entry.get("slot") for entry in plan["actions"] if entry["type"] not in ("admission-refused", "replay-refused")] assert 1 not in planned_slots, ( "the planner proposed a mutation for a slot an unapplied claim owns", plan["actions"]) + +# strict drain RAISES on the wedged claim instead of recording and moving on: +# resume depends on exactly this to refuse resuming over unapplied state. +strict_raised = None +try: + module.drain_pending(menv, slot="9", strict=True) +except module.LifecycleError as exc: + strict_raised = exc +assert strict_raised is not None, "a strict drain swallowed a wedged claim's refusal" PY pass "a wedged slot's claim is reported and retained while the rest of the fleet converges" } From 9e1be513f2d2b586e1eb0ff6ccf2b925e739dc3d Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Wed, 19 Aug 2026 15:31:33 -0400 Subject: [PATCH 3/3] test(pilot): route the provider subprocess-bound probe through the raw path the ban exposes fm-azure-pilot's bound probe called provider_call with 'mutate' directly, which the lock discipline now refuses; the timeout bound under test lives on _provider_call_raw, which every mutation reaches through provider_mutate. The probe now asserts the ban fires AND the raw path carries the bound. --- tests/fm-azure-pilot.test.sh | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/tests/fm-azure-pilot.test.sh b/tests/fm-azure-pilot.test.sh index b08eb0aa61b..9643a7067ab 100755 --- a/tests/fm-azure-pilot.test.sh +++ b/tests/fm-azure-pilot.test.sh @@ -1678,18 +1678,27 @@ env = { "provider_argv": ["/usr/bin/true"], } try: - lifecycle.provider_call(env, "mutate", {"type": "create"}) + # provider_call refuses "mutate" outright under the lock discipline; the + # subprocess bound under test lives on the raw path every mutate reaches + # through provider_mutate. + banned = None + try: + lifecycle.provider_call(env, "mutate", {"type": "create"}) + except lifecycle.LifecycleError as exc: + banned = exc + assert banned is not None and "slot lease" in str(banned), banned + lifecycle._provider_call_raw(env, "mutate", {"type": "create"}) create_timeout = captured["timeout"] - lifecycle.provider_call( + lifecycle._provider_call_raw( env, "mutate", {"type": "execute", "request": {"wall_seconds": 3600}} ) execute_timeout = captured["timeout"] finally: lifecycle.subprocess.run = _real_run -assert create_timeout >= 900, ("provider_call did not bound a create by its action", create_timeout) +assert create_timeout >= 900, ("the raw provider path did not bound a create by its action", create_timeout) assert execute_timeout >= 3600 + 1800, ( - "provider_call did not bound an execute by its guest run", execute_timeout, + "the raw provider path did not bound an execute by its guest run", execute_timeout, ) print("OK") PROVIDERBOUND