test(cloud-api): verify restart recovery

apps/cloud-api/tests/test_app.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import time
+from datetime import timedelta
 
 import pytest
 from fastapi.testclient import TestClient
@@ -8,6 +9,10 @@ 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
+from cloud.database import CloudDatabase
+from cloud.pool import PooledDevice
+from cloud.scheduler import ScheduledTask, TaskConstraints
+from core.models import utc_now
 
 
 def test_create_app_returns_independent_cloud_application() -> None:
@@ -292,6 +297,158 @@ def test_request_correlation_id_is_propagated_without_sensitive_headers(
     assert "never-log-this" not in repr(request_logs)
 
 
+def test_startup_database_failure_closes_database() -> None:
+    closed = False
+
+    class FailingRepository:
+        def health_check(self) -> None:
+            raise RuntimeError("database unavailable")
+
+    class FailingDatabase:
+        repository = FailingRepository()
+
+        def close(self) -> None:
+            nonlocal closed
+            closed = True
+
+    app = create_app(
+        config=CloudControlConfig(database_url="sqlite:///:memory:"),
+        database_factory=lambda _config: FailingDatabase(),  # type: ignore[arg-type,return-value]
+    )
+
+    with pytest.raises(RuntimeError, match="database unavailable"):
+        with TestClient(app):
+            pass
+    assert closed is True
+
+
+def test_persisted_queue_is_recovered_after_control_plane_restart(tmp_path) -> None:
+    database_url = f"sqlite:///{(tmp_path / 'queue-recovery.sqlite3').as_posix()}"
+    seed = CloudDatabase(database_url)
+    now = utc_now()
+    seed.repository.upsert_host("host-a", address=None, last_seen_at=now)
+    seed.repository.replace_host_devices(
+        "host-a",
+        [
+            PooledDevice(
+                device_id="device-a",
+                host_id="host-a",
+                driver_type="wda",
+                status="idle",
+                synced_at=now,
+            )
+        ],
+    )
+    seed.repository.enqueue_task(
+        ScheduledTask(
+            id="persisted-task",
+            goal="resume queued work",
+            workflow_definition_id=None,
+            constraints=TaskConstraints(),
+            created_at=now,
+        )
+    )
+    seed.close()
+
+    app = create_app(
+        config=CloudControlConfig(
+            database_url=database_url,
+            scheduler_interval_seconds=0.01,
+            lease_reaper_interval_seconds=1,
+        )
+    )
+    with TestClient(app):
+        assert _wait_until(
+            lambda: (
+                app.state.cloud_services.repository.get_task("persisted-task").status
+                == "assigned"
+            )
+        )
+
+    verification = CloudDatabase(database_url)
+    try:
+        task = verification.repository.get_task("persisted-task")
+        assert task is not None
+        assert task.status == "assigned"
+        assert task.attempt_count == 1
+    finally:
+        verification.close()
+
+
+def test_expired_lease_is_recovered_after_control_plane_restart(tmp_path) -> None:
+    database_url = f"sqlite:///{(tmp_path / 'lease-recovery.sqlite3').as_posix()}"
+    seed = CloudDatabase(database_url)
+    now = utc_now()
+    seed.repository.upsert_host("host-a", address=None, last_seen_at=now)
+    seed.repository.replace_host_devices(
+        "host-a",
+        [
+            PooledDevice(
+                device_id="device-a",
+                host_id="host-a",
+                driver_type="wda",
+                status="idle",
+                synced_at=now,
+            )
+        ],
+    )
+    seed.repository.enqueue_task(
+        ScheduledTask(
+            id="expired-task",
+            goal="recover expired work",
+            workflow_definition_id=None,
+            constraints=TaskConstraints(),
+            created_at=now,
+        )
+    )
+    seed.repository.assign_task(
+        task_id="expired-task",
+        host_id="host-a",
+        device_id="device-a",
+        lease_id="expired-lease",
+        lease_expires_at=now - timedelta(seconds=1),
+        now=now - timedelta(minutes=1),
+    )
+    seed.close()
+
+    app = create_app(
+        config=CloudControlConfig(
+            database_url=database_url,
+            scheduler_interval_seconds=1,
+            lease_reaper_interval_seconds=0.01,
+            max_task_attempts=1,
+        )
+    )
+    with TestClient(app):
+        assert _wait_until(
+            lambda: (
+                app.state.cloud_services.repository.get_task("expired-task").status
+                == "failed"
+            )
+        )
+
+    verification = CloudDatabase(database_url)
+    try:
+        task = verification.repository.get_task("expired-task")
+        assert task is not None
+        assert task.status == "failed"
+        assert task.failure_reason == "lease expired after 1 attempts"
+        assert verification.repository.list_task_attempts("expired-task")[0].status == (
+            "expired"
+        )
+    finally:
+        verification.close()
+
+
+def _wait_until(predicate, timeout_seconds: float = 1.0) -> bool:
+    deadline = time.monotonic() + timeout_seconds
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return False
+
+
 def test_production_app_rejects_missing_credentials() -> None:
     with pytest.raises(CloudConfigurationError, match="credential"):
         create_app(