perf(azure): observe in-flight worker readiness

bin/fm-azure-worker-provider.py

@@ -164,6 +164,14 @@
 # one, since how long it takes depends on how fast Azure answers. A read count
 # fails the same way every time.
 VM_POWER_MAX_READS = 90
+# ARM can return from the landed deployment while an exact managed child is
+# still moving from Updating to Succeeded. Re-submitting the whole deployment
+# in that state is both slower and more expensive than observing the resources
+# the action already owns. Keep the wait below the caller's existing admission
+# bound and cap both wall time and reads.
+CREATE_READY_TIMEOUT_SECONDS = 240
+CREATE_READY_POLL_SECONDS = 5
+CREATE_READY_MAX_READS = 32
 STEER_CLIENT_TIMEOUT_SECONDS = 600
 # The steer is bracketed by two full inventory sweeps.
 STEER_BUDGET_SECONDS = STEER_CLIENT_TIMEOUT_SECONDS + AZ_TIMEOUT_SECONDS * 2
@@ -172,6 +180,8 @@
     BOOTSTRAP_CLIENT_TIMEOUT_SECONDS
     + VM_START_TIMEOUT_SECONDS   # a converged worker is often stopped by its TTL
     + AZ_TIMEOUT_SECONDS * 16    # instance view, task stub, TTL, blob uploads, sweeps, tagging
