From b8d02abf871a2fbdd283a370fc3fe94345a64807 Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Sun, 12 Jul 2026 18:31:19 +0800
Subject: [PATCH] fix(cloud-api): retry lifecycle iterations

---
 apps/cloud-api/cloud_api/app.py               | 24 ++++++--
 apps/cloud-api/tests/test_app.py              | 56 +++++++++++++++++++
 .../cloud-control-plane-integration/tasks.md  |  2 +-
 3 files changed, 76 insertions(+), 6 deletions(-)

diff --git a/apps/cloud-api/cloud_api/app.py b/apps/cloud-api/cloud_api/app.py
index 11ae7bc..71f9c51 100644
--- a/apps/cloud-api/cloud_api/app.py
+++ b/apps/cloud-api/cloud_api/app.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from collections.abc import Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -26,6 +27,7 @@ from core.models import utc_now
 
 
 DatabaseFactory = Callable[[CloudControlConfig], CloudDatabase]
+logger = logging.getLogger(__name__)
 
 
 class _RepositoryProxy:
@@ -159,7 +161,13 @@ async def _run_scheduler_loop(
     interval_seconds: float,
 ) -> None:
     while not stop.is_set():
-        services.scheduler.assign()
+        try:
+            services.scheduler.assign()
+        except Exception:
+            logger.exception(
+                "cloud lifecycle iteration failed",
+                extra={"worker": "scheduler"},
+            )
         if await _wait_for_stop(stop, interval_seconds):
             return
 
@@ -172,10 +180,16 @@ async def _run_lease_reaper_loop(
     max_attempts: int,
 ) -> None:
     while not stop.is_set():
-        services.repository.reap_expired_leases(
-            now=utc_now(),
-            max_attempts=max_attempts,
-        )
+        try:
+            services.repository.reap_expired_leases(
+                now=utc_now(),
+                max_attempts=max_attempts,
+            )
+        except Exception:
+            logger.exception(
+                "cloud lifecycle iteration failed",
+                extra={"worker": "lease_reaper"},
+            )
         if await _wait_for_stop(stop, interval_seconds):
             return
 
diff --git a/apps/cloud-api/tests/test_app.py b/apps/cloud-api/tests/test_app.py
index e572422..20de669 100644
--- a/apps/cloud-api/tests/test_app.py
+++ b/apps/cloud-api/tests/test_app.py
@@ -5,6 +5,7 @@ import time
 import pytest
 from fastapi.testclient import TestClient
 
+import cloud_api.app as app_module
 from cloud_api.app import create_app
 from cloud.control_config import CloudConfigurationError, CloudControlConfig
 
@@ -126,6 +127,61 @@ def test_lifespan_runs_scheduler_and_reaper_until_shutdown() -> None:
     assert events[-1] == "closed"
 
 
+def test_lifecycle_workers_log_failures_and_continue(monkeypatch) -> None:
+    logged_workers: list[str] = []
+
+    def record_exception(_message: str, *, extra: dict[str, str]) -> None:
+        logged_workers.append(extra["worker"])
+
+    monkeypatch.setattr(app_module.logger, "exception", record_exception)
+    scheduler_calls = 0
+    reaper_calls = 0
+
+    class FakeRepository:
+        def health_check(self) -> None:
+            return None
+
+        def list_queued_tasks(self) -> list[object]:
+            nonlocal scheduler_calls
+            scheduler_calls += 1
+            if scheduler_calls == 1:
+                raise RuntimeError("scheduler transient failure")
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            nonlocal reaper_calls
+            reaper_calls += 1
+            if reaper_calls == 1:
+                raise RuntimeError("reaper transient failure")
+            return []
+
+    class FakeDatabase:
+        repository = FakeRepository()
+
+        def close(self) -> None:
+            return None
+
+    app = create_app(
+        config=CloudControlConfig(
+            database_url="sqlite:///:memory:",
+            scheduler_interval_seconds=0.01,
+            lease_reaper_interval_seconds=0.01,
+        ),
+        database_factory=lambda _config: FakeDatabase(),  # type: ignore[arg-type,return-value]
+    )
+
+    with TestClient(app):
+        time.sleep(0.04)
+        assert scheduler_calls >= 2
+        assert reaper_calls >= 2
+        assert all(not task.done() for task in app.state.worker_tasks)
+
+    assert set(logged_workers) >= {
+        "scheduler",
+        "lease_reaper",
+    }
+
+
 def test_production_app_rejects_missing_credentials() -> None:
     with pytest.raises(CloudConfigurationError, match="credential"):
         create_app(
diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md
index 3727459..e4280db 100644
--- a/openspec/changes/cloud-control-plane-integration/tasks.md
+++ b/openspec/changes/cloud-control-plane-integration/tasks.md
@@ -45,7 +45,7 @@
 
 - [x] 6.1 Compose repository, pool, scheduler, plugin registry, public router, internal router, and auth providers in the cloud app factory.
 - [x] 6.2 Implement FastAPI lifespan startup/shutdown for configuration validation, database checks, scheduler loop, and lease-reaper loop.
-- [ ] 6.3 Ensure lifecycle iteration failures are logged and retried without terminating later iterations.
+- [x] 6.3 Ensure lifecycle iteration failures are logged and retried without terminating later iterations.
 - [ ] 6.4 Add `/health/live` and `/health/ready` with separate process, database/schema, and worker-state semantics.
 - [ ] 6.5 Add structured correlation-aware logging for requests and task lifecycle events with sensitive payload redaction.
 - [ ] 6.6 Add app-level tests for startup failures, readiness transitions, graceful shutdown, persisted queue recovery, and expired-lease recovery.