From c94e08e995ccc87b2f6416dbb8f417b89b476054 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 18:22:12 +0800 Subject: [PATCH] fix(km): retry transient list readbacks --- apps/api/src/services/knowledge_service.py | 110 +++++++++++++----- ...est_knowledge_service_readback_degraded.py | 23 ++++ 2 files changed, 101 insertions(+), 32 deletions(-) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index e1265eb99..3e936a15e 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -462,6 +462,57 @@ class KnowledgeService: repo: IKnowledgeRepository = KnowledgeDBRepository(db) return await repo.delete(entry_id) + async def _list_entries_from_primary( + self, + *, + category: str | None, + entry_type: EntryType | None, + status: EntryStatus | None, + tags: list[str] | None, + q: str | None, + limit: int, + offset: int, + ) -> KnowledgeListResponse: + async with get_db_context() as db: + repo: IKnowledgeRepository = KnowledgeDBRepository(db) + items, total = await repo.list_entries( + category=category, + entry_type=entry_type, + status=status, + tags=tags, + q=q, + limit=limit, + offset=offset, + ) + categories_raw = await repo.get_categories() + categories = [ + CategoryCount(category=cat, count=cnt) for cat, cnt in categories_raw + ] + asset_taxonomy_raw = await repo.get_asset_taxonomy_counts() + asset_taxonomy = [ + KnowledgeAssetTaxonomyCount(key=key, count=count) + for key, count in asset_taxonomy_raw + ] + if total == 0: + return build_knowledge_list_readback_degraded_response( + "knowledge_db_empty", + category=category, + entry_type=entry_type, + status=status, + tags=tags, + q=q, + limit=limit, + offset=offset, + readback_status="source_backed_db_empty", + ) + return KnowledgeListResponse( + items=items, + total=total, + categories=categories, + asset_taxonomy=asset_taxonomy, + primary_readback_ready=True, + ) + async def list_entries( self, category: str | None = None, @@ -474,29 +525,21 @@ class KnowledgeService: ) -> KnowledgeListResponse: """列出知識條目 + 分類統計""" try: - async with get_db_context() as db: - repo: IKnowledgeRepository = KnowledgeDBRepository(db) - items, total = await repo.list_entries( - category=category, - entry_type=entry_type, - status=status, - tags=tags, - q=q, - limit=limit, - offset=offset, - ) - categories_raw = await repo.get_categories() - categories = [ - CategoryCount(category=cat, count=cnt) for cat, cnt in categories_raw - ] - asset_taxonomy_raw = await repo.get_asset_taxonomy_counts() - asset_taxonomy = [ - KnowledgeAssetTaxonomyCount(key=key, count=count) - for key, count in asset_taxonomy_raw - ] - if total == 0: - return build_knowledge_list_readback_degraded_response( - "knowledge_db_empty", + return await self._list_entries_from_primary( + category=category, + entry_type=entry_type, + status=status, + tags=tags, + q=q, + limit=limit, + offset=offset, + ) + except Exception as exc: # noqa: BLE001 - production readback must fail soft + degraded_reason_code = _classify_knowledge_readback_degraded_reason(str(exc)) + 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( category=category, entry_type=entry_type, status=status, @@ -504,19 +547,22 @@ class KnowledgeService: q=q, limit=limit, offset=offset, - readback_status="source_backed_db_empty", ) - return KnowledgeListResponse( - items=items, - total=total, - categories=categories, - asset_taxonomy=asset_taxonomy, - primary_readback_ready=True, - ) - except Exception as exc: # noqa: BLE001 - production readback must fail soft + retry_response.operator_stage = "knowledge_readback_primary_retry_recovered" + return retry_response + except Exception as retry_exc: # noqa: BLE001 - keep source-backed fallback + logger.warning( + "knowledge_list_readback_retry_degraded", + error=str(retry_exc), + degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(retry_exc)), + q=q, + limit=limit, + offset=offset, + ) logger.warning( "knowledge_list_readback_degraded", error=str(exc), + degraded_reason_code=degraded_reason_code, category=category, entry_type=entry_type.value if entry_type else None, status=status.value if status else None, diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index 34565fae7..dfb122036 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/apps/api/tests/test_knowledge_service_readback_degraded.py @@ -1,5 +1,6 @@ import pytest +from src.models.knowledge import KnowledgeListResponse from src.services import knowledge_service as knowledge_service_module from src.services.knowledge_service import KnowledgeService @@ -85,6 +86,28 @@ async def test_knowledge_list_entries_source_backed_filter_and_search(monkeypatc assert response.asset_taxonomy[[row.key for row in response.asset_taxonomy].index("alert")].count == 1 +@pytest.mark.asyncio +async def test_knowledge_list_entries_retries_transient_pool_timeout(monkeypatch) -> None: + service = KnowledgeService.__new__(KnowledgeService) + calls = 0 + + async def fake_primary(**_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("db pool exhausted") + return KnowledgeListResponse(items=[], total=1, readback_status="ready") + + monkeypatch.setattr(service, "_list_entries_from_primary", fake_primary) + + response = await service.list_entries(limit=50) + + assert calls == 2 + assert response.readback_status == "ready" + assert response.primary_readback_ready is True + assert response.operator_stage == "knowledge_readback_primary_retry_recovered" + + @pytest.mark.asyncio async def test_knowledge_search_fails_soft_to_source_backed_entries(monkeypatch) -> None: monkeypatch.setattr(