perf(delivery): overlap bounded readiness queries

backend/tests/test_delivery_validation_work.py

@@ -189,3 +189,67 @@ def test_iterator_failure_joins_in_flight_work_without_dispatching_rest():
         collector.join(5)
     assert not collector.is_alive() and calls == [0]
     assert len(errors) == 1 and str(errors[0]) == "finite inventory refusal"
+
+
+def test_readiness_projection_ports_overlap_with_at_most_two_workers(monkeypatch):
+    """Live ordinary threads over modeled ports, not systemd or VM acceptance."""
+    node_health = importlib.import_module("scripts.release.node_health")
+    lock = threading.Lock()
+    barrier = threading.Barrier(2, timeout=10)
+    active = 0
+    peak = 0
+    started = []
+    finished = []
+    payload = Path("/ordinary-readiness-port-model")
+
+    def query(name, actual_payload):
+        nonlocal active, peak
+        assert actual_payload == payload
+        with lock:
+            active += 1
+            peak = max(peak, active)
+            started.append(name)
+        try:
+            barrier.wait()
+            return {"component": name}
+        finally:
+            with lock:
+                active -= 1
+                finished.append(name)
+
+    monkeypatch.setattr(node_health.operator_units, "run", query)
+    with node_health._status_helpers(payload) as pending:
+        results = {name: future.result(timeout=15) for name, future in pending.items()}
+    names = {"broker-health", "outbox-health", "import-health", "factory-health"}
+    assert peak == 2 and active == 0
+    assert set(started) == set(finished) == set(results) == names
+    assert len(started) == len(finished) == 4
+    assert all(results[name] == {"component": name} for name in names)
+    assert all(future.done() for future in pending.values())
+
+
+def test_readiness_projection_failure_keeps_error_and_joins_all_workers(monkeypatch):
+    """A consumer error cannot abandon a dispatched ordinary helper model."""
+    node_health = importlib.import_module("scripts.release.node_health")
+    lock = threading.Lock()
+    barrier = threading.Barrier(2, timeout=10)
+    finished = []
+
+    def query(name, payload):
+        try:
+            barrier.wait()
+            if name == "broker-health":
+                raise policy.DeliveryError("bounded helper refused")
+            return {"component": name}
+        finally:
+            with lock:
+                finished.append(name)
+
+    monkeypatch.setattr(node_health.operator_units, "run", query)
+    with pytest.raises(RuntimeError, match="consumer refused observation"):
+        with node_health._status_helpers(Path("/ordinary-readiness-port-model")) as pending:
+            raise RuntimeError("consumer refused observation")
+    assert len(finished) == 4 and set(finished) == set(pending)
+    assert all(future.done() for future in pending.values())
+    with pytest.raises(policy.DeliveryError, match="bounded helper refused"):
+        pending["broker-health"].result()

scripts/release/README.md

@@ -572,9 +572,12 @@ end-to-end target still requires measured acceptance.
 
 Fixed health helpers include interpreter startup in their wall-clock deadline.
 Outbox and import queries have 60 seconds each; factory status has 30 seconds and
-broker metadata has 45 seconds. Node readiness has 300 seconds for the sequential
-checks. Each ordinary helper has one CPU within the node's two-CPU ceiling;
-its memory, process and privilege limits remain fixed. These operator budgets
+broker metadata has 45 seconds. Node readiness has 300 seconds. The independent
+broker, outbox, import and factory queries run with at most two fixed helpers
+in flight; every dispatched helper is joined before readiness returns, including
+a failed observation. Results keep their existing validation and failure codes.
+Each ordinary helper has one CPU within the node's two-CPU ceiling; its memory,
+process and privilege limits remain fixed. These operator budgets
 do not change student execution budgets. Startup opens admission only after all
 required readiness queries succeed.
 The proxy's `Type=exec` state only confirms its launcher is running. Lifecycle

scripts/release/node_health.py

@@ -7,6 +7,8 @@ import logging
 import os
 import subprocess
 import urllib.request
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
 from importlib import import_module
 from pathlib import Path
 from uuid import UUID
@@ -128,6 +130,14 @@ def binding() -> tuple[dict, dict, Path]:
     return selected, paths, payload
 
 
+@contextmanager
+def _status_helpers(payload: Path):
+    """Join at most two fixed read-only helpers before returning any observation."""
+    names = ("broker-health", "outbox-health", "import-health", "factory-health")
+    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="kf-health") as executor:
+        yield {name: executor.submit(operator_units.run, name, payload) for name in names}
+
+
 def service_profile(name: str) -> tuple[int, int, int, int, int]:
     if name == "proxy":
         return 1205, 128, 16, 50, 8
@@ -567,53 +577,54 @@ def observe(*, probe: bool) -> dict:
         components["execution"] = counters
     except (OSError, ValueError):
         errors.append("execution_owner_unavailable")
-    try:
-        components["broker"] = operator_units.run("broker-health", payload)
-    except (OSError, ValueError, subprocess.SubprocessError):
-        errors.append("broker_metadata_unavailable")
-    try:
-        state = operator_units.run("outbox-health", payload)
-        validate_outbox(state)
-        components["outbox_pipeline"] = state
-    except (OSError, ValueError, subprocess.SubprocessError):
-        errors.append("outbox_pipeline_unavailable")
-    try:
-        state = operator_units.run("import-health", payload)
-        fields(
-            state,
-            {
-                "schema_version",
-                "worker_ready",
-                "reserved_bytes",
-                "uploaded_capped",
-                "validating_capped",
-                "truncated",
-            },
-            "import status",
-        )
-        if (
-            type(state["schema_version"]) is not int
-            or state["schema_version"] != 1
-            or state["worker_ready"] is not True
-            or state["truncated"] is not False
-        ):
-            raise DeliveryError("import heartbeat or counter unavailable")
-        for key, maximum in (
-            ("reserved_bytes", 268435456),
-            ("uploaded_capped", 128),
-            ("validating_capped", 1),
-        ):
-            if type(state[key]) is not int or not 0 <= state[key] <= maximum:
-                raise DeliveryError("import status bound differs")
-        components["import_pipeline"] = state
-    except (OSError, ValueError, subprocess.SubprocessError):
-        errors.append("import_pipeline_unavailable")
-    try:
-        status = operator_units.run("factory-health", payload)
-        validate_factory(status, digest(paths["runtime.json"]))
-        components["factory_status"] = status
-    except (OSError, ValueError, subprocess.SubprocessError):
-        errors.append("factory_status_unavailable")
+    with _status_helpers(payload) as pending:
+        try:
+            components["broker"] = pending["broker-health"].result()
+        except (OSError, ValueError, subprocess.SubprocessError):
+            errors.append("broker_metadata_unavailable")
+        try:
+            state = pending["outbox-health"].result()
+            validate_outbox(state)
+            components["outbox_pipeline"] = state
+        except (OSError, ValueError, subprocess.SubprocessError):
+            errors.append("outbox_pipeline_unavailable")
+        try:
+            state = pending["import-health"].result()
+            fields(
+                state,
+                {
+                    "schema_version",
+                    "worker_ready",
+                    "reserved_bytes",
+                    "uploaded_capped",
+                    "validating_capped",
+                    "truncated",
+                },
+                "import status",
+            )
+            if (
+                type(state["schema_version"]) is not int
+                or state["schema_version"] != 1
+                or state["worker_ready"] is not True
+                or state["truncated"] is not False
+            ):
+                raise DeliveryError("import heartbeat or counter unavailable")
+            for key, maximum in (
+                ("reserved_bytes", 268435456),
+                ("uploaded_capped", 128),
+                ("validating_capped", 1),
+            ):
+                if type(state[key]) is not int or not 0 <= state[key] <= maximum:
+                    raise DeliveryError("import status bound differs")
+            components["import_pipeline"] = state
+        except (OSError, ValueError, subprocess.SubprocessError):
+            errors.append("import_pipeline_unavailable")
+        try:
+            status = pending["factory-health"].result()
+            validate_factory(status, digest(paths["runtime.json"]))
+            components["factory_status"] = status
+        except (OSError, ValueError, subprocess.SubprocessError):
+            errors.append("factory_status_unavailable")
     if probe:
         try:
             # Blocking module initialization belongs to the fixed 300-second