perf(azure): eliminate fleet lists from slot reads

bin/fm-azure-worker-provider.py

@@ -484,11 +484,52 @@ def resource_record(kind, value, power_state=None, tags_override=None):
 
 def azure_resource_not_found(stderr):
     return bool(re.search(
-        r"(?:\(ResourceNotFound\)|^Code:\s*ResourceNotFound\s*$)",
+        r"(?:\((?:ResourceNotFound|ParentResourceNotFound|ContainerNotFound)\)"
+        r"|^(?:Code|ErrorCode):\s*"
+        r"(?:ResourceNotFound|ParentResourceNotFound|ContainerNotFound)\s*$)",
         str(stderr or ""), re.MULTILINE,
     ))
 
 
+def show_exact_optional(controller, args):
+    """Read one exact Azure object; only a recognized absence becomes None."""
+    value, rc, stderr = az(controller, args, check=False)
+    if rc != 0:
+        if azure_resource_not_found(stderr):
+            return None
+        raise ProviderError("exact Azure inventory read failed or was malformed: {}".format(stderr))
+    if not isinstance(value, dict):
+        raise ProviderError("exact Azure inventory read failed or was malformed: {}".format(stderr))
+    return value
+
+
+def show_exact_optional_many(controller, operations):
+    """Read independent exact objects concurrently, preserving error order."""
+    if not operations:
+        return {}
+    keys = [key for key, _ in operations]
+    if len(keys) != len(set(keys)):
+        raise ProviderError("exact Azure inventory operation keys are not unique")
+    with concurrent.futures.ThreadPoolExecutor(max_workers=min(4, len(operations))) as executor:
+        futures = {
+            key: executor.submit(show_exact_optional, controller, args)
+            for key, args in operations
+        }
+        results = {}
+        failures = []
+        for key in keys:
+            try:
+                results[key] = futures[key].result()
+            except Exception as exc:
+                failures.append((key, exc))
+        if failures:
+            key, exc = failures[0]
+            if isinstance(exc, ProviderError):
+                raise ProviderError("{}: {}".format(key, exc))
+            raise ProviderError("{}: exact Azure inventory read failed: {}".format(key, exc))
+    return results
+
+
 def show_full(controller, resource_id, api_version=None, inventory_missing_ok=False):
     args = ["resource", "show", "--ids", resource_id]
     if api_version:
@@ -1098,37 +1139,116 @@ def inventory(controller, include_metrics=True, target_slot=None):
         raise ProviderError("Azure subscription scope is not the exact enabled controller binding")
 
     prefix = re.escape(controller["prefix"])
-    vms = list_json(
-        controller,
-        ["vm", "list", "--resource-group", controller["resource_group"], "--show-details"],
-        transient_not_found_attempts=4,
-    )
-    nics = list_json(controller, ["network", "nic", "list", "--resource-group", controller["resource_group"]])
-    disks = list_json(controller, ["disk", "list", "--resource-group", controller["resource_group"]])
-    identities = list_json(controller, ["identity", "list", "--resource-group", controller["resource_group"]])
-    extensions = list_json(controller, [
-        "resource", "list", "--resource-group", controller["resource_group"],
-        "--resource-type", "Microsoft.Compute/virtualMachines/extensions",
-    ])
-    run_commands = list_json(controller, [
-        "resource", "list", "--resource-group", controller["resource_group"],
-        "--resource-type", "Microsoft.Compute/virtualMachines/runCommands",
-    ])
-    schedules = list_json(controller, [
-        "resource", "list", "--resource-group", controller["resource_group"],
-        "--resource-type", "Microsoft.DevTestLab/schedules",
-    ])
-    # Scope casing varies across ARM responses, so filter client-side rather
-    # than with a case-sensitive JMESPath query.
-    scope_marker = "/resourcegroups/{}/".format(controller["resource_group"]).lower()
-    roles = [
-        role for role in list_json(controller, ["role", "assignment", "list", "--all"])
-        if scope_marker in str(role.get("scope") or "").lower()
-    ]
-    containers = list_json(controller, [
-        "storage", "container", "list", "--auth-mode", "login", "--include-metadata",
-        "--account-name", os.environ.get("FM_AZURE_STORAGE_NAME", ""),
-    ])
+    if target_slot is None:
+        vms = list_json(
+            controller,
+            ["vm", "list", "--resource-group", controller["resource_group"], "--show-details"],
+            transient_not_found_attempts=4,
+        )
+        nics = list_json(controller, ["network", "nic", "list", "--resource-group", controller["resource_group"]])
+        disks = list_json(controller, ["disk", "list", "--resource-group", controller["resource_group"]])
+        identities = list_json(controller, ["identity", "list", "--resource-group", controller["resource_group"]])
+        extensions = list_json(controller, [
+            "resource", "list", "--resource-group", controller["resource_group"],
+            "--resource-type", "Microsoft.Compute/virtualMachines/extensions",
+        ])
+        run_commands = list_json(controller, [
+            "resource", "list", "--resource-group", controller["resource_group"],
+            "--resource-type", "Microsoft.Compute/virtualMachines/runCommands",
+        ])
+        schedules = list_json(controller, [
+            "resource", "list", "--resource-group", controller["resource_group"],
+            "--resource-type", "Microsoft.DevTestLab/schedules",
+        ])
+        # Scope casing varies across ARM responses, so filter client-side rather
+        # than with a case-sensitive JMESPath query.
+        scope_marker = "/resourcegroups/{}/".format(controller["resource_group"]).lower()
+        roles = [
+            role for role in list_json(controller, ["role", "assignment", "list", "--all"])
+            if scope_marker in str(role.get("scope") or "").lower()
+        ]
+        containers = list_json(controller, [
+            "storage", "container", "list", "--auth-mode", "login", "--include-metadata",
+            "--account-name", os.environ.get("FM_AZURE_STORAGE_NAME", ""),
+        ])
+    else:
+        names = expected_names(controller, target_slot)
+        resource_group = controller["resource_group"]
+        storage = os.environ.get("FM_AZURE_STORAGE_NAME", "")
+
+        vm_id = exact_id(controller, "Microsoft.Compute", "virtualMachines", names["vm"])
+        exact = show_exact_optional_many(controller, [
+            ("vm", [
+                "vm", "show", "--resource-group", resource_group,
+                "--name", names["vm"], "--show-details",
+            ]),
+            ("nic", [
+                "network", "nic", "show", "--resource-group", resource_group,
+                "--name", names["nic"],
+            ]),
+            ("os-disk", [
+                "disk", "show", "--resource-group", resource_group,
+                "--name", names["os-disk"],
+            ]),
+            ("task-disk", [
+                "disk", "show", "--resource-group", resource_group,
+                "--name", names["task-disk"],
+            ]),
+            ("account-disk", [
+                "disk", "show", "--resource-group", resource_group,
+                "--name", names["account-disk"],
+            ]),
+            ("identity", [
+                "identity", "show", "--resource-group", resource_group,
+                "--name", names["identity"],
+            ]),
+            ("monitor-extension", [
+                "resource", "show", "--ids",
+                vm_id + "/extensions/" + names["monitor-extension"],
+            ]),
+            ("bootstrap-command", [
+                "resource", "show", "--ids",
+                vm_id + "/runCommands/" + names["bootstrap-command"],
+            ]),
+            ("task-command", [
+                "resource", "show", "--ids",
+                vm_id + "/runCommands/" + names["task-command"],
+            ]),
+            ("ttl-schedule", [
+                "resource", "show", "--ids", exact_id(
+                    controller, "Microsoft.DevTestLab", "schedules", names["ttl-schedule"],
+                ), "--api-version", "2018-09-15",
+            ]),
+            ("state-container", [
+                "storage", "container", "show", "--auth-mode", "login",
+                "--account-name", storage, "--name", names["state-container"],
+            ]),
+        ])
+
+        def present(kind):
+            return [exact[kind]] if exact[kind] is not None else []
+
+        vms = present("vm")
+        nics = present("nic")
+        disks = [
+            exact[kind] for kind in ("os-disk", "task-disk", "account-disk")
+            if exact[kind] is not None
+        ]
+        identities = present("identity")
+        extensions = present("monitor-extension")
+        run_commands = [
+            exact[kind] for kind in ("bootstrap-command", "task-command")
+            if exact[kind] is not None
+        ]
+        schedules = present("ttl-schedule")
+        containers = present("state-container")
+        container_scope = (
+            exact_id(controller, "Microsoft.Storage", "storageAccounts", storage)
+            + "/blobServices/default/containers/" + names["state-container"]
+        )
+        roles = list_json(controller, [
+            "role", "assignment", "list", "--scope", container_scope,
+        ]) if containers else []
 
     workers = {}
     conflicts = []
@@ -1234,7 +1354,9 @@ def add(kind, value, slot, power=None, tags_override=None):
         )
         # The generic resource listing omits properties and etag; only the full
         # object carries an immutable child identity.
-        value = show_full(controller, extension["id"], inventory_missing_ok=True)
+        value = dict(extension) if target_slot is not None else show_full(
+            controller, extension["id"], inventory_missing_ok=True,
+        )
         if value is None:
             continue
         value["attached_to"] = vm_id
@@ -1250,7 +1372,9 @@ def add(kind, value, slot, power=None, tags_override=None):
         if kind is None:
             conflicts.append({"kind": "run-command", "slot": slot, "reason": "undeclared worker Run Command child"})
             continue
-        value = show_full(controller, command["id"], inventory_missing_ok=True)
+        value = dict(command) if target_slot is not None else show_full(
+            controller, command["id"], inventory_missing_ok=True,
+        )
         if value is None:
             continue
         value["attached_to"] = exact_id(
@@ -1263,7 +1387,7 @@ def add(kind, value, slot, power=None, tags_override=None):
     for schedule in schedules:
         slot = slot_from_name(schedule.get("name"), r"^shutdown-computevm-vm-{}-wkr-".format(prefix))
         if slot is not None and (target_slot is None or slot == target_slot):
-            value = show_full(
+            value = dict(schedule) if target_slot is not None else show_full(
                 controller, schedule["id"], api_version="2018-09-15",
                 inventory_missing_ok=True,
             )

docs/azure-workers.md

@@ -404,7 +404,7 @@ The guest verifies the runtime's exact file inventory, runs the role command wit
 The wrapper verifies the semantic bytes and head binding before writing the controller-facing result; a process exit, missing outcome, malformed outcome, or changed read-only head is a failed result, never `CLEAR` by inference.
 Repair results return one digest-bound single-ref bundle whose head must descend from the requested head, while review and test return no code bundle and must keep the exact requested head.
 The wrapper records a retryable local candidate before cleanup, releases through `service-complete` only after the lifecycle owns the exact execution result, and replays the candidate after a lost response instead of executing the step again.
-Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory expands only that exact slot's Azure children; queued admission retains the whole-fleet quota, spend, and conflict census.
+Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory reads that slot's deterministic Azure object paths in a bounded parallel batch; it neither lists the resource group nor expands peer children. Queued admission retains the whole-fleet quota, spend, and conflict census.
 The guest supervisor marks a no-mistakes Azure execution as the already-isolated test boundary, so the repository test command runs the focused service suite directly instead of recursively provisioning the general validation fleet or a Herdr lab.
 The root-owned supervisor stages the job, then runs the no-mistakes process as the dedicated non-root `fmworker` user with no supplementary groups; the sealed runtime remains root-owned and read-only while the exact repository and projected account are writable only by that service identity.
 `bin/fm-azure-service-test-scope.py` owns that focused inventory and the narrow source set eligible for focused pull-request CI; an empty, mixed, or unknown diff and every push to `main` retain the complete behavior suite.

tests/fm-worker-lifecycle.test.sh

@@ -9223,40 +9223,45 @@ controller = {
 tags = {
     "workload": "firstmate", "deployment-generation": "dep", "cleanup-owner": "owner",
 }
-vms = [
-    {"name": "vm-fixture-wkr-01", "id": "/vm/1", "vmId": "vm-id-1",
-     "powerState": "VM running", "tags": tags},
-    {"name": "vm-fixture-wkr-02", "id": "/vm/2", "vmId": "vm-id-2",
-     "powerState": "VM running", "tags": tags},
-]
-extensions = [
-    {"name": "vm-fixture-wkr-01/AzureMonitorLinuxAgent", "id": "/vm/1/ext"},
-    {"name": "vm-fixture-wkr-02/AzureMonitorLinuxAgent", "id": "/vm/2/ext"},
-]
-expanded = []
-
-provider.az = lambda _controller, args, check=False, timeout=provider.AZ_TIMEOUT_SECONDS: (
-    ({"id": controller["subscription"], "state": "Enabled"}, 0, "")
-    if args[:2] == ["account", "show"] else (None, 1, "unexpected")
+calls = []
+vm_id = provider.exact_id(
+    controller, "Microsoft.Compute", "virtualMachines", "vm-fixture-wkr-02",
+)
+extension_id = vm_id + "/extensions/AzureMonitorLinuxAgent"
+
+def exact_read(_controller, args, check=False, timeout=provider.AZ_TIMEOUT_SECONDS):
+    calls.append(tuple(args))
+    if args[:2] == ["account", "show"]:
+        return {"id": controller["subscription"], "state": "Enabled"}, 0, ""
+    if args[:2] == ["vm", "show"]:
+        assert args[args.index("--name") + 1] == "vm-fixture-wkr-02", args
+        return {
+            "name": "vm-fixture-wkr-02", "id": vm_id, "vmId": "vm-id-2",
+            "powerState": "VM running", "tags": tags, "identity": {},
+        }, 0, ""
+    if args[:3] == ["resource", "show", "--ids"] and args[3] == extension_id:
+        return {
+            "name": "vm-fixture-wkr-02/AzureMonitorLinuxAgent",
+            "id": extension_id, "tags": tags,
+            "properties": {"provisioningState": "Succeeded"},
+        }, 0, ""
+    if args[:3] == ["storage", "container", "show"]:
+        return None, 1, "ErrorCode:ContainerNotFound"
+    return None, 1, "(ResourceNotFound) exact fixture absence"
+
+provider.az = exact_read
+provider.list_json = lambda *_args, **_kwargs: (_ for _ in ()).throw(
+    AssertionError("targeted inventory issued a list operation")
 )
-def listing(_controller, args, transient_not_found_attempts=1):
-    if args[:2] == ["vm", "list"]:
-        return vms
-    if args[:2] == ["resource", "list"] and args[-1] == "Microsoft.Compute/virtualMachines/extensions":
-        return extensions
-    return []
-provider.list_json = listing
-def show(_controller, resource_id, api_version=None, inventory_missing_ok=False):
-    expanded.append(resource_id)
-    return {
-        "id": resource_id, "tags": tags,
-        "properties": {"provisioningState": "Succeeded"},
-    }
-provider.show_full = show
 
 snapshot = provider.inventory_slot(controller, 2)
 assert [worker["slot"] for worker in snapshot["workers"]] == [2], snapshot
-assert expanded == ["/vm/2/ext"], expanded
+assert snapshot["workers"][0]["resources"]["monitor-extension"]["id"] == extension_id
+assert all("wkr-01" not in " ".join(call) for call in calls), calls
+assert not any("list" in call[:3] for call in calls), calls
+assert any(call[:2] == ("vm", "show") for call in calls), calls
+assert any(call[:2] == ("network", "nic") for call in calls), calls
+assert sum(call[:2] == ("disk", "show") for call in calls) == 3, calls
 assert snapshot["capacity_reservations"] == []
 assert snapshot["metrics"]["actual_usd"] is None
 try: