From 3b8d9a2e715a7cf38f5a37e9e258b8e52c8fa589 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Sun, 18 Oct 2026 23:03:10 +0800 Subject: [PATCH] fix(release): close admission after maintenance helper failure --- backend/tests/test_delivery.py | 91 ++++++++++++++++++++- scripts/release/README.md | 4 + scripts/release/lifecycle.py | 130 ++++++++++++++++-------------- scripts/release/operator_units.py | 69 +++++++++------- 4 files changed, 203 insertions(+), 91 deletions(-) diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index cca20b1..a506079 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -2230,16 +2230,103 @@ def test_failed_gate_closure_requires_known_api_stop_without_forging_owner_clean monkeypatch.setattr(lifecycle.availability, "write", unavailable) monkeypatch.setattr( - maintenance, "unit", lambda name: {"ActiveState": "inactive", "MainPID": "0"} + lifecycle, + "selected_unit", + lambda selected, name: {"ActiveState": "inactive", "MainPID": "0"}, ) monkeypatch.setattr(lifecycle.service_install, "systemctl", lambda *args: calls.append(args)) monkeypatch.setattr( - maintenance, "empty_cgroup", lambda name: calls.append(("prove_empty", name)) + lifecycle.Maintenance, + "empty_cgroup", + lambda name: calls.append(("prove_empty", name)), ) maintenance.close_failed_admission() assert calls == [("--no-block", "stop", "kf-api.service"), ("prove_empty", "api")] +@pytest.mark.parametrize("failure", ["exit", "timeout", "invalid_output", "cleanup"]) +def test_dispatched_operator_failure_is_distinct_from_preflight_refusal(monkeypatch, failure): + # Controller result handling only; real timeout/closure is verified in the VM. + from scripts.release import operator_units + + selected = {"release_id": "sd-" + "a" * 12, "configuration_id": "cfg-" + "b" * 24} + payload = installer.RELEASES / selected["release_id"] / "payload" + monkeypatch.setattr(operator_units.os, "getuid", lambda: 0) + monkeypatch.setattr(operator_units.os, "geteuid", lambda: 0) + monkeypatch.setattr(installer, "selection", lambda **_: selected) + observations = [] + + def manager(*args): + observations.append(args) + if len(observations) > 1 and failure == "cleanup": + return "LoadState=loaded\nTransient=yes\nDescription=foreign helper\n" + return "LoadState=not-found\nActiveState=inactive\nMainPID=0\n" + + def dispatch(command, **kwargs): + assert kwargs["timeout"] == 915 + assert "--property=RuntimeMaxSec=900" in command + if failure == "timeout": + raise subprocess.TimeoutExpired(command, 915) + return subprocess.CompletedProcess( + command, + 1 if failure == "exit" else 0, + b"invalid" if failure == "invalid_output" else b'{"success":true}', + b"", + ) + + monkeypatch.setattr(service_install, "systemctl", manager) + monkeypatch.setattr(operator_units.subprocess, "run", dispatch) + with pytest.raises(operator_units.OperatorExecutionError): + operator_units.run("node-maintenance", payload, action="start") + assert len(observations) == 2 + + +@pytest.mark.parametrize("dispatched", [True, False]) +def test_outer_lifecycle_closes_admission_only_for_dispatched_helper_failure( + monkeypatch, + dispatched, +): + from scripts.release import operator_units + + selected = {"release_id": "sd-" + "a" * 12, "configuration_id": "cfg-" + "b" * 24} + gate = ["open"] + monkeypatch.setattr(sys, "argv", ["lifecycle", "--action", "start"]) + monkeypatch.setattr(lifecycle.os, "getuid", lambda: 0) + monkeypatch.setattr(lifecycle.os, "geteuid", lambda: 0) + monkeypatch.setattr(installer, "selection", lambda **_: selected) + monkeypatch.setattr(installer, "directory", lambda *_: None) + + def failure(*args, **kwargs): + error = operator_units.OperatorExecutionError if dispatched else policy.DeliveryError + raise error( + "operator helper unavailable" if dispatched else "operator helper already active" + ) + + monkeypatch.setattr(operator_units, "run", failure) + monkeypatch.setattr(lifecycle.availability, "write", lambda binding, state: gate.append(state)) + with pytest.raises(SystemExit) as exit_info: + lifecycle.main() + assert exit_info.value.code == 1 + assert gate == (["open", "draining"] if dispatched else ["open"]) + + +def test_failed_dispatch_does_not_stop_an_unverified_api_when_gate_write_fails(monkeypatch): + calls = [] + + def unavailable(*_): + raise OSError("publication unavailable") + + def foreign(*_): + raise policy.DeliveryError("lifecycle target is not the selected snapshot") + + monkeypatch.setattr(lifecycle.availability, "write", unavailable) + monkeypatch.setattr(lifecycle, "selected_unit", foreign) + monkeypatch.setattr(service_install, "systemctl", lambda *args: calls.append(args)) + with pytest.raises(policy.DeliveryError, match="selected snapshot"): + lifecycle.close_failed_admission({}) + assert calls == [] + + def proxy_catalog_shape(): # Structural example, not trusted executable bytes or ELF/version evidence. names = ["bin/nginx", "lib/libc.so.6", "lib/ld-linux-x86-64.so.2"] diff --git a/scripts/release/README.md b/scripts/release/README.md index 621fa5f..e09c942 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -311,6 +311,10 @@ API returns 503 for an absent, unknown, closed or wrong-bound availability recor Reads, logout, cancellation and receipt-bound archive uploads keep their existing identity/lease checks. Both startup paths keep admission closed until the opening is read back; failed final readiness closes it again. +If a dispatched maintenance helper exits unsuccessfully or reaches its deadline, +the outer operator also closes admission. If that record cannot be written, it +stops only the verified API unit. A refusal before dispatch, such as an already +active helper, does not change admission or stop another operation. Drain keeps API available to accepted work and polls Coordinator, outbox, import, broker and factory within a finite deadline. Untruncated owner counts, stopped diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py index f20c121..de50a11 100644 --- a/scripts/release/lifecycle.py +++ b/scripts/release/lifecycle.py @@ -226,6 +226,64 @@ def projection_empty(snapshot: dict) -> bool: ) +def selected_unit(selected: dict, name: str) -> dict: + if name not in {*APPS, *DEPENDENCIES}: + raise DeliveryError("unknown lifecycle target") + path = service_install.SYSTEM_UNITS / ("kf-" + name + ".service") + if name in services.PROFILES: + directory = service_install.known_link(path) + base = service_install.UNITS + elif name == "proxy": + directory = proxy_install.known_link(path) + base = proxy_service.UNITS + elif name == "sandbox-factory": + directory = factory_service.known_link(path) + base = factory_service.UNITS + else: + directory = dependency_install.known_link(path) + base = dependency_install.ROOT + if directory != base / (selected["release_id"] + "." + selected["configuration_id"]): + raise DeliveryError("lifecycle target is not the selected snapshot") + values = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", + path.name, + "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath", + ).splitlines() + if "=" in line + ) + if values.get("LoadState") != "loaded" or values.get("FragmentPath") not in { + str(path), + str(directory / path.name), + }: + raise DeliveryError("lifecycle manager target differs") + return values + + +def close_failed_admission(selected: dict) -> None: + try: + availability.write(selected, "draining") + return + except (OSError, ValueError, subprocess.SubprocessError): + pass + # A full/read-only availability filesystem must not leave a successful + # earlier publication accepting mutations. Stop only the verified API + # target without depending on another receipt write. Owners retain work + # and capacity; this is not a factory cleanup or terminal-state proof. + selected_unit(selected, "api") + service_install.systemctl("--no-block", "stop", "kf-api.service") + deadline = time.monotonic() + 50 + while True: + values = selected_unit(selected, "api") + if values["ActiveState"] == "inactive" and values["MainPID"] == "0": + Maintenance.empty_cgroup("api") + return + if time.monotonic() >= deadline or values["ActiveState"] == "failed": + raise DeliveryError("admission closure/API stop unconfirmed; operator required") + time.sleep(0.5) + + class Maintenance: def __init__(self, selected: dict, paths: dict, payload: Path, action: str): self.selected, self.paths, self.payload = selected, paths, payload @@ -268,40 +326,7 @@ class Maintenance: temporary.unlink() def unit(self, name: str) -> dict: - if name not in {*APPS, *DEPENDENCIES}: - raise DeliveryError("unknown lifecycle target") - path = service_install.SYSTEM_UNITS / ("kf-" + name + ".service") - if name in services.PROFILES: - directory = service_install.known_link(path) - base = service_install.UNITS - elif name == "proxy": - directory = proxy_install.known_link(path) - base = proxy_service.UNITS - elif name == "sandbox-factory": - directory = factory_service.known_link(path) - base = factory_service.UNITS - else: - directory = dependency_install.known_link(path) - base = dependency_install.ROOT - if directory != base / ( - self.selected["release_id"] + "." + self.selected["configuration_id"] - ): - raise DeliveryError("lifecycle target is not the selected snapshot") - values = dict( - line.split("=", 1) - for line in service_install.systemctl( - "show", - path.name, - "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath", - ).splitlines() - if "=" in line - ) - if values.get("LoadState") != "loaded" or values.get("FragmentPath") not in { - str(path), - str(directory / path.name), - }: - raise DeliveryError("lifecycle manager target differs") - return values + return selected_unit(self.selected, name) @staticmethod def empty_cgroup(name: str) -> None: @@ -515,26 +540,7 @@ class Maintenance: raise def close_failed_admission(self) -> None: - try: - availability.write(self.selected, "draining") - return - except (OSError, ValueError, subprocess.SubprocessError): - pass - # A full/read-only availability filesystem must not leave a successful - # earlier publication accepting mutations. Stop only the verified API - # target without depending on another receipt write. Owners retain work - # and capacity; this is not a factory cleanup or terminal-state proof. - self.unit("api") - service_install.systemctl("--no-block", "stop", "kf-api.service") - deadline = time.monotonic() + 50 - while True: - values = self.unit("api") - if values["ActiveState"] == "inactive" and values["MainPID"] == "0": - self.empty_cgroup("api") - return - if time.monotonic() >= deadline or values["ActiveState"] == "failed": - raise DeliveryError("admission closure/API stop unconfirmed; operator required") - time.sleep(0.5) + close_failed_admission(self.selected) def run(action: str) -> dict: @@ -614,11 +620,17 @@ def main() -> None: selected = install.selection(check_configuration=False) install.directory(availability.GATE.parent, 0o711) install.directory(STATE, 0o700) - result = operator_units.run( - "node-maintenance", - install.RELEASES / selected["release_id"] / "payload", - action=args.action, - ) + try: + result = operator_units.run( + "node-maintenance", + install.RELEASES / selected["release_id"] / "payload", + action=args.action, + ) + except operator_units.OperatorExecutionError: + # A killed helper cannot run its own exception handler. The + # outer operator closes admission after the fixed helper exits. + close_failed_admission(selected) + raise print(json.dumps(result, sort_keys=True)) except (OSError, ValueError, subprocess.SubprocessError) as error: parser.exit( diff --git a/scripts/release/operator_units.py b/scripts/release/operator_units.py index 4e02e1c..c88d2bb 100644 --- a/scripts/release/operator_units.py +++ b/scripts/release/operator_units.py @@ -9,6 +9,11 @@ from pathlib import Path from . import preflight from .policy import DeliveryError, invalid_constant, unique_object + +class OperatorExecutionError(DeliveryError): + """A fixed helper was dispatched but its result or cleanup was unconfirmed.""" + + PROFILES = { "node-health": ("root", 0, 128, 16, 50, 300), "broker-health": ("rabbitmq", 1204, 256, 64, 100, 45), @@ -232,40 +237,44 @@ def run( elif notification_probe: raise DeliveryError("unexpected broker probe override") try: - # Pipe output is bounded by the fixed helper and the service file/memory - # ceilings; helper tools below cap each pipe before serializing it. - result = subprocess.run( - command, - capture_output=True, - timeout=seconds + 15, - env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}, - ) - if result.returncode or len(result.stderr) > 4096: - raise DeliveryError("operator helper unavailable") - value = decode(result.stdout) - if not isinstance(value, dict): - raise DeliveryError("operator helper object required") - finally: - actual = dict( - line.split("=", 1) - for line in service_install.systemctl( - "show", - unit, - "--property=LoadState,ActiveState,MainPID,Transient,Description", - ).splitlines() - if "=" in line - ) - if actual.get("LoadState") != "not-found": - if actual.get("Transient") != "yes" or actual.get("Description") != description: - raise DeliveryError("foreign operator helper; refused cleanup") - service_install.systemctl("stop", unit) + try: + # Pipe output is bounded by the fixed helper and the service file/memory + # ceilings; helper tools below cap each pipe before serializing it. + result = subprocess.run( + command, + capture_output=True, + timeout=seconds + 15, + env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}, + ) + if result.returncode or len(result.stderr) > 4096: + raise DeliveryError("operator helper unavailable") + value = decode(result.stdout) + if not isinstance(value, dict): + raise DeliveryError("operator helper object required") + finally: actual = dict( line.split("=", 1) for line in service_install.systemctl( - "show", unit, "--property=ActiveState,MainPID" + "show", + unit, + "--property=LoadState,ActiveState,MainPID,Transient,Description", ).splitlines() if "=" in line ) - if actual.get("ActiveState") != "inactive" or actual.get("MainPID") != "0": - raise DeliveryError("operator helper cleanup unconfirmed") + if actual.get("LoadState") != "not-found": + if actual.get("Transient") != "yes" or actual.get("Description") != description: + raise DeliveryError("foreign operator helper; refused cleanup") + service_install.systemctl("stop", unit) + actual = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", unit, "--property=ActiveState,MainPID" + ).splitlines() + if "=" in line + ) + if actual.get("ActiveState") != "inactive" or actual.get("MainPID") != "0": + raise DeliveryError("operator helper cleanup unconfirmed") + except (OSError, ValueError, subprocess.SubprocessError) as error: + message = str(error) if isinstance(error, DeliveryError) else type(error).__name__ + raise OperatorExecutionError(message) from error return value