From a70eda5850b64d86444db526e5360b5011bd3905 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 18:30:25 +0800 Subject: [PATCH] fix(knowledge): preserve primary rows on taxonomy timeout --- apps/api/src/services/knowledge_service.py | 84 ++++++++++++++--- ...est_knowledge_service_readback_degraded.py | 92 +++++++++++++++++++ 2 files changed, 165 insertions(+), 11 deletions(-) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 60b8014f7..e9a5e9e64 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -1059,23 +1059,85 @@ class KnowledgeService: timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS, ) items = [_direct_knowledge_entry_from_row(dict(row)) for row in rows] - categories = await self._read_categories_direct_with_conn( - conn, - project_id=project_id, + degraded_side_reads: list[str] = [] + try: + categories = await self._read_categories_direct_with_conn( + conn, + project_id=project_id, + ) + except Exception as exc: # noqa: BLE001 - preserve primary rows + logger.warning( + "knowledge_direct_categories_readback_degraded", + project_id=project_id, + error_type=type(exc).__name__, + ) + categories = [] + degraded_side_reads.append("categories") + categories_ready = bool( + categories and any(row.count > 0 for row in categories) ) - asset_taxonomy = await self._read_asset_taxonomy_direct_with_conn( - conn, - project_id=project_id, + if not categories_ready and "categories" not in degraded_side_reads: + degraded_side_reads.append("categories") + try: + asset_taxonomy = ( + await self._read_asset_taxonomy_direct_with_conn( + conn, + project_id=project_id, + ) + ) + except Exception as exc: # noqa: BLE001 - preserve primary rows + logger.warning( + "knowledge_direct_asset_taxonomy_readback_degraded", + project_id=project_id, + error_type=type(exc).__name__, + ) + asset_taxonomy = [] + degraded_side_reads.append("asset_taxonomy") + asset_taxonomy_ready = bool( + asset_taxonomy + and any(row.count > 0 for row in asset_taxonomy) ) + if ( + not asset_taxonomy_ready + and "asset_taxonomy" not in degraded_side_reads + ): + degraded_side_reads.append("asset_taxonomy") + source_entries = _source_backed_entries() + partial_taxonomy = bool(degraded_side_reads) return KnowledgeListResponse( items=items, total=total, - categories=categories or _source_category_counts(items), - asset_taxonomy=asset_taxonomy or _source_asset_taxonomy_counts(items), - readback_status="ready_direct_connection_after_session_timeout", + categories=( + categories + if categories_ready + else _source_category_counts(source_entries) + ), + asset_taxonomy=( + asset_taxonomy + if asset_taxonomy_ready + else _source_asset_taxonomy_counts(source_entries) + ), + readback_status=( + "ready_direct_connection_partial_taxonomy_degraded" + if partial_taxonomy + else "ready_direct_connection_after_session_timeout" + ), primary_readback_ready=True, - operator_stage="knowledge_readback_direct_connection_recovered", - next_step="repair_session_pool_readback_without_hiding_primary_km", + degraded_reason_code=( + "primary_km_taxonomy_readback_degraded" + if partial_taxonomy + else None + ), + operator_stage=( + "knowledge_direct_primary_rows_ready_taxonomy_degraded" + if partial_taxonomy + else "knowledge_readback_direct_connection_recovered" + ), + next_step=( + "repair_primary_km_taxonomy_without_hiding_primary_rows" + if partial_taxonomy + else "repair_session_pool_readback_without_hiding_primary_km" + ), writes_on_read=False, manual_review_required=False, ) diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index e51b438eb..31226c481 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/apps/api/tests/test_knowledge_service_readback_degraded.py @@ -1,4 +1,6 @@ import asyncio +import sys +from types import SimpleNamespace import pytest from fastapi import FastAPI, HTTPException @@ -220,6 +222,96 @@ async def test_knowledge_list_entries_uses_direct_readback_after_session_pool_ti assert response.manual_review_required is False +@pytest.mark.asyncio +async def test_direct_list_keeps_primary_rows_when_asset_taxonomy_times_out( + monkeypatch, +) -> None: + service = KnowledgeService.__new__(KnowledgeService) + + class DirectConnection: + async def execute(self, *_args): + return "OK" + + async def fetchval(self, *_args): + return 727 + + async def fetch(self, *_args): + return [ + { + "id": "km-primary-direct-1", + "title": "Primary direct KM row", + "content": "Persistent row survives taxonomy pressure", + "entry_type": "runbook", + "category": "alert_handling", + "tags": ["alert", "telegram"], + "source": "ai_extracted", + "status": "approved", + "view_count": 0, + } + ] + + async def close(self): + return None + + connection = DirectConnection() + + async def connect(_url): + return connection + + async def categories(_conn, *, project_id): + assert project_id == "awoooi" + return [CategoryCount(category="alert_handling", count=113)] + + async def taxonomy_timeout(_conn, *, project_id): + assert project_id == "awoooi" + raise TimeoutError("taxonomy statement timeout") + + monkeypatch.setitem( + sys.modules, + "asyncpg", + SimpleNamespace(connect=connect), + ) + monkeypatch.setattr( + knowledge_service_module.settings, + "DATABASE_URL", + "postgresql+asyncpg://test:test@127.0.0.1:5432/test", + ) + monkeypatch.setattr( + service, + "_read_categories_direct_with_conn", + categories, + ) + monkeypatch.setattr( + service, + "_read_asset_taxonomy_direct_with_conn", + taxonomy_timeout, + ) + + response = await service._list_entries_from_direct( + project_id="awoooi", + category=None, + entry_type=None, + status=None, + tags=None, + q=None, + limit=1, + offset=0, + ) + + assert response is not None + assert response.total == 727 + assert [item.id for item in response.items] == ["km-primary-direct-1"] + assert response.primary_readback_ready is True + assert response.readback_status == ( + "ready_direct_connection_partial_taxonomy_degraded" + ) + assert response.degraded_reason_code == ( + "primary_km_taxonomy_readback_degraded" + ) + assert response.categories[0].category == "alert_handling" + assert any(row.key == "alert" for row in response.asset_taxonomy) + + @pytest.mark.asyncio async def test_knowledge_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None: primary_entry = KnowledgeEntry(