From 23f9c70134fb0c8151f5fe14ad3efe9ccc7af647 Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Tue, 25 Aug 2026 03:36:37 -0400 Subject: [PATCH] fix(runner): wait for shared Azure capacity --- bin/fm-azure-runner.py | 205 ++++++++++++++++++++----- bin/fm-crosscheck-azure-tool-bridge.py | 9 ++ docs/azure-runner.md | 2 +- tests/fm-azure-runner.test.sh | 135 ++++++++++++++-- tests/fm-crosscheck-azure.test.sh | 6 +- 5 files changed, 305 insertions(+), 52 deletions(-) diff --git a/bin/fm-azure-runner.py b/bin/fm-azure-runner.py index 1c7eb70700..3a1f148d34 100755 --- a/bin/fm-azure-runner.py +++ b/bin/fm-azure-runner.py @@ -65,6 +65,11 @@ MAX_BILLABLE_LIFETIME_HOURS = 24 STRICT_COST_ADMISSION_MODE = "strict" COMMISSIONING_COST_ADMISSION_MODE = "commissioning-bounded" +TRANSIENT_SHARED_CAPACITY_REFUSALS = frozenset({ + "exact selected-family observed-plus-reserved capacity is exhausted", + "specialized observed-plus-reserved demand exceeds its shared 40-vCPU shape", + "combined observed-plus-reserved demand would consume the shared East US ceiling", +}) TTL_SCHEDULE_HOURS_AFTER_PREPARATION = 23 AZURE_SCHEDULE_MINIMUM_LEAD_SECONDS = 30 * 60 SHELLCHECK_ARCHIVE_BYTES = 2_559_196 @@ -457,6 +462,26 @@ def environment(): max_concurrency = int(os.environ.get("FM_AZURE_RUNNER_MAX_CONCURRENCY", "4")) if max_concurrency < 1 or max_concurrency > 16: raise RunnerError("FM_AZURE_RUNNER_MAX_CONCURRENCY must be between 1 and 16") + try: + capacity_wait_seconds = int( + os.environ.get("FM_AZURE_RUNNER_CAPACITY_WAIT_SECONDS", "7200") + ) + capacity_poll_seconds = int( + os.environ.get("FM_AZURE_RUNNER_CAPACITY_POLL_SECONDS", "5") + ) + except ValueError: + raise RunnerError( + "FM_AZURE_RUNNER_CAPACITY_WAIT_SECONDS and " + "FM_AZURE_RUNNER_CAPACITY_POLL_SECONDS must be integers" + ) + if not 0 <= capacity_wait_seconds <= 86400: + raise RunnerError( + "FM_AZURE_RUNNER_CAPACITY_WAIT_SECONDS must be between 0 and 86400" + ) + if not 1 <= capacity_poll_seconds <= 60: + raise RunnerError( + "FM_AZURE_RUNNER_CAPACITY_POLL_SECONDS must be between 1 and 60" + ) budget_limit = int(os.environ.get("FM_AZURE_RUNNER_BUDGET_LIMIT_USD", "1000")) if budget_limit not in (1000, 1500): raise RunnerError("FM_AZURE_RUNNER_BUDGET_LIMIT_USD must be 1000 or 1500") @@ -484,6 +509,8 @@ def environment(): "deployment_generation": generation, "resource_group": resource_group, "max_concurrency": max_concurrency, + "capacity_wait_seconds": capacity_wait_seconds, + "capacity_poll_seconds": capacity_poll_seconds, "budget_limit": budget_limit, "cost_admission_mode": cost_admission_mode, "cell_ordinal": cell_ordinal, @@ -2098,40 +2125,107 @@ def shared_capacity_reserve(env, state, cost): "--amount-usd", str(cost["max_increment"]), "--confirm-subscription", env["subscription"], ] - result = run(command, env=shared_capacity_environment(env)) + previous = state.get("shared_capacity_reservation", {}) + wait_deadline_text = previous.get("wait_deadline") + if wait_deadline_text: + try: + wait_deadline = dt.datetime.fromisoformat( + wait_deadline_text.replace("Z", "+00:00") + ) + if wait_deadline.utcoffset() is None: + raise ValueError("capacity wait deadline has no timezone") + except (TypeError, ValueError): + if previous.get("status") == "queued": + shared_capacity_release(env, state) + raise RunnerError("shared allocator capacity wait deadline is malformed") + else: + wait_deadline = now_utc() + dt.timedelta( + seconds=env["capacity_wait_seconds"] + ) + wait_deadline_text = iso_utc(wait_deadline) + monotonic_deadline = time.monotonic() + max( + 0.0, (wait_deadline - now_utc()).total_seconds() + ) + last_reason = "capacity unavailable" + while True: + try: + result = run(command, env=shared_capacity_environment(env)) + except RunnerError: + if state.get("shared_capacity_reservation", {}).get("status") == "queued": + shared_capacity_release(env, state) + raise + try: + reservation = json.loads(result.stdout) + except (TypeError, json.JSONDecodeError): + if state.get("shared_capacity_reservation", {}).get("status") == "queued": + shared_capacity_release(env, state) + raise RunnerError("shared allocator returned a malformed capacity reservation") + if ( + not isinstance(reservation, dict) + or reservation.get("reservation_id") != state["invocation"] + or reservation.get("status") not in ("queued", "reserved") + ): + if state.get("shared_capacity_reservation", {}).get("status") == "queued": + shared_capacity_release(env, state) + raise RunnerError("shared allocator returned a reservation with the wrong identity") + previous = state.get("shared_capacity_reservation", {}) + last_reason = str(reservation.get("reason") or "capacity unavailable")[:500] + state["shared_capacity_reservation"] = { + "reservation_id": state["invocation"], + "fence_binding": fence, + "status": reservation["status"], + "amount_usd": cost["max_increment"], + "sku": limits["sku"], + "sku_family": limits["sku_family"], + "actual_usd": reservation.get("actual_usd"), + "forecast_usd": reservation.get("forecast_usd"), + "admission_limit_usd": reservation.get("admission_limit_usd"), + "reason": last_reason if reservation["status"] == "queued" else "", + "wait_deadline": wait_deadline_text, + "queued_at": previous.get("queued_at") or ( + iso_utc() if reservation["status"] == "queued" else None + ), + } + save_state(env, state) + if reservation["status"] == "reserved": + break + if last_reason not in TRANSIENT_SHARED_CAPACITY_REFUSALS: + shared_capacity_release(env, state) + raise RunnerError( + "shared allocator queued disposable-runner demand: {}".format( + last_reason + ) + ) + remaining = min( + monotonic_deadline - time.monotonic(), + (wait_deadline - now_utc()).total_seconds(), + ) + if remaining <= 0: + shared_capacity_release(env, state) + raise RunnerError( + "shared allocator capacity wait timed out after {} seconds: {}".format( + env["capacity_wait_seconds"], last_reason + ) + ) + time.sleep(min(env["capacity_poll_seconds"], remaining)) try: - reservation = json.loads(result.stdout) - except (TypeError, json.JSONDecodeError): - raise RunnerError("shared allocator returned a malformed capacity reservation") - if ( - not isinstance(reservation, dict) - or reservation.get("reservation_id") != state["invocation"] - or reservation.get("status") not in ("queued", "reserved") - ): - raise RunnerError("shared allocator returned a reservation with the wrong identity") - state["shared_capacity_reservation"] = { - "reservation_id": state["invocation"], - "fence_binding": fence, - "status": reservation["status"], - "amount_usd": cost["max_increment"], - "sku": limits["sku"], - "sku_family": limits["sku_family"], - "actual_usd": reservation.get("actual_usd"), - "forecast_usd": reservation.get("forecast_usd"), - "admission_limit_usd": reservation.get("admission_limit_usd"), - } - save_state(env, state) - if reservation["status"] != "reserved": - raise RunnerError("shared allocator queued disposable-runner demand: {}".format( - str(reservation.get("reason") or "capacity unavailable")[:500] - )) - cost["actual"] = reservation.get("actual_usd") - cost["forecast"] = reservation.get("forecast_usd") - cost["shared_admission_limit"] = reservation.get("admission_limit_usd") - if not isinstance(cost["actual"], (int, float)) or not isinstance(cost["forecast"], (int, float)): - raise RunnerError("shared allocator omitted readable actual or forecast spend evidence") - if max(float(cost["actual"]), float(cost["forecast"])) + float(cost["max_increment"]) >= env["budget_limit"]: - raise RunnerError("shared actual/forecast cost pressure reaches the runner admission limit") + cost["actual"] = reservation.get("actual_usd") + cost["forecast"] = reservation.get("forecast_usd") + cost["shared_admission_limit"] = reservation.get("admission_limit_usd") + if not isinstance(cost["actual"], (int, float)) or not isinstance(cost["forecast"], (int, float)): + raise RunnerError("shared allocator omitted readable actual or forecast spend evidence") + if max(float(cost["actual"]), float(cost["forecast"])) + float(cost["max_increment"]) >= env["budget_limit"]: + raise RunnerError("shared actual/forecast cost pressure reaches the runner admission limit") + except RunnerError as exc: + try: + shared_capacity_release(env, state) + except RunnerError as release_exc: + raise RunnerError( + "{}; shared capacity release also failed: {}".format( + exc, release_exc + ) + ) from release_exc + raise return cost @@ -2390,7 +2484,7 @@ def adopt_vm_identity(env, state, vm): save_state(env, state) -def create_vm(env, state): +def require_compute_deallocation_lead(state): deadline = dt.datetime.fromisoformat( state["request"]["compute_deallocation_deadline"].replace("Z", "+00:00") ) @@ -2401,6 +2495,10 @@ def create_vm(env, state): raise RunnerError( "control-plane TTL schedule has insufficient lead for Azure's 30-minute activation window and bounded deployment" ) + + +def create_vm(env, state): + require_compute_deallocation_lead(state) params = write_private_json(env, ".vm-params-", deployment_parameters(env, state)) try: az_command(env, [ @@ -2975,6 +3073,7 @@ def dispatch_prepared(env, state, confirm_subscription, confirm_cost_admission_m raise RunnerError("--confirm-cost-admission-mode is accepted only for commissioning-bounded") elif env.get("cell_ordinal") is not None: raise RunnerError("FM_AZURE_RUNNER_CELL_ORDINAL is accepted only for commissioning-bounded") + compute_create_attempted = False try: deadline = dt.datetime.fromisoformat( state["request"]["compute_deallocation_deadline"].replace("Z", "+00:00") @@ -2983,7 +3082,6 @@ def dispatch_prepared(env, state, confirm_subscription, confirm_cost_admission_m raise RunnerError("prepared invocation has insufficient time remaining before its control-plane deallocation deadline") scope_gate(env) limits = state["request"]["limits"] - sku_quota_gate(env, limits) validation_capacity_parent_gate(env, state) cost = ( commissioning_cost_gate(env, state, limits) @@ -3022,19 +3120,41 @@ def dispatch_prepared(env, state, confirm_subscription, confirm_cost_admission_m ) if mode == COMMISSIONING_COST_ADMISSION_MODE: commissioning_cost_gate(env, state, limits) - sku_quota_gate(env, limits) + sku_quota_gate(env, limits) foundation_gate(env) validation_capacity_parent_gate(env, state) # The ledger entry recorded above is the durable spend claim; VM # creation only materializes it, so concurrent transports create # their compute in parallel instead of holding the admission lock # through a multi-minute control-plane operation. + require_compute_deallocation_lead(state) + compute_create_attempted = True create_vm(env, state) create_run_command(env, state) poll_run_command(env, state) result = collect_result(env, state) cleanup(env, state) except Exception as exc: + if ( + not compute_create_attempted + and state.get("shared_capacity_reservation", {}).get("status") + in ("queued", "reserved") + ): + if state.get("reservation_recorded"): + spend_ledger_mark_cleaned(env, state) + try: + shared_capacity_release(env, state) + except RunnerError as release_exc: + combined = RunnerError( + "{}; pre-compute shared capacity release also failed: {}".format( + exc, release_exc + ) + ) + if state.get("phase") not in ( + "cleanup-retained", "complete", "absent-fenced" + ): + transition(env, state, "failed-retained", str(combined)[:500]) + raise combined from release_exc if state.get("phase") not in ("cleanup-retained", "complete", "absent-fenced"): transition(env, state, "failed-retained", str(exc)[:500]) raise @@ -3071,6 +3191,19 @@ def resume(env, state): print_logs_and_summary(state, state["result"]) return int(state["result"]["exit_code"]) scope_gate(env) + if ( + phase == "prepared" + and state.get("shared_capacity_reservation", {}).get("status") == "queued" + ): + mode = state["request"].get( + "cost_admission_mode", STRICT_COST_ADMISSION_MODE + ) + confirmation = ( + COMMISSIONING_COST_ADMISSION_MODE + if mode == COMMISSIONING_COST_ADMISSION_MODE + else None + ) + return dispatch_prepared(env, state, env["subscription"], confirmation) if phase in ("result-published", "failed-retained") and state.get("expected_result_digest"): result = collect_result(env, state) cleanup(env, state) diff --git a/bin/fm-crosscheck-azure-tool-bridge.py b/bin/fm-crosscheck-azure-tool-bridge.py index 0d826dee72..d4b6cc1202 100755 --- a/bin/fm-crosscheck-azure-tool-bridge.py +++ b/bin/fm-crosscheck-azure-tool-bridge.py @@ -14,6 +14,7 @@ import hashlib import importlib.util import json +import os from pathlib import Path import re import shlex @@ -210,6 +211,14 @@ def prepare_exact_snapshot( ) runner.normalize_command(arguments) env = runner.environment() + crosscheck_wait = int( + os.environ.get("FM_AZURE_CROSSCHECK_QUEUE_WAIT_SECONDS", "7200") + ) + if not 0 <= crosscheck_wait <= 86400: + raise BridgeError( + "FM_AZURE_CROSSCHECK_QUEUE_WAIT_SECONDS must be between 0 and 86400" + ) + env["capacity_wait_seconds"] = crosscheck_wait state = runner.prepare(env, arguments) repository = state["request"]["repository"] for field, expected in ( diff --git a/docs/azure-runner.md b/docs/azure-runner.md index 6a1a18c06c..ed18701885 100644 --- a/docs/azure-runner.md +++ b/docs/azure-runner.md @@ -175,7 +175,7 @@ Under the ARM CAS admission lease, every invocation gets a finite positive compl The strict runner-local gate remains mandatory whenever the operator has not explicitly selected commissioning mode, and it is intentionally defense in depth rather than a separate capacity owner. Both modes acquire one exact reservation from the durable allocator in [Elastic task workers](azure-workers.md) before any runner management reservation or VM creation. That allocator merges author assignments and disposable-runner reservations into one observed-plus-reserved 128-vCPU East US and exact-family schedule, and applies shared readable actual/forecast spend plus all durable reservation amounts even during commissioning. -A queued shared reservation stops the runner before compute; cleanup releases it only after exact VM/NIC/OS-disk absence, and ambiguity retains it. +A queued shared reservation caused by exact-family or regional capacity pressure waits with the same durable reservation identity for up to `FM_AZURE_RUNNER_CAPACITY_WAIT_SECONDS` (7200 seconds by default, 86400 maximum). Crosscheck evidence runners inherit the Crosscheck queue deadline. Budget, telemetry, identity, and other non-capacity refusals remain immediate. Timeout or any other pre-compute refusal releases the queued reservation after exact zero-compute proof; cleanup releases an admitted reservation only after exact VM/NIC/OS-disk absence, and ambiguity retains it. The normal budget limit is the active $1,000 target. An operator may select the commissioning ceiling of $1,500 through `FM_AZURE_RUNNER_BUDGET_LIMIT_USD=1500` only during the approved commissioning window. diff --git a/tests/fm-azure-runner.test.sh b/tests/fm-azure-runner.test.sh index 951f79e3ba..7d5ab99a19 100755 --- a/tests/fm-azure-runner.test.sh +++ b/tests/fm-azure-runner.test.sh @@ -44,6 +44,8 @@ import importlib.util,os,sys spec=importlib.util.spec_from_file_location("runner",sys.argv[1]); m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m) assert m.environment()["cost_admission_mode"]=="strict" assert m.environment()["operator_data_plane_ip"]=="" +assert m.environment()["capacity_wait_seconds"]==7200 +assert m.environment()["capacity_poll_seconds"]==5 os.environ["FM_AZURE_OPERATOR_DATA_PLANE_IP"]="203.0.113.10/32" assert m.environment()["operator_data_plane_ip"]=="203.0.113.10/32" os.environ.pop("FM_AZURE_OPERATOR_DATA_PLANE_IP") @@ -60,6 +62,16 @@ os.environ["FM_AZURE_RUNNER_MAX_CONCURRENCY"]="17" try: m.environment() except m.RunnerError: pass else: raise AssertionError("environment accepted concurrency above 16") +os.environ["FM_AZURE_RUNNER_MAX_CONCURRENCY"]="4" +for name,value in ( + ("FM_AZURE_RUNNER_CAPACITY_WAIT_SECONDS","86401"), + ("FM_AZURE_RUNNER_CAPACITY_POLL_SECONDS","0"), +): + os.environ[name]=value + try: m.environment() + except m.RunnerError: pass + else: raise AssertionError("environment accepted invalid "+name) + os.environ.pop(name) PY rm -rf "$home" pass "normal environment defaults to strict without commissioning evidence or confirmation variables" @@ -682,49 +694,146 @@ PY2 shared_allocator_bridge_unit() { python3 - "$HOST" <<'PY' || fail "shared allocator runner bridge failed" -import importlib.util, json, pathlib, subprocess, tempfile, sys +import contextlib, importlib.util, json, pathlib, subprocess, tempfile, sys spec=importlib.util.spec_from_file_location("runner_shared",sys.argv[1]); m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m) fixture_home=pathlib.Path(tempfile.mkdtemp()) -env={"subscription":"sub","resource_group":"rg","prefix":"prefix","budget_limit":1500,"state_dir":pathlib.Path(tempfile.mkdtemp()),"azure_operation_count":0} +env={"subscription":"sub","resource_group":"rg","prefix":"prefix","budget_limit":1500,"state_dir":pathlib.Path(tempfile.mkdtemp()),"azure_operation_count":0,"capacity_wait_seconds":10,"capacity_poll_seconds":1} limits={**m.RESOURCE_CLASSES["behavior-heavy"],"sku":"Standard_D4as_v7","sku_family":"StandardDasv7Family"} state={"schema":m.SCHEMA,"invocation":"azr-aaaaaaaaaaaa","resources":{},"request":{"fence":"sha256:"+"a"*64,"resource_class":"behavior-heavy","limits":limits}} m.save_state=lambda *_a,**_k:None calls=[] def completed(value): - return subprocess.CompletedProcess(["python"],0,stdout=json.dumps(value),stderr="") + stdout=value if isinstance(value,str) else json.dumps(value) + return subprocess.CompletedProcess(["python"],0,stdout=stdout,stderr="") +def response(status="reserved",reason="",actual=100.0,forecast=200.0): + return {"reservation_id":"azr-aaaaaaaaaaaa","status":status,"reason":reason,"actual_usd":actual,"forecast_usd":forecast,"admission_limit_usd":1500.0} +sequence=[ + response("queued","exact selected-family observed-plus-reserved capacity is exhausted"), + response(), +] +sleeps=[] def allocator_run(command,**kwargs): assert kwargs["env"]["FM_HOME"]==str(fixture_home) assert kwargs["env"]["FM_AZURE_WORKER_STATE_DIR"]==str((fixture_home/"state"/"azure-workers").resolve()) calls.append(command) - return completed({"reservation_id":"azr-aaaaaaaaaaaa","status":"reserved","reason":"","actual_usd":100.0,"forecast_usd":200.0,"admission_limit_usd":1500.0}) + if "capacity-release" in command:return completed({}) + return completed(sequence.pop(0)) old_home=m.os.environ.get("FM_HOME") m.os.environ["FM_HOME"]=str(fixture_home) m.run=allocator_run +m.time.sleep=lambda seconds:sleeps.append(seconds) cost={"max_increment":25.0} result=m.shared_capacity_reserve(env,state,cost) assert result["actual"]==100.0 and result["forecast"]==200.0 -assert "capacity-reserve" in calls[-1] and calls[-1][calls[-1].index("--sku-family")+1]=="StandardDasv7Family" +reserve_calls=[command for command in calls if "capacity-reserve" in command] +assert len(reserve_calls)==2 and sleeps +assert all(command[command.index("--reservation-id")+1]=="azr-aaaaaaaaaaaa" for command in reserve_calls) +assert len({command[command.index("--fence-binding")+1] for command in reserve_calls})==1 +assert reserve_calls[-1][reserve_calls[-1].index("--sku-family")+1]=="StandardDasv7Family" assert state["shared_capacity_reservation"]["status"]=="reserved" -# Queue, unreadable telemetry, and shared actual/forecast pressure each refuse before compute. -for payload,text in ( - ({"reservation_id":"azr-aaaaaaaaaaaa","status":"queued","reason":"family full","actual_usd":100.0,"forecast_usd":200.0,"admission_limit_usd":1500.0},"queued"), - ({"reservation_id":"azr-aaaaaaaaaaaa","status":"reserved","reason":"","actual_usd":None,"forecast_usd":200.0,"admission_limit_usd":1500.0},"readable"), - ({"reservation_id":"azr-aaaaaaaaaaaa","status":"reserved","reason":"","actual_usd":100.0,"forecast_usd":1490.0,"admission_limit_usd":1500.0},"pressure"), -): - m.run=lambda *_a,payload=payload,**_k:completed(payload) +# A non-capacity queue refusal is immediate and releases its exact row. +state.pop("shared_capacity_reservation") +calls.clear() +def non_capacity(command,**_kwargs): + calls.append(command) + if "capacity-release" in command:return completed({}) + return completed(response("queued","shared actual or forecast spend is unreadable")) +m.run=non_capacity +try:m.shared_capacity_reserve(env,state,{"max_increment":25.0}) +except m.RunnerError as exc:assert "shared actual or forecast" in str(exc) +else:raise AssertionError("non-capacity allocator refusal entered the wait loop") +assert state["shared_capacity_reservation"]["status"]=="released" +assert sum("capacity-release" in command for command in calls)==1 +# A zero-second bounded wait releases a transient queue row before failing. +state.pop("shared_capacity_reservation") +calls.clear() +def always_queued(command,**_kwargs): + calls.append(command) + if "capacity-release" in command:return completed({}) + return completed(response("queued","exact selected-family observed-plus-reserved capacity is exhausted")) +m.run=always_queued +try:m.shared_capacity_reserve({**env,"capacity_wait_seconds":0},state,{"max_increment":25.0}) +except m.RunnerError as exc:assert "timed out after 0 seconds" in str(exc) +else:raise AssertionError("transient capacity timeout was bypassed") +assert state["shared_capacity_reservation"]["status"]=="released" +assert sum("capacity-release" in command for command in calls)==1 +# A malformed retry response after a queue also releases before failing. +state.pop("shared_capacity_reservation") +calls.clear() +sequence=[ + response("queued","exact selected-family observed-plus-reserved capacity is exhausted"), + "not-json", +] +def malformed_after_queue(command,**_kwargs): + calls.append(command) + if "capacity-release" in command:return completed({}) + return completed(sequence.pop(0)) +m.run=malformed_after_queue +try:m.shared_capacity_reserve(env,state,{"max_increment":25.0}) +except m.RunnerError as exc:assert "malformed" in str(exc) +else:raise AssertionError("malformed allocator retry was bypassed") +assert state["shared_capacity_reservation"]["status"]=="released" +# Unreadable telemetry and shared actual/forecast pressure release admitted capacity. +for payload,text in ((response(actual=None),"readable"),(response(forecast=1490.0),"pressure")): + state.pop("shared_capacity_reservation") + calls.clear() + def reserved_then_release(command,**_kwargs): + calls.append(command) + return completed({} if "capacity-release" in command else payload) + m.run=reserved_then_release try:m.shared_capacity_reserve(env,state,{"max_increment":25.0}) except m.RunnerError as exc:assert text in str(exc) else:raise AssertionError("shared allocator failure was bypassed: "+text) + assert state["shared_capacity_reservation"]["status"]=="released" # Release is sent only from the post-cleanup bridge with one digest-bound receipt. m.run=lambda command,**_kwargs:calls.append(command) or completed({}) state["shared_capacity_reservation"]={"status":"reserved"} m.shared_capacity_release(env,state) assert "capacity-release" in calls[-1] and len(calls[-1][calls[-1].index("--cleanup-receipt")+1])==64 assert state["shared_capacity_reservation"]["status"]=="released" +# A refusal after shared admission but before VM creation releases rather than +# retaining an idle reservation behind the local concurrency safety cap. +precompute={ + "phase":"prepared","invocation":"azr-aaaaaaaaaaaa","parent_invocation":None, + "request":{ + "fence":"sha256:"+"a"*64, + "lineage_root_invocation":"azr-aaaaaaaaaaaa", + "cost_admission_mode":m.STRICT_COST_ADMISSION_MODE, + "compute_deallocation_deadline":m.iso_utc(m.now_utc()+m.dt.timedelta(hours=1)), + "limits":limits, + }, +} +dispatch_env={**env,"cost_admission_mode":m.STRICT_COST_ADMISSION_MODE,"max_concurrency":4,"cell_ordinal":None} +released=[] +m.bind_operation_context=lambda *_a:None +m.scope_gate=lambda *_a:None +m.validation_capacity_parent_gate=lambda *_a:None +m.budget_gate=lambda *_a,**_k:{"max_increment":25.0} +m.foundation_gate=lambda *_a:None +def reserve(_env,got_state,cost): + got_state["shared_capacity_reservation"]={"status":"reserved"} + return cost +m.shared_capacity_reserve=reserve +m.reprove_public_request=lambda *_a:None +m.stage_private_snapshot=lambda *_a:None +m.admission_lock=lambda *_a:contextlib.nullcontext() +m.active_runner_vms=lambda *_a:[{}]*4 +m.shared_capacity_release=lambda _env,got_state:released.append(got_state["invocation"]) or got_state["shared_capacity_reservation"].update(status="released") +m.transition=lambda _env,got_state,phase,note=None,**updates:got_state.update({"phase":phase,**updates}) +try:m.dispatch_prepared(dispatch_env,precompute,"sub",None) +except m.RunnerError as exc:assert "bounded concurrency limit" in str(exc) +else:raise AssertionError("pre-compute local concurrency refusal was bypassed") +assert released==["azr-aaaaaaaaaaaa"] and precompute["phase"]=="failed-retained" +# A restarted command re-enters the same queued invocation instead of fencing it. +queued={"phase":"prepared","invocation":"azr-aaaaaaaaaaaa","shared_capacity_reservation":{"status":"queued"},"request":{"cost_admission_mode":m.STRICT_COST_ADMISSION_MODE}} +seen=[] +m.dispatch_prepared=lambda got_env,got_state,subscription,confirmation:seen.append((got_state["invocation"],subscription,confirmation)) or 7 +assert m.resume({"subscription":"sub"},queued)==7 +assert seen==[("azr-aaaaaaaaaaaa","sub",None)] if old_home is None:m.os.environ.pop("FM_HOME",None) else:m.os.environ["FM_HOME"]=old_home PY - pass "runner queues behind the shared allocator and requires actual/forecast evidence before compute" + pass "runner durably waits only on transient shared capacity and releases every pre-compute refusal" } commissioning_admission_unit() { diff --git a/tests/fm-crosscheck-azure.test.sh b/tests/fm-crosscheck-azure.test.sh index dbb33d4f24..765a0d8869 100755 --- a/tests/fm-crosscheck-azure.test.sh +++ b/tests/fm-crosscheck-azure.test.sh @@ -1839,6 +1839,7 @@ PY bridge_private_snapshot_unit() { python3 - "$BRIDGE" <<'PY' || fail "Azure bridge private snapshot contract failed" import importlib.util +import os from pathlib import Path import subprocess import sys @@ -1867,11 +1868,12 @@ with tempfile.TemporaryDirectory() as temporary: ).stdout.strip() observed = {} runner.environment = lambda: {"fixture": True} + os.environ["FM_AZURE_CROSSCHECK_QUEUE_WAIT_SECONDS"] = "17" def prepare(env, arguments): bundle = Path(arguments.private_snapshot_bundle) observed["bundle"] = bundle - assert env == {"fixture": True} + assert env == {"fixture": True, "capacity_wait_seconds": 17} assert arguments.public_ref is None assert arguments.source_ref == "refs/pull/7/head" assert arguments.capacity_parent is None @@ -1905,7 +1907,7 @@ with tempfile.TemporaryDirectory() as temporary: ) assert state["request"]["repository"]["source_head"] == head assert arguments.private_snapshot_bundle - assert env == {"fixture": True} + assert env == {"fixture": True, "capacity_wait_seconds": 17} assert not observed["bundle"].exists() PY pass "Azure evidence bridge privately bundles an exact detached PR checkout without GitHub credentials"