wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(api): retain ready asset matrix on timeout

Browse Source
This commit is contained in:
ogt
2026-07-15 07:21:00 +08:00
parent fc81693c64
commit a088bcc4ea
2 changed files with 269 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
apps/api/src/services/ai_automation_asset_capability_matrix.py
View File

@@ -30,8 +30,14 @@ RECONCILIATION_SUMMARY_OPERATION = "asset_capability_reconciliation_completed"
RECONCILIATION_WORK_ITEM_OPERATION = "asset_capability_reconciliation_work_item_created"
FRESHNESS_SLO_SECONDS = 2 * 60 * 60
RECONCILIATION_FRESHNESS_SLO_SECONDS = 36 * 60 * 60
LIVE_READBACK_TIMEOUT_SECONDS = 8.0
LIVE_READBACK_TIMEOUT_SECONDS = 7.25
LIVE_READBACK_DB_STATEMENT_TIMEOUT_MILLISECONDS = 6_000
PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS = 45
PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS = 15 * 60
PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS = 0.2
_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE = "ai_asset_capability_matrix_fresh"
_PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE = "ai_asset_capability_matrix_fallback"
STAGE_IDS = (
"source_truth",
@@ -1174,6 +1180,133 @@ async def _load_live_rows(
return latest_run, list(assets_by_id.values()), receipt
def _public_matrix_cache_key(project_id: str) -> dict[str, str]:
return {
"project_id": project_id,
"projection": "declared_assets_only",
"schema_version": SCHEMA_VERSION,
}
async def _read_public_matrix_cache(
project_id: str,
*,
namespace: str,
ttl_seconds: int,
) -> dict[str, Any] | None:
from src.core.redis_client import get_redis
from src.services.operator_summary_cache import (
get_cached_operator_summary,
get_cached_operator_summary_async,
)
key_parts = _public_matrix_cache_key(project_id)
try:
get_redis()
except RuntimeError:
return get_cached_operator_summary(
namespace,
key_parts,
ttl_seconds=ttl_seconds,
)
try:
return await asyncio.wait_for(
get_cached_operator_summary_async(
namespace,
key_parts,
ttl_seconds=ttl_seconds,
),
timeout=PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS,
)
except Exception as exc:
logger.debug(
"asset_capability_matrix_cache_read_failed",
namespace=namespace,
error_type=type(exc).__name__,
)
return get_cached_operator_summary(
namespace,
key_parts,
ttl_seconds=ttl_seconds,
)
async def _store_public_matrix_caches(
project_id: str,
payload: dict[str, Any],
) -> None:
from src.core.redis_client import get_redis
from src.services.operator_summary_cache import (
store_operator_summary,
store_operator_summary_async,
)
key_parts = _public_matrix_cache_key(project_id)
cache_targets = (
(
_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE,
PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS,
),
(
_PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE,
PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS,
),
)
for namespace, ttl_seconds in cache_targets:
store_operator_summary(
namespace,
key_parts,
payload,
ttl_seconds=ttl_seconds,
)
try:
get_redis()
except RuntimeError:
return
try:
await asyncio.wait_for(
asyncio.gather(
*(
store_operator_summary_async(
namespace,
key_parts,
payload,
ttl_seconds=ttl_seconds,
)
for namespace, ttl_seconds in cache_targets
)
),
timeout=PUBLIC_MATRIX_CACHE_IO_TIMEOUT_SECONDS,
)
except Exception as exc:
logger.debug(
"asset_capability_matrix_cache_write_failed",
error_type=type(exc).__name__,
)
def _annotate_cached_matrix(
payload: dict[str, Any],
*,
mode: str,
timeout_seconds: float,
error_type: str = "",
) -> dict[str, Any]:
cache_meta = _dict(payload.pop("cache", {}))
payload["live_readback_timeout_seconds"] = timeout_seconds
payload["live_readback_cache"] = {
**cache_meta,
"mode": mode,
"fallback_used": mode == "stale_if_error",
}
if error_type:
payload["live_readback_status"] = "degraded_cached"
payload["live_readback_error_type"] = error_type
return payload
async def build_asset_capability_matrix_with_live_readback(
*,
project_id: str = DEFAULT_PROJECT_ID,
@@ -1187,8 +1320,23 @@ async def build_asset_capability_matrix_with_live_readback(
get_ai_automation_program_scope_classes,
)
scope_classes = get_ai_automation_program_scope_classes()
bounded_timeout_seconds = max(0.001, float(timeout_seconds))
use_public_cache = repo_root is None and not include_live_only_assets
if use_public_cache:
cached_payload = await _read_public_matrix_cache(
project_id,
namespace=_PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE,
ttl_seconds=PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS,
)
if cached_payload is not None:
return _annotate_cached_matrix(
cached_payload,
mode="fresh",
timeout_seconds=bounded_timeout_seconds,
)
scope_classes = get_ai_automation_program_scope_classes()
async def _load_and_build() -> dict[str, Any]:
latest_run, live_assets, receipt = await _load_live_rows(project_id)
@@ -1214,6 +1362,19 @@ async def build_asset_capability_matrix_with_live_readback(
error_type=type(exc).__name__,
timeout_seconds=bounded_timeout_seconds,
)
if use_public_cache:
cached_payload = await _read_public_matrix_cache(
project_id,
namespace=_PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE,
ttl_seconds=PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS,
)
if cached_payload is not None:
return _annotate_cached_matrix(
cached_payload,
mode="stale_if_error",
timeout_seconds=bounded_timeout_seconds,
error_type=type(exc).__name__,
)
payload = build_asset_capability_matrix(
scope_classes,
repo_root=repo_root,
@@ -1223,5 +1384,7 @@ async def build_asset_capability_matrix_with_live_readback(
payload["live_readback_error_type"] = type(exc).__name__
payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
return payload
if use_public_cache:
await _store_public_matrix_caches(project_id, payload)
payload["live_readback_timeout_seconds"] = bounded_timeout_seconds
return payload

104
apps/api/tests/test_ai_automation_asset_capability_matrix.py
View File

@@ -534,3 +534,107 @@ def test_matrix_build_phase_is_inside_the_total_timeout(monkeypatch: Any) -> Non
assert payload["live_readback_status"] == "degraded"
assert payload["live_readback_error_type"] == "TimeoutError"
assert payload["live_readback_timeout_seconds"] == 0.01
def test_public_matrix_returns_fresh_cache_without_live_db_read(
monkeypatch: Any,
) -> None:
async def _cached_read(
_project_id: str,
*,
namespace: str,
ttl_seconds: int,
) -> dict[str, Any] | None:
assert namespace == matrix_service._PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE
assert ttl_seconds == matrix_service.PUBLIC_MATRIX_FRESH_CACHE_TTL_SECONDS
return {
"live_readback_status": "ready",
"summary": {"live_observed_asset_count": 193},
"cache": {"status": "hit", "age_seconds": 2.5},
}
async def _unexpected_live_read(
_project_id: str,
) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
raise AssertionError("fresh cache must bypass the live DB read")
monkeypatch.setattr(matrix_service, "_read_public_matrix_cache", _cached_read)
monkeypatch.setattr(matrix_service, "_load_live_rows", _unexpected_live_read)
payload = asyncio.run(build_asset_capability_matrix_with_live_readback())
assert payload["live_readback_status"] == "ready"
assert payload["summary"]["live_observed_asset_count"] == 193
assert payload["live_readback_cache"]["mode"] == "fresh"
assert payload["live_readback_cache"]["fallback_used"] is False
def test_public_matrix_success_warms_cache_for_next_request(
monkeypatch: Any,
) -> None:
from src.services.operator_summary_cache import clear_operator_summary_cache
live_read_count = 0
async def _live_readback(
_project_id: str,
) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
nonlocal live_read_count
live_read_count += 1
return {}, [], {}
clear_operator_summary_cache()
monkeypatch.setattr(matrix_service, "_load_live_rows", _live_readback)
try:
first = asyncio.run(build_asset_capability_matrix_with_live_readback())
second = asyncio.run(build_asset_capability_matrix_with_live_readback())
finally:
clear_operator_summary_cache()
assert first["live_readback_status"] == "ready"
assert second["live_readback_status"] == "ready"
assert second["live_readback_cache"]["mode"] == "fresh"
assert live_read_count == 1
def test_public_matrix_timeout_uses_last_ready_cache(monkeypatch: Any) -> None:
async def _slow_live_readback(
_project_id: str,
) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
await asyncio.sleep(1)
return {}, [], {}
async def _cached_read(
_project_id: str,
*,
namespace: str,
ttl_seconds: int,
) -> dict[str, Any] | None:
if namespace == matrix_service._PUBLIC_MATRIX_FRESH_CACHE_NAMESPACE:
return None
assert namespace == matrix_service._PUBLIC_MATRIX_FALLBACK_CACHE_NAMESPACE
assert ttl_seconds == matrix_service.PUBLIC_MATRIX_FALLBACK_CACHE_TTL_SECONDS
return {
"live_readback_status": "ready",
"summary": {
"live_inventory_asset_count": 9595,
"live_observed_asset_count": 193,
"stage_coverage_percent": 18,
},
"cache": {"status": "hit", "age_seconds": 51.2},
}
monkeypatch.setattr(matrix_service, "_read_public_matrix_cache", _cached_read)
monkeypatch.setattr(matrix_service, "_load_live_rows", _slow_live_readback)
payload = asyncio.run(
build_asset_capability_matrix_with_live_readback(timeout_seconds=0.01)
)
assert payload["live_readback_status"] == "degraded_cached"
assert payload["live_readback_error_type"] == "TimeoutError"
assert payload["summary"]["live_inventory_asset_count"] == 9595
assert payload["summary"]["live_observed_asset_count"] == 193
assert payload["summary"]["stage_coverage_percent"] == 18
assert payload["live_readback_cache"]["mode"] == "stale_if_error"
assert payload["live_readback_cache"]["fallback_used"] is True