From 937a4646e14c86b904b04d7b1c581eee10b971c7 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Sun, 12 Jul 2026 18:26:41 +0800 Subject: [PATCH] feat(cloud-api): run lifecycle workers --- apps/cloud-api/cloud_api/app.py | 73 ++++++++++++++++++- apps/cloud-api/tests/test_app.py | 67 ++++++++++++++++- .../cloud-control-plane-integration/tasks.md | 2 +- 3 files changed, 136 insertions(+), 6 deletions(-) diff --git a/apps/cloud-api/cloud_api/app.py b/apps/cloud-api/cloud_api/app.py index bc20cc6..11ae7bc 100644 --- a/apps/cloud-api/cloud_api/app.py +++ b/apps/cloud-api/cloud_api/app.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from collections.abc import Callable from contextlib import asynccontextmanager from dataclasses import dataclass @@ -19,7 +20,9 @@ from cloud.internal_api.api import create_internal_router from cloud.plugins import PluginRegistry from cloud.pool import DevicePool from cloud.scheduler import TaskScheduler +from cloud.schema import require_current_schema from cloud.sdk.api import create_cloud_router +from core.models import utc_now DatabaseFactory = Callable[[CloudControlConfig], CloudDatabase] @@ -85,13 +88,43 @@ def create_app( app.state.cloud_config = control_config app.state.database = database app.state.cloud_services = services + stop_workers = asyncio.Event() + worker_tasks: list[asyncio.Task[None]] = [] try: + repository.health_check() + if control_config.environment == "production": + require_current_schema(control_config.database_url) + worker_tasks = [ + asyncio.create_task( + _run_scheduler_loop( + services, + stop_workers, + interval_seconds=control_config.scheduler_interval_seconds, + ), + name="cloud-scheduler", + ), + asyncio.create_task( + _run_lease_reaper_loop( + services, + stop_workers, + interval_seconds=(control_config.lease_reaper_interval_seconds), + max_attempts=control_config.max_task_attempts, + ), + name="cloud-lease-reaper", + ), + ] + app.state.worker_tasks = tuple(worker_tasks) yield finally: + stop_workers.set() try: - database.close() + if worker_tasks: + await asyncio.gather(*worker_tasks) finally: - repository.unbind() + try: + database.close() + finally: + repository.unbind() app = FastAPI(title="Device Cloud API", lifespan=lifespan) app.include_router( @@ -117,3 +150,39 @@ def _default_database_factory(config: CloudControlConfig) -> CloudDatabase: config.database_url, create_schema=config.environment != "production", ) + + +async def _run_scheduler_loop( + services: CloudApplicationServices, + stop: asyncio.Event, + *, + interval_seconds: float, +) -> None: + while not stop.is_set(): + services.scheduler.assign() + if await _wait_for_stop(stop, interval_seconds): + return + + +async def _run_lease_reaper_loop( + services: CloudApplicationServices, + stop: asyncio.Event, + *, + interval_seconds: float, + max_attempts: int, +) -> None: + while not stop.is_set(): + services.repository.reap_expired_leases( + now=utc_now(), + max_attempts=max_attempts, + ) + if await _wait_for_stop(stop, interval_seconds): + return + + +async def _wait_for_stop(stop: asyncio.Event, interval_seconds: float) -> bool: + try: + await asyncio.wait_for(stop.wait(), timeout=interval_seconds) + except TimeoutError: + return False + return True diff --git a/apps/cloud-api/tests/test_app.py b/apps/cloud-api/tests/test_app.py index 273d1ce..e572422 100644 --- a/apps/cloud-api/tests/test_app.py +++ b/apps/cloud-api/tests/test_app.py @@ -1,5 +1,7 @@ from __future__ import annotations +import time + import pytest from fastapi.testclient import TestClient @@ -20,8 +22,20 @@ def test_create_app_returns_independent_cloud_application() -> None: def test_cloud_application_owns_database_lifecycle() -> None: events: list[str] = [] + class FakeRepository: + def health_check(self) -> None: + events.append("healthy") + + def list_queued_tasks(self) -> list[object]: + events.append("scheduled") + return [] + + def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]: + events.append("reaped") + return [] + class FakeDatabase: - repository = object() + repository = FakeRepository() def close(self) -> None: events.append("closed") @@ -37,9 +51,9 @@ def test_cloud_application_owns_database_lifecycle() -> None: assert ( app.state.cloud_services.pool.store is app.state.cloud_services.repository ) - assert events == [] + assert "healthy" in events - assert events == ["closed"] + assert events[-1] == "closed" def test_repository_proxy_is_available_only_during_lifespan() -> None: @@ -47,6 +61,12 @@ def test_repository_proxy_is_available_only_during_lifespan() -> None: def health_check(self) -> None: return None + def list_queued_tasks(self) -> list[object]: + return [] + + def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]: + return [] + class FakeDatabase: repository = FakeRepository() @@ -65,6 +85,47 @@ def test_repository_proxy_is_available_only_during_lifespan() -> None: app.state.cloud_services.repository.health_check() +def test_lifespan_runs_scheduler_and_reaper_until_shutdown() -> None: + events: list[str] = [] + + class FakeRepository: + def health_check(self) -> None: + events.append("healthy") + + def list_queued_tasks(self) -> list[object]: + events.append("scheduled") + return [] + + def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]: + events.append(f"reaped:{max_attempts}") + return [] + + class FakeDatabase: + repository = FakeRepository() + + def close(self) -> None: + events.append("closed") + + app = create_app( + config=CloudControlConfig( + database_url="sqlite:///:memory:", + scheduler_interval_seconds=0.01, + lease_reaper_interval_seconds=0.01, + max_task_attempts=4, + ), + database_factory=lambda _config: FakeDatabase(), # type: ignore[arg-type,return-value] + ) + + with TestClient(app): + time.sleep(0.04) + assert events.count("scheduled") >= 2 + assert events.count("reaped:4") >= 2 + assert all(not task.done() for task in app.state.worker_tasks) + + assert all(task.done() for task in app.state.worker_tasks) + assert events[-1] == "closed" + + 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 e46f1a6..3727459 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -44,7 +44,7 @@ ## 6. Cloud Control Plane Composition - [x] 6.1 Compose repository, pool, scheduler, plugin registry, public router, internal router, and auth providers in the cloud app factory. -- [ ] 6.2 Implement FastAPI lifespan startup/shutdown for configuration validation, database checks, scheduler loop, and lease-reaper loop. +- [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. - [ ] 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.