fix(iwooos): bound control plane source latency

apps/api/tests/test_iwooos_security_asset_control_plane.py

@@ -224,24 +224,28 @@ def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> N
     assert payload["work_items"][0]["work_item_id"] == "AIA-P0-006-00"
 
 
-def test_read_public_source_returns_redacted_reason_code_on_query_failure() -> None:
-    class NestedContext:
+def test_read_public_source_returns_redacted_reason_code_on_query_failure(
+    monkeypatch,
+) -> None:
+    class FailingDb:
+        async def execute(self, statement):
+            raise RuntimeError("private table detail")
+
+    class DbContext:
         async def __aenter__(self):
-            return None
+            return FailingDb()
 
         async def __aexit__(self, exc_type, exc, traceback):
             return False
 
-    class FailingDb:
-        def begin_nested(self):
-            return NestedContext()
-
-        async def execute(self, statement):
-            raise RuntimeError("private table detail")
+    monkeypatch.setattr(
+        "src.services.iwooos_security_asset_control_plane.get_db_context",
+        lambda project_id: DbContext(),
+    )
 
     value, health = asyncio.run(
         _read_public_source(
-            FailingDb(),
+            project_id="awoooi",
             source_id="audit_runtime",
             statement="SELECT 1",
             result_mode="one",
@@ -307,9 +311,51 @@ def test_service_uses_canonical_awoooi_project_scope(monkeypatch) -> None:
     )
     payload = asyncio.run(service.get_snapshot())
 
-    assert requested_projects == ["awoooi"]
+    assert requested_projects == ["awoooi"] * 9
     assert payload["status"] == "degraded"
-    assert payload["reason_code"] == "database_query_failed"
+    assert payload["source_status"] == "live_database_unavailable"
+    assert payload["summary"]["unavailable_source_count"] == 9
+
+
+def test_service_reads_independent_sources_concurrently(monkeypatch) -> None:
+    service = IwoooSSecurityAssetControlPlaneService()
+    concurrency = {"active": 0, "maximum": 0}
+
+    class EmptyResult:
+        def fetchall(self):
+            return []
+
+        def one(self):
+            return {}
+
+        def one_or_none(self):
+            return None
+
+    class SlowDb:
+        async def execute(self, statement):
+            concurrency["active"] += 1
+            concurrency["maximum"] = max(concurrency["maximum"], concurrency["active"])
+            await asyncio.sleep(0.02)
+            concurrency["active"] -= 1
+            return EmptyResult()
+
+    class DbContext:
+        async def __aenter__(self):
+            return SlowDb()
+
+        async def __aexit__(self, exc_type, exc, traceback):
+            return False
+
+    monkeypatch.setattr(
+        "src.services.iwooos_security_asset_control_plane.get_db_context",
+        lambda project_id: DbContext(),
+    )
+    payload = asyncio.run(service._load_snapshot())
+
+    assert concurrency["maximum"] == 9
+    assert payload["summary"]["source_count"] == 9
+    assert payload["summary"]["ready_source_count"] == 9
+    assert payload["summary"]["unavailable_source_count"] == 0
 
 
 def test_public_api_returns_live_aggregate(monkeypatch) -> None: