diff --git a/docs/manage-sandboxes/gateway-lifecycle-control.mdx b/docs/manage-sandboxes/gateway-lifecycle-control.mdx index 867c8346035..b6176f41f1d 100644 --- a/docs/manage-sandboxes/gateway-lifecycle-control.mdx +++ b/docs/manage-sandboxes/gateway-lifecycle-control.mdx @@ -21,6 +21,9 @@ Built-in OpenClaw and Hermes images support two direct-container lifecycle topol The managed controller authenticates the host lifecycle action and prevents PID reuse from redirecting its signal. It cannot prove process provenance against a malicious process running under the same sandbox UID, and it does not create gateway and agent UID isolation. +For `recover` and `gateway restart`, the managed controller acquires the expected-exit lock before it inspects the supervisor or gateway. +Lock acquisition, gateway termination, and replacement health share one recovery deadline. +If lock acquisition reaches that deadline, the controller returns `SUPERVISOR_BUSY` without publishing an expected-exit marker. Mutable managed config retains the trust and time-of-check/time-of-use limits of managed cold start. diff --git a/scripts/managed-gateway-control.py b/scripts/managed-gateway-control.py index 7c2c1e33947..6c698ee3a88 100755 --- a/scripts/managed-gateway-control.py +++ b/scripts/managed-gateway-control.py @@ -47,12 +47,14 @@ import fcntl import hashlib import http.client +import io import importlib.util import os import pwd import re import select import signal +import socket import stat import subprocess import sys @@ -96,6 +98,7 @@ CONTROL_STAGES = frozenset( { "detect-agent", + "acquire-expected-exit-lock", "discover-supervisor", "initial-gateway-proof", "preflight", @@ -223,6 +226,14 @@ class AgentSpec: readiness_checks: tuple[tuple[int, str], ...] = () +@dataclass(frozen=True) +class ExpectedExitLock: + """Held lock for one managed gateway lifecycle operation.""" + + directory_fd: int + lock_fd: int + + @dataclass(frozen=True) class ExpectedExitLease: """Pinned authorization marker and root-only controller lock.""" @@ -417,8 +428,18 @@ def _validate_runtime_regular( raise ControlError("SUPERVISOR_UNAVAILABLE") -def _open_expected_exit_lock(directory_fd: int) -> int: - """Acquire the root-only lock that serializes authorization publication.""" +def _open_expected_exit_lock( + directory_fd: int, + recovery_deadline: float, +) -> int: + """Acquire the root-only lock that serializes lifecycle changes. + + A second host lifecycle request can arrive while the first controller is + still waiting for the gateway replacement. Poll the non-blocking flock + until the shared recovery deadline instead of turning that expected + overlap into an immediate ``SUPERVISOR_BUSY`` failure. The waiting process + has not published a marker and cannot authorize a gateway exit. + """ base_flags = ( os.O_RDWR @@ -466,10 +487,19 @@ def _open_expected_exit_lock(directory_fd: int) -> int: metadata.st_ino, ): raise ControlError("SUPERVISOR_UNAVAILABLE") - try: - fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - except BlockingIOError as exc: - raise ControlError("SUPERVISOR_BUSY") from exc + while True: + if time.monotonic() >= recovery_deadline: + raise ControlError("SUPERVISOR_BUSY") + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError as exc: + remaining = recovery_deadline - time.monotonic() + if remaining <= 0: + raise ControlError("SUPERVISOR_BUSY") from exc + time.sleep(min(POLL_SECONDS, remaining)) + if time.monotonic() >= recovery_deadline: + raise ControlError("SUPERVISOR_BUSY") locked = os.fstat(lock_fd) _validate_runtime_regular(locked, 0o600) if ( @@ -489,6 +519,26 @@ def _open_expected_exit_lock(directory_fd: int) -> int: raise +def _acquire_expected_exit_lock( + recovery_deadline: float, +) -> ExpectedExitLock: + directory_fd = _open_managed_runtime_directory() + lock_fd = None + try: + lock_fd = _open_expected_exit_lock(directory_fd, recovery_deadline) + return ExpectedExitLock(directory_fd=directory_fd, lock_fd=lock_fd) + except Exception: + if lock_fd is not None: + os.close(lock_fd) + os.close(directory_fd) + raise + + +def _close_expected_exit_lock(lock: ExpectedExitLock) -> None: + os.close(lock.lock_fd) + os.close(lock.directory_fd) + + def _trusted_expected_exit_marker( directory_fd: int, ) -> tuple[int, os.stat_result] | None: @@ -517,16 +567,18 @@ def _controller_process_identity(reader: ProcReader) -> ProcessIdentity: def _publish_expected_exit_lease( + lock: ExpectedExitLock, identity: ProcessIdentity, controller: ProcessIdentity, + recovery_deadline: float | None = None, ) -> ExpectedExitLease: """Authorize one exact gateway exit while this root controller is live.""" - directory_fd = _open_managed_runtime_directory() - lock_fd = -1 + _require_recovery_time(recovery_deadline) + directory_fd = lock.directory_fd + lock_fd = lock.lock_fd marker_fd = -1 try: - lock_fd = _open_expected_exit_lock(directory_fd) existing = _trusted_expected_exit_marker(directory_fd) if existing is not None: existing_fd, metadata = existing @@ -539,6 +591,7 @@ def _publish_expected_exit_lease( ) finally: os.close(existing_fd) + _require_recovery_time(recovery_deadline) payload = ( f"v1 {identity.pid} {identity.start_time} " @@ -579,6 +632,7 @@ def _publish_expected_exit_lease( marker_inode, ): raise ControlError("SUPERVISOR_UNAVAILABLE") + _require_recovery_time(recovery_deadline) except Exception: try: if marker_fd >= 0: @@ -595,9 +649,6 @@ def _publish_expected_exit_lease( pass if marker_fd >= 0: os.close(marker_fd) - if lock_fd >= 0: - os.close(lock_fd) - os.close(directory_fd) raise return ExpectedExitLease( directory_fd=directory_fd, @@ -860,13 +911,17 @@ def _read_stable_file_with_proof_grace( identity: ProcessIdentity, name: str, limit: int, + recovery_deadline: float | None = None, ) -> bytes: """Retry an inconsistent proc read only while the pinned process is exact.""" deadline = time.monotonic() + PROCESS_PROOF_GRACE_SECONDS + if recovery_deadline is not None: + deadline = min(deadline, recovery_deadline) + _require_recovery_time(recovery_deadline) while True: try: - return reader.read_stable_file(identity, name, limit) + value = reader.read_stable_file(identity, name, limit) except ControlError as error: if error.code != "SUPERVISOR_UNAVAILABLE": raise @@ -875,6 +930,9 @@ def _read_stable_file_with_proof_grace( if remaining <= 0: raise time.sleep(min(PROCESS_PROOF_RETRY_SECONDS, remaining)) + continue + _require_recovery_time(recovery_deadline) + return value def _basename(value: bytes) -> bytes: @@ -1094,11 +1152,31 @@ def _gateway_matches( return _is_openclaw_gateway(identity, spec.port) +def _recovery_deadline_reached(recovery_deadline: float | None) -> bool: + return bool( + recovery_deadline is not None + and time.monotonic() >= recovery_deadline + ) + + +def _require_recovery_time(recovery_deadline: float | None) -> None: + if _recovery_deadline_reached(recovery_deadline): + raise ControlError("GATEWAY_FAILED") + + def _gateway_candidates( - reader: ProcReader, supervisor: ProcessIdentity, spec: AgentSpec + reader: ProcReader, + supervisor: ProcessIdentity, + spec: AgentSpec, + recovery_deadline: float | None = None, ) -> list[ProcessIdentity]: + if _recovery_deadline_reached(recovery_deadline): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") matches: list[ProcessIdentity] = [] - for pid in reader.pids(): + pids = reader.pids() + for pid in pids: + if _recovery_deadline_reached(recovery_deadline): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") if pid in (1, supervisor.pid): continue try: @@ -1109,7 +1187,15 @@ def _gateway_candidates( matches.append(identity) if len(matches) > 1: break - _recapture_exact_identity(reader, supervisor) + if _recovery_deadline_reached(recovery_deadline): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") + _recapture_exact_identity( + reader, + supervisor, + deadline=recovery_deadline, + ) + if _recovery_deadline_reached(recovery_deadline): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") if len(matches) > 1: raise ControlError("SUPERVISOR_UNAVAILABLE") return matches @@ -1192,16 +1278,112 @@ def _owns_listener( return False -def _http_healthy(port: int, path: str) -> bool: - connection = http.client.HTTPConnection("127.0.0.1", port, timeout=2) +def _http_remaining_time(deadline: float) -> float: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError + return remaining + + +class _DeadlineSocket: + """Apply one deadline to each socket operation used by HTTPResponse.""" + + def __init__(self, transport: socket.socket, deadline: float) -> None: + self._transport = transport + self._deadline = deadline + self._readers = 0 + self._close_requested = False + + def _set_timeout(self) -> None: + self._transport.settimeout(_http_remaining_time(self._deadline)) + + def sendall(self, data: bytes) -> None: + self._set_timeout() + self._transport.sendall(data) + _http_remaining_time(self._deadline) + + def recv_into(self, buffer: bytearray | memoryview) -> int: + self._set_timeout() + received = self._transport.recv_into(buffer) + _http_remaining_time(self._deadline) + return received + + def makefile(self, mode: str) -> io.BufferedReader: + if mode != "rb": + raise ValueError("HTTP response reader requires binary mode") + self._readers += 1 + return io.BufferedReader(_DeadlineSocketReader(self)) + + def _release_reader(self) -> None: + self._readers -= 1 + if self._close_requested and self._readers == 0: + self._transport.close() + + def close(self) -> None: + self._close_requested = True + if self._readers == 0: + self._transport.close() + + +class _DeadlineSocketReader(io.RawIOBase): + def __init__(self, owner: _DeadlineSocket) -> None: + super().__init__() + self._owner = owner + + def readable(self) -> bool: + return True + + def readinto(self, buffer: bytearray | memoryview) -> int: + return self._owner.recv_into(buffer) + + def close(self) -> None: + if self.closed: + return + try: + super().close() + finally: + self._owner._release_reader() + + +def _http_healthy( + port: int, + path: str, + recovery_deadline: float | None = None, +) -> bool: + request_deadline = time.monotonic() + 2.0 + if recovery_deadline is not None: + request_deadline = min(request_deadline, recovery_deadline) + try: + timeout_seconds = _http_remaining_time(request_deadline) + except OSError: + return False + connection = http.client.HTTPConnection( + "127.0.0.1", + port, + timeout=timeout_seconds, + ) + response: http.client.HTTPResponse | None = None try: + connection.connect() + _http_remaining_time(request_deadline) + if connection.sock is None: + return False + connection.sock = _DeadlineSocket( # type: ignore[assignment] + connection.sock, + request_deadline, + ) connection.request("GET", path) + _http_remaining_time(request_deadline) response = connection.getresponse() + _http_remaining_time(request_deadline) response.read(4096) + _http_remaining_time(request_deadline) return response.status in (200, 401) - except OSError: + except (OSError, http.client.HTTPException): return False finally: + if response is not None: + response.close() connection.close() @@ -1210,9 +1392,12 @@ def _http_healthy_in_gateway_namespace( identity: ProcessIdentity, port: int, path: str, + recovery_deadline: float | None = None, ) -> bool: """Probe loopback from the gateway's network namespace, then restore ours.""" + if _recovery_deadline_reached(recovery_deadline): + return False setns = getattr(os, "setns", None) if setns is None: raise ControlError("PRIVILEGED_CONTROL_UNAVAILABLE") @@ -1221,6 +1406,7 @@ def _http_healthy_in_gateway_namespace( pid_fd = -1 target_namespace = -1 switched = False + healthy = False try: pid_fd = _open_pid(reader.fd, identity.pid) pinned = os.fstat(pid_fd) @@ -1232,9 +1418,11 @@ def _http_healthy_in_gateway_namespace( target_namespace = os.open("ns/net", flags, dir_fd=pid_fd) if reader.capture(identity.pid).stable_key() != identity.stable_key(): return False + if _recovery_deadline_reached(recovery_deadline): + return False setns(target_namespace, getattr(os, "CLONE_NEWNET", 0x40000000)) switched = True - return _http_healthy(port, path) + healthy = _http_healthy(port, path, recovery_deadline) except OSError as exc: if exc.errno in (errno.ENOENT, errno.ENOTDIR, errno.ESRCH): return False @@ -1250,44 +1438,97 @@ def _http_healthy_in_gateway_namespace( if pid_fd >= 0: os.close(pid_fd) os.close(current_namespace) + return bool( + healthy + and not _recovery_deadline_reached(recovery_deadline) + ) def _gateway_healthy( - reader: ProcReader, identity: ProcessIdentity, spec: AgentSpec + reader: ProcReader, + identity: ProcessIdentity, + spec: AgentSpec, + recovery_deadline: float | None = None, ) -> bool: return bool( - _owns_listener(reader, identity, spec.port) + not _recovery_deadline_reached(recovery_deadline) + and _owns_listener(reader, identity, spec.port) + and not _recovery_deadline_reached(recovery_deadline) and _http_healthy_in_gateway_namespace( - reader, identity, spec.port, spec.health_path + reader, + identity, + spec.port, + spec.health_path, + recovery_deadline, ) + and not _recovery_deadline_reached(recovery_deadline) and _owns_listener(reader, identity, spec.port) + and not _recovery_deadline_reached(recovery_deadline) ) def _gateway_auxiliaries_healthy( - reader: ProcReader, identity: ProcessIdentity, spec: AgentSpec + reader: ProcReader, + identity: ProcessIdentity, + spec: AgentSpec, + recovery_deadline: float | None = None, ) -> bool: """Prove the public API relay the host probes before completing control.""" for port, path in spec.readiness_checks: - if not _http_healthy_in_gateway_namespace(reader, identity, port, path): + if _recovery_deadline_reached(recovery_deadline): + return False + if not _http_healthy_in_gateway_namespace( + reader, + identity, + port, + path, + recovery_deadline, + ): return False # The public probes can take several seconds. Re-prove the exact gateway # after them so a replacement that exited during auxiliary repair is never # reported as the completed child. - return _gateway_healthy(reader, identity, spec) + return bool( + not _recovery_deadline_reached(recovery_deadline) + and _gateway_healthy( + reader, + identity, + spec, + recovery_deadline, + ) + ) + +def _preflight_timeout(recovery_deadline: float | None) -> float: + timeout_seconds = _remaining_recovery_time(recovery_deadline, 15.0) + if timeout_seconds <= 0: + raise ControlError("GATEWAY_FAILED") + return timeout_seconds -def _run_fixed_validator(script: str, arguments: list[str]) -> None: + +def _run_fixed_validator( + script: str, + arguments: list[str], + recovery_deadline: float | None = None, +) -> None: + _require_recovery_time(recovery_deadline) _validate_trusted_regular(script) - result = subprocess.run( - [sys.executable, "-I", script, *arguments], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - timeout=15, - check=False, - ) + _require_recovery_time(recovery_deadline) + try: + result = subprocess.run( + [sys.executable, "-I", script, *arguments], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=_preflight_timeout(recovery_deadline), + check=False, + ) + except subprocess.TimeoutExpired as exc: + if recovery_deadline is not None: + raise ControlError("GATEWAY_FAILED") from exc + raise ControlError("SECRET_BOUNDARY_REFUSED") from exc + _require_recovery_time(recovery_deadline) if result.returncode != 0: raise ControlError("SECRET_BOUNDARY_REFUSED") @@ -1404,53 +1645,74 @@ def _verify_locked_hermes_hash() -> None: raise ControlError("GATEWAY_CONFIG_HASH_MISMATCH") -def _hermes_preflight(reader: ProcReader, supervisor: ProcessIdentity) -> None: +def _hermes_preflight( + reader: ProcReader, + supervisor: ProcessIdentity, + recovery_deadline: float | None = None, +) -> None: + _require_recovery_time(recovery_deadline) validator = _system_path(HERMES_BOUNDARY_PATH) if not os.path.exists(validator): raise ControlError("SECRET_BOUNDARY_VALIDATOR_MISSING") _run_fixed_validator( validator, ["env-file", _system_path("/sandbox/.hermes/.env")], + recovery_deadline, ) raw_environment = _read_stable_file_with_proof_grace( reader, supervisor, "environ", MAX_ENV_BYTES, + recovery_deadline, ) _validate_runtime_environment(validator, _parse_environment(raw_environment)) + _require_recovery_time(recovery_deadline) _verify_locked_hermes_hash() + _require_recovery_time(recovery_deadline) -def _openclaw_preflight() -> None: +def _openclaw_preflight(recovery_deadline: float | None = None) -> None: + _require_recovery_time(recovery_deadline) guard = _system_path(OPENCLAW_GUARD_PATH) _validate_trusted_regular(guard) - result = subprocess.run( - [ - sys.executable, - "-I", - guard, - "preflight-restart", - "--config-dir", - _system_path("/sandbox/.openclaw"), - ], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - timeout=15, - check=False, - ) + try: + result = subprocess.run( + [ + sys.executable, + "-I", + guard, + "preflight-restart", + "--config-dir", + _system_path("/sandbox/.openclaw"), + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=_preflight_timeout(recovery_deadline), + check=False, + ) + except subprocess.TimeoutExpired as exc: + if recovery_deadline is not None: + raise ControlError("GATEWAY_FAILED") from exc + raise ControlError("GATEWAY_UNSAFE_CONFIG_PATH") from exc + _require_recovery_time(recovery_deadline) if result.returncode != 0: raise ControlError("GATEWAY_UNSAFE_CONFIG_PATH") def _preflight( - spec: AgentSpec, reader: ProcReader, supervisor: ProcessIdentity + spec: AgentSpec, + reader: ProcReader, + supervisor: ProcessIdentity, + recovery_deadline: float | None = None, ) -> None: + _require_recovery_time(recovery_deadline) if spec.name == "hermes": - _hermes_preflight(reader, supervisor) + _hermes_preflight(reader, supervisor, recovery_deadline) else: - _openclaw_preflight() + _openclaw_preflight(recovery_deadline) + _require_recovery_time(recovery_deadline) def _pidfd_open(pid: int) -> int | None: @@ -1485,29 +1747,68 @@ def _send_pidfd(pidfd: int, signum: signal.Signals) -> bool: return True -def _terminate_gateway(reader: ProcReader, identity: ProcessIdentity) -> None: +def _remaining_recovery_time( + recovery_deadline: float | None, + maximum: float, +) -> float: + if recovery_deadline is None: + return maximum + return min( + maximum, + max(0.0, recovery_deadline - time.monotonic()), + ) + + +def _terminate_gateway( + reader: ProcReader, + identity: ProcessIdentity, + recovery_deadline: float | None = None, +) -> None: pidfd = _pidfd_open(identity.pid) if pidfd is None: return try: + _require_recovery_time(recovery_deadline) try: - _recapture_exact_identity(reader, identity) + _recapture_exact_identity( + reader, + identity, + deadline=recovery_deadline, + ) except ControlError: # The pidfd is readable only when the exact process opened above has # exited. Accept that race without ever falling back to a PID signal. if _pidfd_exited(pidfd, 0): return + _require_recovery_time(recovery_deadline) raise + _require_recovery_time(recovery_deadline) if not _send_pidfd(pidfd, signal.SIGTERM): return - if _pidfd_exited(pidfd, STOP_GRACE_SECONDS): + if _pidfd_exited( + pidfd, + _remaining_recovery_time( + recovery_deadline, + STOP_GRACE_SECONDS, + ), + ): return + if _recovery_deadline_reached(recovery_deadline): + if _pidfd_exited(pidfd, 0): + return + raise ControlError("GATEWAY_FAILED") # The pidfd already pins the proven gateway across exit and PID reuse. # Re-reading /proc here races with the normal live-to-zombie transition # and adds no signal-target safety. if not _send_pidfd(pidfd, signal.SIGKILL): return - if not _pidfd_exited(pidfd, KILL_GRACE_SECONDS): + if not _pidfd_exited( + pidfd, + _remaining_recovery_time( + recovery_deadline, + KILL_GRACE_SECONDS, + ), + ): raise ControlError("GATEWAY_FAILED") finally: os.close(pidfd) @@ -1520,11 +1821,19 @@ def _wait_for_healthy_gateway( old_identity: ProcessIdentity | None, timeout_seconds: float = RECOVERY_TIMEOUT_SECONDS, require_auxiliary_health: bool = False, + recovery_deadline: float | None = None, ) -> ProcessIdentity: deadline = time.monotonic() + timeout_seconds + if recovery_deadline is not None: + deadline = min(deadline, recovery_deadline) while time.monotonic() < deadline: try: - candidates = _gateway_candidates(reader, supervisor, spec) + candidates = _gateway_candidates( + reader, + supervisor, + spec, + deadline, + ) except (FileNotFoundError, ProcessLookupError, PermissionError): raise ControlError("SUPERVISOR_UNAVAILABLE") for candidate in candidates: @@ -1534,17 +1843,31 @@ def _wait_for_healthy_gateway( ) == (old_identity.pid, old_identity.start_time): continue try: - if _gateway_healthy(reader, candidate, spec) and ( + if _gateway_healthy( + reader, + candidate, + spec, + deadline, + ) and ( not require_auxiliary_health - or _gateway_auxiliaries_healthy(reader, candidate, spec) + or _gateway_auxiliaries_healthy( + reader, + candidate, + spec, + deadline, + ) ): - return candidate + if time.monotonic() < deadline: + return candidate except (FileNotFoundError, ProcessLookupError): continue except ControlError as error: if error.code != "SUPERVISOR_UNAVAILABLE": raise - time.sleep(POLL_SECONDS) + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(POLL_SECONDS, remaining)) raise ControlError("GATEWAY_HEALTH_TIMEOUT") @@ -1553,8 +1876,9 @@ def _wait_for_recovery_candidate( supervisor: ProcessIdentity, spec: AgentSpec, initial_identity: ProcessIdentity, + recovery_deadline: float | None = None, ) -> tuple[ProcessIdentity | None, ProcessIdentity | None]: - """Give the observed candidate, and at most one successor, a full grace.""" + """Give the observed candidate and at most one successor bounded grace.""" observed = initial_identity for _attempt in range(2): @@ -1564,14 +1888,23 @@ def _wait_for_recovery_candidate( supervisor, spec, None, - RECOVER_EXISTING_GRACE_SECONDS, + _remaining_recovery_time( + recovery_deadline, + RECOVER_EXISTING_GRACE_SECONDS, + ), + recovery_deadline=recovery_deadline, ) return healthy, None except ControlError as error: if error.code != "GATEWAY_HEALTH_TIMEOUT": raise - candidates = _gateway_candidates(reader, supervisor, spec) + candidates = _gateway_candidates( + reader, + supervisor, + spec, + recovery_deadline, + ) current = candidates[0] if candidates else None if current is None: return None, None @@ -1583,60 +1916,98 @@ def _wait_for_recovery_candidate( def _control(action: str, nonce: str) -> tuple[str, int, int]: - with _control_stage("detect-agent"): - agent = _detect_agent() - with ProcReader() as reader: - with _control_stage("discover-supervisor"): - supervisor = _discover_supervisor(reader) - with _control_stage("initial-gateway-proof"): - spec = _agent_spec(agent, reader, supervisor) - candidates = _gateway_candidates(reader, supervisor, spec) - old_identity = candidates[0] if candidates else None - - with _control_stage("preflight"): - _preflight(spec, reader, supervisor) - - if action == "probe": - if old_identity is None: - raise ControlError("GATEWAY_HEALTH_TIMEOUT") - if not _gateway_healthy(reader, old_identity, spec): - raise ControlError("GATEWAY_HEALTH_TIMEOUT") - if not _gateway_auxiliaries_healthy(reader, old_identity, spec): - raise ControlError("GATEWAY_HEALTH_TIMEOUT") - healthy = reader.capture(old_identity.pid) - if healthy.stable_key() != old_identity.stable_key(): - raise ControlError("GATEWAY_HEALTH_TIMEOUT") - return "already-running", old_identity.pid, healthy.pid - - if action == "recover" and old_identity is not None: - # PID 1 continuously supervises the managed gateway. A host - # recovery request can arrive after PID 1 has launched a - # replacement but before its listener is healthy. Give that proven - # child a short grace period. If its identity changes during that - # grace, give the one successor its own bounded grace before any - # signal so recovery cannot churn a newly launched replacement. - original_identity = old_identity - with _control_stage("await-existing-gateway"): - existing, old_identity = _wait_for_recovery_candidate( + recovery_deadline = ( + None + if action == "probe" + else time.monotonic() + RECOVERY_TIMEOUT_SECONDS + ) + expected_exit_lock = None + expected_exit_lease = None + try: + if recovery_deadline is not None: + with _control_stage("acquire-expected-exit-lock"): + expected_exit_lock = _acquire_expected_exit_lock( + recovery_deadline + ) + + with _control_stage("detect-agent"): + agent = _detect_agent() + with ProcReader() as reader: + with _control_stage("discover-supervisor"): + supervisor = _discover_supervisor(reader) + with _control_stage("initial-gateway-proof"): + spec = _agent_spec(agent, reader, supervisor) + candidates = _gateway_candidates( reader, supervisor, spec, - original_identity, + recovery_deadline, ) - if existing is not None: + old_identity = candidates[0] if candidates else None + + with _control_stage("preflight"): + _preflight( + spec, + reader, + supervisor, + recovery_deadline, + ) + + if action == "probe": + if old_identity is None: + raise ControlError("GATEWAY_HEALTH_TIMEOUT") + if not _gateway_healthy(reader, old_identity, spec): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") + if not _gateway_auxiliaries_healthy( + reader, + old_identity, + spec, + ): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") + healthy = reader.capture(old_identity.pid) + if healthy.stable_key() != old_identity.stable_key(): + raise ControlError("GATEWAY_HEALTH_TIMEOUT") + return "already-running", old_identity.pid, healthy.pid + + assert recovery_deadline is not None + assert expected_exit_lock is not None + if action == "recover" and old_identity is not None: + # PID 1 continuously supervises the managed gateway. A host + # recovery request can arrive after PID 1 has launched a + # replacement but before its listener is healthy. Give that + # proven child a short grace period. If its identity changes + # during that grace, give the one successor its own bounded + # grace before any signal so recovery cannot churn a newly + # launched replacement. + original_identity = old_identity with _control_stage("await-existing-gateway"): - completed = _wait_for_healthy_gateway( + existing, old_identity = _wait_for_recovery_candidate( reader, supervisor, spec, - None, - RECOVERY_TIMEOUT_SECONDS, - True, + original_identity, + recovery_deadline, + ) + if existing is not None: + with _control_stage("await-existing-gateway"): + completed = _wait_for_healthy_gateway( + reader, + supervisor, + spec, + None, + _remaining_recovery_time( + recovery_deadline, + RECOVERY_TIMEOUT_SECONDS, + ), + True, + recovery_deadline=recovery_deadline, + ) + return ( + "already-running", + original_identity.pid, + completed.pid, ) - return "already-running", original_identity.pid, completed.pid - expected_exit_lease = None - try: if old_identity is not None: # The nonroot entrypoint owns the child and its crash budget, # and SIGTERM is indistinguishable from a self-requested @@ -1651,12 +2022,19 @@ def _control(action: str, nonce: str) -> tuple[str, int, int]: with _control_stage("publish-expected-exit"): controller_identity = _controller_process_identity(reader) expected_exit_lease = _publish_expected_exit_lease( + expected_exit_lock, old_identity, controller_identity, + recovery_deadline, ) + expected_exit_lock = None if old_identity is not None: with _control_stage("terminate-gateway"): - _terminate_gateway(reader, old_identity) + _terminate_gateway( + reader, + old_identity, + recovery_deadline, + ) with _control_stage("await-replacement"): replacement = _wait_for_healthy_gateway( @@ -1664,14 +2042,20 @@ def _control(action: str, nonce: str) -> tuple[str, int, int]: supervisor, spec, old_identity, - RECOVERY_TIMEOUT_SECONDS, + _remaining_recovery_time( + recovery_deadline, + RECOVERY_TIMEOUT_SECONDS, + ), True, + recovery_deadline=recovery_deadline, ) return "ok", old_identity.pid if old_identity else 0, replacement.pid - finally: - if expected_exit_lease is not None: - with _control_stage("cleanup-expected-exit"): - _clear_expected_exit_lease(expected_exit_lease) + finally: + if expected_exit_lease is not None: + with _control_stage("cleanup-expected-exit"): + _clear_expected_exit_lease(expected_exit_lease) + elif expected_exit_lock is not None: + _close_expected_exit_lock(expected_exit_lock) def _sanitize_start_log_diagnostic_line(line: str) -> str | None: diff --git a/test/managed-gateway-control-deadline.test.ts b/test/managed-gateway-control-deadline.test.ts new file mode 100644 index 00000000000..cf961a0f0b3 --- /dev/null +++ b/test/managed-gateway-control-deadline.test.ts @@ -0,0 +1,466 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { spawnSync } from "node:child_process"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; + +const HELPER = path.join(import.meta.dirname, "..", "scripts", "managed-gateway-control.py"); + +const CONTROL_DEADLINE_HARNESS = String.raw` +import importlib.util +import json +import os +import sys +import tempfile + +spec = importlib.util.spec_from_file_location("managed_control_deadline", sys.argv[1]) +control = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = control +spec.loader.exec_module(control) + + +def identity(pid, start_time, parent_pid, role): + uid = os.geteuid() + return control.ProcessIdentity( + pid=pid, + start_time=str(start_time), + parent_pid=parent_pid, + state="S", + uids=(uid,) * 4, + namespace_pid=pid, + namespace_inode=1, + cmdline=(role.encode("ascii"),), + proc_device=1, + proc_inode=pid, + ) + + +def error_code(operation): + try: + operation() + return "accepted" + except control.ControlError as error: + return error.code + + +gateway = identity(41, 333, 40, "gateway") +replacement = identity(43, 555, 40, "gateway") +supervisor = identity(40, 222, 1, "supervisor") +controller = identity(999, 777, 1, "controller") +forwarding_clock = [10.0] +control.time.monotonic = lambda: forwarding_clock[0] +lock = object() +lease = object() +observed = { + "lock": [], + "candidates": [], + "preflight": [], + "publication": [], + "termination": [], + "replacement": [], + "cleared": [], +} + + +class FakeProcReader: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + +def record_candidates(_reader, _supervisor, _spec, recovery_deadline=None): + observed["candidates"].append(recovery_deadline) + return [gateway] + + +def record_preflight(_spec, _reader, _supervisor, recovery_deadline=None): + observed["preflight"].append(recovery_deadline) + + +def record_publication( + received_lock, + received_gateway, + received_controller, + recovery_deadline=None, +): + assert received_lock is lock + assert received_gateway is gateway + assert received_controller is controller + observed["publication"].append(recovery_deadline) + return lease + + +def record_termination(_reader, received_gateway, recovery_deadline=None): + assert received_gateway is gateway + observed["termination"].append(recovery_deadline) + + +def record_replacement( + _reader, + _supervisor, + _spec, + received_gateway, + _timeout_seconds=control.RECOVERY_TIMEOUT_SECONDS, + _require_auxiliary_health=False, + recovery_deadline=None, +): + assert received_gateway is gateway + observed["replacement"].append(recovery_deadline) + return replacement + + +real_acquire_lock = control._acquire_expected_exit_lock +real_publish_lease = control._publish_expected_exit_lease +real_close_lock = control._close_expected_exit_lock +real_clear_lease = control._clear_expected_exit_lease +real_terminate_gateway = control._terminate_gateway +control.ProcReader = FakeProcReader +control._acquire_expected_exit_lock = lambda recovery_deadline: ( + observed["lock"].append(recovery_deadline) or lock +) +control._close_expected_exit_lock = lambda _lock: None +control._clear_expected_exit_lease = lambda received_lease: observed["cleared"].append( + received_lease is lease +) +control._detect_agent = lambda: "hermes" +control._discover_supervisor = lambda _reader: supervisor +control._agent_spec = lambda *_args: control.AgentSpec("hermes", 18642) +control._gateway_candidates = record_candidates +control._preflight = record_preflight +control._controller_process_identity = lambda _reader: controller +control._publish_expected_exit_lease = record_publication +control._terminate_gateway = record_termination +control._wait_for_healthy_gateway = record_replacement + +forwarded_result = control._control("restart", "a" * 64) +forwarded_deadlines = [ + *observed["lock"], + *observed["candidates"], + *observed["preflight"], + *observed["publication"], + *observed["termination"], + *observed["replacement"], +] +shared_deadline = ( + len(forwarded_deadlines) == 6 + and len(set(forwarded_deadlines)) == 1 + and forwarded_deadlines[0] > forwarding_clock[0] + and observed["cleared"] == [True] +) + +preflight_clock = [0.0] +control.time.monotonic = lambda: preflight_clock[0] +fixed_deadlines = [] +hash_checks = [] + + +class EnvironmentReader: + def read_stable_file(self, _identity, _name, _limit): + return b"SAFE=1\0" + + +real_system_path = control._system_path +real_exists = control.os.path.exists +control._system_path = lambda value: value +control.os.path.exists = lambda _path: True +control._run_fixed_validator = ( + lambda _script, _arguments, recovery_deadline=None: fixed_deadlines.append( + recovery_deadline + ) +) + + +def expire_runtime_validation(_script, _environment): + preflight_clock[0] = 1.0 + + +control._validate_runtime_environment = expire_runtime_validation +control._verify_locked_hermes_hash = lambda: hash_checks.append("called") +preflight_after_validation = error_code( + lambda: control._hermes_preflight(EnvironmentReader(), supervisor, 1.0) +) +after_validation_result = [ + preflight_after_validation, + fixed_deadlines == [1.0], + len(hash_checks), +] +fixed_deadlines.clear() +preflight_before_validation = error_code( + lambda: control._hermes_preflight(EnvironmentReader(), supervisor, 1.0) +) +before_validation_result = [ + preflight_before_validation, + len(fixed_deadlines), +] +control._system_path = real_system_path +control.os.path.exists = real_exists + +control._acquire_expected_exit_lock = real_acquire_lock +control._publish_expected_exit_lease = real_publish_lease +control._close_expected_exit_lock = real_close_lock +control._clear_expected_exit_lease = real_clear_lease +control._terminate_gateway = real_terminate_gateway +with tempfile.TemporaryDirectory() as root: + proc_root = os.path.join(root, "proc") + system_root = os.path.join(root, "system") + os.makedirs(proc_root) + os.makedirs(os.path.join(system_root, "run")) + os.environ["NEMOCLAW_MANAGED_CONTROL_ALLOW_NONROOT_TEST"] = "1" + os.environ["NEMOCLAW_MANAGED_CONTROL_PROC_ROOT"] = proc_root + os.environ["NEMOCLAW_MANAGED_CONTROL_SYSTEM_ROOT"] = system_root + marker_path = os.path.join( + system_root, + "run/nemoclaw", + control.EXPECTED_EXIT_MARKER_NAME, + ) + marker_clock = [0.0] + control.time.monotonic = lambda: marker_clock[0] + publication_lock = control._acquire_expected_exit_lock(1.0) + real_path_matches = control._lease_path_matches + + def expire_after_marker_validation(*arguments): + result = real_path_matches(*arguments) + marker_clock[0] = 1.0 + return result + + control._lease_path_matches = expire_after_marker_validation + late_lease = None + try: + late_lease = control._publish_expected_exit_lease( + publication_lock, + gateway, + controller, + 1.0, + ) + publication_status = "accepted" + except control.ControlError as error: + publication_status = error.code + finally: + control._lease_path_matches = real_path_matches + if late_lease is None: + control._close_expected_exit_lock(publication_lock) + else: + control._clear_expected_exit_lease(late_lease) + publication_result = [ + publication_status, + not os.path.exists(marker_path), + ] + +closed_pidfds = [] +control.os.close = lambda pidfd: closed_pidfds.append(pidfd) +control._pidfd_open = lambda _pid: 50 +control._pidfd_exited = lambda _pidfd, _timeout: False +termination_signals = [] +control._send_pidfd = lambda _pidfd, signum: ( + termination_signals.append(int(signum)) or True +) +termination_clock = [1.0] +control.time.monotonic = lambda: termination_clock[0] +recapture_calls = [] + + +def record_recapture(_reader, _identity, *, deadline=None): + recapture_calls.append(deadline) + return gateway + + +control._recapture_exact_identity = record_recapture +before_recapture_status = error_code( + lambda: control._terminate_gateway(object(), gateway, 1.0) +) +before_recapture_result = [ + before_recapture_status, + list(recapture_calls), + list(termination_signals), +] + +termination_clock[0] = 0.0 +recapture_calls.clear() +termination_signals.clear() + + +def expire_during_recapture(_reader, _identity, *, deadline=None): + recapture_calls.append(deadline) + termination_clock[0] = 1.0 + return gateway + + +control._recapture_exact_identity = expire_during_recapture +during_recapture_status = error_code( + lambda: control._terminate_gateway(object(), gateway, 1.0) +) +during_recapture_result = [ + during_recapture_status, + list(recapture_calls), + list(termination_signals), +] + +print(json.dumps({ + "forwarding": [forwarded_result, shared_deadline], + "preflight_after_validation": after_validation_result, + "preflight_before_validation": before_validation_result, + "late_publication": publication_result, + "termination_before_recapture": before_recapture_result, + "termination_during_recapture": during_recapture_result, +}, sort_keys=True)) +`; + +const HTTP_DEADLINE_HARNESS = String.raw` +import importlib.util +import json +import sys + +spec = importlib.util.spec_from_file_location("managed_control_http_deadline", sys.argv[1]) +control = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = control +spec.loader.exec_module(control) + +real_connection = control.http.client.HTTPConnection +clock = [0.0] +control.time.monotonic = lambda: clock[0] +active = {} + + +def advance(duration, timeout=None): + if timeout is not None and duration >= timeout: + clock[0] += timeout + raise control.socket.timeout("deadline reached") + clock[0] += duration + + +class ScriptedSocket: + def __init__(self, chunks, request_delay): + self.chunks = list(chunks) + self.request_delay = request_delay + self.timeout = None + self.closed = False + + def settimeout(self, timeout): + self.timeout = timeout + + def sendall(self, _data): + advance(self.request_delay, self.timeout) + + def recv_into(self, buffer): + if not self.chunks: + return 0 + delay, payload = self.chunks.pop(0) + advance(delay, self.timeout) + count = min(len(buffer), len(payload)) + buffer[:count] = payload[:count] + if count < len(payload): + self.chunks.insert(0, (0.0, payload[count:])) + return count + + def close(self): + self.closed = True + + +class ScriptedConnection(real_connection): + def connect(self): + advance(active["connect_delay"], self.timeout) + self.sock = ScriptedSocket( + active["chunks"], + active["request_delay"], + ) + + +def byte_chunks(payload, delay): + return [(delay, bytes((value,))) for value in payload] + + +status = b"HTTP/1.1 200 OK\r\n" +unauthorized = b"HTTP/1.1 401 Unauthorized\r\n" +headers = b"Content-Length: 4\r\n\r\n" +body = b"pong" +complete = status + headers + body +control.http.client.HTTPConnection = ScriptedConnection + + +def check(scenario): + active.clear() + active.update(scenario) + clock[0] = 0.0 + return control._http_healthy(18642, "/health", 1.0) + + +results = { + "healthy": check({ + "connect_delay": 0.0, + "request_delay": 0.0, + "chunks": [(0.0, complete)], + }), + "unauthorized": check({ + "connect_delay": 0.0, + "request_delay": 0.0, + "chunks": [(0.0, unauthorized + headers + body)], + }), + "slow_connect": check({ + "connect_delay": 1.0, + "request_delay": 0.0, + "chunks": [(0.0, complete)], + }), + "slow_request": check({ + "connect_delay": 0.0, + "request_delay": 1.0, + "chunks": [(0.0, complete)], + }), + "slow_status": check({ + "connect_delay": 0.0, + "request_delay": 0.0, + "chunks": byte_chunks(status, 0.2) + [(0.0, headers + body)], + }), + "slow_headers": check({ + "connect_delay": 0.0, + "request_delay": 0.0, + "chunks": [(0.0, status)] + byte_chunks(headers, 0.2) + [(0.0, body)], + }), + "slow_body": check({ + "connect_delay": 0.0, + "request_delay": 0.0, + "chunks": [(0.0, status + headers)] + byte_chunks(body, 0.3), + }), +} + +print(json.dumps(results, sort_keys=True)) +`; + +function runHarness(source: string): unknown { + const result = spawnSync("python3", ["-c", source, HELPER], { + encoding: "utf8", + timeout: 30_000, + killSignal: "SIGKILL", + }); + + expect(result.error, result.error?.stack ?? result.stderr).toBeUndefined(); + expect(result.status, result.stderr).toBe(0); + return JSON.parse(result.stdout); +} + +describe("managed gateway recovery deadline", () => { + it("stops preflight, marker publication, and signaling at the recovery deadline (#8262)", () => { + expect(runHarness(CONTROL_DEADLINE_HARNESS)).toEqual({ + forwarding: [["ok", 41, 43], true], + preflight_after_validation: ["GATEWAY_FAILED", true, 0], + preflight_before_validation: ["GATEWAY_FAILED", 0], + late_publication: ["GATEWAY_FAILED", true], + termination_before_recapture: ["GATEWAY_FAILED", [], []], + termination_during_recapture: ["GATEWAY_FAILED", [1], []], + }); + }); + + it("applies one recovery deadline to every HTTP health check phase (#8262)", () => { + expect(runHarness(HTTP_DEADLINE_HARNESS)).toEqual({ + healthy: true, + slow_body: false, + slow_connect: false, + slow_headers: false, + slow_request: false, + slow_status: false, + unauthorized: true, + }); + }); +}); diff --git a/test/managed-gateway-control-locking.test.ts b/test/managed-gateway-control-locking.test.ts new file mode 100644 index 00000000000..22249a8b7b9 --- /dev/null +++ b/test/managed-gateway-control-locking.test.ts @@ -0,0 +1,372 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { spawnSync } from "node:child_process"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; + +const HELPER = path.join(import.meta.dirname, "..", "scripts", "managed-gateway-control.py"); + +const LOCKING_HARNESS = String.raw` +import importlib.util +import json +import os +import sys +import tempfile +import threading + +spec = importlib.util.spec_from_file_location("managed_control_locking", sys.argv[1]) +control = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = control +spec.loader.exec_module(control) + +def identity(pid, start_time, parent_pid, role): + uid = os.geteuid() + return control.ProcessIdentity( + pid=pid, + start_time=str(start_time), + parent_pid=parent_pid, + state="S", + uids=(uid,) * 4, + namespace_pid=pid, + namespace_inode=1, + cmdline=(role.encode("ascii"),), + proc_device=1, + proc_inode=pid, + ) + +with tempfile.TemporaryDirectory() as root: + proc_root = os.path.join(root, "proc") + system_root = os.path.join(root, "system") + os.makedirs(proc_root) + os.makedirs(os.path.join(system_root, "run")) + os.environ["NEMOCLAW_MANAGED_CONTROL_ALLOW_NONROOT_TEST"] = "1" + os.environ["NEMOCLAW_MANAGED_CONTROL_PROC_ROOT"] = proc_root + os.environ["NEMOCLAW_MANAGED_CONTROL_SYSTEM_ROOT"] = system_root + + supervisor = identity(40, 222, 1, "supervisor") + gateway_41 = identity(41, 333, 40, "gateway") + gateway_43 = identity(43, 555, 40, "gateway") + gateway_44 = identity(44, 666, 40, "gateway") + controller = identity(999, 777, 1, "controller") + marker_path = os.path.join( + system_root, + "run/nemoclaw", + control.EXPECTED_EXIT_MARKER_NAME, + ) + lock_path = os.path.join( + system_root, + "run/nemoclaw", + control.EXPECTED_EXIT_LOCK_NAME, + ) + + def publish(identity_to_authorize): + lock = control._acquire_expected_exit_lock( + control.time.monotonic() + control.RECOVERY_TIMEOUT_SECONDS + ) + try: + return control._publish_expected_exit_lease( + lock, + identity_to_authorize, + controller, + ) + except Exception: + control._close_expected_exit_lock(lock) + raise + + active_lease = publish(gateway_41) + contention_seen = threading.Event() + release_complete = threading.Event() + real_flock = control.fcntl.flock + def observe_contention(fd, operation): + try: + return real_flock(fd, operation) + except BlockingIOError: + contention_seen.set() + raise + def release_after_contention(): + if contention_seen.wait(1.0): + control._clear_expected_exit_lease(active_lease) + release_complete.set() + control.fcntl.flock = observe_contention + release_thread = threading.Thread(target=release_after_contention) + release_thread.start() + try: + waited_lease = publish(gateway_41) + finally: + release_thread.join(1.0) + control.fcntl.flock = real_flock + if release_thread.is_alive() or not release_complete.is_set(): + raise AssertionError("active expected-exit lease was not released") + serialized_lease = ["waited", contention_seen.is_set()] + control._clear_expected_exit_lease(waited_lease) + + previous_timeout = control.RECOVERY_TIMEOUT_SECONDS + control.RECOVERY_TIMEOUT_SECONDS = 0.01 + timeout_clock = [0.0] + original_time = control.time.monotonic, control.time.sleep + control.time.monotonic = lambda: timeout_clock[0] + control.time.sleep = lambda duration: timeout_clock.__setitem__( + 0, + timeout_clock[0] + duration, + ) + timeout_lease = publish(gateway_41) + try: + try: + publish(gateway_41) + lock_timeout = "accepted" + except control.ControlError as error: + lock_timeout = error.code + finally: + control._clear_expected_exit_lease(timeout_lease) + control.RECOVERY_TIMEOUT_SECONDS = previous_timeout + control.time.monotonic, control.time.sleep = original_time + lock_timeout_result = [lock_timeout, timeout_clock[0]] + + late_clock = [0.0] + original_monotonic = control.time.monotonic + real_flock = control.fcntl.flock + def acquire_at_deadline(fd, operation): + result = real_flock(fd, operation) + late_clock[0] = 0.01 + return result + control.time.monotonic = lambda: late_clock[0] + control.fcntl.flock = acquire_at_deadline + late_lock = None + try: + try: + late_lock = control._acquire_expected_exit_lock(0.01) + late_acquisition = "accepted" + except control.ControlError as error: + late_acquisition = error.code + finally: + if late_lock is not None: + control._close_expected_exit_lock(late_lock) + control.fcntl.flock = real_flock + control.time.monotonic = original_monotonic + + orphaned_lease = publish(gateway_41) + orphaned_inode = os.stat(marker_path, follow_symlinks=False).st_ino + os.close(orphaned_lease.marker_fd) + os.close(orphaned_lease.lock_fd) + os.close(orphaned_lease.directory_fd) + untrusted_marker_fd = os.open(marker_path, os.O_RDONLY) + control.fcntl.flock(untrusted_marker_fd, control.fcntl.LOCK_SH) + recovered_lease = publish(gateway_41) + marker_flock_cannot_pin = ( + os.stat(marker_path, follow_symlinks=False).st_ino != orphaned_inode + ) + control.fcntl.flock(untrusted_marker_fd, control.fcntl.LOCK_UN) + os.close(untrusted_marker_fd) + control._clear_expected_exit_lease(recovered_lease) + + original_lease = publish(gateway_41) + os.unlink(marker_path) + with open(marker_path, "w", encoding="ascii") as stream: + stream.write("replacement\n") + os.chmod(marker_path, 0o444) + replacement_inode = os.stat(marker_path, follow_symlinks=False).st_ino + control._clear_expected_exit_lease(original_lease) + inode_safe_cleanup = ( + os.path.exists(marker_path) + and os.stat(marker_path, follow_symlinks=False).st_ino + == replacement_inode + ) + os.unlink(marker_path) + + os.unlink(lock_path) + original_umask = os.umask(0o777) + try: + restrictive_umask_lease = publish(gateway_41) + finally: + os.umask(original_umask) + restrictive_umask_modes = [ + os.stat(marker_path, follow_symlinks=False).st_mode & 0o777, + os.stat(lock_path, follow_symlinks=False).st_mode & 0o777, + ] + control._clear_expected_exit_lease(restrictive_umask_lease) + + original_open_runtime_directory = control._open_managed_runtime_directory + original_open_expected_exit_lock = control._open_expected_exit_lock + original_expected_exit_lock = control.ExpectedExitLock + original_close = control.os.close + constructor_cleanup = [] + control._open_managed_runtime_directory = lambda: 101 + control._open_expected_exit_lock = lambda *_args: 202 + control.ExpectedExitLock = lambda **_kwargs: (_ for _ in ()).throw( + RuntimeError("lock record construction failed") + ) + control.os.close = lambda fd: constructor_cleanup.append(fd) + try: + try: + control._acquire_expected_exit_lock( + control.time.monotonic() + control.RECOVERY_TIMEOUT_SECONDS + ) + raise AssertionError("lock record construction unexpectedly succeeded") + except RuntimeError as error: + if str(error) != "lock record construction failed": + raise + finally: + control._open_managed_runtime_directory = original_open_runtime_directory + control._open_expected_exit_lock = original_open_expected_exit_lock + control.ExpectedExitLock = original_expected_exit_lock + control.os.close = original_close + + current_gateway = [gateway_41] + control._proc_root = lambda: proc_root + control._detect_agent = lambda: "hermes" + control._discover_supervisor = lambda _reader: supervisor + control._agent_spec = lambda *_args: control.AgentSpec("hermes", 18789) + control._gateway_candidates = lambda *_args: [current_gateway[0]] + control._preflight = lambda *_args: None + control._controller_process_identity = lambda _reader: controller + control._wait_for_healthy_gateway = lambda *_args, **_kwargs: current_gateway[0] + + observed_marker = [] + def replace_after_wait(_reader, gateway, _recovery_deadline=None): + if gateway.stable_key() != gateway_43.stable_key(): + raise AssertionError("second controller used a stale gateway proof") + with open(marker_path, "r", encoding="ascii") as stream: + observed_marker.extend(stream.read().split()) + current_gateway[0] = gateway_44 + original_terminate_gateway = control._terminate_gateway + control._terminate_gateway = replace_after_wait + + first_controller_lock = control._acquire_expected_exit_lock( + control.time.monotonic() + control.RECOVERY_TIMEOUT_SECONDS + ) + controller_waiting = threading.Event() + real_flock = control.fcntl.flock + def observe_controller_contention(fd, operation): + try: + return real_flock(fd, operation) + except BlockingIOError: + controller_waiting.set() + raise + control.fcntl.flock = observe_controller_contention + contended_results = [] + contended_errors = [] + def run_contended_restart(): + try: + contended_results.append(control._control("restart", "9" * 64)) + except BaseException as error: + contended_errors.append(error) + contended_thread = threading.Thread(target=run_contended_restart) + first_lock_closed = False + try: + contended_thread.start() + if not controller_waiting.wait(1.0): + raise AssertionError("second controller did not wait for the first") + current_gateway[0] = gateway_43 + control._close_expected_exit_lock(first_controller_lock) + first_lock_closed = True + contended_thread.join(5.0) + finally: + if not first_lock_closed: + control._close_expected_exit_lock(first_controller_lock) + contended_thread.join(5.0) + control.fcntl.flock = real_flock + control._terminate_gateway = original_terminate_gateway + if contended_thread.is_alive(): + raise AssertionError("second controller did not finish") + if contended_errors: + raise contended_errors[0] + contended_restart = [ + contended_results[0], + observed_marker, + not os.path.exists(marker_path), + ] + + read_fd, write_fd = os.pipe() + deadline_clock = [0.0] + deadline_waits = [] + deadline_signals = [] + control._pidfd_open = lambda _pid: os.dup(read_fd) + control._recapture_exact_identity = lambda *_args, **_kwargs: gateway_41 + control.time.monotonic = lambda: deadline_clock[0] + control._send_pidfd = lambda _pidfd, signum: ( + deadline_signals.append(int(signum)) or True + ) + def record_deadline_wait(_pidfd, timeout_seconds): + deadline_waits.append(timeout_seconds) + deadline_clock[0] = 3.0 + return False + control._pidfd_exited = record_deadline_wait + try: + try: + control._terminate_gateway(object(), gateway_41, 3.0) + termination_deadline = "accepted" + except control.ControlError as error: + termination_deadline = [ + error.code, + deadline_waits, + deadline_signals, + ] + + accounted_clock = [0.0] + accounted_waits = [] + accounted_signals = [] + control.time.monotonic = lambda: accounted_clock[0] + control._send_pidfd = lambda _pidfd, signum: ( + accounted_signals.append(int(signum)) or True + ) + def record_accounted_exit(_pidfd, timeout_seconds): + accounted_waits.append(timeout_seconds) + accounted_clock[0] = 3.0 + return len(accounted_waits) > 1 + control._pidfd_exited = record_accounted_exit + try: + control._terminate_gateway(object(), gateway_41, 3.0) + termination_accounted = [ + "accounted", + accounted_waits, + accounted_signals, + ] + except control.ControlError as error: + termination_accounted = [ + error.code, + accounted_waits, + accounted_signals, + ] + finally: + os.close(read_fd) + os.close(write_fd) + + print(json.dumps({ + "serialized_lease": serialized_lease, + "lock_timeout": lock_timeout_result, + "late_acquisition": late_acquisition, + "marker_flock_cannot_pin": marker_flock_cannot_pin, + "inode_safe_cleanup": inode_safe_cleanup, + "restrictive_umask_modes": restrictive_umask_modes, + "constructor_cleanup": constructor_cleanup, + "contended_restart": contended_restart, + "termination_deadline": termination_deadline, + "termination_accounted": termination_accounted, + })) +`; + +describe("managed gateway lifecycle locking", () => { + it("acquires the expected-exit lock before gateway inspection and enforces one recovery deadline (#8262)", () => { + const result = spawnSync("python3", ["-c", LOCKING_HARNESS, HELPER], { + encoding: "utf8", + timeout: 30_000, + killSignal: "SIGKILL", + }); + + expect(result.error, result.error?.stack ?? result.stderr).toBeUndefined(); + expect(result.status, result.stderr).toBe(0); + expect(JSON.parse(result.stdout)).toEqual({ + serialized_lease: ["waited", true], + lock_timeout: ["SUPERVISOR_BUSY", 0.01], + late_acquisition: "SUPERVISOR_BUSY", + marker_flock_cannot_pin: true, + inode_safe_cleanup: true, + restrictive_umask_modes: [0o444, 0o600], + constructor_cleanup: [202, 101], + contended_restart: [["ok", 43, 44], ["v1", "43", "555", "999", "777"], true], + termination_deadline: ["GATEWAY_FAILED", [3, 0], [15]], + termination_accounted: ["accounted", [3, 0], [15]], + }); + }); +}); diff --git a/test/managed-gateway-control.test.ts b/test/managed-gateway-control.test.ts index 862edd3e293..f31ee4d5fa0 100644 --- a/test/managed-gateway-control.test.ts +++ b/test/managed-gateway-control.test.ts @@ -181,7 +181,7 @@ with tempfile.TemporaryDirectory() as root: control._sandbox_uid = lambda: 1000 control._http_healthy_in_gateway_namespace = ( - lambda _reader, _identity, port, path: (port, path) in { + lambda _reader, _identity, port, path, *_args: (port, path) in { (18642, "/health"), (8642, "/health"), } @@ -416,7 +416,7 @@ with tempfile.TemporaryDirectory() as root: real_validator = control._run_fixed_validator real_runtime_validator = control._validate_runtime_environment real_hash_check = control._verify_locked_hermes_hash - control._run_fixed_validator = lambda script, arguments: preflight_steps.append({ + control._run_fixed_validator = lambda script, arguments, _recovery_deadline=None: preflight_steps.append({ "script": script, "arguments": arguments, }) @@ -665,8 +665,6 @@ with tempfile.TemporaryDirectory() as root: control._preflight = lambda *_args: None control._http_healthy_in_gateway_namespace = lambda *_args: True real_terminate = control._terminate_gateway - with control.ProcReader(proc_root) as controller_reader: - controller_identity = control._controller_process_identity(controller_reader) lease_path = os.path.join( system_root, "run/nemoclaw", @@ -703,7 +701,7 @@ with tempfile.TemporaryDirectory() as root: and lock_metadata.st_nlink == 1 ), }) - def replace_gateway(_reader, identity): + def replace_gateway(_reader, identity, _recovery_deadline=None): assert identity.pid == 41 observe_expected_exit_lease(identity, "restart") remove_process(proc_root, 41) @@ -718,69 +716,6 @@ with tempfile.TemporaryDirectory() as root: listener_inode="77777", ) - active_lease = control._publish_expected_exit_lease( - expected_gateway, - controller_identity, - ) - try: - control._publish_expected_exit_lease(expected_gateway, controller_identity) - active_controller_lock = "replaced" - except control.ControlError as error: - active_controller_lock = error.code - control._clear_expected_exit_lease(active_lease) - - orphaned_lease = control._publish_expected_exit_lease( - expected_gateway, - controller_identity, - ) - orphaned_inode = os.stat(lease_path, follow_symlinks=False).st_ino - os.close(orphaned_lease.marker_fd) - os.close(orphaned_lease.lock_fd) - os.close(orphaned_lease.directory_fd) - untrusted_marker_fd = os.open(lease_path, os.O_RDONLY) - control.fcntl.flock(untrusted_marker_fd, control.fcntl.LOCK_SH) - recovered_lease = control._publish_expected_exit_lease( - expected_gateway, - controller_identity, - ) - marker_flock_cannot_pin = ( - os.stat(lease_path, follow_symlinks=False).st_ino != orphaned_inode - ) - control.fcntl.flock(untrusted_marker_fd, control.fcntl.LOCK_UN) - os.close(untrusted_marker_fd) - control._clear_expected_exit_lease(recovered_lease) - - original_lease = control._publish_expected_exit_lease( - expected_gateway, - controller_identity, - ) - os.unlink(lease_path) - with open(lease_path, "w", encoding="ascii") as stream: - stream.write(f"v1 41 333 {controller_pid} {controller_start_time}\n") - os.chmod(lease_path, 0o444) - replacement_inode = os.stat(lease_path, follow_symlinks=False).st_ino - control._clear_expected_exit_lease(original_lease) - inode_safe_cleanup = ( - os.path.exists(lease_path) - and os.stat(lease_path, follow_symlinks=False).st_ino == replacement_inode - ) - os.unlink(lease_path) - - os.unlink(lock_path) - original_umask = os.umask(0o777) - try: - restrictive_umask_lease = control._publish_expected_exit_lease( - expected_gateway, - controller_identity, - ) - finally: - os.umask(original_umask) - restrictive_umask_modes = [ - os.stat(lease_path, follow_symlinks=False).st_mode & 0o777, - os.stat(lock_path, follow_symlinks=False).st_mode & 0o777, - ] - control._clear_expected_exit_lease(restrictive_umask_lease) - control._terminate_gateway = replace_gateway try: restarted = control._control("restart", "a" * 64) @@ -795,9 +730,13 @@ with tempfile.TemporaryDirectory() as root: control._detect_agent = lambda: "openclaw" control._agent_spec = lambda *_args: control.AgentSpec("openclaw", 18642) control._gateway_candidates = lambda reader, *_args: [reader.capture(43)] - control._wait_for_healthy_gateway = lambda reader, *_args: reader.capture(43) - control._terminate_gateway = lambda _reader, identity: observe_expected_exit_lease( - identity, "openclaw-restart" + control._wait_for_healthy_gateway = ( + lambda reader, *_args, **_kwargs: reader.capture(43) + ) + control._terminate_gateway = ( + lambda _reader, identity, _recovery_deadline=None: ( + observe_expected_exit_lease(identity, "openclaw-restart") + ) ) try: openclaw_restart = control._control("restart", "f" * 64) @@ -819,6 +758,7 @@ with tempfile.TemporaryDirectory() as root: old_identity, timeout_seconds=control.RECOVERY_TIMEOUT_SECONDS, require_auxiliary_health=False, + **_kwargs, ): timeout_refresh_waits.append([ old_identity.pid if old_identity else 0, @@ -841,7 +781,11 @@ with tempfile.TemporaryDirectory() as root: if len(timeout_refresh_waits) == 2: raise control.ControlError("GATEWAY_HEALTH_TIMEOUT") return reader.capture(45) - def terminate_refreshed_gateway(_reader, identity): + def terminate_refreshed_gateway( + _reader, + identity, + _recovery_deadline=None, + ): timeout_refresh_signals.append(identity.pid) assert identity.pid == 44 observe_expected_exit_lease(identity, "unhealthy-recover") @@ -909,7 +853,13 @@ with tempfile.TemporaryDirectory() as root: real_http_health = control._http_healthy_in_gateway_namespace public_health_attempts = [] - def delayed_public_health(_reader, _identity, port, path): + def delayed_public_health( + _reader, + _identity, + port, + path, + _recovery_deadline=None, + ): if (port, path) == (8642, "/health"): public_health_attempts.append("attempt") return len(public_health_attempts) >= 2 @@ -931,9 +881,61 @@ with tempfile.TemporaryDirectory() as root: finally: control._http_healthy_in_gateway_namespace = real_http_health + deadline_clock = [0.0] + deadline_health_calls = [] + real_monotonic = control.time.monotonic + real_owns_listener = control._owns_listener + real_http_health = control._http_healthy_in_gateway_namespace + control.time.monotonic = lambda: deadline_clock[0] + control._owns_listener = lambda *_args: True + def health_finishes_after_deadline( + _reader, + _identity, + port, + path, + recovery_deadline=None, + ): + deadline_health_calls.append([port, path, recovery_deadline]) + if (port, path) == (8642, "/health"): + deadline_clock[0] = 1.1 + return True + control._http_healthy_in_gateway_namespace = health_finishes_after_deadline + try: + with control.ProcReader(proc_root) as deadline_reader: + try: + control._wait_for_healthy_gateway( + deadline_reader, + supervisor, + control.AgentSpec( + "hermes", + 18642, + readiness_checks=((8642, "/health"),), + ), + None, + 1.0, + True, + 1.0, + ) + deadline_health = "accepted" + except control.ControlError as error: + deadline_health = [ + error.code, + deadline_clock[0], + deadline_health_calls, + ] + finally: + control.time.monotonic = real_monotonic + control._owns_listener = real_owns_listener + control._http_healthy_in_gateway_namespace = real_http_health + auxiliary_attempts = [] real_auxiliary_health = control._gateway_auxiliaries_healthy - def replace_during_auxiliary_check(_reader, identity, _spec): + def replace_during_auxiliary_check( + _reader, + identity, + _spec, + _recovery_deadline=None, + ): auxiliary_attempts.append(identity.pid) if identity.pid == 43: remove_process(proc_root, 43) @@ -1226,12 +1228,6 @@ with tempfile.TemporaryDirectory() as root: "recovered": recovered, "probed": probed, "openclaw_restart": openclaw_restart, - "lease_races": [ - active_controller_lock, - marker_flock_cannot_pin, - inode_safe_cleanup, - restrictive_umask_modes, - ], "expected_exit_leases": [ lease_observations, restart_lease_cleared, @@ -1246,6 +1242,7 @@ with tempfile.TemporaryDirectory() as root: "inflight_recovery": [inflight_recovery, len(inflight_health_attempts)], "transient_retry": [retried_pid, len(health_attempts)], "public_readiness_retry": [readiness_pid, len(public_health_attempts)], + "deadline_health": deadline_health, "auxiliary_replacement": [auxiliary_replacement, auxiliary_attempts], "source_seams": [source_proc, source_system], "disabled_source_seams": [disabled_source_proc, disabled_source_system], @@ -1353,7 +1350,6 @@ describe("managed gateway root control", () => { recovered: ["already-running", 43, 43], probed: ["already-running", 43, 43], openclaw_restart: ["ok", 43, 43], - lease_races: ["SUPERVISOR_BUSY", true, true, [0o444, 0o600]], expected_exit_leases: [ [ { @@ -1382,12 +1378,20 @@ describe("managed gateway root control", () => { [ [0, 10, false], [0, 10, false], - [44, 150, true], + [44, expect.any(Number), true], ], ], inflight_recovery: [["already-running", 43, 43], 4], transient_retry: [43, 2], public_readiness_retry: [43, 2], + deadline_health: [ + "GATEWAY_HEALTH_TIMEOUT", + 1.1, + [ + [18642, "/health", 1], + [8642, "/health", 1], + ], + ], auxiliary_replacement: [44, [43, 44]], source_seams: ["/attacker/proc", "/attacker/root"], disabled_source_seams: ["/proc", "/"], @@ -1457,6 +1461,8 @@ describe("managed gateway root control", () => { [1, ["SUPERVISOR_NOT_RUNNING"]], ], }); + expect(output.timeout_refresh[2][2][1]).toBeGreaterThan(0); + expect(output.timeout_refresh[2][2][1]).toBeLessThanOrEqual(150); expect(output.start_log_security.installed_topology[0]).not.toBe( output.start_log_security.installed_topology[1], ); diff --git a/test/openclaw-managed-restart-respawn.test.ts b/test/openclaw-managed-restart-respawn.test.ts index 99e56171d65..eb6d9ea6c87 100644 --- a/test/openclaw-managed-restart-respawn.test.ts +++ b/test/openclaw-managed-restart-respawn.test.ts @@ -128,26 +128,36 @@ with tempfile.TemporaryDirectory() as root: # openclaw is the detected agent, and its preflight must not gate the lease. control._detect_agent = lambda: "openclaw" - control._openclaw_preflight = lambda: None + control._openclaw_preflight = lambda _recovery_deadline=None: None control._sandbox_uid = lambda: 1000 - control._http_healthy_in_gateway_namespace = lambda _reader, _identity, port, path: True + control._http_healthy_in_gateway_namespace = ( + lambda _reader, _identity, port, path, _recovery_deadline=None: True + ) os.environ["NEMOCLAW_MANAGED_CONTROL_ALLOW_NONROOT_TEST"] = "1" os.environ["NEMOCLAW_MANAGED_CONTROL_SYSTEM_ROOT"] = system_root os.environ["NEMOCLAW_MANAGED_CONTROL_PROC_ROOT"] = proc_root lease_path = os.path.join(system_root, "run/nemoclaw", control.EXPECTED_EXIT_MARKER_NAME) - observed = {"lease_live_during_terminate": False, "payload": None} + observed = { + "lease_live_during_terminate": False, + "payload": None, + "recovery_deadline_provided": False, + } - def fake_terminate(_reader, identity): + def fake_terminate(_reader, identity, recovery_deadline=None): # The entrypoint reads the lease while the controller waits, so it must # exist and name this exact gateway at signal time — not after the wait. + observed["recovery_deadline_provided"] = ( + recovery_deadline is not None + and recovery_deadline > control.time.monotonic() + ) observed["lease_live_during_terminate"] = os.path.exists(lease_path) if observed["lease_live_during_terminate"]: with open(lease_path, "r", encoding="ascii") as stream: version, pid, start_time, controller, _controller_start = stream.read().split() observed["payload"] = [version, int(pid), start_time, int(controller)] - def fake_wait(_reader, _supervisor, _spec, old, _timeout=0, _aux=False): + def fake_wait(_reader, _supervisor, _spec, old, *_args, **_kwargs): return replace(old, pid=43, start_time="555", namespace_pid=43) control._terminate_gateway = fake_terminate @@ -158,6 +168,7 @@ with tempfile.TemporaryDirectory() as root: "result": result, "old_pid": old_pid, "new_pid": new_pid, + "recovery_deadline_provided": observed["recovery_deadline_provided"], "lease_live_during_terminate": observed["lease_live_during_terminate"], "payload": observed["payload"], "controller_pid": controller_pid, @@ -220,6 +231,7 @@ describe("openclaw managed restart respawn (#6868)", () => { // openclaw's SIGTERM exit read as an intentional stop and never respawned. expect(observed.lease_live_during_terminate).toBe(true); expect(observed.payload).toEqual(["v1", 41, "333", observed.controller_pid]); + expect(observed.recovery_deadline_provided).toBe(true); expect(observed.result).toBe("ok"); expect(observed.old_pid).toBe(41); expect(observed.new_pid).toBe(43);