diff --git a/bin/fm-azure-validation.py b/bin/fm-azure-validation.py index d50ceba2cd7..dc89b6bcc2e 100755 --- a/bin/fm-azure-validation.py +++ b/bin/fm-azure-validation.py @@ -39,6 +39,7 @@ CREDENTIAL_EXPIRY = ROOT / "bin" / "fm-credential-expiry.py" CONTAINER = "validation-shards" SCHEMA = "fm.azure-validation/v1" +PURGE_SCHEMA = "fm.azure-validation-purge/v1" RESULT_SCHEMA = "fm.azure-validation-result/v1" CREDENTIALS_SCHEMA = "fm.azure-validation-credentials/v1" RUNTIME_SCHEMA = "fm.azure-validation-runtime/v1" @@ -240,8 +241,12 @@ def canonical_bytes(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") +def sha256_hex(value): + return hashlib.sha256(value).hexdigest() + + def sha256_bytes(value): - return "sha256:" + hashlib.sha256(value).hexdigest() + return "sha256:" + sha256_hex(value) def sha256_file(path): @@ -2111,6 +2116,21 @@ def runner_module(): return _RUNNER_MODULE +_WORKER_LIFECYCLE_MODULE = None + + +def worker_lifecycle_module(): + global _WORKER_LIFECYCLE_MODULE + if _WORKER_LIFECYCLE_MODULE is None: + spec = importlib.util.spec_from_file_location( + "worker_lifecycle_module", str(ROOT / "bin" / "fm-worker-lifecycle.py") + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + _WORKER_LIFECYCLE_MODULE = module + return _WORKER_LIFECYCLE_MODULE + + _CREDENTIAL_EXPIRY_MODULE = None @@ -2407,6 +2427,23 @@ def release_shape_constituent(env, state, reservation_id, evidence): )) +def retire_purge_capacity_fence(env, retirement): + arguments = [ + "capacity-retire-fence", + "--fence-binding", retirement["fence_binding"], + "--retirement-receipt", retirement["retirement_receipt"], + "--confirm-subscription", env["subscription"], + ] + for reservation_id in retirement["reservation_ids"]: + arguments += ["--reservation-id", reservation_id] + result = lifecycle_command(env, arguments) + if result.returncode != 0: + detail = (result.stderr or result.stdout or "").strip() + raise ValidationError("shared capacity fence retirement refused: {}".format( + detail[-500:] + )) + + RETAIL_RATE_CACHE_FRESH_SECONDS = 7 * 24 * 3600 @@ -3997,6 +4034,883 @@ def fail_retain(env, args): print("AZURE VALIDATION RETAINED cell={} compute=zero worktree=retained".format(state["cell"])) +def purge_role_identity(item): + identity = { + "id": str(item.get("id", "")), + "scope": str(item.get("scope", "")), + "principal_id": str(item.get("principalId", "")), + "role_definition_id": str(item.get("roleDefinitionId", "")), + } + if any(not value for value in identity.values()): + raise ValidationError("purge RBAC identity is incomplete") + return identity + + +def same_purge_role(left, right): + return all( + str(left.get(key, "")).lower() == str(right.get(key, "")).lower() + for key in ("id", "scope", "principal_id", "role_definition_id") + ) + + +def purge_compute_ids(resources): + required = ( + ("vm", resources.get("vm_id")), + ("nic", resources.get("nic_id")), + ("disk", resources.get("os_disk_id")), + ("ttl-schedule", resources.get("ttl_schedule_id")), + ) + if any(not resource_id for _, resource_id in required): + raise ValidationError("purge compute identity inventory is incomplete") + values = list(required) + safety = resources.get("safety_run_command_id") + if safety: + values.append(("run-command", safety)) + elif resources.get("vm_id"): + values.append(("run-command", resources["vm_id"] + "/runCommands/safety-shutdown")) + for record in resources.get("run_commands") or []: + if not isinstance(record, dict) or not record.get("id"): + raise ValidationError("purge run-command inventory is incomplete") + values.append(("run-command", record["id"])) + managed = resources.get("run_command_name") + if managed: + values.append(("run-command", resources["vm_id"] + "/runCommands/" + managed)) + return sorted(set(values)) + + +def prove_compute_zero(env, compute_ids, label): + for kind, resource_id in compute_ids: + exists, _ = read_resource(env, resource_id, kind) + if exists: + raise ValidationError("{} still has live {} compute".format(label, kind)) + + +def shared_capacity_authority(env): + configured = Path(os.environ.get( + "FM_AZURE_VALIDATION_LIFECYCLE", str(ROOT / "bin" / "fm-worker-lifecycle.sh") + )).resolve() + if configured != (ROOT / "bin" / "fm-worker-lifecycle.sh").resolve(): + raise ValidationError("purge cannot inspect an overridden shared capacity authority") + module = worker_lifecycle_module() + try: + lifecycle_env = module.environment() + if lifecycle_env["subscription"] != env["subscription"]: + raise ValidationError("shared capacity authority subscription differs during purge") + return module, lifecycle_env + except module.LifecycleError as exc: + raise ValidationError("shared capacity authority is unreadable during purge: {}".format(exc)) + + +def capacity_authority_snapshot(env): + module, lifecycle_env = shared_capacity_authority(env) + try: + with module.controller_lock(lifecycle_env): + lifecycle_state = module.load_state(lifecycle_env) + reservations = lifecycle_state.get("capacity_reservations") or {} + reservations = json.loads(json.dumps(reservations)) + response = module.provider_call(lifecycle_env, "inventory") + provider_reservations = response["inventory"]["capacity_reservations"] + return reservations, json.loads(json.dumps(provider_reservations)) + except module.LifecycleError as exc: + raise ValidationError("shared capacity authority is unreadable during purge: {}".format(exc)) + + +def require_complete_purge_capacity_census( + reservations, provider_reservations, fence, allowed_ids, boundary +): + provider_active = { + item.get("reservation_id"): item + for item in provider_reservations + if isinstance(item, dict) and item.get("active") is True + } + for reservation_id, provider in provider_active.items(): + controller = reservations.get(reservation_id) + if ( + not isinstance(controller, dict) + or controller.get("schema") != "fm.capacity-reservation/v1" + or controller.get("reservation_id") != reservation_id + or not isinstance(controller.get("fence_binding"), str) + or not controller.get("fence_binding") + or not isinstance(controller.get("workload_role"), str) + or not controller.get("workload_role") + or controller.get("discretionary") is not True + or controller.get("role") != provider.get("role") + or controller.get("sku") != provider.get("sku") + or str(controller.get("sku_family", "")).lower() + != str(provider.get("sku_family", "")).lower() + or controller.get("vcpus") != provider.get("vcpus") + or isinstance(controller.get("amount_usd"), bool) + or not isinstance(controller.get("amount_usd"), (int, float)) + or abs( + float(controller["amount_usd"]) + - float(provider.get("amount_usd", -1.0)) + ) > 1e-6 + ): + raise ValidationError( + "provider-active capacity constituent {} lacks exact controller " + "identity at the {} boundary".format(reservation_id, boundary) + ) + for reservation_key, reservation in reservations.items(): + if ( + isinstance(reservation, dict) + and reservation.get("fence_binding") == fence + and ( + reservation.get("reservation_id") != reservation_key + or reservation_key not in allowed_ids + ) + ): + raise ValidationError( + "same-fence capacity constituent {} is outside the exact " + "purge census at the {} boundary".format(reservation_key, boundary) + ) + + +def exact_purge_capacity_constituents( + state, shard_plan, runner_states, reservations, provider_reservations +): + admission = state["admission"] + allocation = state.get("allocation") or {} + expected = [{ + "reservation_id": state["cell"], + "sku": allocation.get("sku"), + "sku_family": allocation.get("sku_family"), + "vcpus": 8, + "amount_usd": admission.get("control_amount_usd"), + }] + [{ + "reservation_id": entry.get("invocation"), + "sku": entry.get("sku"), + "sku_family": entry.get("sku_family"), + "vcpus": 4, + "amount_usd": entry.get("amount_usd"), + } for entry in shard_plan] + fence = state["request"]["fence"].split(":", 1)[-1] + census = {item["reservation_id"] for item in expected} | set(runner_states) + require_complete_purge_capacity_census( + reservations, provider_reservations, fence, census, "initial-plan" + ) + snapshots = {} + for item in expected: + reservation = reservations.get(item["reservation_id"]) + if ( + not isinstance(reservation, dict) + or reservation.get("schema") != "fm.capacity-reservation/v1" + or reservation.get("reservation_id") != item["reservation_id"] + or reservation.get("fence_binding") != fence + or reservation.get("shape_id") != state["cell"] + or reservation.get("role") != "specialized" + or reservation.get("workload_role") != "validation" + or reservation.get("discretionary") is not True + or reservation.get("sku") != item["sku"] + or str(reservation.get("sku_family", "")).lower() != str(item["sku_family"] or "").lower() + or reservation.get("vcpus") != item["vcpus"] + or not isinstance(item["amount_usd"], (int, float)) + or isinstance(item["amount_usd"], bool) + or abs(float(reservation.get("amount_usd", -1.0)) - float(item["amount_usd"])) > 1e-6 + or reservation.get("status") not in ("queued", "reserved", "released") + or ( + reservation.get("status") == "released" + and not re.match( + r"^(?:sha256:)?[0-9a-f]{64}$", + str(reservation.get("cleanup_receipt", "")), + ) + ) + ): + raise ValidationError( + "shared capacity constituent {} has no exact durable shape identity".format( + item["reservation_id"] + ) + ) + snapshots[item["reservation_id"]] = { + key: reservation.get(key) for key in ( + "schema", "reservation_id", "fence_binding", "shape_id", "role", + "workload_role", "discretionary", "sku", "sku_family", "vcpus", "amount_usd", + "status", "cleanup_receipt", + ) + } + return snapshots + + +def bind_purge_provider_capacity_absence(capacity, provider_reservations): + by_id = { + item["reservation_id"]: item + for item in provider_reservations + if isinstance(item, dict) and isinstance(item.get("reservation_id"), str) + } + for reservation_id, expected in capacity.items(): + observed = by_id.get(reservation_id) + if observed is None: + expected["provider"] = {"present": False, "active": False} + continue + if ( + observed.get("reservation_id") != reservation_id + or observed.get("role") != expected.get("role") + or observed.get("sku") != expected.get("sku") + or str(observed.get("sku_family", "")).lower() + != str(expected.get("sku_family", "")).lower() + or observed.get("vcpus") != expected.get("vcpus") + or isinstance(observed.get("amount_usd"), bool) + or not isinstance(observed.get("amount_usd"), (int, float)) + or abs( + float(observed["amount_usd"]) + - float(expected.get("amount_usd", -1.0)) + ) > 1e-6 + or observed.get("active") is not False + ): + raise ValidationError( + "provider capacity constituent {} is active or has drifted from " + "the exact purge lineage".format(reservation_id) + ) + expected["provider"] = { + key: observed.get(key) for key in ( + "reservation_id", "role", "sku", "sku_family", "vcpus", + "amount_usd", "active", + ) + } + + +def prove_purge_capacity_released( + env, capacity_constituents, release_ids, require_released=True +): + reservations, provider_reservations = capacity_authority_snapshot(env) + expected = { + item["reservation_id"]: json.loads(json.dumps(item)) + for item in capacity_constituents + } + fences = {item.get("fence_binding") for item in expected.values()} + if len(fences) != 1 or not next(iter(fences)): + raise ValidationError("sealed purge capacity fence is incomplete") + require_complete_purge_capacity_census( + reservations, provider_reservations, next(iter(fences)), set(expected), + "retry/pre-artifact", + ) + release_ids = set(release_ids) + for reservation_id, planned in expected.items(): + observed = reservations.get(reservation_id) + receipt = str((observed or {}).get("cleanup_receipt", "")) + if ( + (reservation_id in release_ids and not isinstance(observed, dict)) + or ( + isinstance(observed, dict) + and any( + str(observed.get(key) or "").lower() + != str(planned.get(key) or "").lower() + for key in ( + "schema", "reservation_id", "fence_binding", "shape_id", "role", + "workload_role", "sku", "sku_family", + ) + ) + ) + or ( + isinstance(observed, dict) + and ( + observed.get("discretionary") is not True + or observed.get("vcpus") != planned.get("vcpus") + or isinstance(observed.get("amount_usd"), bool) + or not isinstance(observed.get("amount_usd"), (int, float)) + or abs( + float(observed["amount_usd"]) + - float(planned.get("amount_usd", -1.0)) + ) > 1e-6 + ) + ) + or ( + isinstance(observed, dict) + and reservation_id in release_ids + and require_released + and ( + observed.get("status") != "released" + or not re.match(r"^(?:sha256:)?[0-9a-f]{64}$", receipt) + ) + ) + or ( + isinstance(observed, dict) + and reservation_id in release_ids + and not require_released + and ( + observed.get("status") not in ("queued", "reserved", "released") + or ( + observed.get("status") == "released" + and not re.match(r"^(?:sha256:)?[0-9a-f]{64}$", receipt) + ) + ) + ) + or ( + isinstance(observed, dict) + and reservation_id not in release_ids + and ( + observed.get("status") != "released" + or not re.match(r"^(?:sha256:)?[0-9a-f]{64}$", receipt) + ) + ) + ): + raise ValidationError( + "purge capacity constituent {} lacks its exact durable {} identity".format( + reservation_id, + "release" if require_released or reservation_id not in release_ids + else "pre-release", + ) + ) + bind_purge_provider_capacity_absence(expected, provider_reservations) + + +def exact_purge_retry_constituent(state, value, reservations): + invocation = value["invocation"] + request = value.get("request") or {} + limits = request.get("limits") or {} + runner_reservation = value.get("shared_capacity_reservation") or {} + reservation = reservations.get(invocation) + fence = state["request"]["fence"].split(":", 1)[-1] + if ( + not isinstance(reservation, dict) + or reservation.get("schema") != "fm.capacity-reservation/v1" + or reservation.get("reservation_id") != invocation + or reservation.get("fence_binding") != fence + or reservation.get("shape_id") is not None + or reservation.get("role") != "specialized" + or reservation.get("workload_role") != "validation" + or reservation.get("discretionary") is not True + or reservation.get("sku") != limits.get("sku") + or str(reservation.get("sku_family", "")).lower() != str(limits.get("sku_family", "")).lower() + or reservation.get("vcpus") != 4 + or not isinstance(runner_reservation.get("amount_usd"), (int, float)) + or isinstance(runner_reservation.get("amount_usd"), bool) + or abs( + float(reservation.get("amount_usd", -1.0)) + - float(runner_reservation["amount_usd"]) + ) > 1e-6 + or reservation.get("status") != "released" + or reservation.get("cleanup_receipt") != runner_reservation.get("cleanup_receipt") + ): + raise ValidationError( + "retry shard constituent {} has no exact durable released identity".format(invocation) + ) + return { + key: reservation.get(key) for key in ( + "schema", "reservation_id", "fence_binding", "shape_id", "role", + "workload_role", "discretionary", "sku", "sku_family", "vcpus", "amount_usd", + "status", "cleanup_receipt", + ) + } + + +def load_purge_runner_states(env, state): + directory = runner_state_dir(env) + if not directory.is_dir(): + raise ValidationError("shard runner state directory is absent") + values = {} + for path in sorted(directory.glob("azr-*.json")): + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValidationError("shard runner state is unreadable during purge: {}".format(exc)) + invocation = value.get("invocation") + if value.get("schema") != "fm.azure-command/v1" or not RUNNER_INVOCATION.match(str(invocation or "")): + raise ValidationError("shard runner state identity is corrupt during purge") + if path.stem != invocation or invocation in values: + raise ValidationError("shard runner state filename or invocation is ambiguous") + values[invocation] = value + return { + invocation: value + for invocation, value in values.items() + if (value.get("request") or {}).get("capacity_parent") == state["cell"] + } + + +def purge_runner_compute_ids(value): + resources = value.get("resources") or {} + values = purge_compute_ids(resources) + vm_id = resources.get("vm_id") + execute_name = resources.get("run_command_name") or "execute" + safety_name = resources.get("safety_run_command_name") or "safety-shutdown" + values.extend(( + ("run-command", vm_id + "/runCommands/" + execute_name), + ("run-command", vm_id + "/runCommands/" + safety_name), + )) + return sorted(set(values)) + + +def plan_purge_shards(env, state): + admission = state.get("admission") or {} + shard_plan = admission.get("shard_plan") or [] + roots = [entry.get("invocation") for entry in shard_plan] + expected_shards = (state.get("request") or {}).get("limits", {}).get("behavior_shards") + if ( + admission.get("shape_id") != state["cell"] + or not isinstance(expected_shards, int) + or len(shard_plan) != expected_shards + or {entry.get("shard") for entry in shard_plan} != set(range(1, expected_shards + 1)) + or any(not RUNNER_INVOCATION.match(str(root or "")) for root in roots) + or len(roots) != len(set(roots)) + ): + raise ValidationError("purge shard capacity plan is incomplete or ambiguous") + runner_states = load_purge_runner_states(env, state) + reservations, provider_reservations = capacity_authority_snapshot(env) + capacity = exact_purge_capacity_constituents( + state, shard_plan, runner_states, reservations, provider_reservations + ) + fence = state["request"]["fence"].split(":", 1)[-1] + planned = [] + for invocation, value in sorted(runner_states.items()): + request = value.get("request") or {} + root = request.get("lineage_root_invocation") or invocation + parent = value.get("parent_invocation") + reservation = value.get("shared_capacity_reservation") or {} + if invocation not in capacity: + capacity[invocation] = exact_purge_retry_constituent(state, value, reservations) + if ( + root not in roots + or request.get("schema") != "fm.azure-command/v1" + or request.get("invocation") != invocation + or request.get("parent_invocation") != parent + or request.get("capacity_fence") != fence + or value.get("request_digest") != request.get("request_digest") + or not SHA256.match(str(request.get("request_digest", ""))) + or not SHA256.match(str(request.get("command_digest", ""))) + or value.get("phase") not in ("complete", "absent-fenced") + or reservation.get("reservation_id") != invocation + or reservation.get("fence_binding") != fence + or reservation.get("status") != "released" + or not re.match(r"^(?:sha256:)?[0-9a-f]{64}$", str(reservation.get("cleanup_receipt", ""))) + or capacity[invocation].get("status") != "released" + or capacity[invocation].get("cleanup_receipt") != reservation.get("cleanup_receipt") + ): + raise ValidationError("shard lineage {} is not terminal, compute-zero, and released".format(invocation)) + if parent: + parent_state = runner_states.get(parent) + if not parent_state or (parent_state.get("request") or {}).get("lineage_root_invocation", parent) != root: + raise ValidationError("shard retry lineage is incomplete during purge") + elif invocation != root: + raise ValidationError("shard lineage root identity is inconsistent") + compute_ids = purge_runner_compute_ids(value) + prove_compute_zero(env, compute_ids, "shard {}".format(invocation)) + planned.append({ + "invocation": invocation, + "lineage_root_invocation": root, + "parent_invocation": parent, + "phase": value["phase"], + "request_digest": request["request_digest"], + "command_digest": request["command_digest"], + "cleanup_receipt": reservation["cleanup_receipt"], + "compute_ids": [[kind, resource_id] for kind, resource_id in compute_ids], + }) + shard_runs = state.get("shard_runs") or {} + if not isinstance(shard_runs, dict): + raise ValidationError("cell shard dispatch ledger is corrupt") + for record in shard_runs.values(): + if not isinstance(record, dict): + raise ValidationError("cell shard dispatch record is corrupt") + live = runner_states.get(record.get("invocation")) + if not live or (live.get("request") or {}).get("command_digest") != record.get("command_digest"): + raise ValidationError("dispatched shard lineage lacks exact terminal runner evidence") + dispatched_roots = { + (value.get("request") or {}).get("lineage_root_invocation") or invocation + for invocation, value in runner_states.items() + } + releases = [] + if capacity[state["cell"]]["status"] != "released": + releases.append({"reservation_id": state["cell"], "evidence": "purge-control-compute-absent"}) + recorded_invocations = {record.get("invocation") for record in shard_runs.values()} + for root in roots: + if root not in dispatched_roots: + if root in recorded_invocations: + raise ValidationError("recorded shard dispatch has no runner lineage") + if capacity[root]["status"] != "released": + releases.append({"reservation_id": root, "evidence": "purge-shard-never-dispatched"}) + bind_purge_provider_capacity_absence(capacity, provider_reservations) + ordered_capacity = [state["cell"]] + roots + sorted(set(capacity) - {state["cell"]} - set(roots)) + return planned, releases, [capacity[key] for key in ordered_capacity] + + +def purge_container_scope(env, state): + return ( + "/subscriptions/{}/resourceGroups/{}/providers/Microsoft.Storage/storageAccounts/{}" + "/blobServices/default/containers/{}" + ).format(env["subscription"], env["resource_group"], env["storage"], state["staging"]["container"]) + + +def build_purge_plan(env, state): + control_compute = purge_compute_ids(state["resources"]) + prove_compute_zero(env, control_compute, "control cell") + recorded_worktree = (state.get("resources") or {}).get("identities", {}).get("worktree") + worktree_id = state["resources"].get("worktree_disk_id") + exists, worktree = read_resource(env, worktree_id, "disk") + if not exists: + raise ValidationError("retained worktree disk is absent before purge planning") + verify_cleanup_resource(state, worktree, "disk", "worktree") + worktree_identity = immutable_identity(worktree, "disk") + worktree_etag = worktree.get("etag") or worktree.get("properties", {}).get("etag") + if ( + not same_stable_identity(recorded_worktree, worktree_identity, "disk") + or not worktree_etag + or worktree.get("managedBy") + or worktree.get("properties", {}).get("managedBy") + ): + raise ValidationError("retained worktree disk is not exact and detached") + worktree_identity["etag"] = worktree_etag + identity_id = state["resources"].get("identity_id") + exists, identity_resource = read_resource(env, identity_id, "identity") + if not exists: + raise ValidationError("cell storage identity is absent before purge planning") + verify_cleanup_resource(state, identity_resource, "identity") + identity = immutable_identity(identity_resource, "identity") + principal = identity["principal_id"] + if principal.lower() != str(state["resources"].get("identity_principal_id", "")).lower(): + raise ValidationError("cell storage principal changed before purge planning") + scope = purge_container_scope(env, state) + exists, container = read_resource(env, scope, "container") + if not exists: + raise ValidationError("cell private container is absent before purge planning") + properties = container.get("properties", container) + container_etag = container.get("etag") or properties.get("etag") + if properties.get("publicAccess") not in (None, "None") or not container_etag: + raise ValidationError("cell container is not private with an exact ETag") + blob_role = "/subscriptions/{}/providers/Microsoft.Authorization/roleDefinitions/{}".format( + env["subscription"], BLOB_DATA_CONTRIBUTOR_ROLE + ) + assignments, _, _ = az_command(env, ["role", "assignment", "list", "--scope", scope, "--all"]) + direct = [ + purge_role_identity(item) for item in assignments or [] + if str(item.get("scope", "")).lower() == scope.lower() + ] + expected_principals = {env["operator_object_id"].lower(), principal.lower()} + if ( + len(direct) != 2 + or {item["principal_id"].lower() for item in direct} != expected_principals + or any(item["role_definition_id"].lower() != blob_role.lower() for item in direct) + ): + raise ValidationError("cell container RBAC is foreign or incomplete before purge") + account_scope = storage_account_scope(env) + file_role = "/subscriptions/{}/providers/Microsoft.Authorization/roleDefinitions/{}".format( + env["subscription"], FILE_DATA_PRIVILEGED_CONTRIBUTOR_ROLE + ) + account_assignments, _, _ = az_command( + env, ["role", "assignment", "list", "--scope", account_scope, "--all"] + ) + auth_roles = [ + purge_role_identity(item) for item in account_assignments or [] + if str(item.get("scope", "")).lower() == account_scope.lower() + and str(item.get("principalId", "")).lower() == principal.lower() + ] + expected_auth_count = 1 if auth_share_name() else 0 + if len(auth_roles) != expected_auth_count or any( + item["role_definition_id"].lower() != file_role.lower() for item in auth_roles + ): + raise ValidationError("cell auth-share RBAC is foreign or incomplete before purge") + effective, _, _ = az_command(env, [ + "role", "assignment", "list", "--assignee-object-id", principal, + "--all", "--include-inherited", "--include-groups", + ]) + expected_effective = [item for item in direct if item["principal_id"].lower() == principal.lower()] + auth_roles + effective_roles = [purge_role_identity(item) for item in effective or []] + if len(effective_roles) != len(expected_effective) or any( + not any(same_purge_role(item, expected) for expected in expected_effective) + for item in effective_roles + ): + raise ValidationError("cell identity effective RBAC exceeds the purge plan") + shard_lineages, capacity_releases, capacity_constituents = plan_purge_shards(env, state) + capacity_fences = {item.get("fence_binding") for item in capacity_constituents} + if len(capacity_fences) != 1 or not next(iter(capacity_fences)): + raise ValidationError("purge capacity fence identity is incomplete") + retirement_identity = { + "schema": "fm.azure-validation-capacity-retirement/v1", + "cell": state["cell"], + "request_digest": state["request_digest"], + "fence_binding": next(iter(capacity_fences)), + "reservation_ids": sorted( + item["reservation_id"] for item in capacity_constituents + ), + } + immutable = { + "cell": state["cell"], + "subscription": env["subscription"], + "request_digest": state["request_digest"], + "control_compute_ids": [[kind, resource_id] for kind, resource_id in control_compute], + "shard_lineages": shard_lineages, + "capacity_constituents": capacity_constituents, + "capacity_fence_retirement": dict( + retirement_identity, + # The shared allocator's binding contract is deliberately the + # narrow raw lowercase digest, unlike validation protocol digests + # that carry an explicit sha256: prefix. + retirement_receipt=sha256_hex(canonical_bytes(retirement_identity)), + ), + "worktree": {"resource_id": worktree_id, "identity": worktree_identity}, + "storage": { + "container_scope": scope, + "container_etag": container_etag, + "container_roles": sorted(direct, key=lambda item: item["id"].lower()), + "auth_share_roles": sorted(auth_roles, key=lambda item: item["id"].lower()), + "identity_resource_id": identity_id, + "identity": identity, + }, + "capacity_releases": capacity_releases, + } + return { + "schema": PURGE_SCHEMA, + "created_at": iso_utc(), + "plan": immutable, + "plan_digest": sha256_bytes(canonical_bytes(immutable)), + "progress": { + "worktree_absent": False, + "container_roles_absent": False, + "container_absent": False, + "auth_share_roles_absent": False, + "identity_absent": False, + "capacity_fence_retired": False, + "released_capacity": [], + }, + } + + +def verify_purge_record(state, env): + purge = state.get("purge") or {} + plan = purge.get("plan") or {} + if ( + purge.get("schema") != PURGE_SCHEMA + or purge.get("plan_digest") != sha256_bytes(canonical_bytes(plan)) + or plan.get("cell") != state["cell"] + or plan.get("subscription") != env["subscription"] + or plan.get("request_digest") != state.get("request_digest") + or not isinstance(purge.get("progress"), dict) + ): + raise ValidationError("stored retained-purge plan is corrupt or rebound") + if state.get("phase") == "purged": + terminal = purge.get("terminal") or {} + if ( + terminal.get("phase") != "purged" + or terminal.get("plan_digest") != purge["plan_digest"] + or not terminal.get("completed_at") + ): + raise ValidationError("terminal retained-purge tombstone is incomplete") + return purge + + +def save_purge_progress(env, state, key, value=True): + state["purge"]["progress"][key] = value + save_state(env, state) + + +def delete_planned_purge_resource(env, state, resource_id, kind, planned_identity, label): + exists, resource = read_resource(env, resource_id, kind) + if not exists: + return + verify_cleanup_resource(state, resource, kind, "worktree" if label == "worktree" else None) + live = immutable_identity(resource, kind) + if not same_stable_identity(planned_identity, live, kind): + raise ValidationError("planned {} stable identity changed".format(label)) + if planned_identity.get("etag"): + raw_etag = resource.get("etag") or resource.get("properties", {}).get("etag") + if raw_etag != planned_identity["etag"]: + raise ValidationError("planned {} ETag changed".format(label)) + if kind == "disk" and (resource.get("managedBy") or resource.get("properties", {}).get("managedBy")): + raise ValidationError("planned {} disk reattached before deletion".format(label)) + arguments = [ + "rest", "--method", "delete", + "--url", "https://management.azure.com{}?api-version={}".format(resource_id, RESOURCE_API[kind]), + ] + if planned_identity.get("etag"): + arguments += ["--headers", "If-Match={}".format(planned_identity["etag"])] + _, rc, stderr = az_command(env, arguments, check=False) + if rc != 0: + raise ValidationError("exact planned {} deletion failed: {}".format(label, stderr)) + for _ in range(60): + remains, _ = read_resource(env, resource_id, kind) + if not remains: + return + time.sleep(5) + raise ValidationError("exact planned {} remains after bounded deletion".format(label)) + + +def current_planned_roles(env, state, plan): + storage = plan["storage"] + scope = storage["container_scope"] + assignments, _, _ = az_command(env, ["role", "assignment", "list", "--scope", scope, "--all"]) + current_container = [ + purge_role_identity(item) for item in assignments or [] + if str(item.get("scope", "")).lower() == scope.lower() + ] + planned_container = storage["container_roles"] + if any(not any(same_purge_role(item, expected) for expected in planned_container) for item in current_container): + raise ValidationError("cell container gained foreign RBAC after purge planning") + account_scope = storage_account_scope(env) + account, _, _ = az_command(env, ["role", "assignment", "list", "--scope", account_scope, "--all"]) + principal = storage["identity"]["principal_id"] + current_auth = [ + purge_role_identity(item) for item in account or [] + if str(item.get("scope", "")).lower() == account_scope.lower() + and str(item.get("principalId", "")).lower() == principal.lower() + ] + planned_auth = storage["auth_share_roles"] + if any(not any(same_purge_role(item, expected) for expected in planned_auth) for item in current_auth): + raise ValidationError("cell identity gained foreign auth-share RBAC after purge planning") + effective, _, _ = az_command(env, [ + "role", "assignment", "list", "--assignee-object-id", principal, + "--all", "--include-inherited", "--include-groups", + ]) + planned_effective = [ + item for item in planned_container if item["principal_id"].lower() == principal.lower() + ] + planned_auth + current_effective = [purge_role_identity(item) for item in effective or []] + if any(not any(same_purge_role(item, expected) for expected in planned_effective) for item in current_effective): + raise ValidationError("cell identity gained foreign effective RBAC after purge planning") + return current_container, current_auth + + +def purge_storage(env, state, purge): + plan = purge["plan"] + progress = purge["progress"] + storage = plan["storage"] + container_roles, auth_roles = current_planned_roles(env, state, plan) + for item in container_roles: + _, rc, stderr = az_command(env, ["role", "assignment", "delete", "--ids", item["id"]], check=False) + if rc != 0: + raise ValidationError("planned container RBAC deletion failed: {}".format(stderr)) + for _ in range(60): + remaining, _ = current_planned_roles(env, state, plan) + if not remaining: + break + time.sleep(5) + else: + raise ValidationError("planned container RBAC remains after bounded deletion") + if not progress.get("container_roles_absent"): + save_purge_progress(env, state, "container_roles_absent") + exists, container = read_resource(env, storage["container_scope"], "container") + if exists: + properties = container.get("properties", container) + live_etag = container.get("etag") or properties.get("etag") + if live_etag != storage["container_etag"] or properties.get("publicAccess") not in (None, "None"): + raise ValidationError("planned private container changed before deletion") + _, rc, stderr = az_command(env, [ + "rest", "--method", "delete", + "--url", "https://management.azure.com{}?api-version={}".format( + storage["container_scope"], RESOURCE_API["container"] + ), + "--headers", "If-Match={}".format(storage["container_etag"]), + ], check=False) + if rc != 0: + raise ValidationError("planned private container deletion failed: {}".format(stderr)) + for _ in range(60): + remains, _ = read_resource(env, storage["container_scope"], "container") + if not remains: + break + time.sleep(5) + else: + raise ValidationError("planned private container remains after bounded deletion") + if not progress.get("container_absent"): + save_purge_progress(env, state, "container_absent") + _, auth_roles = current_planned_roles(env, state, plan) + for item in auth_roles: + _, rc, stderr = az_command(env, ["role", "assignment", "delete", "--ids", item["id"]], check=False) + if rc != 0: + raise ValidationError("planned auth-share RBAC deletion failed: {}".format(stderr)) + for _ in range(60): + _, remaining = current_planned_roles(env, state, plan) + if not remaining: + break + time.sleep(5) + else: + raise ValidationError("planned auth-share RBAC remains after bounded deletion") + if not progress.get("auth_share_roles_absent"): + save_purge_progress(env, state, "auth_share_roles_absent") + delete_planned_purge_resource( + env, state, storage["identity_resource_id"], "identity", storage["identity"], "storage identity" + ) + if not progress.get("identity_absent"): + save_purge_progress(env, state, "identity_absent") + + +def execute_purge_plan(env, state, purge): + plan = purge["plan"] + progress = purge["progress"] + prove_compute_zero(env, [(kind, resource_id) for kind, resource_id in plan["control_compute_ids"]], "control cell") + for lineage in plan["shard_lineages"]: + prove_compute_zero( + env, [(kind, resource_id) for kind, resource_id in lineage["compute_ids"]], + "shard {}".format(lineage["invocation"]), + ) + released = set(progress.get("released_capacity") or []) + planned_release_ids = [entry["reservation_id"] for entry in plan["capacity_releases"]] + prove_purge_capacity_released( + env, plan["capacity_constituents"], planned_release_ids, + require_released=False, + ) + for entry in plan["capacity_releases"]: + reservation_id = entry["reservation_id"] + if reservation_id not in released: + release_shape_constituent(env, state, reservation_id, entry["evidence"]) + released.add(reservation_id) + save_purge_progress(env, state, "released_capacity", sorted(released)) + prove_purge_capacity_released( + env, plan["capacity_constituents"], planned_release_ids + ) + # This command performs one last entire allocator/provider census while + # holding the shared admission lock, then durably retires the exact fence. + # Both reserve entry points reject the tombstone before they can insert or + # re-admit capacity, closing the former proof-to-disk-delete race. + retire_purge_capacity_fence(env, plan["capacity_fence_retirement"]) + if not progress.get("capacity_fence_retired"): + save_purge_progress(env, state, "capacity_fence_retired") + worktree = plan["worktree"] + delete_planned_purge_resource( + env, state, worktree["resource_id"], "disk", worktree["identity"], "worktree" + ) + if not progress.get("worktree_absent"): + save_purge_progress(env, state, "worktree_absent") + purge_storage(env, state, purge) + + +def purge_retained(env, args): + cell = require_cell(args.cell) + if ( + not args.confirm_purge + or args.confirm_subscription != env["subscription"] + or args.confirm_cell != cell + ): + raise ValidationError("purge-retained requires exact purge, subscription, and cell confirmation") + require_sha256("purge request digest confirmation", args.confirm_request_digest) + with lock(env, cell + "-shards"): + with lock(env, cell): + state = load_state(env, cell) + if args.confirm_request_digest != state.get("request_digest"): + raise ValidationError("purge request digest confirmation does not match the retained cell") + if (state.get("result") or {}).get("outcome") in ("passed", "checks-passed"): + raise ValidationError("a passed result cannot enter retained-failure purge") + if state["phase"] == "purged": + purge = verify_purge_record(state, env) + print("AZURE VALIDATION PURGED cell={} plan={} compute=zero retained=zero".format( + cell, purge["plan_digest"] + )) + return + if state["phase"] == "failed-retained": + purge = build_purge_plan(env, state) + transition( + env, state, "purging", + "immutable retained-resource purge plan sealed before destructive mutation", + purge=purge, + ) + elif state["phase"] == "purging": + purge = verify_purge_record(state, env) + else: + raise ValidationError("purge-retained owns only an exact failed-retained cell or its purge retry") + try: + execute_purge_plan(env, state, purge) + except ValidationError as exc: + state.setdefault("events", []).append({ + "at": iso_utc(), "phase": "purging", + "note": "retained purge remains resumable: {}".format(str(exc)[:300]), + }) + save_state(env, state) + raise + purge["terminal"] = { + "phase": "purged", "completed_at": iso_utc(), + "plan_digest": purge["plan_digest"], + } + transition(env, state, "purged", "sealed retained-resource purge completed", purge=purge) + print("AZURE VALIDATION PURGED cell={} plan={} compute=zero retained=zero".format( + cell, state["purge"]["plan_digest"] + )) + + def list_cell_blobs(env, state, prefix="shards/"): values, _, _ = az_command(env, [ "storage", "blob", "list", "--auth-mode", "login", "--account-name", env["storage"], @@ -4596,6 +5510,12 @@ def parser(): retain_parser.add_argument("--cell", required=True) retain_parser.add_argument("--confirm-retain", action="store_true") retain_parser.add_argument("--confirm-subscription") + purge_parser = commands.add_parser("purge-retained") + purge_parser.add_argument("--cell", required=True) + purge_parser.add_argument("--confirm-purge", action="store_true") + purge_parser.add_argument("--confirm-subscription") + purge_parser.add_argument("--confirm-cell", required=True) + purge_parser.add_argument("--confirm-request-digest", required=True) commands.add_parser("queue") seed_parser = commands.add_parser("auth-seed") seed_parser.add_argument("--codex") @@ -4617,7 +5537,10 @@ def main(): if args.command == "build-runtime-bundle": build_runtime_bundle(args) return 0 - cloud = args.command in ("dispatch", "drive", "observe", "collect", "respond", "replace", "close", "retain-failure") + cloud = args.command in ( + "dispatch", "drive", "observe", "collect", "respond", "replace", + "close", "retain-failure", "purge-retained", + ) # Planning a seed is a purely local credential read; only the upload # needs a cloud scope, so a plan works without Azure environment. if args.command == "auth-seed" and args.apply: @@ -4641,6 +5564,8 @@ def main(): close(env, args) elif args.command == "retain-failure": fail_retain(env, args) + elif args.command == "purge-retained": + purge_retained(env, args) elif args.command == "queue": queue(env) elif args.command == "status": diff --git a/bin/fm-azure-validation.sh b/bin/fm-azure-validation.sh index 82db0661050..dffd01f100a 100755 --- a/bin/fm-azure-validation.sh +++ b/bin/fm-azure-validation.sh @@ -48,6 +48,9 @@ # --confirm-subscription --confirm-head # fm-azure-validation.sh retain-failure --cell --confirm-retain \ # --confirm-subscription +# fm-azure-validation.sh purge-retained --cell --confirm-purge \ +# --confirm-subscription --confirm-cell \ +# --confirm-request-digest # fm-azure-validation.sh queue # fm-azure-validation.sh auth-seed [--codex ] [--claude ] # [--apply --confirm-seed --confirm-subscription ] @@ -63,14 +66,14 @@ set -euo pipefail SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P) usage() { - sed -n '2,59p' "$0" | sed 's/^# \{0,1\}//' + sed -n '2,62p' "$0" | sed 's/^# \{0,1\}//' } case "${1:-}" in help|-h|--help|"") usage ;; - build-runtime-bundle|submit|dispatch|drive|observe|collect|status|respond|replace|close|retain-failure|queue|auth-seed) + build-runtime-bundle|submit|dispatch|drive|observe|collect|status|respond|replace|close|retain-failure|purge-retained|queue|auth-seed) exec python3 "$SCRIPT_DIR/fm-azure-validation.py" "$@" ;; *) diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index f0f905b9c3a..298f2fccf99 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -38,7 +38,8 @@ # account home is exactly how a credential stager and its remover once resolved # different directories and leaked a credential. PI_ACCOUNT_HOME_TOOL = ROOT / "bin" / "fm-pi-account-home.py" -STATE_SCHEMA = "fm.worker-lifecycle/v1" +LEGACY_STATE_SCHEMA = "fm.worker-lifecycle/v1" +STATE_SCHEMA = "fm.worker-lifecycle/v2" # The scalar pending_action slot this schema carried is superseded by the # per-slot pending_actions map. The sentinel is deliberately a string an OLD # binary's verify_state refuses ("pending provider action is malformed"), so a @@ -52,6 +53,7 @@ RELEASE_SCHEMA = "fm.worker-release/v2" AUTHORITY_SCHEMA = "fm.worker-authority/v1" CAPACITY_RESERVATION_SCHEMA = "fm.capacity-reservation/v1" +CAPACITY_FENCE_RETIREMENT_SCHEMA = "fm.capacity-fence-retirement/v1" SPECIALIZED_WORKLOAD_ROLES = ("validation", "review", "browser", "networkless-verifier", "crosscheck") PROVIDER_REQUEST_SCHEMA = "fm.worker-provider-request/v1" PROVIDER_RESPONSE_SCHEMA = "fm.worker-provider-response/v1" @@ -504,6 +506,7 @@ def empty_state(env): "queue": {}, "workers": {}, "capacity_reservations": {}, + "retired_capacity_fences": {}, "completed_worker_seconds": 0.0, "pending_action": LEGACY_PENDING_SENTINEL, "pending_actions": {}, @@ -525,6 +528,7 @@ def verify_state(env, state): not isinstance(state.get("queue"), dict) or not isinstance(state.get("workers"), dict) or not isinstance(state.get("capacity_reservations"), dict) + or not isinstance(state.get("retired_capacity_fences"), dict) or not isinstance(state.get("executions"), dict) ): raise LifecycleError("lifecycle queue, worker, or shared capacity inventory is malformed") @@ -560,6 +564,26 @@ def verify_state(env, state): require_binding("capacity reservation fence", reservation.get("fence_binding")) if "shape_id" in reservation: require_id("capacity shape id", reservation.get("shape_id")) + for fence, retirement in state["retired_capacity_fences"].items(): + if ( + not isinstance(retirement, dict) + or retirement.get("schema") != CAPACITY_FENCE_RETIREMENT_SCHEMA + or retirement.get("fence_binding") != fence + or not isinstance(retirement.get("reservation_ids"), list) + or not retirement.get("reservation_ids") + or retirement.get("reservation_ids") + != sorted(set(retirement.get("reservation_ids") or [])) + or len(retirement.get("reservation_ids") or []) > 256 + or not isinstance(retirement.get("retired_at"), str) + or not retirement.get("retired_at") + ): + raise LifecycleError("durable specialized capacity fence retirement is malformed") + require_binding("retired capacity fence", fence) + require_binding( + "capacity fence retirement receipt", retirement.get("retirement_receipt") + ) + for reservation_id in retirement["reservation_ids"]: + require_id("retired capacity reservation id", reservation_id) legacy = state.get("pending_action") if legacy is not None and legacy != LEGACY_PENDING_SENTINEL: # A dict here means load_state's migration did not run; anything else @@ -599,7 +623,14 @@ def load_state(env): if "is absent" not in str(exc): raise state = empty_state(env) + if state.get("schema") == LEGACY_STATE_SCHEMA: + # A v1 document has no fence-retirement authority to preserve. Upgrade + # it in memory; the next locked save makes the v2 rollback fence + # durable, after which a v1 binary refuses instead of reopening a + # retired fence it does not understand. + state["schema"] = STATE_SCHEMA state.setdefault("capacity_reservations", {}) + state.setdefault("retired_capacity_fences", {}) state.setdefault("executions", {}) state.setdefault("pending_actions", {}) state.setdefault("revision", 0) @@ -2763,6 +2794,15 @@ def parser(): capacity_release.add_argument("--cleanup-receipt", required=True) capacity_release.add_argument("--confirm-subscription", required=True) + capacity_retire = sub.add_parser( + "capacity-retire-fence", + help="permanently close one exact specialized capacity fence after release", + ) + capacity_retire.add_argument("--fence-binding", required=True) + capacity_retire.add_argument("--reservation-id", action="append", required=True) + capacity_retire.add_argument("--retirement-receipt", required=True) + capacity_retire.add_argument("--confirm-subscription", required=True) + execute = sub.add_parser("execute", help="run one exact private task command and collect its bound result") execute.add_argument("--task", required=True) execute.add_argument("--task-generation", required=True) @@ -3339,6 +3379,11 @@ def specialized_reservation_from_args(args): ) +def refuse_retired_capacity_fence(state, fence): + if fence in state["retired_capacity_fences"]: + raise LifecycleError("retired capacity fence cannot admit another reservation") + + def command_capacity_reserve(env, args): if args.confirm_subscription != env["subscription"]: raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") @@ -3346,6 +3391,7 @@ def command_capacity_reserve(env, args): reservation_id = candidate["reservation_id"] with controller_lock(env): state = load_state(env) + refuse_retired_capacity_fence(state, candidate["fence_binding"]) existing = state["capacity_reservations"].get(reservation_id) readmission_id = None identity_fields = ( @@ -3477,6 +3523,7 @@ def command_capacity_reserve_shape(env, args): ) with controller_lock(env): state = load_state(env) + refuse_retired_capacity_fence(state, args.fence_binding) entries = [] for candidate in constituents: existing = state["capacity_reservations"].get(candidate["reservation_id"]) @@ -3606,6 +3653,103 @@ def command_capacity_release(env, args): print("specialized capacity reservation released after exact zero-compute proof") +def exact_provider_capacity_identity(reservation, provider): + return ( + isinstance(reservation, dict) + and reservation.get("schema") == CAPACITY_RESERVATION_SCHEMA + and reservation.get("reservation_id") == provider.get("reservation_id") + and reservation.get("role") == provider.get("role") + and reservation.get("sku") == provider.get("sku") + and str(reservation.get("sku_family", "")).lower() + == str(provider.get("sku_family", "")).lower() + and reservation.get("vcpus") == provider.get("vcpus") + and not isinstance(reservation.get("amount_usd"), bool) + and isinstance(reservation.get("amount_usd"), (int, float)) + and not isinstance(provider.get("amount_usd"), bool) + and isinstance(provider.get("amount_usd"), (int, float)) + and math.isclose( + float(reservation["amount_usd"]), float(provider["amount_usd"]), + rel_tol=0.0, abs_tol=1e-6, + ) + ) + + +def command_capacity_retire_fence(env, args): + """Atomically close a released fence against every future admission. + + Provider inventory and the complete same-fence ledger census occur while + the shared controller lock excludes both reservation entry points. The v2 + retirement tombstone is committed before that lock opens, so successful + return is the irreversible admission barrier an artifact purge can rely on. + """ + if args.confirm_subscription != env["subscription"]: + raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID") + fence = require_binding("capacity reservation fence", args.fence_binding) + receipt = require_binding("capacity fence retirement receipt", args.retirement_receipt) + reservation_ids = sorted(set( + require_id("retired capacity reservation id", value) + for value in args.reservation_id + )) + if len(reservation_ids) != len(args.reservation_id) or len(reservation_ids) > 256: + raise LifecycleError("capacity fence retirement reservation ids are not exact and distinct") + expected = { + "schema": CAPACITY_FENCE_RETIREMENT_SCHEMA, + "fence_binding": fence, + "reservation_ids": reservation_ids, + "retirement_receipt": receipt, + } + with controller_lock(env): + state = load_state(env) + prior = state["retired_capacity_fences"].get(fence) + if prior is not None and any(prior.get(key) != value for key, value in expected.items()): + raise LifecycleError("capacity fence already has a different retirement identity") + allowed = set(reservation_ids) + same_fence = { + reservation_id: reservation + for reservation_id, reservation in state["capacity_reservations"].items() + if isinstance(reservation, dict) and reservation.get("fence_binding") == fence + } + outside = sorted(set(same_fence) - allowed) + if outside: + raise LifecycleError( + "capacity fence retirement census found an unplanned reservation: {}".format( + outside[0] + ) + ) + for reservation_id, reservation in same_fence.items(): + if ( + reservation.get("reservation_id") != reservation_id + or reservation.get("status") != "released" + or not HEX_BINDING.match(str(reservation.get("cleanup_receipt", "")).split(":")[-1]) + ): + raise LifecycleError( + "capacity fence retirement requires every exact reservation released" + ) + inventory = provider_call(env, "inventory")["inventory"] + provider_reservations = inventory.get("capacity_reservations") + if not isinstance(provider_reservations, list): + raise LifecycleError("provider capacity inventory is malformed") + for provider in provider_reservations: + if not isinstance(provider, dict) or provider.get("active") is not True: + continue + reservation_id = provider.get("reservation_id") + controller = state["capacity_reservations"].get(reservation_id) + if not exact_provider_capacity_identity(controller, provider): + raise LifecycleError( + "provider-active capacity lacks exact controller identity during fence retirement" + ) + if reservation_id in allowed or controller.get("fence_binding") == fence: + raise LifecycleError( + "provider still observes active capacity on the retiring fence" + ) + if prior is None: + state["retired_capacity_fences"][fence] = dict(expected, retired_at=iso_utc()) + save_state(env, state) + print("specialized capacity fence retired after exact release census") + else: + print("specialized capacity fence already retired with exact identity") + + # What an ORDINARY crewmate payload may contain: the repository as a # credential-free bundle, plus the one task file its entrypoint reads. This set # is deliberately NOT widened for the compartment lane; see below. @@ -4663,6 +4807,8 @@ def main(argv=None): command_capacity_reserve_shape(env, args) elif args.command == "capacity-release": command_capacity_release(env, args) + elif args.command == "capacity-retire-fence": + command_capacity_retire_fence(env, args) elif args.command == "execute": command_execute(env, args) elif args.command == "authority-receipt": diff --git a/bin/fm-worker-lifecycle.sh b/bin/fm-worker-lifecycle.sh index a2a8d315cc3..4bb2b918bac 100755 --- a/bin/fm-worker-lifecycle.sh +++ b/bin/fm-worker-lifecycle.sh @@ -45,6 +45,7 @@ # fm-worker-lifecycle.sh message-put --file | --attach # fm-worker-lifecycle.sh message-collect --output-dir # fm-worker-lifecycle.sh compartment-chain-tip --sequence --chain-digest +# fm-worker-lifecycle.sh capacity-retire-fence # fm-worker-lifecycle.sh status [--live] [--json] # fm-worker-lifecycle.sh acceptance-plan set -euo pipefail @@ -137,7 +138,7 @@ fm_worker_receipt_credential_remains() { # ' \ + --confirm-purge \ + --confirm-subscription "$FM_AZURE_SUBSCRIPTION_ID" \ + --confirm-cell '' \ + --confirm-request-digest '' +``` + +The command acquires the cell's shard-driver lock before its cell-state lock, so it cannot purge while the same cell is driving child invocations. +Before any destructive call, it proves the control VM, NIC, OS disk, Run Commands, and shutdown schedule absent. +It also reads every runner state owned by the cell, requires every retry lineage to be terminal with Azure compute absent, and requires every dispatched reservation to carry a durable released receipt. +Every control and shard constituent must still have the exact durable allocator id, fence, shape, SKU, family, vCPU, and cost identity recorded at admission. +The allocator's entire exact-fence ledger must be inside the sealed control, planned-root, and runner-state census; any outside row refuses before the purge plan is sealed, regardless of release status, provider inactivity, or recorded workload role. +The pinned shared-provider inventory must show every exact censused reservation inactive or absent with matching SKU, family, vCPU, and cost identity; a provider-active exact id, including one hidden behind a stale released allocator row, refuses. +Because provider inventory is not fence-bound, any provider-active id without an exact controller reservation also refuses; an unrelated active reservation is ignored only when its controller record proves a different fence. +An already-released exact constituent is accepted with its cleanup receipt, while any admitted constituent with no dispatch lineage and a queued or reserved status is listed separately for exact release. + +It then proves the worktree disk is detached with its recorded stable identity and current ETag, proves the private container and its complete two-role inventory, and proves the cell identity has only its exact container and auth-share grants. +Those identities, every verified shard lineage, and only the remaining capacity constituents are sealed into an immutable purge plan. +The state durably enters non-replaceable `purging` with the plan digest before the first deletion. +Any remaining exact capacity constituent is released first and then read back as durably released and provider-inactive or absent before the retained disk, RBAC, container, or identity is touched. +Every retry repeats the complete fresh allocator/provider census before attempting a remaining release, and the census is repeated after release before artifact deletion, so a new same-fence reservation can never fall outside the sealed plan. + +Retries never rebuild or widen that plan. +They accept only remaining subsets of its disk, role, container, identity, and capacity identities, use the stored ETags for conditional deletion, and persist progress after every boundary. +A partial or ambiguous attempt remains `purging`; `replace` and `retain-failure` cannot reclaim it. +Completion retains the local state as a `purged` tombstone with the immutable plan and digest, and repeating the exact command is a no-op. + ## Cleanup order Successful close occurs only after complete result/report/evidence collection, CI-green proof, exact remote-current head proof, and explicit head confirmation. diff --git a/docs/azure-workers.md b/docs/azure-workers.md index d5bf53df9c1..c0faa838877 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -103,7 +103,7 @@ It does not bypass unreadable cost or quota evidence, identity checks, the sixte ## Desired capacity and admission -The same durable allocator also owns specialized reservations through `capacity-reserve`, `capacity-reserve-shape`, and `capacity-release`. +The same durable allocator also owns specialized reservations through `capacity-reserve`, `capacity-reserve-shape`, `capacity-release`, and the cleanup-only `capacity-retire-fence` barrier. Every caller uses the canonical Firstmate control home and its one shared state directory; `FM_AZURE_SHARED_CAPACITY_STATE_DIR` may relocate that directory only for an explicitly configured installation and must be identical for all callers. A validation, review, browser, networkless-verifier, or Crosscheck caller submits one exact reservation ID and fence binding, a reviewed SKU/family pair, the reviewed four-vCPU worker shape or reviewed eight-vCPU control shape, and finite worst-case cost before creating compute. Admission returns `reserved` or leaves the request durably `queued`; callers must not create compute for a queued reservation. @@ -114,6 +114,7 @@ A shape retry never demotes constituents that are already reserved, and each con The complete shape total may never exceed the shared 40-vCPU specialized envelope, and no shape or constituent bypasses cumulative actual/forecast admission in commissioning mode. The existing disposable runner invokes this path before its Azure management reservation and VM creation, then releases the shared reservation only after exact VM/NIC/OS-disk absence. Restarting either controller is idempotent under the same reservation ID and fence, while a changed identity refuses. +An exact failed-retained validation purge closes its capacity fence only after every sealed constituent is released and provider-inactive. `capacity-retire-fence` repeats the complete same-fence ledger and provider census while holding the allocator lock, then durably records a permanent retirement tombstone before unlocking. Both single and shape reservation entry points refuse a retired fence before any insertion or re-admission, so retained disks, private storage, and RBAC can be deleted only after no later same-fence capacity can enter. The lifecycle state advances to v2 when this authority is first written, causing an older allocator that does not understand fence retirement to refuse the document rather than reopen it. The controller computes desired active capacity from all eligible queued and assigned work, current exact assignments, the sixteen-worker software cap, one shared East US regional ceiling, live exact-family quota, actual and forecast spend, and durable per-assignment cost reservations. Quota is only capacity and never creates demand. diff --git a/tests/fm-azure-validation.test.sh b/tests/fm-azure-validation.test.sh index b844a12c9a5..3b6b399bb28 100755 --- a/tests/fm-azure-validation.test.sh +++ b/tests/fm-azure-validation.test.sh @@ -1145,6 +1145,471 @@ PY pass "partial container/role cleanup resumes idempotently from its exact persisted plan" } +purge_retained_contract() { + local tmp + fm_test_tmproot_into tmp fm-azure-validation-purge + python3 - "$HOST" "$tmp" <<'PY' || fail "retained-failure purge contract failed" +import contextlib,copy,hashlib,importlib.util,json,pathlib,sys,types +spec=importlib.util.spec_from_file_location("validation",sys.argv[1]) +m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m) +root=pathlib.Path(sys.argv[2]); state_dir=root/"state"/"azure-validation" +runner_dir=root/"state"/"azure-runner" +state_dir.mkdir(parents=True); runner_dir.mkdir(parents=True) +cell="azv-aaaaaaaaaaaa"; fence="sha256:"+"a"*64; request_digest="sha256:"+"b"*64 +sub="11111111-1111-4111-8111-111111111111" +principal="33333333-3333-4333-8333-333333333333" +client="44444444-4444-4444-8444-444444444444" +operator="22222222-2222-4222-8222-222222222222" +root_one="azr-111111111111"; root_two="azr-222222222222"; retry_one="azr-333333333333-a2" +command_digest="sha256:"+"c"*64 +env={"home":root,"state_dir":state_dir,"subscription":sub,"resource_group":"rg", + "storage":"storage","operator_object_id":operator} +worktree_id="/subscriptions/{}/resourceGroups/rg/providers/Microsoft.Compute/disks/work".format(sub) +identity_id="/subscriptions/{}/resourceGroups/rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/cell".format(sub) +container_scope="/subscriptions/{}/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/storage/blobServices/default/containers/fmvalaaaaaaaaaaaa".format(sub) +account_scope="/subscriptions/{}/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/storage".format(sub) +blob_role="/subscriptions/{}/providers/Microsoft.Authorization/roleDefinitions/{}".format(sub,m.BLOB_DATA_CONTRIBUTOR_ROLE) +file_role="/subscriptions/{}/providers/Microsoft.Authorization/roleDefinitions/{}".format(sub,m.FILE_DATA_PRIVILEGED_CONTRIBUTOR_ROLE) +state={ + "schema":m.SCHEMA,"cell":cell,"phase":"failed-retained","request_digest":request_digest, + "request":{"fence":fence,"limits":{"behavior_shards":2}}, + "staging":{"container":"fmvalaaaaaaaaaaaa"}, + "allocation":{"sku":"Standard_D8as_v6","sku_family":"standardDasv6Family"}, + "resources":{ + "vm_id":"/control/vm","nic_id":"/control/nic","os_disk_id":"/control/os", + "ttl_schedule_id":"/control/ttl","safety_run_command_id":"/control/safety", + "worktree_disk_id":worktree_id,"identity_id":identity_id,"identity_principal_id":principal, + "identities":{ + "worktree":{"id":worktree_id.lower(),"etag":"work-etag","unique_id":"work-unique"}, + "identity":{"id":identity_id.lower(),"etag":None,"client_id":client,"principal_id":principal}, + }, + }, + "admission":{"shape_id":cell,"control_amount_usd":25.0,"shard_plan":[ + {"invocation":root_one,"shard":1,"sku":"Standard_D4s_v6","sku_family":"standardDsv6Family","amount_usd":10.0}, + {"invocation":root_two,"shard":2,"sku":"Standard_D4ds_v6","sku_family":"standardDdsv6Family","amount_usd":11.0}, + ]}, + "shard_runs":{"sha256:"+"d"*64:{"invocation":root_one,"command_digest":command_digest}}, + "events":[], +} +m.ensure_dirs(env); m.save_state(env,state,create=True) +runner_resources={ + "vm_id":"/runner/vm","nic_id":"/runner/nic","os_disk_id":"/runner/os", + "ttl_schedule_id":"/runner/ttl","safety_run_command_id":"/runner/safety", + "run_command_name":"execute","safety_run_command_name":"safety-shutdown", +} +runner={ + "schema":"fm.azure-command/v1","invocation":root_one,"parent_invocation":None,"phase":"complete", + "request_digest":"sha256:"+"e"*64, + "request":{"schema":"fm.azure-command/v1","invocation":root_one,"parent_invocation":None, + "request_digest":"sha256:"+"e"*64,"command_digest":command_digest, + "capacity_parent":cell,"capacity_fence":fence.split(":",1)[1], + "lineage_root_invocation":root_one}, + "resources":runner_resources, + "shared_capacity_reservation":{"reservation_id":root_one,"fence_binding":fence.split(":",1)[1], + "status":"released","cleanup_receipt":"sha256:"+"f"*64}, +} +(runner_dir/(root_one+".json")).write_text(json.dumps(runner)) +retry_runner=copy.deepcopy(runner) +retry_runner.update({"invocation":retry_one,"parent_invocation":root_one}) +retry_runner["request"].update({"invocation":retry_one,"parent_invocation":root_one, + "request_digest":"sha256:"+"7"*64, + "limits":{"sku":"Standard_D4s_v6","sku_family":"standardDsv6Family"}}) +retry_runner["request_digest"]="sha256:"+"7"*64 +retry_runner["resources"]={key:value.replace("/runner/","/retry/") if isinstance(value,str) else value + for key,value in runner_resources.items()} +retry_runner["shared_capacity_reservation"]={"reservation_id":retry_one, + "fence_binding":fence.split(":",1)[1],"status":"released","amount_usd":9.0, + "cleanup_receipt":"sha256:"+"8"*64} +(runner_dir/(retry_one+".json")).write_text(json.dumps(retry_runner)) +resources={ + worktree_id:{"id":worktree_id,"etag":"work-etag","properties":{"uniqueId":"work-unique"}, + "tags":{"validation-cell":cell,"fence":fence}}, + identity_id:{"id":identity_id,"properties":{"clientId":client,"principalId":principal}, + "tags":{"validation-cell":cell,"fence":fence}}, + container_scope:{"id":container_scope,"etag":"container-etag","properties":{"publicAccess":"None"}}, +} +roles=[ + {"id":"/roles/operator","scope":container_scope,"principalId":operator,"roleDefinitionId":blob_role}, + {"id":"/roles/cell","scope":container_scope,"principalId":principal,"roleDefinitionId":blob_role}, + {"id":"/roles/auth","scope":account_scope,"principalId":principal,"roleDefinitionId":file_role}, +] +mutations=[]; lock_stack=[]; lock_events=[]; release_fail=[True] +retired_fences=set(); retirement_tombstones={}; delete_seam_attempts=[] +def read_resource(_env,resource_id,kind): + value=resources.get(resource_id) + return (value is not None,copy.deepcopy(value) if value is not None else None) +def assert_sealed_before_mutation(): + durable=json.loads((state_dir/(cell+".json")).read_text()) + assert durable["phase"]=="purging", "destructive mutation preceded the durable purging transition" + purge=durable["purge"] + assert purge["plan_digest"]==m.sha256_bytes(m.canonical_bytes(purge["plan"])) +def az(_env,args,**kwargs): + if args[:3]==["role","assignment","list"]: + if "--assignee-object-id" in args: + wanted=args[args.index("--assignee-object-id")+1].lower() + return ([copy.deepcopy(item) for item in roles if item["principalId"].lower()==wanted],0,"") + scope=args[args.index("--scope")+1].lower() + return ([copy.deepcopy(item) for item in roles if item["scope"].lower()==scope],0,"") + if args[:3]==["role","assignment","delete"]: + assert_sealed_before_mutation(); wanted=args[-1] + mutations.append(("role",wanted)); roles[:]=[item for item in roles if item["id"]!=wanted] + return (None,0,"") + if args[:2]==["rest","--method"] and args[2]=="delete": + assert_sealed_before_mutation(); url=args[args.index("--url")+1] + wanted=next(resource_id for resource_id in list(resources) if resource_id in url) + if wanted==worktree_id: + assert "If-Match=work-etag" in args + # Reproduce the former TOCTOU at the exact first irreversible artifact + # mutation. The allocator must already have made this fence permanently + # inadmissible, so the attempted reserve cannot enter the fresh census. + if fence.split(":",1)[1] in retired_fences: + delete_seam_attempts.append("refused-retired-fence") + else: + late_id="azr-delete-seam0001" + capacity_records[late_id]=reservation( + late_id,"Standard_D4s_v6","standardDsv6Family",4,13.5,"reserved" + ) + delete_seam_attempts.append("admitted") + assert delete_seam_attempts[-1]=="refused-retired-fence" + if wanted==container_scope: + assert "If-Match=container-etag" in args + mutations.append(("resource",wanted)); resources.pop(wanted) + return (None,0,"") + raise AssertionError(args) +@contextlib.contextmanager +def lock(_env,name="queue"): + if name==cell+"-shards": + assert not lock_stack + else: + assert name==cell and lock_stack==[cell+"-shards"] + lock_stack.append(name); lock_events.append(("enter",name)) + try: yield + finally: + lock_events.append(("exit",name)); assert lock_stack.pop()==name +def release(_env,_state,reservation_id,evidence): + assert_sealed_before_mutation(); mutations.append(("capacity",reservation_id,evidence)) + if release_fail[0]: + release_fail[0]=False + raise m.ValidationError("injected capacity release boundary") + capacity_records[reservation_id]["status"]="released" + capacity_records[reservation_id]["cleanup_receipt"]="sha256:"+"6"*64 + provider_capacity_records[reservation_id]["active"]=False +def retire(_env,retirement): + assert_sealed_before_mutation() + exact_fence=fence.split(":",1)[1] + assert retirement["fence_binding"]==exact_fence + prior=retirement_tombstones.get(exact_fence) + if prior is not None and prior!=retirement: + raise m.ValidationError("conflicting durable retirement tombstone") + assert retirement["reservation_ids"]==sorted( + key for key,value in capacity_records.items() if value.get("fence_binding")==exact_fence + ) + assert all( + capacity_records[key]["status"]=="released" + and provider_capacity_records[key]["active"] is False + for key in retirement["reservation_ids"] + ) + retirement_tombstones.setdefault(exact_fence,copy.deepcopy(retirement)) + retired_fences.add(exact_fence); mutations.append(("retire",exact_fence)) +m.read_resource=read_resource; m.az_command=az; m.lock=lock +m.release_shape_constituent=release; m.time.sleep=lambda _seconds:None +real_retire_purge_capacity_fence=m.retire_purge_capacity_fence +m.retire_purge_capacity_fence=retire +def reservation(reservation_id,sku,family,vcpus,amount,status,receipt=None): + return {"schema":"fm.capacity-reservation/v1","reservation_id":reservation_id, + "fence_binding":fence.split(":",1)[1],"shape_id":cell,"role":"specialized", + "workload_role":"validation","discretionary":True,"sku":sku,"sku_family":family, + "vcpus":vcpus,"amount_usd":amount,"status":status,"cleanup_receipt":receipt} +capacity_records={ + cell:reservation(cell,"Standard_D8as_v6","standardDasv6Family",8,25.0,"reserved"), + root_one:reservation(root_one,"Standard_D4s_v6","standardDsv6Family",4,10.0,"released","sha256:"+"f"*64), + root_two:reservation(root_two,"Standard_D4ds_v6","standardDdsv6Family",4,11.0,"reserved"), + retry_one:reservation(retry_one,"Standard_D4s_v6","standardDsv6Family",4,9.0,"released","sha256:"+"8"*64), +} +capacity_records[retry_one].pop("shape_id") +def provider_capacity(record,active=False): + return {key:record[key] for key in ( + "reservation_id","role","sku","sku_family","vcpus","amount_usd" + )}|{"active":active} +provider_capacity_records={ + key:provider_capacity(value) for key,value in capacity_records.items() +} +unrelated="azr-555555555555" +capacity_records[unrelated]=reservation( + unrelated,"Standard_D4s_v6","standardDsv6Family",4,12.0,"reserved" +) +capacity_records[unrelated]["fence_binding"]="4"*64 +capacity_records[unrelated]["shape_id"]="azv-cccccccccccc" +provider_capacity_records[unrelated]=provider_capacity( + capacity_records[unrelated],active=True +) +m.capacity_authority_snapshot=lambda _env:( + copy.deepcopy(capacity_records),copy.deepcopy(list(provider_capacity_records.values())) +) +def args(**updates): + values={"cell":cell,"confirm_purge":True,"confirm_subscription":sub, + "confirm_cell":cell,"confirm_request_digest":request_digest} + values.update(updates); return types.SimpleNamespace(**values) + +# An already-released control constituent is accepted only through the same +# exact id/fence/shape ledger proof, and is not planned for a second release. +capacity_records[cell]["status"]="released" +capacity_records[cell]["cleanup_receipt"]="sha256:"+"9"*64 +_, released_control_plan, _=m.plan_purge_shards(env,m.load_state(env,cell)) +assert [item["reservation_id"] for item in released_control_plan]==[root_two] +capacity_records[cell]["status"]="reserved"; capacity_records[cell]["cleanup_receipt"]=None + +# Admit-red confirmations and phase checks perform no destructive mutation. +for bad in ( + args(confirm_purge=False),args(confirm_subscription="99999999-9999-4999-8999-999999999999"), + args(confirm_cell="azv-bbbbbbbbbbbb"),args(confirm_request_digest="sha256:"+"0"*64), +): + try: m.purge_retained(env,bad) + except m.ValidationError: pass + else: raise AssertionError("purge admitted an inexact destructive confirmation") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +state=m.load_state(env,cell); state["phase"]="running"; m.save_state(env,state) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "only an exact failed-retained" in str(exc) +else: raise AssertionError("purge admitted a live phase") +assert not mutations +state=m.load_state(env,cell); state["phase"]="failed-retained"; m.save_state(env,state) + +# A corrupt phase cannot convert a passing result into purge authority. +state=m.load_state(env,cell); state["result"]={"outcome":"checks-passed"}; m.save_state(env,state) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "passed result" in str(exc) +else: raise AssertionError("purge admitted a passed result under a corrupt retained phase") +assert not mutations +state=m.load_state(env,cell); state.pop("result"); m.save_state(env,state) + +# Provider inventory is not fence-bound, so an active id without an exact +# controller row cannot be classified as unrelated and must refuse. The +# unrelated active id above remains admissible because its controller fence is +# exact and different. +orphan="azr-666666666666" +provider_capacity_records[orphan]=provider_capacity(capacity_records[root_one],active=True) +provider_capacity_records[orphan]["reservation_id"]=orphan +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "lacks exact controller identity" in str(exc) +else: raise AssertionError("purge admitted provider-active capacity with no controller identity") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +provider_capacity_records.pop(orphan) + +# A live allocator reservation on this cell's exact fence cannot disappear +# merely because its retry runner file is missing. The preflight must refuse +# before sealing a plan or deleting any retained resource. +untracked_retry="azr-444444444444-a3" +capacity_records[untracked_retry]=reservation( + untracked_retry,"Standard_D4s_v6","standardDsv6Family",4,9.5,"reserved" +) +capacity_records[untracked_retry].pop("shape_id") +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "outside the exact purge census" in str(exc) +else: raise AssertionError("purge ignored a live same-fence retry without runner state") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +capacity_records.pop(untracked_retry) + +# A released allocator row cannot hide provider drift either. If the same exact +# reservation remains provider-active without runner evidence, purge still +# refuses before sealing or deleting anything. +capacity_records[untracked_retry]=reservation( + untracked_retry,"Standard_D4s_v6","standardDsv6Family",4,9.5,"released","sha256:"+"5"*64 +) +capacity_records[untracked_retry].pop("shape_id") +provider_capacity_records[untracked_retry]=provider_capacity( + capacity_records[untracked_retry],active=True +) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "outside the exact purge census" in str(exc) +else: raise AssertionError("purge ignored provider-active capacity hidden by a released allocator row") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +capacity_records.pop(untracked_retry); provider_capacity_records.pop(untracked_retry) + +# Released and provider-inactive does not make an untracked same-fence row +# safe to omit: the permanent allocator retirement performs the same entire +# fence census and would refuse it after the cell had already entered purging. +# Reject it before the immutable plan and non-replaceable transition instead. +capacity_records[untracked_retry]=reservation( + untracked_retry,"Standard_D4s_v6","standardDsv6Family",4,9.5,"released","sha256:"+"5"*64 +) +capacity_records[untracked_retry].pop("shape_id") +provider_capacity_records[untracked_retry]=provider_capacity( + capacity_records[untracked_retry],active=False +) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "outside the exact purge census" in str(exc) +else: raise AssertionError("purge omitted released provider-inactive same-fence history") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +capacity_records.pop(untracked_retry); provider_capacity_records.pop(untracked_retry) + +# Even a fully tracked terminal retry cannot pass while provider inventory +# still observes its exact capacity id active. +provider_capacity_records[retry_one]["active"]=True +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "active or has drifted" in str(exc) +else: raise AssertionError("purge ignored provider-active tracked retry capacity") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +provider_capacity_records[retry_one]["active"]=False + +# A nonterminal shard is an admit-red condition before the immutable plan. +runner["phase"]="running"; (runner_dir/(root_one+".json")).write_text(json.dumps(runner)) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "not terminal" in str(exc) +else: raise AssertionError("purge admitted a nonterminal shard lineage") +assert not mutations and m.load_state(env,cell)["phase"]=="failed-retained" +runner["phase"]="complete"; (runner_dir/(root_one+".json")).write_text(json.dumps(runner)) + +# The first destructive attempt fails at capacity release before artifact +# cleanup. It must stay non-replaceable and retain the same plan and resources. +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "injected capacity" in str(exc) +else: raise AssertionError("injected destructive boundary did not fail") +partial=m.load_state(env,cell); assert partial["phase"]=="purging" +sealed=copy.deepcopy(partial["purge"]["plan"]); digest=partial["purge"]["plan_digest"] +assert digest==m.sha256_bytes(m.canonical_bytes(sealed)) +assert partial["purge"]["progress"]=={ + "worktree_absent":False,"container_roles_absent":False,"container_absent":False, + "auth_share_roles_absent":False,"identity_absent":False, + "capacity_fence_retired":False,"released_capacity":[], +} +assert m.replacement_allowed(partial,"absent-proven")[0] is False +assert set(resources)=={worktree_id,identity_id,container_scope} and len(roles)==3 +assert lock_events[-4:]==[("enter",cell+"-shards"),("enter",cell),("exit",cell),("exit",cell+"-shards")] + +# Exercise the real bridge end to end: this retirement was emitted by the real +# build_purge_plan above, then the real validation helper constructs argv for +# the actual lifecycle parser and command against a private controller state. +# A prefixed receipt passes the mocked purge path but fails require_binding; +# pinning the raw digest here prevents that false green. +retirement=sealed["capacity_fence_retirement"] +assert len(retirement["retirement_receipt"])==64 +assert all(ch in "0123456789abcdef" for ch in retirement["retirement_receipt"]) +lifecycle=m.worker_lifecycle_module(); bridge_dir=root/"bridge-workers" +bridge_env={ + "home_binding":"1"*64,"subscription":sub,"deployment_generation":"dep-one", + "owner":"owner","prefix":"fmtest","state_dir":bridge_dir, + "state_path":bridge_dir/"controller.json","lock_path":bridge_dir/".lock", +} +with lifecycle.controller_lock(bridge_env): + bridge_state=lifecycle.load_state(bridge_env) + for planned in sealed["capacity_constituents"]: + # Released history is bounded and may be absent by retirement time. Keep + # two real sealed shape rows here so the command still exercises its locked + # same-fence census while legitimately treating the remaining ids as pruned. + if planned["reservation_id"] not in (root_one,root_two): continue + item=copy.deepcopy(planned); item["status"]="released" + item["released_at"]="2026-08-22T00:00:00Z"; item["cleanup_receipt"]="5"*64 + bridge_state["capacity_reservations"][item["reservation_id"]]=item + outsider=copy.deepcopy(bridge_state["capacity_reservations"][root_one]) + outsider["reservation_id"]="azr-untracked0001" + bridge_state["capacity_reservations"][outsider["reservation_id"]]=outsider + lifecycle.save_state(bridge_env,bridge_state) +saved_provider_call=lifecycle.provider_call; saved_lifecycle_command=m.lifecycle_command +lifecycle.provider_call=lambda _env,_operation:{ + "inventory":{"capacity_reservations":[]} +} +def actual_retirement_bridge(_env,arguments): + parsed=lifecycle.parser().parse_args(arguments) + assert parsed.command=="capacity-retire-fence" + lifecycle.command_capacity_retire_fence(bridge_env,parsed) + return types.SimpleNamespace(returncode=0,stdout="",stderr="") +m.lifecycle_command=actual_retirement_bridge +try: + try: real_retire_purge_capacity_fence(env,retirement) + except lifecycle.LifecycleError as exc: assert "unplanned reservation" in str(exc) + else: raise AssertionError("real retirement accepted omitted released same-fence history") + with lifecycle.controller_lock(bridge_env): + bridge_state=lifecycle.load_state(bridge_env) + del bridge_state["capacity_reservations"]["azr-untracked0001"] + lifecycle.save_state(bridge_env,bridge_state) + real_retire_purge_capacity_fence(env,retirement) +finally: + m.lifecycle_command=saved_lifecycle_command; lifecycle.provider_call=saved_provider_call +with lifecycle.controller_lock(bridge_env): + retired=lifecycle.load_state(bridge_env)["retired_capacity_fences"] +assert retired[retirement["fence_binding"]]["retirement_receipt"]==retirement["retirement_receipt"] + +# A retry must re-census the entire fresh authority before repeating the +# failed release or touching artifacts. A new same-fence active retry appearing +# after the sealed boundary refuses with no additional mutation. +late_retry="azr-777777777777-a4" +capacity_records[late_retry]=reservation( + late_retry,"Standard_D4s_v6","standardDsv6Family",4,13.0,"reserved" +) +capacity_records[late_retry].pop("shape_id") +provider_capacity_records[late_retry]=provider_capacity( + capacity_records[late_retry],active=True +) +before_late=copy.deepcopy(mutations) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "outside the exact purge census" in str(exc) +else: raise AssertionError("purge retry ignored a late same-fence capacity reservation") +late=m.load_state(env,cell) +assert mutations==before_late and late["phase"]=="purging" +assert late["purge"]["plan"]==sealed and late["purge"]["plan_digest"]==digest +assert set(resources)=={worktree_id,identity_id,container_scope} and len(roles)==3 +capacity_records.pop(late_retry); provider_capacity_records.pop(late_retry) + +# A retry cannot alter, widen, or rediscover a deletion target after the +# irreversible boundary; the stored digest fences the entire immutable plan. +tampered=copy.deepcopy(partial); tampered["purge"]["plan"]["worktree"]["resource_id"]="/foreign" +m.save_state(env,tampered); before_tamper=copy.deepcopy(mutations) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "plan is corrupt or rebound" in str(exc) +else: raise AssertionError("purge retry accepted a widened stored plan") +assert mutations==before_tamper +m.save_state(env,partial) + +# Crash after the shared allocator durably retires the fence but before the +# cell progress save must leave every artifact intact. The next invocation +# presents the same sealed identity/receipt, gets the idempotent retirement, +# and continues without reopening admission. +original_save_progress=m.save_purge_progress; fail_retirement_progress=[True] +def save_progress(_env,_state,key,value=True): + if key=="capacity_fence_retired" and fail_retirement_progress[0]: + fail_retirement_progress[0]=False + raise m.ValidationError("injected post-retirement progress crash") + return original_save_progress(_env,_state,key,value) +m.save_purge_progress=save_progress +before_retirement=len(mutations) +try: m.purge_retained(env,args()) +except m.ValidationError as exc: assert "post-retirement progress crash" in str(exc) +else: raise AssertionError("purge crossed an injected post-retirement crash") +after_retirement=m.load_state(env,cell) +assert after_retirement["phase"]=="purging" +assert after_retirement["purge"]["progress"]["capacity_fence_retired"] is False +assert fence.split(":",1)[1] in retired_fences +assert set(resources)=={worktree_id,identity_id,container_scope} and len(roles)==3 +assert not any(item[0] in ("resource","role") for item in mutations[before_retirement:]) + +# The crashing invocation released control plus the one never-dispatched +# constituent once and never released the dispatched child. +capacity=[item[1] for item in mutations[before_retirement:] if item[0]=="capacity"] +assert capacity==[cell,root_two], capacity +assert root_one not in capacity + +# Retry consumes only the stored plan, presents the exact retirement receipt, +# and does not release any constituent twice. +before_retry=len(mutations); m.purge_retained(env,args()) +done=m.load_state(env,cell); assert done["phase"]=="purged" +assert done["purge"]["plan"]==sealed and done["purge"]["plan_digest"]==digest +capacity=[item[1] for item in mutations[before_retry:] if item[0]=="capacity"] +assert capacity==[], capacity +assert done["purge"]["progress"]["released_capacity"]==sorted([cell,root_two]) +assert done["purge"]["progress"]["capacity_fence_retired"] is True +assert delete_seam_attempts==["refused-retired-fence"] +retirements=[item for item in mutations if item[0]=="retire"] +assert retirements==[("retire",fence.split(":",1)[1])]*2 + +# The terminal tombstone is an idempotent no-op under the same exact locks. +before=copy.deepcopy(mutations); m.purge_retained(env,args()) +again=m.load_state(env,cell); assert mutations==before +assert again["phase"]=="purged" and again["purge"]["plan_digest"]==digest +PY + pass "retained purge admits only exact terminal cells, seals before mutation, resumes boundaries, and is idempotent" +} + multi_lane_queue_contract() { local tmp home out rc fm_test_tmproot_into tmp fm-azure-validation-lanes @@ -2092,6 +2557,7 @@ identity_and_recovery_contract trusted_manifest_verifier_contract shard_runner_integration_contract cleanup_recovery_contract +purge_retained_contract multi_lane_queue_contract operator_documentation_contract shard_receipt_demotion_contract diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index 998fc227669..192dd632f4e 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -62,7 +62,8 @@ for marker in ( "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", + "capacity-reserve", "capacity-reserve-shape", "capacity-release", "capacity-retire-fence", + "retired_capacity_fences", "merged_specialized_reservations", "command_withdraw", "command_surrender", "WORKER AUTHORITY REFUSED", "--confirm-discard-unlanded", "REVIEWED_CONTROL_SKU_FAMILY", "command_capacity_reserve_shape", @@ -2288,7 +2289,7 @@ PY } shared_specialized_cli() { - local tmp provider fixture home fence receipt out state_file + local tmp provider fixture home fence receipt retirement out state_file fm_test_tmproot_into tmp fm-shared-specialized provider="$tmp/provider.py" fixture="$tmp/provider-state.json" @@ -2297,6 +2298,7 @@ shared_specialized_cli() { write_fixture_provider "$provider" fence=$(printf reservation-fence | shasum -a 256 | awk '{print $1}') receipt=$(printf cleanup-proof | shasum -a 256 | awk '{print $1}') + retirement=$(printf fence-retirement | shasum -a 256 | awk '{print $1}') out=$(env \ FM_HOME="$home" \ FM_AZURE_SUBSCRIPTION_ID="$SUB" \ @@ -2425,6 +2427,110 @@ state = json.load(open(sys.argv[1])) assert state["capacity_reservations"]["azr-123456789abc"]["status"] == "released" assert state["capacity_reservations"]["azr-abcdef123456"]["status"] == "reserved" assert state["capacity_reservations"]["azr-fedcba654321"]["status"] == "queued" +PY + # The first retirement may encounter the released controller's v1 document. + # Migration must preserve every reservation and become rollback-safe at the + # same save that lands the retirement tombstone. + python3 - "$state_file" <<'PY' +import json +import sys +from pathlib import Path +path = Path(sys.argv[1]) +state = json.loads(path.read_text()) +state["schema"] = "fm.worker-lifecycle/v1" +state.pop("retired_capacity_fences", None) +path.write_text(json.dumps(state, sort_keys=True, separators=(",", ":"))) +PY + env \ + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID="$SUB" \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + FM_WORKER_PROVIDER_COMMAND="python3 $provider" \ + FIXTURE_STATE="$fixture" \ + "$WRAPPER" capacity-retire-fence \ + --fence-binding "$fence" \ + --reservation-id azr-123456789abc \ + --retirement-receipt "$retirement" \ + --confirm-subscription "$SUB" >/dev/null + if env \ + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID="$SUB" \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + FM_WORKER_PROVIDER_COMMAND="python3 $provider" \ + FIXTURE_STATE="$fixture" \ + "$WRAPPER" capacity-retire-fence \ + --fence-binding "$fence" \ + --reservation-id azr-123456789abc \ + --retirement-receipt "$(printf conflicting-retirement | shasum -a 256 | awk '{print $1}')" \ + --confirm-subscription "$SUB" >/dev/null 2>&1; then + fail "conflicting capacity fence retirement identity was accepted" + fi + # The durable retirement is idempotent, but neither reserve entry point may + # insert or re-admit on that fence after the shared lock opens. + env \ + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID="$SUB" \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + FM_WORKER_PROVIDER_COMMAND="python3 $provider" \ + FIXTURE_STATE="$fixture" \ + "$WRAPPER" capacity-retire-fence \ + --fence-binding "$fence" \ + --reservation-id azr-123456789abc \ + --retirement-receipt "$retirement" \ + --confirm-subscription "$SUB" >/dev/null + if env \ + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID="$SUB" \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + FM_WORKER_PROVIDER_COMMAND="python3 $provider" \ + FIXTURE_STATE="$fixture" \ + "$WRAPPER" capacity-reserve \ + --reservation-id azr-retired000001 \ + --fence-binding "$fence" \ + --role validation \ + --sku Standard_D4as_v7 \ + --sku-family StandardDasv7Family \ + --vcpus 4 \ + --amount-usd 25 \ + --confirm-subscription "$SUB" >/dev/null 2>&1; then + fail "single reservation entered a retired capacity fence" + fi + if env \ + FM_HOME="$home" \ + FM_AZURE_SUBSCRIPTION_ID="$SUB" \ + FM_AZURE_DEPLOYMENT_GENERATION=dep-one \ + FM_AZURE_OWNER_TAG=owner \ + FM_AZURE_NAMING_PREFIX=fmtest \ + FM_WORKER_PROVIDER_COMMAND="python3 $provider" \ + FIXTURE_STATE="$fixture" \ + "$WRAPPER" capacity-reserve-shape \ + --shape-id shape-retired \ + --fence-binding "$fence" \ + --constituent "reservation-id=azr-retired000002,role=validation,sku=Standard_D4as_v7,sku-family=StandardDasv7Family,vcpus=4,amount-usd=25" \ + --confirm-subscription "$SUB" >/dev/null 2>&1; then + fail "shape reservation entered a retired capacity fence" + fi + python3 - "$state_file" "$fence" "$retirement" <<'PY' \ + || fail "capacity fence retirement was not durable and exact" +import json +import sys +state = json.load(open(sys.argv[1])) +retirement = state["retired_capacity_fences"][sys.argv[2]] +assert state["schema"] == "fm.worker-lifecycle/v2" +assert retirement["reservation_ids"] == ["azr-123456789abc"] +assert retirement["retirement_receipt"] == sys.argv[3] +assert state["capacity_reservations"]["azr-abcdef123456"]["status"] == "reserved" +assert state["capacity_reservations"]["azr-fedcba654321"]["status"] == "queued" +assert "azr-retired000001" not in state["capacity_reservations"] +assert "azr-retired000002" not in state["capacity_reservations"] PY pass "specialized CLI durably reserves, queues exact-family excess, and releases only exact fenced capacity" }