From c5f02e0dbacb51cca6079de4c34934b02cdbc759 Mon Sep 17 00:00:00 2001 From: Nick Huo Date: Mon, 21 Sep 2026 17:41:33 -0700 Subject: [PATCH] fix(core): fail deploy when controllers stay unhealthy --- packages/cli/canyonos/deploy.py | 10 +-- packages/cli/tests/test_deploy_progress.py | 18 ++++- .../controller/global_controller.py | 29 +++++--- .../tests/test_global_controller_readiness.py | 72 +++++++++++++++++++ 4 files changed, 113 insertions(+), 16 deletions(-) create mode 100644 packages/core/tests/test_global_controller_readiness.py diff --git a/packages/cli/canyonos/deploy.py b/packages/cli/canyonos/deploy.py index d8bd696b..ba97f3be 100644 --- a/packages/cli/canyonos/deploy.py +++ b/packages/cli/canyonos/deploy.py @@ -141,9 +141,7 @@ def feed(self, line): return None, None, False def agents_ready_message(self): - """(message, all_ready). `_wait_for_healthy` gives up after its timeout and - lets the controller start anyway, so the workflow can come up short. - """ + """Summarize observed readiness, including incomplete older runtimes.""" ready = len(self.replicas_ready) if not self.replicas_total: return "Workflow ready", True @@ -351,8 +349,12 @@ def _interrupted(): def _tail_verbose(stream, state, api_port, config_path, serve): """Every log line, verbatim, until the workflow is up -- what `-v` restores.""" - for line in stream: + tracker = PhaseTracker() + for line in _drain(_queued_lines(stream), state): print(line, end="") + _, _, is_error = tracker.feed(line) + if is_error: + return None # Logged exactly once, right after the workflow finishes coming up. if "Global controller started, polling every" in line: return _deploy_summary(state, api_port, config_path, serve) diff --git a/packages/cli/tests/test_deploy_progress.py b/packages/cli/tests/test_deploy_progress.py index 12a2c854..94e0224b 100644 --- a/packages/cli/tests/test_deploy_progress.py +++ b/packages/cli/tests/test_deploy_progress.py @@ -120,9 +120,7 @@ def test_a_re_read_ready_line_does_not_double_count(): def test_coming_up_short_of_the_announced_replicas_is_not_reported_as_success(): - """`_wait_for_healthy` gives up after its timeout and the controller starts - anyway, so the up-marker can arrive with agents still unhealthy. - """ + """Keep partial-readiness output safe for logs from older runtimes.""" tracker, _, _, _ = drive( [ "INFO:canyonos_core.controller.global_controller:Waiting for 3 replica(s) to become healthy (timeout=300s)...\n", @@ -174,6 +172,20 @@ def test_benign_warnings_do_not_trip_the_error_path(line): assert not errored +def test_verbose_tail_stops_on_a_fatal_line(monkeypatch, capsys): + lines = [ + "INFO:canyonos_core:Deploying from config: config.yaml\n", + "CRITICAL:canyonos_core.controller.global_controller:Controller readiness timed out\n", + ] + monkeypatch.setattr(deploy_cmd, "_queued_lines", lambda stream: stream) + monkeypatch.setattr(deploy_cmd, "_drain", lambda stream, state: iter(stream)) + + result = deploy_cmd._tail_verbose(lines, {}, None, "config.yaml", serve=False) + + assert result is None + assert "Controller readiness timed out" in capsys.readouterr().out + + def test_the_deploy_is_only_declared_dead_after_two_consecutive_checks(monkeypatch): """One dropped request shouldn't end a deploy that is merely busy.""" replies = iter([None, {"running": True}, None, None]) diff --git a/packages/core/canyonos_core/controller/global_controller.py b/packages/core/canyonos_core/controller/global_controller.py index 4335dea7..5e99d95e 100644 --- a/packages/core/canyonos_core/controller/global_controller.py +++ b/packages/core/canyonos_core/controller/global_controller.py @@ -575,15 +575,26 @@ def _wait_for_healthy(self, timeout=30, interval=2): if pending: time.sleep(interval) - if pending: - for instance in pending: - logger.warning( - "Controller %s (%s:%s) not ready after %ds.", - instance["agent_name"], - instance["host"], - instance["host_port"], - timeout, - ) + not_ready = [] + for instance in pending: + name = instance["agent_name"] + host = instance["host"] + port = instance["host_port"] + node_redis = self._get_node_redis_for(host) + endpoint = self.instance_manager._routing_endpoint_for(instance) + status = node_redis.get(f"controller:{endpoint}:status") + if status == "healthy": + logger.info("Controller %s (%s:%s) is ready.", name, host, port) + self._last_status[(host, port)] = "healthy" + else: + not_ready.append(f"{name} ({host}:{port})={status or 'unknown'}") + + if not_ready: + message = f"Controller readiness timed out after {timeout}s: " + ", ".join( + not_ready + ) + logger.critical(message) + raise RuntimeError(message) # ------------------------------------------------------------------ # # Polling loop # diff --git a/packages/core/tests/test_global_controller_readiness.py b/packages/core/tests/test_global_controller_readiness.py new file mode 100644 index 00000000..1b8c73c0 --- /dev/null +++ b/packages/core/tests/test_global_controller_readiness.py @@ -0,0 +1,72 @@ +import os +import sys +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from canyonos_core.controller.global_controller import GlobalController + + +class _FakeRedis: + def __init__(self, statuses): + self.statuses = statuses + + def get(self, key): + return self.statuses.get(key) + + +class _FakeInstanceManager: + def __init__(self, instances): + self.instances = instances + + def list_instances(self): + return list(self.instances) + + def _routing_endpoint_for(self, instance): + return instance["endpoint"] + + +def _controller(statuses): + instances = [ + { + "agent_name": "ResearchAgent", + "host": "localhost", + "host_port": "8000", + "endpoint": "localhost:8000", + } + ] + controller = GlobalController.__new__(GlobalController) + controller.redis = _FakeRedis(statuses) + controller.node_redis = {} + controller.instance_manager = _FakeInstanceManager(instances) + controller._last_status = {} + return controller + + +class GlobalControllerReadinessTests(unittest.TestCase): + def test_healthy_controller_completes_readiness(self): + controller = _controller({"controller:localhost:8000:status": "healthy"}) + + controller._wait_for_healthy(timeout=0, interval=0) + + self.assertEqual(controller._last_status[("localhost", "8000")], "healthy") + + def test_failed_controller_fails_deploy_after_deadline(self): + controller = _controller({"controller:localhost:8000:status": "failed"}) + + with self.assertRaisesRegex( + RuntimeError, + r"Controller readiness timed out after 0s: ResearchAgent " + r"\(localhost:8000\)=failed", + ): + controller._wait_for_healthy(timeout=0, interval=0) + + def test_missing_status_fails_deploy_after_deadline(self): + controller = _controller({}) + + with self.assertRaisesRegex(RuntimeError, r"ResearchAgent .*unknown"): + controller._wait_for_healthy(timeout=0, interval=0) + + +if __name__ == "__main__": + unittest.main()