fix(azure): recover no-mistakes worker completion

Author: ruby-dlee

bin/fm-azure-worker-provider.py

@@ -3109,10 +3109,29 @@ def execute_terminal_disposition(controller, action, resources):
         )
     names = expected_names(controller, action["slot"])
     view = view or run_command_instance_view(controller, names["vm"], names["task-command"])
-    if view.get("executionState") != "Succeeded":
+    execution_state = view.get("executionState")
+    if str(execution_state).lower() in ("failed", "canceled"):
+        exit_code = view.get("exitCode")
+        if isinstance(exit_code, bool) or not isinstance(exit_code, int):
+            raise ProviderError(
+                "exact terminal worker execution has no integer exit code: state={}".format(
+                    execution_state
+                )
+            )
+        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,
+            "execution_state": execution_state,
+            "exit_code": exit_code,
+            "task_command_id": task_command_resource["id"],
+        }
+    if execution_state != "Succeeded":
         raise ProviderError(
             "exact worker execution has no recoverable terminal result: state={}".format(
-                view.get("executionState")
+                execution_state
             )
         )
     execution = exact_execution_marker(view)

bin/fm-worker-lifecycle.py

@@ -2384,12 +2384,20 @@ def apply_action_result(env, state, action, result):
             expected_task_command_id = (
                 ((action.get("resources") or {}).get("task-command") or {}).get("id")
             )
+            provisioning_state = str(execution.get("provisioning_state", "")).lower()
+            execution_state = str(execution.get("execution_state", "")).lower()
+            provisioning_terminal = provisioning_state in ("failed", "canceled")
+            execution_terminal = (
+                provisioning_state == "succeeded"
+                and execution_state in ("failed", "canceled")
+                and isinstance(execution.get("exit_code"), int)
+                and not isinstance(execution.get("exit_code"), bool)
+            )
             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 (provisioning_terminal or execution_terminal)
                 or not isinstance(execution.get("task_command_id"), str)
                 or not execution["task_command_id"]
             ):
@@ -2398,9 +2406,10 @@ def apply_action_result(env, state, action, result):
                 raise ProviderResultIdentityRefused(
                     "provider terminal execution task-command identity differs from the claimed action"
                 )
+            terminal_state = execution_state if execution_terminal else provisioning_state
             raise LifecycleError(
                 "provider-terminal {}: exact execution is {} and cannot be applied".format(
-                    execution["request_digest"], execution["provisioning_state"]
+                    execution["request_digest"], terminal_state
                 )
             )
         if not isinstance(execution, dict) or execution.get("schema") != EXECUTION_RESULT_SCHEMA:
@@ -4523,16 +4532,53 @@ def command_service_complete(env, args):
             require_id("task generation", args.task_generation),
         )
         item = state["queue"].get(key)
-        worker = state["workers"].get(str((item or {}).get("slot")))
-        if item is None or item.get("status") != "assigned" or worker is None:
-            raise LifecycleError("service completion requires one exact assigned worker")
-        if item.get("role") != "no-mistakes" or worker.get("role") != "no-mistakes":
+        if item is None or item.get("role") != "no-mistakes":
             raise LifecycleError("service completion is owned by no-mistakes workers only")
-        if worker.get("assignment_generation") != args.assignment_generation:
-            raise LifecycleError("service completion assignment generation is not exact")
         execution = state["executions"].get(request_digest)
         if not isinstance(execution, dict):
             raise LifecycleError("service completion has no exact recorded execution")
