diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index b96fb49..6adc676 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -972,3 +972,60 @@ def test_journal_profile_never_substitutes_rotation_settings_for_a_dedicated_har observed[field] = value with pytest.raises(policy.DeliveryError): journal.validate_storage(observed, path) + + +@pytest.mark.parametrize("failure", ["publication", "receipt", "readback"]) +def test_maintenance_does_not_leave_admission_open_when_final_confirmation_fails( + monkeypatch, failure +): + # Pure controller error flow, no Root, manager or filesystem operations. + # Publication atomicity, actual stop and disk failure remain VM-only cases. + gate = [] + maintenance = object.__new__(lifecycle.Maintenance) + maintenance.selected = {} + maintenance.payload = Path("/unused") + monkeypatch.setattr(maintenance, "change", lambda *_: None) + + def publish(_, state): + gate.append(state) + if state == "open" and failure == "publication": + raise OSError("publication readback unavailable") + + def save(_): + if failure == "receipt": + raise OSError("maintenance receipt unavailable") + + monkeypatch.setattr(lifecycle.availability, "write", publish) + monkeypatch.setattr(maintenance, "save", save) + monkeypatch.setattr( + lifecycle.operator_units, + "run", + lambda *_: { + "service_preconditions_met": True, + "ready_for_admission": failure != "readback", + }, + ) + with pytest.raises((OSError, policy.DeliveryError)): + maintenance.start(resume=True) + assert gate == ["draining", "open", "draining"] + + +def test_failed_gate_closure_requires_known_api_stop_without_forging_owner_cleanup(monkeypatch): + # Controller flow only; actual cgroup/stop/full-volume evidence is VM-only. + maintenance = object.__new__(lifecycle.Maintenance) + maintenance.selected = {} + calls = [] + + def unavailable(*_): + raise OSError("publication unavailable") + + monkeypatch.setattr(lifecycle.availability, "write", unavailable) + monkeypatch.setattr( + maintenance, "unit", lambda 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)) + ) + maintenance.close_failed_admission() + assert calls == [("--no-block", "stop", "kf-api.service"), ("prove_empty", "api")] diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py index 7c72f33..d421f28 100644 --- a/scripts/release/lifecycle.py +++ b/scripts/release/lifecycle.py @@ -446,15 +446,37 @@ class Maintenance: ready = operator_units.run("node-health", self.payload) if ready.get("service_preconditions_met") is not True: raise DeliveryError("service preconditions unavailable; admission remains closed") - availability.write(self.selected, "open") try: + availability.write(self.selected, "open") ready = operator_units.run("node-health", self.payload) if ready.get("ready_for_admission") is not True: raise DeliveryError("admission readiness readback unconfirmed") + self.save("accepting") except (OSError, ValueError, subprocess.SubprocessError): - availability.write(self.selected, "draining") + self.close_failed_admission() raise - self.save("accepting") + + 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) def run(action: str) -> dict: @@ -473,9 +495,17 @@ def run(action: str) -> dict: install.exclusive_file(lockpath, b"", 0o600) preflight.protected(lockpath) configuration.private(lockpath) + if lockpath.stat().st_size != 0: + raise DeliveryError("unrecorded maintenance lock contents") fd = os.open(lockpath, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + # Refuse unknown history before publishing availability, including + # resume where every known service could already be running. + if LAST.exists() or LAST.is_symlink(): + preflight.protected(LAST) + configuration.private(LAST) + validate_record(read_json(LAST, 32768)) maintenance = Maintenance(selected, paths, payload, action) try: if action in {"drain", "stop", "restart"}: