fix(km): bound knowledge readbacks under pool pressure

apps/api/src/services/knowledge_service.py

@@ -52,6 +52,10 @@ _DEGRADED_CATEGORY_FALLBACKS = (
 )
 
 _ASSET_TAXONOMY_FALLBACKS = _DEGRADED_CATEGORY_FALLBACKS[:-1]
+_PRIMARY_KM_LIST_TIMEOUT_SECONDS = 2.0
+_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS = 1.0
+_PRIMARY_KM_RETRY_DELAY_SECONDS = 0.1
+_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS = 1.2
 _PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS = 1.5
 _ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
     "service": (
@@ -228,7 +232,13 @@ _SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = (
 
 def _classify_knowledge_readback_degraded_reason(reason: str) -> str:
     normalized = reason.lower()
-    if "pool" in normalized or "timeout" in normalized or "timed out" in normalized:
+    if (
+        not normalized
+        or "pool" in normalized
+        or "timeout" in normalized
+        or "timed out" in normalized
+        or "timeouterror" in normalized
+    ):
         return "primary_km_db_timeout_or_pool_exhausted"
     if "missing tenant context" in normalized or "project_id" in normalized or "unauthorized" in normalized:
         return "primary_km_project_context_missing"
@@ -238,6 +248,12 @@ def _classify_knowledge_readback_degraded_reason(reason: str) -> str:
         return "primary_km_legacy_row_validation"
     return "primary_km_readback_exception"
 
+
+def _knowledge_readback_exception_reason(exc: BaseException) -> str:
+    if isinstance(exc, TimeoutError):
+        return "TimeoutError"
+    return str(exc) or type(exc).__name__
+
 # =============================================================================
 # Singleton
 # =============================================================================
@@ -589,6 +605,40 @@ class KnowledgeService:
                 for key, count in asset_taxonomy_raw
             ]
 
+    async def _search_primary_entries(
+        self,
+        query: str,
+        limit: int,
+    ) -> list[KnowledgeEntry]:
+        async with get_db_context() as db:
+            repo: IKnowledgeRepository = KnowledgeDBRepository(db)
+            return await repo.search(query, limit)
+
+    async def _list_entries_from_primary_bounded(
+        self,
+        *,
+        category: str | None,
+        entry_type: EntryType | None,
+        status: EntryStatus | None,
+        tags: list[str] | None,
+        q: str | None,
+        limit: int,
+        offset: int,
+        timeout: float,
+    ) -> KnowledgeListResponse:
+        return await asyncio.wait_for(
+            self._list_entries_from_primary(
+                category=category,
+                entry_type=entry_type,
+                status=status,
+                tags=tags,
+                q=q,
+                limit=limit,
+                offset=offset,
+            ),
+            timeout=timeout,
+        )
+
     async def list_entries(
         self,
         category: str | None = None,
@@ -601,7 +651,7 @@ class KnowledgeService:
     ) -> KnowledgeListResponse:
         """列出知識條目 + 分類統計"""
         try:
-            return await self._list_entries_from_primary(
+            return await self._list_entries_from_primary_bounded(
                 category=category,
                 entry_type=entry_type,
                 status=status,
@@ -609,13 +659,15 @@ class KnowledgeService:
                 q=q,
                 limit=limit,
                 offset=offset,
+                timeout=_PRIMARY_KM_LIST_TIMEOUT_SECONDS,
             )
         except Exception as exc:  # noqa: BLE001 - production readback must fail soft
-            degraded_reason_code = _classify_knowledge_readback_degraded_reason(str(exc))
+            degraded_reason = _knowledge_readback_exception_reason(exc)
+            degraded_reason_code = _classify_knowledge_readback_degraded_reason(degraded_reason)
             if degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted":
                 try:
-                    await asyncio.sleep(0.25)
-                    retry_response = await self._list_entries_from_primary(
+                    await asyncio.sleep(_PRIMARY_KM_RETRY_DELAY_SECONDS)
+                    retry_response = await self._list_entries_from_primary_bounded(
                         category=category,
                         entry_type=entry_type,
                         status=status,
@@ -623,21 +675,23 @@ class KnowledgeService:
                         q=q,
                         limit=limit,
                         offset=offset,
+                        timeout=_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS,
                     )
                     retry_response.operator_stage = "knowledge_readback_primary_retry_recovered"
                     return retry_response
                 except Exception as retry_exc:  # noqa: BLE001 - keep source-backed fallback
+                    retry_reason = _knowledge_readback_exception_reason(retry_exc)
                     logger.warning(
                         "knowledge_list_readback_retry_degraded",
-                        error=str(retry_exc),
-                        degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(retry_exc)),
+                        error=retry_reason,
+                        degraded_reason_code=_classify_knowledge_readback_degraded_reason(retry_reason),
                         q=q,
                         limit=limit,
                         offset=offset,
                     )
             logger.warning(
                 "knowledge_list_readback_degraded",
-                error=str(exc),
+                error=degraded_reason,
                 degraded_reason_code=degraded_reason_code,
                 category=category,
                 entry_type=entry_type.value if entry_type else None,
@@ -647,7 +701,7 @@ class KnowledgeService:
                 offset=offset,
             )
             return build_knowledge_list_readback_degraded_response(
-                str(exc),
+                degraded_reason,
                 category=category,
                 entry_type=entry_type,
                 status=status,
@@ -660,50 +714,56 @@ class KnowledgeService:
     async def get_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]:
         """取得 AI 自動化資產維度統計。"""
         try:
-            async with get_db_context() as db:
-                repo: IKnowledgeRepository = KnowledgeDBRepository(db)
-                rows = await repo.get_asset_taxonomy_counts()
-                taxonomy = [
-                    KnowledgeAssetTaxonomyCount(key=key, count=count)
-                    for key, count in rows
-                ]
-                if taxonomy and any(row.count > 0 for row in taxonomy):
-                    return taxonomy
-                return _source_asset_taxonomy_counts(_source_backed_entries())
+            taxonomy = await asyncio.wait_for(
+                self._read_primary_asset_taxonomy(),
+                timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
+            )
+            if taxonomy and any(row.count > 0 for row in taxonomy):
+                return taxonomy
+            return _source_asset_taxonomy_counts(_source_backed_entries())
         except Exception as exc:  # noqa: BLE001 - taxonomy must not 500 the KM UI
-            logger.warning("knowledge_asset_taxonomy_readback_degraded", error=str(exc))
+            reason = _knowledge_readback_exception_reason(exc)
+            logger.warning(
+                "knowledge_asset_taxonomy_readback_degraded",
+                error=reason,
+                degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
+            )
             return _source_asset_taxonomy_counts(_source_backed_entries())
 
     async def get_categories(self) -> list[CategoryCount]:
         """取得分類統計（直接呼叫 repo，不走 list_entries）"""
         try:
-            async with get_db_context() as db:
-                repo: IKnowledgeRepository = KnowledgeDBRepository(db)
-                categories_raw = await repo.get_categories()
-                categories = [
-                    CategoryCount(category=cat, count=cnt)
-                    for cat, cnt in categories_raw
-                ]
-                if categories:
-                    return categories
-                return _source_category_counts(_source_backed_entries())
+            categories = await asyncio.wait_for(
+                self._read_primary_categories(),
+                timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
+            )
+            if categories:
+                return categories
+            return _source_category_counts(_source_backed_entries())
         except Exception as exc:  # noqa: BLE001 - categories must not 500 the KM UI
-            logger.warning("knowledge_categories_readback_degraded", error=str(exc))
+            reason = _knowledge_readback_exception_reason(exc)
+            logger.warning(
+                "knowledge_categories_readback_degraded",
+                error=reason,
+                degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
+            )
             return _source_category_counts(_source_backed_entries())
 
     async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]:
         """關鍵字搜尋"""
         try:
-            async with get_db_context() as db:
-                repo: IKnowledgeRepository = KnowledgeDBRepository(db)
-                return await repo.search(query, limit)
+            return await asyncio.wait_for(
+                self._search_primary_entries(query, limit),
+                timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS,
+            )
         except Exception as exc:  # noqa: BLE001 - KM search must not 500 the UI
+            reason = _knowledge_readback_exception_reason(exc)
             logger.warning(
                 "knowledge_search_readback_degraded",
-                error=str(exc),
+                error=reason,
                 q=query,
                 limit=limit,
-                degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(exc)),
+                degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
             )
             return _filter_source_backed_entries(q=query)[:limit]
 

apps/api/tests/test_knowledge_service_readback_degraded.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 from src.models.knowledge import (
@@ -122,6 +124,30 @@ async def test_knowledge_list_entries_retries_transient_pool_timeout(monkeypatch
     assert response.operator_stage == "knowledge_readback_primary_retry_recovered"
 
 
+@pytest.mark.asyncio
+async def test_knowledge_list_entries_bounds_primary_timeout_to_source_backed(monkeypatch) -> None:
+    service = KnowledgeService.__new__(KnowledgeService)
+    calls = 0
+
+    async def never_finishes_primary(**_kwargs):
+        nonlocal calls
+        calls += 1
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(service, "_list_entries_from_primary", never_finishes_primary)
+    monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_LIST_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_RETRY_DELAY_SECONDS", 0)
+
+    response = await service.list_entries(limit=50)
+
+    assert calls == 2
+    assert response.readback_status == "source_backed_degraded"
+    assert response.primary_readback_ready is False
+    assert response.degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted"
+    assert response.total == 13
+
+
 @pytest.mark.asyncio
 async def test_knowledge_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None:
     primary_entry = KnowledgeEntry(
@@ -247,3 +273,54 @@ async def test_knowledge_asset_taxonomy_fails_soft_when_readback_breaks(monkeypa
         "schedule",
     ]
     assert all(row.count >= 1 for row in taxonomy)
+
+
+@pytest.mark.asyncio
+async def test_knowledge_side_readbacks_bound_timeouts_to_source_backed(monkeypatch) -> None:
+    service = KnowledgeService.__new__(KnowledgeService)
+
+    async def never_finishes(*_args):
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr(service, "_read_primary_categories", never_finishes)
+    monkeypatch.setattr(service, "_read_primary_asset_taxonomy", never_finishes)
+    monkeypatch.setattr(service, "_search_primary_entries", never_finishes)
+
+    categories = await service.get_categories()
+    taxonomy = await service.get_asset_taxonomy()
+    entries = await service.search("Telegram", limit=10)
+
+    assert [row.category for row in categories] == [
+        "project",
+        "product",
+        "website",
+        "service",
+        "package",
+        "tool",
+        "log",
+        "alert",
+        "playbook",
+        "rag",
+        "mcp",
+        "schedule",
+        "general",
+    ]
+    assert [row.key for row in taxonomy] == [
+        "project",
+        "product",
+        "website",
+        "service",
+        "package",
+        "tool",
+        "log",
+        "alert",
+        "playbook",
+        "rag",
+        "mcp",
+        "schedule",
+    ]
+    assert {entry.id for entry in entries} >= {
+        "source-backed-service-telegram-alert-receipts",
+        "source-backed-alert-telegram-monitoring-coverage",
+    }