+        receipt = item.get("service_completion_receipt")
+        if receipt is not None:
+            if item.get("status") not in ("releasing", "complete"):
+                raise LifecycleError("service completion receipt exists in an invalid queue state")
+            if not isinstance(receipt, dict):
+                raise LifecycleError("service completion receipt is malformed")
+            unsigned_receipt = dict(receipt)
+            receipt_digest = unsigned_receipt.pop("proof_digest", None)
+            if (
+                receipt.get("schema") != "fm.worker-service-release/v1"
+                or receipt.get("task") != args.task
+                or receipt.get("task_generation") != args.task_generation
+                or receipt.get("assignment_generation") != args.assignment_generation
+                or receipt.get("request_digest") != request_digest
+                or receipt.get("result_digest") != execution.get("result_digest")
+                or execution.get("request_digest") != request_digest
+                or execution.get("assignment_generation") != args.assignment_generation
+                or receipt.get("verdict") != "proved"
+                or receipt_digest != digest_value(unsigned_receipt)
+            ):
+                raise LifecycleError("service completion receipt identity differs")
+            worker = state["workers"].get(str(item.get("slot")))
+            worker_owns_item = (
+                worker is not None and worker.get("queue_key") == key
+            )
+            if item.get("status") == "releasing" and not worker_owns_item:
+                raise LifecycleError("releasing service completion lost its exact worker")
+            if worker_owns_item and (
+                worker.get("role") != "no-mistakes"
+                or worker.get("assignment_generation") != args.assignment_generation
+                or worker.get("release_proof") != receipt
+            ):
+                raise LifecycleError("service completion worker receipt identity differs")
+            print("service release proof already recorded with exact identity")
+            return
+        worker = state["workers"].get(str(item.get("slot")))
+        if item.get("status") != "assigned" or worker is None:
+            raise LifecycleError("service completion requires one exact assigned worker")
+        if worker.get("role") != "no-mistakes":
+            raise LifecycleError("service completion is owned by no-mistakes workers only")
+        if worker.get("assignment_generation") != args.assignment_generation:
+            raise LifecycleError("service completion assignment generation is not exact")
         if (
             execution.get("request_digest") != request_digest
             or execution.get("result_digest") != worker.get("last_execution_digest")
@@ -4556,9 +4602,10 @@ def command_service_complete(env, args):
                 raise LifecycleError("worker already has a different service release proof")
             print("service release proof already recorded with exact identity")
             return
-        worker["release_proof"] = proof
+        worker["release_proof"] = copy.deepcopy(proof)
         worker["released_at"] = iso_utc()
         worker["phase"] = "release-proved"
+        item["service_completion_receipt"] = copy.deepcopy(proof)
         item["status"] = "releasing"
         save_state(env, state)
     print("service execution proved; exact idle capacity is eligible for cleanup")

bin/fm-worker-lifecycle.sh

@@ -35,6 +35,7 @@
 #   fm-worker-lifecycle.sh capacity-release <exact fence and cleanup receipt>
 #   fm-worker-lifecycle.sh execute <exact assignment flags> [--existing-task-disk --return-kind <ship|scout> --outcome-dir <dir>] -- <argv...>
 #   fm-worker-lifecycle.sh authority-receipt <exact assignment flags> --output <json>
+#   fm-worker-lifecycle.sh service-complete <exact no-mistakes execution binding>
 #   fm-worker-lifecycle.sh proof-template --task <id> --task-generation <id>
 #   fm-worker-lifecycle.sh release --task <id> --task-generation <id> --proof-file <json>
 #   fm-worker-lifecycle.sh withdraw --task <id> --task-generation <id> --confirm-withdraw --confirm-subscription <uuid>
@@ -138,7 +139,7 @@ fm_worker_receipt_credential_remains() {  # <task home file> <task id> <controll
 }
 
 case "${1:-}" in
-  request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|capacity-retire-fence|abandon-claim|message-put|message-collect|compartment-chain-tip)
+  request|release|resume|steer|execute|authority-receipt|service-complete|capacity-reserve|capacity-reserve-shape|capacity-release|capacity-retire-fence|abandon-claim|message-put|message-collect|compartment-chain-tip)
     fm_refuse_if_gate_agent
     exec python3 "$SCRIPT_DIR/fm-worker-lifecycle.py" "$@"
     ;;

bin/fm-worker-supervisor.py

@@ -29,6 +29,7 @@
 MAX_RESULT_BYTES = 8 * 1024 * 1024
 MAX_WALL_SECONDS = 6 * 60 * 60
 MAX_OUTPUT_BYTES = 4 * 1024 * 1024
+MAX_NO_MISTAKES_RUNTIME_FILES = 20000
 
 
 class SupervisorError(RuntimeError):
@@ -386,7 +387,7 @@ def stage_no_mistakes_runtime(source, target, enforce_linux=True):
     try:
         with tarfile.open(source, mode="r:gz") as archive:
             members = archive.getmembers()
-            if not members or len(members) > 4096:
+            if not members or len(members) > MAX_NO_MISTAKES_RUNTIME_FILES:
                 raise SupervisorError("no-mistakes runtime member inventory is unbounded")
             for member in members:
                 parts = Path(member.name).parts

tests/fm-no-mistakes-runtime.test.sh

@@ -74,6 +74,8 @@ def load(name, path):
 
 builder = load("runtime_builder", root / "bin/fm-no-mistakes-runtime.py")
 supervisor = load("worker_supervisor", root / "bin/fm-worker-supervisor.py")
+assert supervisor.MAX_NO_MISTAKES_RUNTIME_FILES == builder.MAX_FILES, (
+    supervisor.MAX_NO_MISTAKES_RUNTIME_FILES, builder.MAX_FILES)
 bundle = temporary / "runtime.tar.gz"
 args = argparse.Namespace(
     no_mistakes=str(temporary / "no-mistakes"), node=str(temporary / "node"),

tests/fm-worker-lifecycle.test.sh

@@ -16,6 +16,149 @@ AUTHORITY="$ROOT/bin/fm-worker-authority.py"
 DOC="$ROOT/docs/azure-workers.md"
 SUB=11111111-1111-4111-8111-111111111111
 
+service_complete_front_door() {
+  local tmp help
+  fm_test_tmproot_into tmp fm-worker-service-complete-front-door
+  mkdir -p "$tmp/home"
+  help=$(FM_HOME="$tmp/home" "$WRAPPER" service-complete --help) \
+    || fail "supported lifecycle wrapper rejected service-complete"
+  case "$help" in
+    *--request-digest*--confirm-subscription*) ;;
+    *) fail "service-complete help lost its exact execution binding" ;;
+  esac
+  pass "supported lifecycle wrapper exposes service-complete"
+}
+
+service_complete_replay_contract() {
+  python3 - "$CONTROLLER" <<'PY' || fail "service completion replay contract failed"
+import contextlib
+import importlib.util
+from types import SimpleNamespace
+import sys
+
+spec = importlib.util.spec_from_file_location("lifecycle", sys.argv[1])
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+bindings = {
+    "home_binding": "1" * 64,
+    "task": "service-task",
+    "task_generation": "service-generation",
+    "assignment_generation": "asg-00000001",
+    "account_binding": "2" * 64,
+    "worktree_binding": "3" * 64,
+    "repository_binding": "4" * 64,
+    "repository_generation": "repository-generation",
+}
+request_digest = "5" * 64
+result_digest = "6" * 64
+item = {
+    **bindings,
+    "role": "no-mistakes",
+    "status": "assigned",
+    "slot": 1,
+}
+worker = {
+    "role": "no-mistakes",
+    "queue_key": "service-task@service-generation",
+    "assignment_generation": bindings["assignment_generation"],
+    "bindings": bindings,
+    "cloud_instance_id": "worker-instance",
+    "resources": {"vm": {"id": "/exact/vm"}},
+    "last_execution_digest": result_digest,
+    "release_proof": None,
+}
+execution = {
+    "request_digest": request_digest,
+    "result_digest": result_digest,
+    "assignment_generation": bindings["assignment_generation"],
+}
+state = {
+    "queue": {"service-task@service-generation": item},
+    "workers": {"1": worker},
+    "executions": {request_digest: execution},
+}
+module.controller_lock = lambda _env: contextlib.nullcontext()
+module.load_state = lambda _env: state
+module.save_state = lambda _env, _state: None
+args = SimpleNamespace(
+    task="service-task",
+    task_generation="service-generation",
+    assignment_generation=bindings["assignment_generation"],
+    request_digest=request_digest,
+    confirm_subscription="subscription",
+)
+env = {"subscription": "subscription"}
+
+# Crash window one: the first call durably moved the item to releasing, but
+# the caller died before observing success. The exact retry is idempotent.
+module.command_service_complete(env, args)
+assert item["status"] == "releasing", item
+assert item["service_completion_receipt"] == worker["release_proof"], item
+module.command_service_complete(env, args)
+
+# Crash window two: reconcile completed the reset and removed the worker, but
+# the caller died before publishing the cached result. The queue-owned exact
+# receipt survives reset and admits only the same bound completion request.
+item["status"] = "complete"
+# The released slot may already belong to a later task. Its presence cannot
+# invalidate the old queue item's exact, self-digested completion receipt.
+state["workers"] = {"1": {
+    "queue_key": "later-task@later-generation",
+    "role": "author",
+    "assignment_generation": "asg-00000002",
+    "release_proof": None,
+}}
+module.command_service_complete(env, args)
+wrong = SimpleNamespace(**vars(args))
+wrong.assignment_generation = "asg-99999999"
+try:
+    module.command_service_complete(env, wrong)
+except module.LifecycleError as exc:
+    assert "identity differs" in str(exc), exc
+else:
+    raise AssertionError("completed service receipt admitted a foreign assignment")
+
+# The provider-side regression below emits this exact terminal shape. Prove
+# the lifecycle consumer recognizes it as terminal (rather than malformed),
+# fails closed, and leaves the durable execute claim available for explicit
+# abandonment/recovery.
+terminal_action = {
+    "type": "execute",
+    "slot": 1,
+    "request_digest": "7" * 64,
+    "idempotency_key": "8" * 64,
+    "resources": {"task-command": {"id": "/exact/task-command"}},
+}
+terminal_state = {
+    "queue": {"terminal-task@terminal-generation": {"status": "assigned"}},
+    "workers": {"1": {"queue_key": "terminal-task@terminal-generation"}},
+    "executions": {},
+    "pending_actions": {"1": terminal_action},
+    "completed_worker_seconds": 0.0,
+}
+state = terminal_state
+terminal_result = {"execution": {
+    "schema": "fm.worker-execution-terminal/v1",
+    "request_digest": terminal_action["request_digest"],
+    "idempotency_key": terminal_action["idempotency_key"],
+    "disposition": "provider-terminal",
+    "provisioning_state": "Succeeded",
+    "execution_state": "Failed",
+    "exit_code": 2,
+    "task_command_id": "/exact/task-command",
+}}
+try:
+    module.apply_pending(env, terminal_action, terminal_result)
+except module.LifecycleError as exc:
+    assert "provider-terminal" in str(exc) and "failed" in str(exc), exc
+else:
+    raise AssertionError("failed guest execution was applied as a successful result")
+assert terminal_state["pending_actions"]["1"] == terminal_action, terminal_state
+PY
+  pass "service completion replays across releasing and completed crash windows"
+}
+
 static_contract() {
   python3 - "$CONTROLLER" "$AZURE" "$SUPERVISOR" "$AUTHORITY" "$DOC" <<'PY' || fail "elastic worker static contract failed"
 from pathlib import Path
@@ -859,6 +1002,23 @@ recovered_kind, recovered_execution = module.execute_terminal_disposition(
 )
 assert recovered_kind == module.EXECUTE_DISPOSITION_RECOVERED, recovered_kind
 assert recovered_execution == execution, recovered_execution
+module.run_command_instance_view = lambda *_args, **_kwargs: {
+    "executionState": "Failed", "exitCode": 2, "output": "", "error": "guest failed",
+}
+failed_kind, failed_execution = module.execute_terminal_disposition(
+    controller, execute_action, worker["resources"]
+)
+assert failed_kind == module.EXECUTE_DISPOSITION_TERMINAL, failed_kind
+assert failed_execution == {
+    "schema": "fm.worker-execution-terminal/v1",
+    "request_digest": execute_action["request_digest"],
+    "idempotency_key": execute_action["idempotency_key"],
+    "disposition": "provider-terminal",
+    "provisioning_state": "Succeeded",
+    "execution_state": "Failed",
+    "exit_code": 2,
+    "task_command_id": task_command["id"],
+}, failed_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
@@ -8692,6 +8852,8 @@ PY
 }
 
 static_contract
+service_complete_front_door
+service_complete_replay_contract
 compartment_payload_contract
 classification_and_admission_matrix
 azure_provider_refusal_matrix