+    + CREATE_READY_TIMEOUT_SECONDS
+    + AZ_TIMEOUT_SECONDS         # one exact readiness read may cross the wall deadline
 )
 RESOURCE_API = {
     "vm": "2024-03-01",
@@ -1616,6 +1626,57 @@ def recorded_exact(
     return resources
 
 
+def wait_exact_create_ready(controller, action, initial_worker):
+    """Observe an exact in-flight create instead of resubmitting its template."""
+    deadline = time.monotonic() + CREATE_READY_TIMEOUT_SECONDS
+    worker = initial_worker
+    reads = 0
+    while True:
+        resources = recorded_exact(
+            action, worker, require_ready_children=False
+        )
+        if set(resources) != set(REQUIRED_RESOURCE_KINDS):
+            raise ProviderError(
+                "worker create did not produce the complete exact resource set"
+            )
+        states = {
+            kind: str(resources[kind].get("provisioning_state", "")).lower()
+            for kind in READY_CHILD_KINDS
+        }
+        failed = sorted(
+            kind for kind, state in states.items()
+            if state in ("failed", "canceled")
+        )
+        if failed:
+            raise ProviderError(
+                "exact worker create child reached terminal provisioning state: {}"
+                .format(", ".join(
+                    "{}={}".format(kind, states[kind]) for kind in failed
+                ))
+            )
+        if all(state == "succeeded" for state in states.values()):
+            return worker
+        reads += 1
+        if reads >= CREATE_READY_MAX_READS or time.monotonic() >= deadline:
+            raise ProviderError(
+                "exact worker create readiness did not converge: {}".format(
+                    ", ".join(
+                        "{}={}".format(kind, states[kind] or "unknown")
+                        for kind in sorted(states)
+                    )
+                )
+            )
+        time.sleep(CREATE_READY_POLL_SECONDS)
+        snapshot = inventory_slot(controller, action["slot"])
+        if snapshot["conflicts"]:
+            raise ProviderError(
+                "in-flight worker inventory contains foreign or unsafe resources"
+            )
+        worker = worker_by_slot(snapshot, action["slot"])
+        if worker is None:
+            raise ProviderError("exact in-flight worker slot disappeared")
+
+
 def cleanup_recorded_exact(
     action, worker, allow_missing=(), skip_immutable=(), require_ready_children=True,
 ):
@@ -2620,10 +2681,7 @@ def converge_create_tags(controller, action):
     if snapshot["conflicts"]:
         raise ProviderError("tagged worker inventory contains foreign or unsafe resources")
     worker = worker_by_slot(snapshot, action["slot"])
-    resources = recorded_exact(action, worker)
-    if set(resources) != set(REQUIRED_RESOURCE_KINDS):
-        raise ProviderError("worker create did not produce the complete exact resource set")
-    return worker
+    return wait_exact_create_ready(controller, action, worker)
 
 
 def create_or_resume(controller, action):
@@ -2636,7 +2694,21 @@ def create_or_resume(controller, action):
         resources = existing.get("resources") or {}
         if resources.get("vm"):
             try:
-                recorded_exact(action, existing)
+                recorded_exact(action, existing, require_ready_children=False)
+            except ProviderError:
+                # A submitted create can be visible with the template's exact VM
+                # bindings before child tag convergence. It is safe to replay the
+                # same landed incremental deployment and complete the same action.
+                vm_tags = resources["vm"].get("tags") or {}
+                bindings = action["bindings"]
+                if not (
+                    vm_tags.get("home-binding") == bindings["home_binding"]
+                    and vm_tags.get("task-binding") == bindings["task"]
+                    and vm_tags.get("invocation-binding") == bindings["assignment_generation"]
+                ):
+                    raise ProviderError("visible worker belongs to another task or generation")
+            else:
+                existing = wait_exact_create_ready(controller, action, existing)
                 # A fully converged worker returns from here without ever
                 # reaching create_lifecycle_children, and the TTL schedule
                 # deallocates idle workers daily. Returning one as-is reports
@@ -2651,18 +2723,6 @@ def create_or_resume(controller, action):
                         inventory_slot(controller, action["slot"]), action["slot"]
                     )
                 return existing
-            except ProviderError:
-                # A submitted create can be visible with the template's exact VM
-                # bindings before child tag convergence. It is safe to replay the
-                # same landed incremental deployment and complete the same action.
-                vm_tags = resources["vm"].get("tags") or {}
-                bindings = action["bindings"]
-                if not (
-                    vm_tags.get("home-binding") == bindings["home_binding"]
-                    and vm_tags.get("task-binding") == bindings["task"]
-                    and vm_tags.get("invocation-binding") == bindings["assignment_generation"]
-                ):
-                    raise ProviderError("visible worker belongs to another task or generation")
         elif reuse:
             recorded_exact(
                 action, existing, allow_missing=(

tests/fm-azure-pilot.test.sh

@@ -1303,7 +1303,8 @@ provider.inventory = lambda controller, include_metrics=True, target_slot=None:
     "conflicts": [], "workers": [existing_worker],
 }
 provider.worker_by_slot = lambda snapshot, slot: existing_worker
-provider.recorded_exact = lambda action, existing: existing
+provider.recorded_exact = lambda action, existing, **kwargs: existing["resources"]
+provider.wait_exact_create_ready = lambda controller, action, existing: existing
 provider.expected_names = lambda controller, slot: {"vm": "vm-x"}
 provider.create_or_resume(controller, {"slot": 1, "bindings": {}})
 assert [c for c in calls if c[:2] == ["vm", "start"]], (

tests/fm-worker-lifecycle.test.sh

@@ -1651,6 +1651,39 @@ except partial_module.ProviderError as exc:
 else:
     raise AssertionError("foreign partial slot was inherited")
 
+# A replay that sees the complete, exactly bound resource set in Azure's
+# ordinary Updating -> Succeeded transition observes it in place. It must not
+# pay for or wait on a second full subscription deployment.
+ready_spec=importlib.util.spec_from_file_location("azure_provider_ready", sys.argv[1])
+ready_module=importlib.util.module_from_spec(ready_spec); ready_spec.loader.exec_module(ready_module)
+ready_action=copy.deepcopy(partial_action)
+ready_resources={kind:{
+    "id":"/{}".format(kind), "immutable_id":"i-{}".format(kind), "tags":{}
+} for kind in ready_module.REQUIRED_RESOURCE_KINDS}
+ready_worker={"slot":1,"resources":ready_resources}
+ready_calls=[]
+def ready_recorded(_action, worker, require_ready_children=True, **_kwargs):
+    state=worker["resources"]["bootstrap-command"]["provisioning_state"]
+    if require_ready_children and state != "succeeded":
+        raise ready_module.ProviderError("bootstrap-command provisioning state is not succeeded")
+    return worker["resources"]
+def ready_inventory(_controller, _slot):
+    ready_calls.append("read")
+    ready_worker["resources"]["bootstrap-command"]["provisioning_state"]="succeeded"
+    ready_worker["resources"]["monitor-extension"]["provisioning_state"]="succeeded"
+    return {"workers":[ready_worker],"conflicts":[],"capacity_reservations":[],"metrics":{}}
+ready_worker["resources"]["bootstrap-command"]["provisioning_state"]="updating"
+ready_worker["resources"]["monitor-extension"]["provisioning_state"]="succeeded"
+ready_module.recorded_exact=ready_recorded
+ready_module.inventory_slot=ready_inventory
+ready_module.worker_by_slot=lambda snapshot, slot: snapshot["workers"][0]
+ready_module.ensure_worker_running=lambda *_args: False
+ready_module.time.sleep=lambda _seconds: None
+ready_module.run_pilot_create=lambda *_args: (_ for _ in ()).throw(
+    AssertionError("transitional exact create resubmitted the deployment"))
+resumed=ready_module.create_or_resume({"prefix":"fmtest"}, ready_action)
+assert resumed is ready_worker and ready_calls == ["read"], (resumed, ready_calls)
+
 # A pre-convergence container has empty metadata; it inherits a same-slot
 # exact-fleet sibling's tags (VM first) instead of classifying as foreign,
 # while a bare orphan container keeps its emptiness and still refuses.