From 6505e2cede2d9e25114052882a054a1c84eab882 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 10:38:32 +0800 Subject: [PATCH] fix(automation): stabilize asset reconciliation source --- .../asset_capability_reconciliation_job.py | 105 +++++++++++++- apps/api/src/jobs/asset_change_tracker_job.py | 11 +- apps/api/src/jobs/asset_scanner_job.py | 6 +- apps/api/src/jobs/rule_catalog_sync_job.py | 29 ++-- ...t_ai_automation_asset_capability_matrix.py | 114 ++++++++++++++- .../tests/test_asset_change_tracker_job.py | 54 +++++++ apps/api/tests/test_asset_scanner_job.py | 6 + apps/api/tests/test_rule_catalog_sync_job.py | 133 ++++++++++++++++++ ..._signal_worker_ansible_executor_binding.py | 1 + k8s/awoooi-prod/08-deployment-worker.yaml | 4 + 10 files changed, 442 insertions(+), 21 deletions(-) create mode 100644 apps/api/tests/test_asset_change_tracker_job.py create mode 100644 apps/api/tests/test_rule_catalog_sync_job.py diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index 446da976e..c4dd07b0b 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -13,6 +13,7 @@ import time import uuid from collections import Counter from collections.abc import Mapping, Sequence +from datetime import UTC, datetime from typing import Any import structlog @@ -29,9 +30,12 @@ from src.services.ai_automation_asset_capability_matrix import ( logger = structlog.get_logger(__name__) -_FIRST_DELAY_SECONDS = 30 +_FIRST_DELAY_SECONDS = 7 * 60 _LOOP_BACKOFF_SECONDS = 5 * 60 _RECONCILIATION_POLL_SECONDS = 5 * 60 +_SOURCE_SETTLE_SECONDS = 5 * 60 +_SOURCE_STABILITY_DELAY_SECONDS = 5 +_SOURCE_STABILITY_RETRY_SECONDS = 60 _RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS = 30.0 _WORK_ITEM_INSERT_BATCH_SIZE = 250 _MAX_RECONCILIATION_CANDIDATES = 20_000 @@ -39,6 +43,8 @@ _RETRYABLE_RESULT_STATUSES = { "blocked_safe_no_write", "degraded_cache_refresh", "degraded_no_write", + "source_snapshot_settling", + "source_snapshot_unstable", } @@ -50,6 +56,24 @@ def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) +def _latest_discovery_age_seconds( + matrix: Mapping[str, Any], + *, + now: datetime | None = None, +) -> float | None: + ended_at = str(_dict(matrix.get("latest_discovery_run")).get("ended_at") or "") + if not ended_at: + return None + try: + parsed = datetime.fromisoformat(ended_at.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + current = (now or datetime.now(UTC)).astimezone(UTC) + return max(0.0, (current - parsed.astimezone(UTC)).total_seconds()) + + def build_reconciliation_summary_payload( matrix: Mapping[str, Any], candidates: Sequence[Mapping[str, Any]], @@ -177,11 +201,13 @@ def build_reconciliation_work_item_input( async def run_asset_capability_reconciliation_loop() -> None: - """Reconcile shortly after startup and retry each discovery fingerprint.""" + """Reconcile settled source snapshots and retry each fingerprint.""" logger.info( "asset_capability_reconciliation_loop_started", + first_delay_seconds=_FIRST_DELAY_SECONDS, poll_interval_seconds=_RECONCILIATION_POLL_SECONDS, + source_settle_seconds=_SOURCE_SETTLE_SECONDS, ) await asyncio.sleep(_FIRST_DELAY_SECONDS) triggered_by = "startup" @@ -197,13 +223,20 @@ async def run_asset_capability_reconciliation_loop() -> None: triggered_by = "retry" continue if str(result.get("status") or "") in _RETRYABLE_RESULT_STATUSES: + retry_seconds = max( + 30, + min( + _LOOP_BACKOFF_SECONDS, + int(result.get("retry_seconds") or _LOOP_BACKOFF_SECONDS), + ), + ) logger.warning( "asset_capability_reconciliation_retry_scheduled", status=result.get("status"), reason=result.get("reason"), - retry_seconds=_LOOP_BACKOFF_SECONDS, + retry_seconds=retry_seconds, ) - await asyncio.sleep(_LOOP_BACKOFF_SECONDS) + await asyncio.sleep(retry_seconds) triggered_by = "retry" continue await asyncio.sleep(_RECONCILIATION_POLL_SECONDS) @@ -235,6 +268,70 @@ async def reconcile_once( "candidate_count": 0, } + discovery_age_seconds = _latest_discovery_age_seconds(matrix) + if discovery_age_seconds is None or discovery_age_seconds < _SOURCE_SETTLE_SECONDS: + remaining_seconds = ( + _SOURCE_STABILITY_RETRY_SECONDS + if discovery_age_seconds is None + else max( + _SOURCE_STABILITY_RETRY_SECONDS, + int(_SOURCE_SETTLE_SECONDS - discovery_age_seconds) + 1, + ) + ) + logger.info( + "asset_capability_reconciliation_source_snapshot_settling", + discovery_age_seconds=discovery_age_seconds, + required_age_seconds=_SOURCE_SETTLE_SECONDS, + retry_seconds=remaining_seconds, + ) + return { + "status": "source_snapshot_settling", + "reason": "latest_discovery_run_not_settled", + "discovery_age_seconds": discovery_age_seconds, + "required_age_seconds": _SOURCE_SETTLE_SECONDS, + "retry_seconds": remaining_seconds, + "external_runtime_write_performed": False, + } + + await asyncio.sleep(_SOURCE_STABILITY_DELAY_SECONDS) + stable_matrix = await build_asset_capability_matrix_with_live_readback( + project_id=project_id, + timeout_seconds=_RECONCILIATION_LIVE_READBACK_TIMEOUT_SECONDS, + include_live_only_assets=True, + ) + if stable_matrix.get("live_readback_status") != "ready": + return { + "status": "degraded_no_write", + "reason": "stability_verifier_live_readback_degraded", + "live_readback_status": stable_matrix.get("live_readback_status"), + "candidate_count": 0, + "retry_seconds": _SOURCE_STABILITY_RETRY_SECONDS, + "external_runtime_write_performed": False, + } + + first_identity = ( + str(matrix.get("matrix_fingerprint") or ""), + str(_dict(matrix.get("latest_discovery_run")).get("run_id") or ""), + ) + stable_identity = ( + str(stable_matrix.get("matrix_fingerprint") or ""), + str(_dict(stable_matrix.get("latest_discovery_run")).get("run_id") or ""), + ) + if not all(stable_identity) or stable_identity != first_identity: + logger.warning( + "asset_capability_reconciliation_source_snapshot_unstable", + discovery_run_changed=first_identity[1] != stable_identity[1], + matrix_fingerprint_changed=first_identity[0] != stable_identity[0], + retry_seconds=_SOURCE_STABILITY_RETRY_SECONDS, + ) + return { + "status": "source_snapshot_unstable", + "reason": "matrix_changed_during_stability_verifier", + "retry_seconds": _SOURCE_STABILITY_RETRY_SECONDS, + "external_runtime_write_performed": False, + } + matrix = stable_matrix + candidates = build_reconciliation_candidates(matrix) if len(candidates) > _MAX_RECONCILIATION_CANDIDATES: logger.error( diff --git a/apps/api/src/jobs/asset_change_tracker_job.py b/apps/api/src/jobs/asset_change_tracker_job.py index e46b7875c..5b8dc0761 100644 --- a/apps/api/src/jobs/asset_change_tracker_job.py +++ b/apps/api/src/jobs/asset_change_tracker_job.py @@ -29,7 +29,6 @@ from __future__ import annotations import asyncio import json as _json import time as _time -from typing import Any import structlog @@ -38,6 +37,7 @@ logger = structlog.get_logger(__name__) _TRACK_INTERVAL_SEC = 3600 _FIRST_DELAY_SEC = 360 _LOOP_BACKOFF_SEC = 600 +_PROJECT_ID = "awoooi" async def run_asset_change_tracker_loop() -> None: @@ -92,9 +92,10 @@ async def track_once() -> dict[str, int]: async def _get_recent_runs(limit: int = 2) -> list[str]: """取最近 N 個 success 的 run_id (降序).""" from sqlalchemy import text as _sql + from src.db.base import get_db_context - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: rows = await db.execute( _sql("SELECT run_id FROM asset_discovery_run WHERE status='success' ORDER BY ended_at DESC LIMIT :lim"), {"lim": limit}, @@ -113,11 +114,12 @@ async def _diff_runs(newer_run: str, older_run: str) -> dict[str, int]: - newer ∩ older 且 metadata 變 = modified """ from sqlalchemy import text as _sql + from src.db.base import get_db_context stats = {"added": 0, "removed": 0, "modified": 0} - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: # 1. Added: newer run 有但 older run 沒有 result = await db.execute( _sql(""" @@ -216,10 +218,11 @@ async def _diff_runs(newer_run: str, older_run: str) -> dict[str, int]: async def _log_aol(stats: dict[str, int], duration_ms: int, error: str | None) -> None: try: from sqlalchemy import text as _sql + from src.db.base import get_db_context aol_status = "failed" if error else "success" - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" INSERT INTO automation_operation_log ( diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index cef3d872f..5cc49c313 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -206,7 +206,7 @@ async def run_asset_scanner_loop() -> None: while True: try: - await scan_once(triggered_by="cron") + await scan_once(triggered_by="cron", raise_on_failure=True) except Exception as e: logger.exception("asset_scanner_loop_error", error=str(e)) await asyncio.sleep(_LOOP_BACKOFF_SEC) @@ -227,6 +227,7 @@ async def scan_once( "domain_tls_inventory", ), scan_depth: str = "shallow", + raise_on_failure: bool = False, ) -> str: """ 執行一次資產盤點。 @@ -235,6 +236,7 @@ async def scan_once( triggered_by: cron / ai / human / incident scope: 本次掃描範圍標籤 (寫入 asset_discovery_run.scope) scan_depth: shallow (僅 list) / deep (含 describe) / full + raise_on_failure: durable terminal 寫入後拋錯，供排程進入 bounded retry Returns: run_id (UUID 字串)。無法建立 durable header 時會拋出例外。 @@ -324,6 +326,8 @@ async def scan_once( disappeared=disappeared_count, duration_ms=duration_ms, ) + if error_msg and raise_on_failure: + raise RuntimeError(f"asset_scan_failed:{run_id}:{error_msg}") return run_id diff --git a/apps/api/src/jobs/rule_catalog_sync_job.py b/apps/api/src/jobs/rule_catalog_sync_job.py index 30655e93b..facbf3589 100644 --- a/apps/api/src/jobs/rule_catalog_sync_job.py +++ b/apps/api/src/jobs/rule_catalog_sync_job.py @@ -48,6 +48,7 @@ _SYNC_INTERVAL_SEC = 3600 # 每 1 小時 _FIRST_DELAY_SEC = 90 # 啟動後等 90s (等 Prometheus 就緒) _HTTP_TIMEOUT_SEC = 10 _LOOP_BACKOFF_SEC = 300 +_PROJECT_ID = "awoooi" _PROM_RULES_ENDPOINT = "/api/v1/rules" @@ -65,11 +66,19 @@ async def run_rule_catalog_sync_loop() -> None: while True: try: - await sync_once() + stats = await sync_once() except Exception as e: logger.exception("rule_catalog_sync_loop_error", error=str(e)) await asyncio.sleep(_LOOP_BACKOFF_SEC) continue + if stats.get("failed", 0) > 0: + logger.warning( + "rule_catalog_sync_retry_scheduled", + failed=stats["failed"], + retry_seconds=_LOOP_BACKOFF_SEC, + ) + await asyncio.sleep(_LOOP_BACKOFF_SEC) + continue await asyncio.sleep(_SYNC_INTERVAL_SEC) @@ -78,11 +87,11 @@ async def sync_once() -> dict[str, int]: 執行一次 Prometheus → alert_rule_catalog 同步. Returns: - {"total": N, "new": M, "updated": K, "unchanged": L} - 失敗時所有值為 0,error 會寫 aol. + {"total": N, "new": M, "updated": K, "unchanged": L, "failed": F} + 失敗時 failed > 0，error 會寫 AOL 並由排程 bounded retry. """ started_ms = _time.time() - stats = {"total": 0, "new": 0, "updated": 0, "unchanged": 0} + stats = {"total": 0, "new": 0, "updated": 0, "unchanged": 0, "failed": 0} error_msg: str | None = None try: @@ -98,6 +107,8 @@ async def sync_once() -> dict[str, int]: stats["unchanged"] += 1 except Exception as e: error_msg = f"{type(e).__name__}: {e}"[:1000] + processed = stats["new"] + stats["updated"] + stats["unchanged"] + stats["failed"] = max(1, stats["total"] - processed) logger.exception("rule_catalog_sync_once_failed", error=error_msg) duration_ms = int((_time.time() - started_ms) * 1000) @@ -159,7 +170,7 @@ async def _fetch_prometheus_rules() -> list[dict[str, Any]]: def _parse_duration(d: Any) -> int: """Prometheus 的 duration 可能是 秒 (int/float) 或 '5m' 字串.""" - if isinstance(d, (int, float)): + if isinstance(d, int | float): return int(d) if isinstance(d, str) and d: try: @@ -189,10 +200,11 @@ async def _upsert_rule(rule: dict[str, Any]) -> str: 比較 expr/severity/labels/annotations,有變化 → updated,否則 unchanged. """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: row = await db.execute( _sql(""" INSERT INTO alert_rule_catalog ( @@ -237,7 +249,7 @@ async def _upsert_rule(rule: dict[str, Any]) -> str: return "new" if inserted else "updated" except Exception as e: logger.warning("rule_upsert_failed", rule_name=rule["rule_name"], error=str(e)) - return "unchanged" + raise async def _log_aol(stats: dict[str, int], duration_ms: int, error: str | None) -> None: @@ -256,9 +268,10 @@ async def _log_aol(stats: dict[str, int], duration_ms: int, error: str | None) - try: from sqlalchemy import text as _sql + from src.db.base import get_db_context - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" INSERT INTO automation_operation_log ( diff --git a/apps/api/tests/test_ai_automation_asset_capability_matrix.py b/apps/api/tests/test_ai_automation_asset_capability_matrix.py index 0e9d3185f..8235dbb42 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -454,13 +454,24 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( monkeypatch: Any, ) -> None: live_readback_calls: list[dict[str, object]] = [] + settled_at = (datetime.now(UTC) - timedelta(minutes=10)).isoformat() async def _fake_live_readback(**kwargs: object) -> dict[str, object]: live_readback_calls.append(dict(kwargs)) if kwargs["include_live_only_assets"] is True: - return {"live_readback_status": "ready"} + return { + "live_readback_status": "ready", + "matrix_fingerprint": "stable-fingerprint", + "latest_discovery_run": { + "run_id": "discovery-run-1", + "ended_at": settled_at, + }, + } return {"live_readback_status": "ready", "closed": True} + async def _no_wait(_seconds: int) -> None: + return None + async def _fake_persist( _matrix: object, _candidates: object, @@ -481,6 +492,7 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( lambda _matrix: [], ) monkeypatch.setattr(reconciliation_job, "_persist_reconciliation", _fake_persist) + monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _no_wait) result = asyncio.run( reconciliation_job.reconcile_once( @@ -495,6 +507,11 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( "timeout_seconds": 30.0, "include_live_only_assets": True, }, + { + "project_id": "awoooi", + "timeout_seconds": 30.0, + "include_live_only_assets": True, + }, { "project_id": "awoooi", "timeout_seconds": 30.0, @@ -505,6 +522,95 @@ def test_reconcile_once_publishes_closed_public_matrix_cache( assert result["public_cache_refreshed"] is True +def test_reconcile_once_waits_for_settled_discovery_before_write( + monkeypatch: Any, +) -> None: + async def _fake_live_readback(**_kwargs: object) -> dict[str, object]: + return { + "live_readback_status": "ready", + "matrix_fingerprint": "too-fresh", + "latest_discovery_run": { + "run_id": "discovery-run-1", + "ended_at": datetime.now(UTC).isoformat(), + }, + } + + async def _unexpected_persist(*_args: object, **_kwargs: object) -> None: + raise AssertionError("unsettled source must not persist reconciliation") + + monkeypatch.setattr( + reconciliation_job, + "build_asset_capability_matrix_with_live_readback", + _fake_live_readback, + ) + monkeypatch.setattr( + reconciliation_job, + "_persist_reconciliation", + _unexpected_persist, + ) + + result = asyncio.run( + reconciliation_job.reconcile_once( + triggered_by="controlled_replay", + project_id="awoooi", + ) + ) + + assert result["status"] == "source_snapshot_settling" + assert result["reason"] == "latest_discovery_run_not_settled" + assert result["retry_seconds"] >= 60 + assert result["external_runtime_write_performed"] is False + + +def test_reconcile_once_retries_when_source_changes_during_stability_check( + monkeypatch: Any, +) -> None: + fingerprints = iter(("first-fingerprint", "second-fingerprint")) + settled_at = (datetime.now(UTC) - timedelta(minutes=10)).isoformat() + + async def _fake_live_readback(**_kwargs: object) -> dict[str, object]: + return { + "live_readback_status": "ready", + "matrix_fingerprint": next(fingerprints), + "latest_discovery_run": { + "run_id": "discovery-run-1", + "ended_at": settled_at, + }, + } + + async def _no_wait(_seconds: int) -> None: + return None + + async def _unexpected_persist(*_args: object, **_kwargs: object) -> None: + raise AssertionError("unstable source must not persist reconciliation") + + monkeypatch.setattr( + reconciliation_job, + "build_asset_capability_matrix_with_live_readback", + _fake_live_readback, + ) + monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _no_wait) + monkeypatch.setattr( + reconciliation_job, + "_persist_reconciliation", + _unexpected_persist, + ) + + result = asyncio.run( + reconciliation_job.reconcile_once( + triggered_by="controlled_replay", + project_id="awoooi", + ) + ) + + assert result == { + "status": "source_snapshot_unstable", + "reason": "matrix_changed_during_stability_verifier", + "retry_seconds": 60, + "external_runtime_write_performed": False, + } + + def test_reconciliation_loop_retries_degraded_result_before_poll( monkeypatch: Any, ) -> None: @@ -524,9 +630,9 @@ def test_reconciliation_loop_retries_degraded_result_before_poll( monkeypatch.setattr(reconciliation_job, "reconcile_once", _fake_reconcile_once) monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep) - monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90) + monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 420) monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800) - monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 1) + monkeypatch.setattr(reconciliation_job, "_RECONCILIATION_POLL_SECONDS", 3_600) try: asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop()) @@ -536,7 +642,7 @@ def test_reconciliation_loop_retries_degraded_result_before_poll( raise AssertionError("loop should stop at the test cancellation point") assert triggered_by_values == ["startup", "retry"] - assert sleep_values == [90, 1_800, 1] + assert sleep_values == [420, 1_800, 3_600] def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None: diff --git a/apps/api/tests/test_asset_change_tracker_job.py b/apps/api/tests/test_asset_change_tracker_job.py new file mode 100644 index 000000000..11f473995 --- /dev/null +++ b/apps/api/tests/test_asset_change_tracker_job.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import src.jobs.asset_change_tracker_job as tracker + + +class _FakeResult: + def __init__(self, rows: list[tuple[str]] | None = None) -> None: + self._rows = rows or [] + self.rowcount = 0 + + def fetchall(self) -> list[tuple[str]]: + return self._rows + + +class _FakeDb: + async def execute( + self, + statement: Any, + _params: dict[str, Any] | None = None, + ) -> _FakeResult: + if "SELECT run_id FROM asset_discovery_run" in str(statement): + return _FakeResult([("newer-run",), ("older-run",)]) + return _FakeResult() + + +class _FakeDbContext: + async def __aenter__(self) -> _FakeDb: + return _FakeDb() + + async def __aexit__(self, *_args: object) -> None: + return None + + +def test_change_tracker_scopes_every_database_path_to_awoooi(monkeypatch: Any) -> None: + project_ids: list[str] = [] + + def _get_db_context(project_id: str) -> _FakeDbContext: + project_ids.append(project_id) + return _FakeDbContext() + + monkeypatch.setattr("src.db.base.get_db_context", _get_db_context) + + assert asyncio.run(tracker._get_recent_runs()) == ["newer-run", "older-run"] + assert asyncio.run(tracker._diff_runs("newer-run", "older-run")) == { + "added": 0, + "removed": 0, + "modified": 0, + } + asyncio.run(tracker._log_aol({"added": 0, "removed": 0, "modified": 0}, 1, None)) + + assert project_ids == ["awoooi", "awoooi", "awoooi"] diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index 8ccb262d9..adaefb78b 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -4,6 +4,8 @@ import asyncio import json from pathlib import Path +import pytest + from src.jobs import asset_scanner_job @@ -1547,6 +1549,10 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) - "RuntimeError: asset_collector_failures=prometheus_targets" ) + with pytest.raises(RuntimeError, match="asset_scan_failed"): + asyncio.run(asset_scanner_job.scan_once(raise_on_failure=True)) + assert terminal["status"] == "failed" + def test_database_catalog_collector_reads_metadata_only(monkeypatch) -> None: requested_projects: list[str | None] = [] diff --git a/apps/api/tests/test_rule_catalog_sync_job.py b/apps/api/tests/test_rule_catalog_sync_job.py new file mode 100644 index 000000000..5c8f182c7 --- /dev/null +++ b/apps/api/tests/test_rule_catalog_sync_job.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import src.jobs.rule_catalog_sync_job as rule_sync + + +class _FakeResult: + def one_or_none(self) -> tuple[str, bool]: + return "rule-id", True + + +class _FakeDb: + async def execute( + self, + _statement: Any, + _parameters: dict[str, Any] | None = None, + ) -> _FakeResult: + return _FakeResult() + + +class _FakeDbContext: + async def __aenter__(self) -> _FakeDb: + return _FakeDb() + + async def __aexit__(self, *_args: object) -> None: + return None + + +def _rule(name: str = "ExampleRule") -> dict[str, Any]: + return { + "rule_name": name, + "expr": "vector(1)", + "duration_seconds": 60, + "severity": "warning", + "labels": {}, + "annotations": {}, + "group_name": "example", + } + + +def test_rule_catalog_database_paths_are_tenant_scoped(monkeypatch: Any) -> None: + project_ids: list[str] = [] + + def _get_db_context(project_id: str) -> _FakeDbContext: + project_ids.append(project_id) + return _FakeDbContext() + + monkeypatch.setattr("src.db.base.get_db_context", _get_db_context) + + assert asyncio.run(rule_sync._upsert_rule(_rule())) == "new" + asyncio.run( + rule_sync._log_aol( + {"total": 1, "new": 1, "updated": 0, "unchanged": 0, "failed": 0}, + duration_ms=1, + error=None, + ) + ) + + assert project_ids == ["awoooi", "awoooi"] + + +def test_rule_catalog_failure_is_visible_and_durably_logged(monkeypatch: Any) -> None: + captured: dict[str, Any] = {} + + async def _fetch() -> list[dict[str, Any]]: + return [_rule("RuleA"), _rule("RuleB")] + + async def _fail_upsert(_item: dict[str, Any]) -> str: + raise RuntimeError("database unavailable") + + async def _capture_log( + stats: dict[str, int], + duration_ms: int, + error: str | None, + ) -> None: + captured.update(stats=stats.copy(), duration_ms=duration_ms, error=error) + + monkeypatch.setattr(rule_sync, "_fetch_prometheus_rules", _fetch) + monkeypatch.setattr(rule_sync, "_upsert_rule", _fail_upsert) + monkeypatch.setattr(rule_sync, "_log_aol", _capture_log) + + result = asyncio.run(rule_sync.sync_once()) + + assert result == { + "total": 2, + "new": 0, + "updated": 0, + "unchanged": 0, + "failed": 2, + } + assert str(captured["error"]).startswith("RuntimeError: database unavailable") + assert captured["stats"] == result + + +def test_rule_catalog_loop_retries_failed_sync_before_hourly_wait( + monkeypatch: Any, +) -> None: + sync_count = 0 + sleep_values: list[int] = [] + + async def _sync_once() -> dict[str, int]: + nonlocal sync_count + sync_count += 1 + return { + "total": 1, + "new": 0, + "updated": 0, + "unchanged": 0 if sync_count == 1 else 1, + "failed": 1 if sync_count == 1 else 0, + } + + async def _sleep(seconds: int) -> None: + sleep_values.append(seconds) + if len(sleep_values) == 3: + raise asyncio.CancelledError + + monkeypatch.setattr(rule_sync, "sync_once", _sync_once) + monkeypatch.setattr(rule_sync.asyncio, "sleep", _sleep) + monkeypatch.setattr(rule_sync, "_FIRST_DELAY_SEC", 1) + monkeypatch.setattr(rule_sync, "_LOOP_BACKOFF_SEC", 2) + monkeypatch.setattr(rule_sync, "_SYNC_INTERVAL_SEC", 3) + + try: + asyncio.run(rule_sync.run_rule_catalog_sync_loop()) + except asyncio.CancelledError: + pass + else: + raise AssertionError("loop should stop at the test cancellation point") + + assert sync_count == 2 + assert sleep_values == [1, 2, 3] diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 7eaf00e8a..086d00987 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -93,6 +93,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168" assert worker_env["DATABASE_NULL_POOL"] == "true" assert worker_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "1" + assert worker_env["DATABASE_POOL_TIMEOUT_SECONDS"] == "30" def test_execution_broker_is_only_workload_with_ssh_transport() -> None: diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index c9a589790..33118d88e 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -174,6 +174,10 @@ spec: # NullPool releases each connection after use, while this cap # keeps concurrent background loops inside the worker role budget. value: "1" + - name: DATABASE_POOL_TIMEOUT_SECONDS + # Preserve the one-connection budget while allowing serialized + # background jobs to wait through a bounded long query. + value: "30" - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA