diff --git a/backend/tests/test_delivery_validation_work.py b/backend/tests/test_delivery_validation_work.py index ef13c54..a4529fe 100644 --- a/backend/tests/test_delivery_validation_work.py +++ b/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() diff --git a/scripts/release/README.md b/scripts/release/README.md index 9453050..a1b0e83 100644 --- a/scripts/release/README.md +++ b/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 diff --git a/scripts/release/node_health.py b/scripts/release/node_health.py index a10d82f..a03609c 100644 --- a/scripts/release/node_health.py +++ b/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