From 8a0c916d82b11f936e93c478fd53cdb38cae1a40 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 19:53:21 +0800 Subject: [PATCH] fix(incidents): prefer durable closure readback --- apps/api/src/api/v1/incidents.py | 2 +- apps/api/src/services/incident_service.py | 111 +++++++++++++++++- .../test_incident_readback_source_truth.py | 104 ++++++++++++++++ 3 files changed, 212 insertions(+), 5 deletions(-) create mode 100644 apps/api/tests/test_incident_readback_source_truth.py diff --git a/apps/api/src/api/v1/incidents.py b/apps/api/src/api/v1/incidents.py index e362079d9..1a7f06800 100644 --- a/apps/api/src/api/v1/incidents.py +++ b/apps/api/src/api/v1/incidents.py @@ -306,7 +306,7 @@ async def get_incident(incident_id: str) -> IncidentResponse: incident_service = get_incident_service() try: - incident = await incident_service.get_from_working_memory(incident_id) + incident = await incident_service.get_for_readback(incident_id) if incident is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, diff --git a/apps/api/src/services/incident_service.py b/apps/api/src/services/incident_service.py index d1290b230..45a5c54e1 100644 --- a/apps/api/src/services/incident_service.py +++ b/apps/api/src/services/incident_service.py @@ -664,15 +664,110 @@ class IncidentService: ) return None + async def project_terminal_status_to_working_memory( + self, + incident_id: str, + *, + project_id: str, + status: IncidentStatus, + updated_at: datetime, + resolved_at: datetime | None, + ) -> Literal["updated", "current", "missing", "failed"]: + """Project a durable terminal DB state into the Redis read model.""" + + if status not in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED): + raise ValueError("working_memory_terminal_projection_requires_terminal_status") + + incident = await self.get_from_working_memory(incident_id) + hydrated_from_episodic = incident is None + if incident is None: + incident = await self.get_from_episodic_memory( + incident_id, + project_id=project_id, + ) + if incident is None: + logger.warning( + "working_memory_terminal_projection_missing_incident", + incident_id=incident_id, + project_id=project_id, + ) + return "missing" + + current_updated_at = incident.updated_at + if current_updated_at.tzinfo is None: + current_updated_at = current_updated_at.replace(tzinfo=UTC) + durable_updated_at = updated_at + if durable_updated_at.tzinfo is None: + durable_updated_at = durable_updated_at.replace(tzinfo=UTC) + + already_current = bool( + not hydrated_from_episodic + and incident.status in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED) + and current_updated_at >= durable_updated_at + and (resolved_at is None or incident.resolved_at is not None) + ) + if already_current: + return "current" + + if incident.status != IncidentStatus.CLOSED: + incident.status = status + incident.updated_at = updated_at + incident.resolved_at = resolved_at or incident.resolved_at + if not await self.save_to_working_memory(incident): + logger.warning( + "working_memory_terminal_projection_failed", + incident_id=incident_id, + project_id=project_id, + ) + return "failed" + + logger.info( + "working_memory_terminal_projection_updated", + incident_id=incident_id, + project_id=project_id, + status=incident.status.value, + hydrated_from_episodic=hydrated_from_episodic, + ) + return "updated" + + async def get_for_readback(self, incident_id: str) -> Incident | None: + """Return durable incident truth, with Redis as an availability fallback.""" + incident = await self.get_from_episodic_memory(incident_id) + if incident is not None: + return incident + + logger.warning( + "incident_readback_falling_back_to_working_memory", + incident_id=incident_id, + ) + return await self.get_from_working_memory(incident_id) + async def get_active_incidents(self) -> list[Incident]: """ - 列出所有活躍的 Incidents (從 Working Memory) + 列出所有活躍的 Incidents。 - 方案 C: 解析時正規化舊格式 Enum 值 + PostgreSQL 是 durable source of truth；只有 repository 暫時不可用時， + 才退回 Redis working memory 維持讀取可用性。 Returns: list[Incident]: 活躍事件列表 (investigating 或 mitigating) """ + try: + from src.repositories.incident_repository import get_incident_repository + + incidents = await get_incident_repository().get_active() + logger.info( + "get_active_incidents_from_episodic_memory", + count=len(incidents), + ) + return incidents + except Exception as e: + logger.warning( + "active_incidents_episodic_read_failed", + error=str(e), + fallback="working_memory", + ) + redis_client = get_redis() incidents: list[Incident] = [] @@ -805,7 +900,12 @@ class IncidentService: ) return False - async def get_from_episodic_memory(self, incident_id: str) -> Incident | None: + async def get_from_episodic_memory( + self, + incident_id: str, + *, + project_id: str | None = None, + ) -> Incident | None: """ 從 Episodic Memory 讀取 Incident @@ -813,7 +913,10 @@ class IncidentService: Incident | None: 事件資料，若不存在則返回 None """ try: - async with get_db_context() as db: + db_context = ( + get_db_context(project_id) if project_id else get_db_context() + ) + async with db_context as db: from sqlalchemy import select stmt = select(IncidentRecord).where( diff --git a/apps/api/tests/test_incident_readback_source_truth.py b/apps/api/tests/test_incident_readback_source_truth.py new file mode 100644 index 000000000..b16ea0541 --- /dev/null +++ b/apps/api/tests/test_incident_readback_source_truth.py @@ -0,0 +1,104 @@ +"""Incident readback must not regress durable closure to stale cache state.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock + +import pytest + +from src.api.v1 import incidents as incidents_api +from src.models.incident import Incident, IncidentStatus, Severity +from src.repositories import incident_repository as incident_repository_module +from src.services.incident_service import IncidentService + + +def _incident(status: IncidentStatus, *, minute: int) -> Incident: + timestamp = datetime(2026, 7, 11, 11, minute, tzinfo=UTC) + return Incident( + incident_id="INC-20260711-SOURCE-TRUTH", + status=status, + severity=Severity.P2, + affected_services=["awoooi-api"], + created_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC), + updated_at=timestamp, + resolved_at=timestamp if status is IncidentStatus.RESOLVED else None, + ) + + +@pytest.mark.asyncio +async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monkeypatch): + service = IncidentService() + durable = _incident(IncidentStatus.RESOLVED, minute=38) + cached = _incident(IncidentStatus.INVESTIGATING, minute=11) + monkeypatch.setattr( + service, + "get_from_episodic_memory", + AsyncMock(return_value=durable), + ) + working_read = AsyncMock(return_value=cached) + monkeypatch.setattr(service, "get_from_working_memory", working_read) + + result = await service.get_for_readback(durable.incident_id) + + assert result is durable + assert result.status is IncidentStatus.RESOLVED + working_read.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_readback_falls_back_to_working_memory_when_durable_record_unavailable( + monkeypatch, +): + service = IncidentService() + cached = _incident(IncidentStatus.INVESTIGATING, minute=11) + monkeypatch.setattr( + service, + "get_from_episodic_memory", + AsyncMock(return_value=None), + ) + monkeypatch.setattr( + service, + "get_from_working_memory", + AsyncMock(return_value=cached), + ) + + result = await service.get_for_readback(cached.incident_id) + + assert result is cached + + +@pytest.mark.asyncio +async def test_incident_detail_endpoint_uses_durable_readback(monkeypatch): + durable = _incident(IncidentStatus.RESOLVED, minute=38) + + class _Service: + get_for_readback = AsyncMock(return_value=durable) + + service = _Service() + monkeypatch.setattr(incidents_api, "get_incident_service", lambda: service) + + result = await incidents_api.get_incident(durable.incident_id) + + assert result.incident_id == durable.incident_id + assert result.status == IncidentStatus.RESOLVED.value + service.get_for_readback.assert_awaited_once_with(durable.incident_id) + + +@pytest.mark.asyncio +async def test_active_incidents_come_from_durable_repository(monkeypatch): + service = IncidentService() + durable_active = _incident(IncidentStatus.INVESTIGATING, minute=38) + + class _Repository: + get_active = AsyncMock(return_value=[durable_active]) + + repository = _Repository() + monkeypatch.setattr( + incident_repository_module, + "get_incident_repository", + lambda: repository, + ) + + incidents = await service.get_active_incidents() + + assert incidents == [durable_active] + repository.get_active.assert_awaited_once_with()