From da237789ce80832a13e9bc9e20be9ab4b7cd72b5 Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Fri, 21 Aug 2026 10:18:18 -0400 Subject: [PATCH] fix(azure): recover terminal worker claims --- bin/fm-azure-worker-provider.py | 221 +++++++++++---- bin/fm-worker-lifecycle.py | 117 ++++++-- docs/azure-workers.md | 5 +- tests/fm-worker-lifecycle.test.sh | 447 ++++++++++++++++++++++++++++-- 4 files changed, 686 insertions(+), 104 deletions(-) diff --git a/bin/fm-azure-worker-provider.py b/bin/fm-azure-worker-provider.py index bd0393dcda7..f9b90cc37cf 100755 --- a/bin/fm-azure-worker-provider.py +++ b/bin/fm-azure-worker-provider.py @@ -53,6 +53,10 @@ REQUEST_SCHEMA = "fm.worker-provider-request/v1" RESPONSE_SCHEMA = "fm.worker-provider-response/v1" INVENTORY_SCHEMA = "fm.worker-provider-inventory/v1" +EXECUTION_TERMINAL_SCHEMA = "fm.worker-execution-terminal/v1" +EXECUTE_DISPOSITION_SUBMIT = "submit" +EXECUTE_DISPOSITION_TERMINAL = "terminal" +EXECUTE_DISPOSITION_RECOVERED = "recovered" MAX_INPUT_BYTES = 2 * 1024 * 1024 # The one blob name the guest may create, and the ceiling the supervisor # enforces before it uploads; both sides bound the same transfer. The name @@ -250,6 +254,10 @@ class ProviderError(RuntimeError): pass +class ProviderIdentityRefusal(ProviderError): + pass + + def canonical_bytes(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") @@ -1284,17 +1292,21 @@ def recorded_exact( raise ProviderError("exact {} resource is absent".format(kind)) if prior is not None: if current.get("id") != prior.get("id"): - raise ProviderError("{} resource ID differs from the recorded assignment".format(kind)) - # Staging request/result blobs are per-execution transport: every - # execute rewrites them under the same stable blob path and binds - # their content through the request and result digests, so only - # their path identity is fenced here. + raise ProviderIdentityRefusal( + "{} resource ID differs from the recorded assignment".format(kind) + ) + # Task commands and staging request/result blobs are per-execution + # transport: every execute rewrites them under the same stable + # resource path and binds their content through the request and + # result digests, so only their path identity is fenced here. if ( kind not in skip_immutable - and kind not in ("staging-request", "staging-result") + and kind not in ("task-command", "staging-request", "staging-result") and current.get("immutable_id") != prior.get("immutable_id") ): - raise ProviderError("{} immutable identity differs from the recorded assignment".format(kind)) + raise ProviderIdentityRefusal( + "{} immutable identity differs from the recorded assignment".format(kind) + ) for key, value in tags.items(): if kind in ( "role-assignment", "state-container", "global-reservation", @@ -1328,7 +1340,7 @@ def recorded_exact( str(ttl.get("status", "")).lower() != "enabled" or not ttl.get("deadline") ): raise ProviderError("worker TTL schedule is disabled or has no exact deadline") - for kind in ("bootstrap-command", "task-command", "monitor-extension"): + for kind in ("bootstrap-command", "monitor-extension"): child = resources.get(kind) if child is not None and str(child.get("provisioning_state", "")).lower() != "succeeded": raise ProviderError("{} provisioning state is not succeeded".format(kind)) @@ -2557,17 +2569,150 @@ def mutate_reset(controller, action): return None +def execute_generation_line(action): + bindings = action["bindings"] + return "export FM_WORKER_ASSIGNMENT_GENERATION='{}' FM_WORKER_ACCOUNT_BINDING='{}'".format( + bindings["assignment_generation"], bindings["account_binding"] + ) + + +def build_execute_script(action): + request = action["request"] + request_json = json.dumps(request, sort_keys=True, separators=(",", ":")) + bindings = action["bindings"] + return """set -eu +umask 077 +install -d -m 0700 /var/lib/firstmate-worker +cat > /var/lib/firstmate-worker/request.json <<'JSON' +{request} +JSON +export FM_WORKER_HOME_BINDING='{home}' FM_WORKER_TASK='{task}' FM_WORKER_TASK_GENERATION='{task_generation}' +{generation_line} +export FM_WORKER_WORKTREE_BINDING='{worktree}' FM_WORKER_REPOSITORY_BINDING='{repository}' +export FM_WORKER_REPOSITORY_GENERATION='{repository_generation}' FM_WORKER_CLOUD_INSTANCE_ID='{cloud}' +export FM_WORKER_WORKTREE=/mnt/task FM_WORKER_ACCOUNT_HOME=/mnt/account +/usr/local/libexec/fm-worker-supervisor execute --request /var/lib/firstmate-worker/request.json --result /var/lib/firstmate-worker/result.json +printf 'FM-WORKER-RESULT:%s\\n' "$(cat /var/lib/firstmate-worker/result.json)" +""".format( + request=request_json, home=bindings["home_binding"], task=bindings["task"], + task_generation=bindings["task_generation"], generation_line=execute_generation_line(action), + worktree=bindings["worktree_binding"], + repository=bindings["repository_binding"], repository_generation=bindings["repository_generation"], + cloud=action["cloud_instance_id"], + ) + + +def execute_terminal_disposition(controller, action, task_command_resource): + if not isinstance(task_command_resource, dict) or not task_command_resource.get("id"): + return EXECUTE_DISPOSITION_SUBMIT, None + live = show_full(controller, task_command_resource["id"]) + properties = live.get("properties") or {} + source = properties.get("source") or live.get("source") or {} + stored_script = source.get("script") if isinstance(source, dict) else None + if not isinstance(stored_script, str) or not stored_script.strip(): + raise ProviderError("existing worker task Run Command source script is unreadable") + expected_script = build_execute_script(action) + request_digest = action.get("request_digest") + exact_script = stored_script == expected_script + fallback_bound = ( + isinstance(request_digest, str) + and re.fullmatch(r"[0-9a-f]{64}", request_digest) + and request_digest in stored_script + and execute_generation_line(action) in stored_script.splitlines() + ) + if not exact_script and not fallback_bound: + return EXECUTE_DISPOSITION_SUBMIT, None + provisioning_state = properties.get("provisioningState") or live.get("provisioningState") + if str(provisioning_state).lower() in ("failed", "canceled"): + return EXECUTE_DISPOSITION_TERMINAL, { + "schema": EXECUTION_TERMINAL_SCHEMA, + "request_digest": request_digest, + "idempotency_key": action.get("idempotency_key"), + "disposition": "provider-terminal", + "provisioning_state": provisioning_state, + "task_command_id": task_command_resource["id"], + } + if str(provisioning_state).lower() != "succeeded": + raise ProviderError( + "exact worker execution remains bound and nonterminal: state={}".format( + provisioning_state + ) + ) + names = expected_names(controller, action["slot"]) + view = run_command_instance_view(controller, names["vm"], names["task-command"]) + if view.get("executionState") != "Succeeded": + raise ProviderError( + "exact worker execution has no recoverable terminal result: state={}".format( + view.get("executionState") + ) + ) + execution = marker_payload( + "{}\n{}".format(view.get("output", ""), view.get("error", "")), + "FM-WORKER-RESULT:", + ) + if not isinstance(execution, dict) or execution.get("request_digest") != request_digest: + raise ProviderError("exact worker execution has no request-bound result marker") + supplied = execution.get("result_digest") + unsigned = dict(execution) + unsigned.pop("result_digest", None) + if supplied != hashlib.sha256(canonical_bytes(unsigned)).hexdigest(): + raise ProviderError("recovered private worker result digest is not exact") + return EXECUTE_DISPOSITION_RECOVERED, execution + + +def persist_execute_result(controller, action, names, tags, execution): + request = action["request"] + if request.get("outcome_expected") and execution.get("outcome_present"): + outcome_target = action.get("outcome_dir") + if not outcome_target: + raise ProviderError("execution collected an outcome with no controller directory to land it in") + # The guest records where it actually put the bytes. Anything but the + # staging blob means the upload was diverted (a test sink, an injected + # unprotected FM_WORKER_OUTCOME_FILE), and the result must not be + # treated as a collectable outcome. + if execution.get("outcome_sink", "") != "blob": + raise ProviderError( + "execution claims an outcome written to {!r} rather than the staging blob".format( + execution.get("outcome_sink") + ) + ) + digest_claim = execution.get("outcome_sha256") + bytes_claim = execution.get("outcome_bytes") + if not isinstance(digest_claim, str) or not re.fullmatch(r"[0-9a-f]{64}", digest_claim): + raise ProviderError("execution outcome digest is malformed") + if not isinstance(bytes_claim, int) or isinstance(bytes_claim, bool) or not 0 < bytes_claim <= MAX_OUTCOME_BYTES: + raise ProviderError("execution outcome size is malformed or unbounded") + download_outcome_bundle( + controller, os.environ.get("FM_AZURE_STORAGE_NAME", ""), names["state-container"], + outcome_blob_name(request["request_digest"]), digest_claim, bytes_claim, + Path(outcome_target) / "outcome.bundle", + ) + upload_json_blob( + controller, os.environ.get("FM_AZURE_STORAGE_NAME", ""), names["state-container"], + names["staging-result"], execution, tags, overwrite=True, + ) + + def mutate_execute(controller, action): snapshot = inventory(controller, include_metrics=False) worker = worker_by_slot(snapshot, action["slot"]) resources = recorded_exact(action, worker) - if "deallocated" in str(resources["vm"].get("power_state", "")).lower(): - raise ProviderError("execute refuses deallocated worker compute") request = action.get("request") if not isinstance(request, dict) or request.get("request_digest") != action.get("request_digest"): raise ProviderError("execution request identity is not exact") names = expected_names(controller, action["slot"]) tags = action_tags(controller, action) + disposition, recovered = execute_terminal_disposition( + controller, action, resources.get("task-command") + ) + if disposition in (EXECUTE_DISPOSITION_TERMINAL, EXECUTE_DISPOSITION_RECOVERED): + if disposition == EXECUTE_DISPOSITION_RECOVERED: + persist_execute_result(controller, action, names, tags, recovered) + return worker, recovered + if disposition != EXECUTE_DISPOSITION_SUBMIT: + raise ProviderError("worker execution disposition is unsupported") + if "deallocated" in str(resources["vm"].get("power_state", "")).lower(): + raise ProviderError("execute refuses deallocated worker compute") upload_json_blob( controller, os.environ.get("FM_AZURE_STORAGE_NAME", ""), names["state-container"], names["staging-request"], request, tags, overwrite=True, @@ -2620,28 +2765,7 @@ def mutate_execute(controller, action): if not re.fullmatch(r"https://[A-Za-z0-9.:/_?&=%+-]+", outcome_sas): raise ProviderError("outcome staging SAS carries unsupported characters") protected_parameters.append("FM_WORKER_OUTCOME_URL=" + outcome_sas) - request_json = json.dumps(request, sort_keys=True, separators=(",", ":")) - bindings = action["bindings"] - script = """set -eu -umask 077 -install -d -m 0700 /var/lib/firstmate-worker -cat > /var/lib/firstmate-worker/request.json <<'JSON' -{request} -JSON -export FM_WORKER_HOME_BINDING='{home}' FM_WORKER_TASK='{task}' FM_WORKER_TASK_GENERATION='{task_generation}' -export FM_WORKER_ASSIGNMENT_GENERATION='{assignment}' FM_WORKER_ACCOUNT_BINDING='{account}' -export FM_WORKER_WORKTREE_BINDING='{worktree}' FM_WORKER_REPOSITORY_BINDING='{repository}' -export FM_WORKER_REPOSITORY_GENERATION='{repository_generation}' FM_WORKER_CLOUD_INSTANCE_ID='{cloud}' -export FM_WORKER_WORKTREE=/mnt/task FM_WORKER_ACCOUNT_HOME=/mnt/account -/usr/local/libexec/fm-worker-supervisor execute --request /var/lib/firstmate-worker/request.json --result /var/lib/firstmate-worker/result.json -printf 'FM-WORKER-RESULT:%s\\n' "$(cat /var/lib/firstmate-worker/result.json)" -""".format( - request=request_json, home=bindings["home_binding"], task=bindings["task"], - task_generation=bindings["task_generation"], assignment=bindings["assignment_generation"], - account=bindings["account_binding"], worktree=bindings["worktree_binding"], - repository=bindings["repository_binding"], repository_generation=bindings["repository_generation"], - cloud=action["cloud_instance_id"], - ) + script = build_execute_script(action) update_command = [ "vm", "run-command", "update", "--resource-group", controller["resource_group"], "--vm-name", names["vm"], "--name", names["task-command"], @@ -2681,35 +2805,7 @@ def mutate_execute(controller, action): unsigned.pop("result_digest", None) if supplied != hashlib.sha256(canonical_bytes(unsigned)).hexdigest(): raise ProviderError("private worker result digest is not exact") - if request.get("outcome_expected") and execution.get("outcome_present"): - outcome_target = action.get("outcome_dir") - if not outcome_target: - raise ProviderError("execution collected an outcome with no controller directory to land it in") - # The guest records where it actually put the bytes. Anything but the - # staging blob means the upload was diverted (a test sink, an injected - # unprotected FM_WORKER_OUTCOME_FILE), and the result must not be - # treated as a collectable outcome. - if execution.get("outcome_sink", "") != "blob": - raise ProviderError( - "execution claims an outcome written to {!r} rather than the staging blob".format( - execution.get("outcome_sink") - ) - ) - digest_claim = execution.get("outcome_sha256") - bytes_claim = execution.get("outcome_bytes") - if not isinstance(digest_claim, str) or not re.fullmatch(r"[0-9a-f]{64}", digest_claim): - raise ProviderError("execution outcome digest is malformed") - if not isinstance(bytes_claim, int) or isinstance(bytes_claim, bool) or not 0 < bytes_claim <= MAX_OUTCOME_BYTES: - raise ProviderError("execution outcome size is malformed or unbounded") - download_outcome_bundle( - controller, os.environ.get("FM_AZURE_STORAGE_NAME", ""), names["state-container"], - outcome_blob_name(request["request_digest"]), digest_claim, bytes_claim, - Path(outcome_target) / "outcome.bundle", - ) - upload_json_blob( - controller, os.environ.get("FM_AZURE_STORAGE_NAME", ""), names["state-container"], - names["staging-result"], execution, tags, overwrite=True, - ) + persist_execute_result(controller, action, names, tags, execution) return worker_by_slot(inventory(controller, include_metrics=False), action["slot"]), execution @@ -2858,6 +2954,9 @@ def main(): if __name__ == "__main__": try: main() + except ProviderIdentityRefusal as exc: + print("AZURE WORKER PROVIDER REFUSED-IDENTITY: {}".format(exc), file=sys.stderr) + raise SystemExit(3) except ProviderError as exc: print("AZURE WORKER PROVIDER REFUSED: {}".format(exc), file=sys.stderr) raise SystemExit(2) diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index 46e0ce59b5d..e1c5763d1d5 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -48,6 +48,7 @@ REQUEST_SCHEMA = "fm.worker-request/v1" EXECUTION_SCHEMA = "fm.worker-execution/v1" EXECUTION_RESULT_SCHEMA = "fm.worker-execution-result/v1" +EXECUTION_TERMINAL_SCHEMA = "fm.worker-execution-terminal/v1" RELEASE_SCHEMA = "fm.worker-release/v2" AUTHORITY_SCHEMA = "fm.worker-authority/v1" CAPACITY_RESERVATION_SCHEMA = "fm.capacity-reservation/v1" @@ -173,6 +174,14 @@ class LifecycleError(RuntimeError): pass +class ProviderIdentityRefused(LifecycleError): + pass + + +class ProviderResultIdentityRefused(LifecycleError): + pass + + def canonical_bytes(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") @@ -996,6 +1005,10 @@ def _provider_call_raw(env, operation, action=None): raise LifecycleError("provider response exceeded its bounded output allowance") if result.returncode != 0: detail = result.stderr.decode("utf-8", errors="replace").strip()[-1000:] + if result.returncode == 3 and detail.startswith( + "AZURE WORKER PROVIDER REFUSED-IDENTITY:" + ): + raise ProviderIdentityRefused(detail) raise LifecycleError("provider {} failed{}".format(operation, ": " + detail if detail else "")) try: response = json.loads(result.stdout.decode("utf-8")) @@ -1159,13 +1172,13 @@ def resources_exact(worker, cloud, allow_missing_compute=False): continue missing.append(kind) continue - # Staging request/result blobs are per-execution transport: every - # execute rewrites them and binds their content through the request - # and result digests, so their blob identity legitimately changes - # while every other kind stays immutable for the worker's lifetime. + # Task commands and staging request/result blobs are per-execution + # transport: every execute rewrites them and binds their content + # through the request and result digests, so their transport identity + # legitimately changes while every other kind stays immutable. if ( prior is not None - and kind not in ("staging-request", "staging-result") + and kind not in ("task-command", "staging-request", "staging-result") and resource_identity(current) != resource_identity(prior) ): return False, "{} immutable identity changed".format(kind) @@ -2037,6 +2050,29 @@ def apply_action_result(env, state, action, result): if worker is None: raise LifecycleError("execute result has no durable worker owner") execution = result.get("execution") + if isinstance(execution, dict) and execution.get("schema") == EXECUTION_TERMINAL_SCHEMA: + expected_task_command_id = ( + ((action.get("resources") or {}).get("task-command") or {}).get("id") + ) + if ( + execution.get("request_digest") != action.get("request_digest") + or execution.get("idempotency_key") != action.get("idempotency_key") + or execution.get("disposition") != "provider-terminal" + or str(execution.get("provisioning_state", "")).lower() + not in ("failed", "canceled") + or not isinstance(execution.get("task_command_id"), str) + or not execution["task_command_id"] + ): + raise LifecycleError("provider terminal execution disposition is not exact") + if execution["task_command_id"] != expected_task_command_id: + raise ProviderResultIdentityRefused( + "provider terminal execution task-command identity differs from the claimed action" + ) + raise LifecycleError( + "provider-terminal {}: exact execution is {} and cannot be applied".format( + execution["request_digest"], execution["provisioning_state"] + ) + ) if not isinstance(execution, dict) or execution.get("schema") != EXECUTION_RESULT_SCHEMA: raise LifecycleError("provider execution result schema is not supported") if execution.get("request_digest") != action.get("request_digest"): @@ -4203,9 +4239,13 @@ def command_abandon_claim(env, args): 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. + that succeeds ends the claim normally. A script-bound terminal execution + whose apply refuses is recorded with its result digest before clearing, + but only when it names the task-command resource in the claimed action; a + foreign terminal resource identity retains the claim. + An exact-key ProviderIdentityRefused replay is recorded verbatim before + clearing because that recorded resource identity can never bind again. + Every other provider failure leaves the claim untouched. """ if not args.confirm_abandon: raise LifecycleError("--confirm-abandon is required") @@ -4232,24 +4272,49 @@ def command_abandon_claim(env, args): 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: - 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) + identity_refusal = None + try: + result = provider_mutate(env, action, lease) + except ProviderIdentityRefused as exc: + identity_refusal = exc + if identity_refusal is not None: + with controller_lock(env): + 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: {}".format(identity_refusal) + )) + clean["pending_actions"].pop(slot, None) + save_state(env, clean) + else: + with controller_lock(env): + try: + apply_pending(env, action, result) + print("claim applied cleanly; nothing was abandoned") + return + except ProviderResultIdentityRefused: + raise + 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") diff --git a/docs/azure-workers.md b/docs/azure-workers.md index 0271726c992..6d472ff7459 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -325,7 +325,10 @@ bin/fm-worker-lifecycle.sh reconcile \ ``` 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. +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. +`abandon-claim` can record and clear either an exact-key provider result whose apply deterministically refuses, including a script-bound Failed or Canceled execution disposition that names the claimed task-command resource, or an exact-key `REFUSED-IDENTITY` replay whose recorded resource identity can never bind again. +Both dispositions are recorded in `cleanup_refusals` before the claim is cleared, while an ordinary transient provider failure retains the claim unchanged. +An existing Run Command whose source script is missing, non-string, empty, or whitespace-only cannot prove that a first submission is safe and retains the claim. An exact-bound Run Command that is still Updating or Running, reports Succeeded without a digest-valid request-bound result marker, or returns a terminal disposition naming a foreign task-command resource likewise retains the claim without submitting the command again. 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/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index b4f1a56271f..b7110d0a0a8 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -61,6 +61,7 @@ for marker in ( "pending_actions", "LEGACY_PENDING_SENTINEL", "superseded-by-pending-actions", "revision moved from", "FencedState", "slot_lease", "LOCK_NB", "provider_mutate", "drain_pending", "claim_pending", "apply_pending", "command_abandon_claim", + "ProviderIdentityRefused", "fm.worker-execution-terminal/v1", "capacity-reserve", "capacity-reserve-shape", "capacity-release", "merged_specialized_reservations", "command_withdraw", "command_surrender", "WORKER AUTHORITY REFUSED", "--confirm-discard-unlanded", @@ -91,6 +92,8 @@ for marker in ( "/dev/disk/azure/scsi1/lun", "/dev/disk/azure/data/by-lun/", '"bootstrap-command"', '"task-command"', '"ttl-schedule"', '"global-reservation"', '"staging-request"', '"staging-result"', + "ProviderIdentityRefusal", "fm.worker-execution-terminal/v1", + "build_execute_script", "REFUSED-IDENTITY", "worker NIC has a public IP relation", "VM cloud identity set is not exactly one slot identity", ): assert marker in azure, marker @@ -231,13 +234,23 @@ foreign["resources"]["task-disk"]["immutable_id"] = "foreign" assert module.classify_worker(worker, foreign)[0] == "retained-for-investigation" assert module.classify_worker(worker, None)[0] == "retained-for-investigation" -# Staging blobs are per-execution transport: their identity changes after an -# execute and must not wedge classification, while every other kind still -# fences on identity (the task-disk case above). +# Run commands and staging blobs are per-execution transport: their identities +# change after an execute and must not wedge classification, while every other +# kind still fences on identity (the task-disk case above). executed = copy.deepcopy(cloud) +executed["resources"]["task-command"]["immutable_id"] = "post-execute-state" +executed["resources"]["task-command"]["provisioning_state"] = "Failed" executed["resources"]["staging-request"]["immutable_id"] = "post-execute-etag" executed["resources"]["staging-result"]["immutable_id"] = "post-execute-etag-2" assert module.classify_worker(worker, executed)[0] == "assigned" +for kind in module.REQUIRED_RESOURCE_KINDS: + changed = copy.deepcopy(cloud) + changed["resources"][kind]["immutable_id"] = "foreign-" + kind + classification = module.classify_worker(worker, changed)[0] + if kind in ("task-command", "staging-request", "staging-result"): + assert classification == "assigned", (kind, classification) + else: + assert classification == "retained-for-investigation", (kind, classification) released_executed = copy.deepcopy(released) released_executed["phase"] = "assigned" executed_dark = copy.deepcopy(executed) @@ -410,6 +423,7 @@ import copy import hashlib import importlib.util import json +import subprocess import sys spec = importlib.util.spec_from_file_location("azure_provider", sys.argv[1]) @@ -454,6 +468,16 @@ action["resources"] = { } worker = {"slot": 1, "resources": resources} module.recorded_exact(action, worker) +# Ordinary provider failures stay on exit 2. If every ProviderError were +# mislabeled as a permanent identity refusal, abandon-claim could clear a +# transiently failed claim. +plain_refusal = subprocess.run( + [sys.executable, sys.argv[1]], input=b"{}\n", + stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert plain_refusal.returncode == 2, plain_refusal +assert b"AZURE WORKER PROVIDER REFUSED:" in plain_refusal.stderr, plain_refusal.stderr +assert b"REFUSED-IDENTITY" not in plain_refusal.stderr, plain_refusal.stderr for child in ("monitor-extension", "bootstrap-command", "task-command", "ttl-schedule"): changed = copy.deepcopy(worker) changed["resources"][child]["attached_to"] = "/foreign-vm" @@ -461,6 +485,7 @@ for child in ("monitor-extension", "bootstrap-command", "task-command", "ttl-sch module.recorded_exact(action, changed) except module.ProviderError as exc: assert "exact worker VM" in str(exc) + assert not isinstance(exc, module.ProviderIdentityRefusal), exc else: raise AssertionError("foreign {} target accepted".format(child)) changed = copy.deepcopy(worker) @@ -482,33 +507,187 @@ else: for kind in module.REQUIRED_RESOURCE_KINDS: changed = copy.deepcopy(worker) changed["resources"][kind]["immutable_id"] = "foreign" - if kind in ("staging-request", "staging-result"): - # Transport blobs rewrite on every execute; only their path is fenced. + if kind in ("task-command", "staging-request", "staging-result"): + # Execute transport rewrites on every run; only its path is fenced. module.recorded_exact(action, changed) moved = copy.deepcopy(worker) moved["resources"][kind]["id"] = "/slot/1/elsewhere" try: module.recorded_exact(action, moved) - except module.ProviderError: + except module.ProviderIdentityRefusal: pass else: - raise AssertionError("relocated {} blob path accepted".format(kind)) + raise AssertionError("relocated {} transport path accepted".format(kind)) continue try: module.recorded_exact(action, changed) - except module.ProviderError: - pass + except module.ProviderError as exc: + assert isinstance(exc, module.ProviderIdentityRefusal), exc else: raise AssertionError("foreign {} immutable identity accepted".format(kind)) +changed = copy.deepcopy(worker) +changed["resources"]["task-command"]["immutable_id"] = "post-execute-state" +changed["resources"]["task-command"]["provisioning_state"] = "Failed" +module.recorded_exact(action, changed) +for kind in ("bootstrap-command", "monitor-extension"): + changed = copy.deepcopy(worker) + changed["resources"][kind]["provisioning_state"] = "Failed" + try: + module.recorded_exact(action, changed) + except module.ProviderError as exc: + assert "provisioning state" in str(exc), exc + else: + raise AssertionError("failed {} provisioning state accepted".format(kind)) for key in tags: changed = copy.deepcopy(worker) changed["resources"]["task-disk"]["tags"][key] = "foreign" try: module.recorded_exact(action, changed) + except module.ProviderError as exc: + assert not isinstance(exc, module.ProviderIdentityRefusal), exc + else: + raise AssertionError("foreign task-disk tag accepted: {}".format(key)) + +# A terminal probe is bound to the exact execute script (or its exact request +# digest plus assignment-generation line), recovers a completed real result, +# and never converts an unrelated stale Failed Run Command into finality. +execute_action = copy.deepcopy(action) +execute_action["type"] = "execute" +execute_action["request_digest"] = "5" * 64 +execute_action["idempotency_key"] = "6" * 64 +execute_action["request"] = dict(execute_action["bindings"], **{ + "schema": "fm.worker-execution/v1", + "cloud_instance_id": execute_action["cloud_instance_id"], + "argv": ["/usr/bin/true"], "wall_seconds": 60, + "request_digest": execute_action["request_digest"], +}) +script = module.build_execute_script(execute_action) +task_command = worker["resources"]["task-command"] +real_show_full = module.show_full +real_run_command_instance_view = module.run_command_instance_view +module.show_full = lambda *_args, **_kwargs: { + "properties": {"source": {"script": script}, "provisioningState": "Failed"} +} +terminal_kind, terminal = module.execute_terminal_disposition( + controller, execute_action, task_command +) +assert terminal_kind == module.EXECUTE_DISPOSITION_TERMINAL, terminal_kind +assert terminal == { + "schema": "fm.worker-execution-terminal/v1", + "request_digest": execute_action["request_digest"], + "idempotency_key": execute_action["idempotency_key"], + "disposition": "provider-terminal", + "provisioning_state": "Failed", + "task_command_id": task_command["id"], +}, terminal +module.show_full = lambda *_args, **_kwargs: { + "properties": {"source": {"script": "# normalized by Azure\n" + script}, + "provisioningState": "Canceled"} +} +fallback_kind, fallback_terminal = module.execute_terminal_disposition( + controller, execute_action, task_command +) +assert fallback_kind == module.EXECUTE_DISPOSITION_TERMINAL, fallback_kind +assert fallback_terminal["disposition"] == "provider-terminal", fallback_terminal +stale_script = script.replace(execute_action["request_digest"], "9" * 64) +module.show_full = lambda *_args, **_kwargs: { + "properties": {"source": {"script": stale_script}, "provisioningState": "Failed"} +} +assert module.execute_terminal_disposition(controller, execute_action, task_command) == ( + module.EXECUTE_DISPOSITION_SUBMIT, None +) +module.show_full = lambda *_args, **_kwargs: { + "properties": {"source": {"script": "echo unrelated"}, "provisioningState": "Failed"} +} +assert module.execute_terminal_disposition(controller, execute_action, task_command) == ( + module.EXECUTE_DISPOSITION_SUBMIT, None +) +execution = { + "schema": "fm.worker-execution-result/v1", + "request_digest": execute_action["request_digest"], + "task": "task", "task_generation": "task-gen", + "assignment_generation": "asg-00000001", "cloud_instance_id": "vm-instance", + "repository_binding": "4" * 64, "repository_generation": "repo-gen", + "exit_code": 0, "timed_out": False, + "stdout_sha256": "7" * 64, "stderr_sha256": "8" * 64, + "stdout_truncated": False, "stderr_truncated": False, +} +execution["result_digest"] = hashlib.sha256(module.canonical_bytes(execution)).hexdigest() +module.show_full = lambda *_args, **_kwargs: { + "properties": {"source": {"script": script}, "provisioningState": "Succeeded"} +} +module.run_command_instance_view = lambda *_args, **_kwargs: { + "executionState": "Succeeded", + "output": "FM-WORKER-RESULT:" + json.dumps(execution, sort_keys=True, separators=(",", ":")), + "error": "", +} +recovered_kind, recovered_execution = module.execute_terminal_disposition( + controller, execute_action, task_command +) +assert recovered_kind == module.EXECUTE_DISPOSITION_RECOVERED, recovered_kind +assert recovered_execution == execution, recovered_execution + +# Once the exact request owns the Run Command, a replay may only recover its +# terminal disposition or exact result. Updating/Running and a Succeeded +# command without a valid marker/digest are ordinary retryable refusals; they +# must never fall through to the update that would execute the guest twice. +real_inventory = module.inventory +real_worker_by_slot = module.worker_by_slot +real_recorded_exact = module.recorded_exact +real_upload_json_blob = module.upload_json_blob +real_az = module.az +module.inventory = lambda *_args, **_kwargs: {"workers": [worker]} +module.worker_by_slot = lambda _snapshot, _slot: worker +module.recorded_exact = lambda _action, _worker: worker["resources"] +updates = [] +def forbidden_update(*_args, **_kwargs): + updates.append("update") + raise AssertionError("exact-bound execution reached Run Command update") +module.upload_json_blob = forbidden_update +module.az = forbidden_update + +def retained_execute(properties, view=None): + module.show_full = lambda *_args, **_kwargs: {"properties": properties} + module.run_command_instance_view = lambda *_args, **_kwargs: dict(view or {}) + try: + module.mutate_execute(controller, execute_action) except module.ProviderError: pass else: - raise AssertionError("foreign task-disk tag accepted: {}".format(key)) + raise AssertionError("exact-bound incomplete execution was submitted again") + assert updates == [], updates + +retained_execute({"provisioningState": "Succeeded"}) +retained_execute({"source": {"script": None}, "provisioningState": "Succeeded"}) +retained_execute({"source": {"script": ""}, "provisioningState": "Succeeded"}) +retained_execute({"source": {"script": " \n\t"}, "provisioningState": "Succeeded"}) +retained_execute({"source": {"script": script}, "provisioningState": "Updating"}) +retained_execute( + {"source": {"script": script}, "provisioningState": "Succeeded"}, + {"executionState": "Running", "output": "", "error": ""}, +) +retained_execute( + {"source": {"script": script}, "provisioningState": "Succeeded"}, + {"executionState": "Succeeded", "output": "", "error": ""}, +) +bad_execution = dict(execution, result_digest="0" * 64) +retained_execute( + {"source": {"script": script}, "provisioningState": "Succeeded"}, + { + "executionState": "Succeeded", + "output": "FM-WORKER-RESULT:" + json.dumps( + bad_execution, sort_keys=True, separators=(",", ":") + ), + "error": "", + }, +) +module.inventory = real_inventory +module.worker_by_slot = real_worker_by_slot +module.recorded_exact = real_recorded_exact +module.upload_json_blob = real_upload_json_blob +module.az = real_az +module.show_full = real_show_full +module.run_command_instance_view = real_run_command_instance_view # Specialized validation demand and its durable reservation are exact shared # capacity inputs; active-without-reservation and foreign reservation identities fail closed. @@ -891,7 +1070,8 @@ if path.exists(): state = json.loads(path.read_text()) else: state = { - "workers": {}, "seen": {}, "calls": [], + "workers": {}, "seen": {}, "calls": [], "execute_updates": 0, + "execute_terminal_probes": 0, "metrics": {"actual_usd": 100.0, "forecast_usd": 150.0}, } @@ -979,19 +1159,87 @@ if request["operation"] == "mutate": ) ) raise SystemExit(1) + slot = str(action["slot"]) + kind = action["type"] + if kind == "execute" and os.environ.get("FIXTURE_TRANSIENT_EXECUTE_REFUSAL"): + sys.stderr.write("FIXTURE PROVIDER REFUSED: transient execute refusal\n") + raise SystemExit(2) + live_worker = state["workers"].get(slot) + if live_worker is not None: + live_resources = live_worker.get("resources", {}) + for resource_kind, recorded in (action.get("resources") or {}).items(): + current = live_resources.get(resource_kind) + if not isinstance(current, dict) or not isinstance(recorded, dict): + continue + identity_changed = ( + resource_kind not in ("task-command", "staging-request", "staging-result") + and current.get("immutable_id") != recorded.get("immutable_id") + ) + if current.get("id") != recorded.get("id") or identity_changed: + sys.stderr.write( + "AZURE WORKER PROVIDER REFUSED-IDENTITY: {} identity differs from " + "the recorded assignment\n".format(resource_kind) + ) + raise SystemExit(3) + task_command = (live_worker or {}).get("resources", {}).get("task-command", {}) + if kind == "execute" and os.environ.get("FIXTURE_UNREADABLE_EXECUTE_SCRIPT"): + state["execute_terminal_probes"] = state.get("execute_terminal_probes", 0) + 1 + save() + sys.stderr.write( + "FIXTURE PROVIDER REFUSED: existing worker task Run Command source " + "script is unreadable\n" + ) + raise SystemExit(2) + bound_state = os.environ.get("FIXTURE_BOUND_EXECUTE_STATE") + if kind == "execute" and bound_state: + task_command["provisioning_state"] = bound_state + task_command["request_digest"] = action["request_digest"] + state["execute_terminal_probes"] = state.get("execute_terminal_probes", 0) + 1 + save() + sys.stderr.write( + "FIXTURE PROVIDER REFUSED: exact worker execution remains bound and " + "nonterminal: state={}\n".format(bound_state) + ) + raise SystemExit(2) state["calls"].append({ - "type": action["type"], "slot": action["slot"], "key": key, + "type": kind, "slot": action["slot"], "key": key, "outcome_expected": bool((action.get("request") or {}).get("outcome_expected")), "outcome_dir": action.get("outcome_dir"), # Additive: the digest-bound staged manifest the provider actually # receives, so a caller can assert what a lane staged. "payload_files": (action.get("request") or {}).get("payload_files"), }) - if key in state["seen"]: + if kind == "execute" and os.environ.get("FIXTURE_TERMINAL_EXECUTE"): + task_command["immutable_id"] = "task-command-terminal-{}".format(key[:8]) + task_command["provisioning_state"] = "Failed" + task_command["request_digest"] = action["request_digest"] + task_state = str(task_command.get("provisioning_state", "")).lower() + if ( + kind == "execute" + and task_state in ("failed", "canceled") + and task_command.get("request_digest") == action.get("request_digest") + ): + result = { + "idempotency_key": key, + "action": kind, + "worker": live_worker, + "execution": { + "schema": "fm.worker-execution-terminal/v1", + "request_digest": action["request_digest"], + "idempotency_key": key, + "disposition": "provider-terminal", + "provisioning_state": task_command["provisioning_state"], + "task_command_id": ( + "/fixture/foreign/task-command" + if os.environ.get("FIXTURE_FOREIGN_TERMINAL_ID") + else task_command["id"] + ), + }, + } + state["seen"][key] = result + elif key in state["seen"]: result = state["seen"][key] else: - slot = str(action["slot"]) - kind = action["type"] if kind == "create": assert slot not in state["workers"] worker = complete_worker(action) @@ -1024,7 +1272,12 @@ if request["operation"] == "mutate": state["workers"].pop(slot) result = {"idempotency_key": key, "action": kind} elif kind == "execute": + state["execute_updates"] = state.get("execute_updates", 0) + 1 request_value = action["request"] + task_command = state["workers"][slot]["resources"]["task-command"] + task_command["immutable_id"] = "task-command-execute-{}".format(key[:8]) + task_command["provisioning_state"] = "Succeeded" + task_command["request_digest"] = action["request_digest"] execution = { "schema": "fm.worker-execution-result/v1", "request_digest": action["request_digest"], @@ -1379,6 +1632,21 @@ refused_abandon = run("abandon-claim", "--slot", skew_slot, "--idempotency-key", "--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 +# THE negative safety pin: an ordinary transient provider failure is not the +# structured permanent identity refusal, so abandonment must fail and retain +# both the claim and the absence of any abandonment record. +before_refusals = list(skew_state["cleanup_refusals"]) +transient_env = dict(env, FIXTURE_TRANSIENT_EXECUTE_REFUSAL="1") +transient_abandon = subprocess.run( + [wrapper, "abandon-claim", "--slot", skew_slot, + "--idempotency-key", skew_claim["idempotency_key"], "--confirm-abandon", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]], + env=transient_env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert transient_abandon.returncode != 0, transient_abandon +after_transient = controller_state() +assert after_transient["pending_actions"][skew_slot]["idempotency_key"] == skew_claim["idempotency_key"] +assert after_transient["cleanup_refusals"] == before_refusals, after_transient["cleanup_refusals"] 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() @@ -1398,6 +1666,153 @@ armed_action = [ assert armed_action["outcome_expected"] is True, armed_action assert armed_action["outcome_dir"] == str(outcome_dir.resolve()), armed_action +# An existing task-command whose live source script is unreadable cannot prove +# that it is safe to submit, and an exact request still Updating is retained as +# an ordinary provider failure. Neither the initial execute nor abandon-claim +# may reach the fixture's Run Command update; removing the injected hostile +# state lets the same durable claim make its one first submission and apply. +before_bound = fixture_state() +bound_updates = before_bound.get("execute_updates", 0) +bound_probes = before_bound.get("execute_terminal_probes", 0) +bound_execute = armed_execute( + "--payload-dir", str(payload_dir), "--account-dir", str(account_dir), + "--outcome-dir", str(outcome_dir), + overrides={"FIXTURE_UNREADABLE_EXECUTE_SCRIPT": "1"}, + command="/usr/bin/false", +) +assert bound_execute.returncode != 0 and "source script is unreadable" in bound_execute.stderr, ( + bound_execute.stderr +) +bound_state = controller_state() +bound_slot = str(bound_state["queue"]["task-2@gen-2"]["slot"]) +bound_claim = bound_state["pending_actions"][bound_slot] +after_bound_execute = fixture_state() +assert after_bound_execute.get("execute_updates", 0) == bound_updates, after_bound_execute +assert after_bound_execute.get("execute_terminal_probes", 0) == bound_probes + 1, after_bound_execute +bound_abandon = subprocess.run( + [wrapper, "abandon-claim", "--slot", bound_slot, + "--idempotency-key", bound_claim["idempotency_key"], "--confirm-abandon", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]], + env=dict(env, FIXTURE_BOUND_EXECUTE_STATE="Running"), text=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert bound_abandon.returncode != 0 and "remains bound" in bound_abandon.stderr, bound_abandon.stderr +after_bound_abandon = controller_state() +assert after_bound_abandon["pending_actions"][bound_slot]["idempotency_key"] == bound_claim["idempotency_key"] +bound_fixture = fixture_state() +assert bound_fixture.get("execute_updates", 0) == bound_updates, bound_fixture +assert bound_fixture.get("execute_terminal_probes", 0) == bound_probes + 2, bound_fixture +run("abandon-claim", "--slot", bound_slot, + "--idempotency-key", bound_claim["idempotency_key"], "--confirm-abandon", + "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) +assert bound_slot not in controller_state()["pending_actions"] +assert fixture_state().get("execute_updates", 0) == bound_updates + 1 + +# Reproduce the Azure wedge: the exact execute claim owns a Run Command whose +# per-execution identity and provisioning state moved to Failed. Reconcile +# records the terminal disposition but retains the claim; abandon replays that +# same terminal result, records before clearing, and ordinary release still +# traverses deallocate -> delete-compute -> reset cleanly. +task3 = controller_state()["queue"]["task-3@gen-3"] +task3_worker = controller_state()["workers"][str(task3["slot"])] +terminal_env = dict(env, FIXTURE_TERMINAL_EXECUTE="1") +terminal_execute = subprocess.run( + [wrapper, "execute", "--task", "task-3", "--task-generation", "gen-3", + "--assignment-generation", task3_worker["assignment_generation"], + "--wall-seconds", "60", "--confirm-execute", "--confirm-subscription", + env["FM_AZURE_SUBSCRIPTION_ID"], "--", "/usr/bin/true"], + env=terminal_env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert terminal_execute.returncode != 0 and "provider-terminal" in terminal_execute.stderr, terminal_execute.stderr +terminal_state = controller_state() +terminal_slot = str(task3["slot"]) +terminal_claim = terminal_state["pending_actions"][terminal_slot] +reconciled = json.loads(run( + "reconcile", "--apply", "--json", "--confirm-subscription", + env["FM_AZURE_SUBSCRIPTION_ID"], +).stdout) +assert any(item["type"] == "replay-refused" for item in reconciled["actions"]), reconciled +retained = controller_state() +assert terminal_slot in retained["pending_actions"], retained["pending_actions"] +assert any("provider-terminal" in str(entry.get("note", "")) + for entry in retained["cleanup_refusals"]), retained["cleanup_refusals"] +# The terminal result must identify the exact task-command path stored in the +# claimed action. A foreign path is an untrusted provider result: ordinary +# apply refuses it, and even explicit abandonment retains the claim. +foreign_terminal_env = dict( + terminal_env, FIXTURE_FOREIGN_TERMINAL_ID="1" +) +foreign_abandon = subprocess.run( + [wrapper, "abandon-claim", "--slot", terminal_slot, + "--idempotency-key", terminal_claim["idempotency_key"], + "--confirm-abandon", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]], + env=foreign_terminal_env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert foreign_abandon.returncode != 0 and "task-command identity differs" in foreign_abandon.stderr, ( + foreign_abandon.stderr +) +after_foreign_terminal = controller_state() +assert after_foreign_terminal["pending_actions"][terminal_slot]["idempotency_key"] == ( + terminal_claim["idempotency_key"] +), after_foreign_terminal["pending_actions"] +run("abandon-claim", "--slot", terminal_slot, + "--idempotency-key", terminal_claim["idempotency_key"], + "--confirm-abandon", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) +cleared = controller_state() +assert terminal_slot not in cleared["pending_actions"], cleared["pending_actions"] +assert any("claim abandoned by operator" in str(entry.get("note", "")) + and "provider-terminal" in str(entry.get("note", "")) + for entry in cleared["cleanup_refusals"]), cleared["cleanup_refusals"] +status = json.loads(run("status", "--live", "--json").stdout) +assert status["classification_counts"]["assigned"] == 4, status["classification_counts"] +release(3) +run("reconcile", "--apply", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) +released_state = controller_state() +released_fixture = fixture_state() +assert terminal_slot not in released_state["workers"], released_state["workers"] +assert terminal_slot not in released_fixture["workers"], released_fixture["workers"] +assert [entry["type"] for entry in released_fixture["calls"]][-3:] == [ + "deallocate", "delete-compute", "reset", +], released_fixture["calls"][-3:] + +# The other abandon-only disposition is an exact provider identity refusal. +# A transient/plain ProviderError was retained above; exit 3 plus the explicit +# marker alone permits the operator to record the refusal before clearing. +task4 = controller_state()["queue"]["task-4@gen-4"] +task4_slot = str(task4["slot"]) +task4_worker = controller_state()["workers"][task4_slot] +identity_wedge = subprocess.run( + [wrapper, "execute", "--task", "task-4", "--task-generation", "gen-4", + "--assignment-generation", task4_worker["assignment_generation"], + "--wall-seconds", "60", "--payload-dir", str(payload_dir), + "--account-dir", str(account_dir), "--outcome-dir", str(outcome_dir), + "--confirm-execute", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"], + "--", "/usr/bin/true"], + env=dict(env, FIXTURE_OMIT_OUTCOME="1"), text=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, +) +assert identity_wedge.returncode != 0 and "no outcome disposition" in identity_wedge.stderr +identity_state = controller_state() +identity_claim = identity_state["pending_actions"][task4_slot] +identity_fixture = fixture_state() +identity_fixture["workers"][task4_slot]["resources"]["task-disk"]["immutable_id"] = "foreign-task-disk" +Path(fixture_path).write_text(json.dumps(identity_fixture, sort_keys=True, separators=(",", ":")) + "\n") +run("abandon-claim", "--slot", task4_slot, + "--idempotency-key", identity_claim["idempotency_key"], + "--confirm-abandon", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) +identity_cleared = controller_state() +assert task4_slot not in identity_cleared["pending_actions"], identity_cleared["pending_actions"] +assert any("claim abandoned by operator: AZURE WORKER PROVIDER REFUSED-IDENTITY:" in + str(entry.get("note", "")) for entry in identity_cleared["cleanup_refusals"]), ( + identity_cleared["cleanup_refusals"]) +# Identity abandonment clears only the impossible claim. It never adopts or +# clears the foreign resource itself; repair the fixture before the ordinary +# release coverage later in this same scenario. +identity_fixture = fixture_state() +identity_fixture["workers"][task4_slot]["resources"]["task-disk"]["immutable_id"] = ( + identity_cleared["workers"][task4_slot]["resources"]["task-disk"]["immutable_id"]) +Path(fixture_path).write_text(json.dumps(identity_fixture, sort_keys=True, separators=(",", ":")) + "\n") + # A waiting task can use the slot only after deallocate, disposable deletion, # complete reset, and a new assignment generation. release(1) @@ -1415,7 +1830,7 @@ sequence = actions[-4:] assert sequence == ["deallocate", "delete-compute", "reset", "create"], sequence # Drain every active task and prove queue/compute/disposable capacity reach zero. -for number in (2, 3, 4, 5): +for number in (2, 4, 5): release(number) run("reconcile", "--apply", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"]) state = controller_state()