From abc512a7b3911f44c48668d06965423f56471ab1 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 1 Jul 2026 08:45:40 +0800 Subject: [PATCH] fix(db): split inbound truth-chain hot lookup --- ..._event_content_preview_trgm_2026-07-01.sql | 15 ++ ...t_content_preview_trgm_2026-07-01_down.sql | 5 + .../services/awooop_truth_chain_service.py | 164 ++++++++++++------ ...oop_conversation_event_hot_path_indexes.py | 32 ++++ .../tests/test_awooop_truth_chain_service.py | 27 ++- docs/LOGBOOK.md | 24 +++ ...re-drain-readback-2026-07-01.snapshot.json | 43 ++++- 7 files changed, 245 insertions(+), 65 deletions(-) create mode 100644 apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql create mode 100644 apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql diff --git a/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql b/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql new file mode 100644 index 000000000..dcd777a7b --- /dev/null +++ b/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql @@ -0,0 +1,15 @@ +-- P0 post-reboot CPU pressure: bounded fallback index for truth-chain source lookup. +-- Run outside an explicit transaction because CREATE INDEX CONCURRENTLY requires it. + +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS btree_gin; + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_conv_event_content_preview_trgm +ON awooop_conversation_event +USING gin (content_preview gin_trgm_ops) +WHERE content_preview IS NOT NULL; + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_conv_event_project_content_preview_trgm +ON awooop_conversation_event +USING gin (project_id, content_preview gin_trgm_ops) +WHERE content_preview IS NOT NULL; diff --git a/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql b/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql new file mode 100644 index 000000000..3a743759e --- /dev/null +++ b/apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql @@ -0,0 +1,5 @@ +-- Roll back only the index owned by the P0 content_preview fallback migration. +-- Do not drop pg_trgm; the extension may be shared by other indexes. + +DROP INDEX CONCURRENTLY IF EXISTS idx_awooop_conv_event_content_preview_trgm; +DROP INDEX CONCURRENTLY IF EXISTS idx_awooop_conv_event_project_content_preview_trgm; diff --git a/apps/api/src/services/awooop_truth_chain_service.py b/apps/api/src/services/awooop_truth_chain_service.py index 9662247ed..e33266f41 100644 --- a/apps/api/src/services/awooop_truth_chain_service.py +++ b/apps/api/src/services/awooop_truth_chain_service.py @@ -34,6 +34,9 @@ from src.services.operator_summary_cache import ( logger = structlog.get_logger(__name__) _MAX_ROWS = 100 +_INBOUND_LOOKUP_BRANCH_LIMIT = int( + os.getenv("AWOOOP_TRUTH_CHAIN_INBOUND_BRANCH_LIMIT", "25") +) _JSON_TEXT_FIELDS = {"gate_result", "source_envelope"} _QUALITY_SUMMARY_CACHE_TTL_SECONDS = int( os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "30") @@ -130,6 +133,108 @@ async def _fetch_one( return _clean_row(row) if row else None +def _inbound_lookup_branch_limit(limit: int) -> int: + return max(1, min(int(limit), _INBOUND_LOOKUP_BRANCH_LIMIT)) + + +async def _fetch_inbound_conversation_event_rows( + db: Any, + *, + project_id: str, + source_id: str, + fingerprint_needle: str, + fingerprint_value: str, + limit: int = _MAX_ROWS, +) -> list[dict[str, Any]]: + columns = """ + event_id, + project_id, + channel_type, + provider_event_id, + platform_subject_id, + channel_user_id, + channel_chat_id, + run_id, + content_type, + content_hash, + content_preview, + content_redacted, + redaction_version, + source_envelope, + attachment_sha256, + is_duplicate, + provider_ts, + received_at + """ + branch_limit = _inbound_lookup_branch_limit(limit) + rows_by_id: dict[str, dict[str, Any]] = {} + source_ref_paths = ( + "{source_refs,event_ids}", + "{source_refs,incident_ids}", + "{source_refs,approval_ids}", + "{source_refs,alert_ids}", + "{source_refs,sentry_issue_ids}", + "{source_refs,signoz_alerts}", + ) + + async def fetch_branch(where_sql: str, params: dict[str, Any]) -> None: + if len(rows_by_id) >= limit: + return + rows = await _fetch_all( + db, + f""" + SELECT + {columns} + FROM awooop_conversation_event + WHERE project_id = :project_id + AND ({where_sql}) + ORDER BY received_at DESC + LIMIT :limit + """, + { + "project_id": project_id, + "source_id": source_id, + "limit": min(branch_limit, limit - len(rows_by_id)), + **params, + }, + ) + for row in rows: + event_id = str(row.get("event_id") or "") + if event_id: + rows_by_id.setdefault(event_id, row) + + await fetch_branch("run_id::text = :source_id", {}) + await fetch_branch("provider_event_id = :source_id", {}) + for source_ref_path in source_ref_paths: + await fetch_branch( + f"coalesce(source_envelope #> '{source_ref_path}', '[]'::jsonb) ? :source_id", + {}, + ) + + if fingerprint_value: + await fetch_branch( + "coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value", + {"fingerprint_value": fingerprint_value}, + ) + + if len(rows_by_id) < branch_limit: + await fetch_branch( + "content_preview ILIKE :source_needle", + {"source_needle": f"%{source_id}%"}, + ) + if fingerprint_needle and len(rows_by_id) < branch_limit: + await fetch_branch( + "provider_event_id ILIKE :fingerprint_needle OR content_preview ILIKE :fingerprint_needle", + {"fingerprint_needle": fingerprint_needle}, + ) + + return sorted( + rows_by_id.values(), + key=lambda row: str(row.get("received_at") or ""), + reverse=True, + )[:limit] + + def _source_type(source_id: str, incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str: if incident is not None: return "incident" @@ -1745,60 +1850,13 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[ else "" ) fingerprint_value = incident_fingerprints[0] if incident_fingerprints else "" - inbound_rows = await _fetch_all( + inbound_rows = await _fetch_inbound_conversation_event_rows( db, - """ - SELECT - event_id, - project_id, - channel_type, - provider_event_id, - platform_subject_id, - channel_user_id, - channel_chat_id, - run_id, - content_type, - content_hash, - content_preview, - content_redacted, - redaction_version, - source_envelope, - attachment_sha256, - is_duplicate, - provider_ts, - received_at - FROM awooop_conversation_event - WHERE project_id = :project_id - AND ( - run_id::text = :source_id - OR provider_event_id = :source_id - OR content_preview ILIKE :source_needle - OR coalesce(source_envelope #> '{source_refs,event_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,approval_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,alert_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,sentry_issue_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,signoz_alerts}', '[]'::jsonb) ? :source_id - OR ( - :fingerprint_needle != '' - AND ( - provider_event_id ILIKE :fingerprint_needle - OR content_preview ILIKE :fingerprint_needle - OR coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value - ) - ) - ) - ORDER BY received_at DESC - LIMIT :limit - """, - { - "source_id": source_id, - "project_id": project_id, - "source_needle": f"%{source_id}%", - "fingerprint_needle": fingerprint_needle, - "fingerprint_value": fingerprint_value, - "limit": _MAX_ROWS, - }, + project_id=project_id, + source_id=source_id, + fingerprint_needle=fingerprint_needle, + fingerprint_value=fingerprint_value, + limit=_MAX_ROWS, ) outbound_rows = await _fetch_all( db, diff --git a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py index 3be19aa64..9b7453841 100644 --- a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py +++ b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py @@ -3,6 +3,8 @@ from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MIGRATION = "awooop_conversation_event_hot_path_indexes_2026-07-01.sql" ROLLBACK = "awooop_conversation_event_hot_path_indexes_2026-07-01_down.sql" +CONTENT_PREVIEW_MIGRATION = "awooop_conversation_event_content_preview_trgm_2026-07-01.sql" +CONTENT_PREVIEW_ROLLBACK = "awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql" def _read(path: str) -> str: @@ -17,6 +19,14 @@ def _rollback_sql() -> str: return _read(f"migrations/{ROLLBACK}") +def _content_preview_migration_sql() -> str: + return _read(f"migrations/{CONTENT_PREVIEW_MIGRATION}") + + +def _content_preview_rollback_sql() -> str: + return _read(f"migrations/{CONTENT_PREVIEW_ROLLBACK}") + + def test_hot_path_indexes_are_concurrent_and_transactionless() -> None: sql = _migration_sql() @@ -93,3 +103,25 @@ def test_rollback_only_drops_indexes_owned_by_hot_path_migration() -> None: "idx_awooop_conv_event_source_refs_fingerprints_gin", ): assert f"DROP INDEX CONCURRENTLY IF EXISTS {index_name};" in rollback + + +def test_content_preview_trgm_index_bounds_truth_chain_preview_fallback() -> None: + sql = _content_preview_migration_sql() + rollback = _content_preview_rollback_sql() + truth_chain_source = _read("src/services/awooop_truth_chain_service.py") + + assert "BEGIN;" not in sql + assert "COMMIT;" not in sql + assert "CREATE EXTENSION IF NOT EXISTS pg_trgm" in sql + assert "CREATE EXTENSION IF NOT EXISTS btree_gin" in sql + assert "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_conv_event_content_preview_trgm" in sql + assert "USING gin (content_preview gin_trgm_ops)" in sql + assert "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_conv_event_project_content_preview_trgm" in sql + assert "USING gin (project_id, content_preview gin_trgm_ops)" in sql + assert "WHERE content_preview IS NOT NULL" in sql + assert "content_preview ILIKE :source_needle" in truth_chain_source + assert "content_preview ILIKE :fingerprint_needle" in truth_chain_source + + assert "DROP INDEX CONCURRENTLY IF EXISTS idx_awooop_conv_event_content_preview_trgm;" in rollback + assert "DROP INDEX CONCURRENTLY IF EXISTS idx_awooop_conv_event_project_content_preview_trgm;" in rollback + assert "DROP EXTENSION" not in rollback diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index f186e3b11..22d2318f4 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -38,6 +38,7 @@ from src.services.awooop_truth_chain_service import ( _automation_quality_score_bucket, _clean_row, _execution_backend_summary, + _fetch_inbound_conversation_event_rows, _incident_fingerprints, _summarize_gateway_mcp, _truth_status, @@ -131,13 +132,29 @@ def test_ansible_transport_cooldown_uses_asyncpg_safe_interval_parameter() -> No def test_fetch_truth_chain_returns_inbound_redacted_envelope_fields() -> None: source = inspect.getsource(fetch_truth_chain) + helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows) assert "content_redacted" in source - assert "source_envelope" in source - assert "source_refs,event_ids" in source - assert "source_refs,incident_ids" in source - assert "source_refs,sentry_issue_ids" in source - assert "source_refs,signoz_alerts" in source + assert "source_envelope" in helper_source + assert "source_refs,event_ids" in helper_source + assert "source_refs,incident_ids" in helper_source + assert "source_refs,sentry_issue_ids" in helper_source + assert "source_refs,signoz_alerts" in helper_source + + +def test_fetch_truth_chain_splits_inbound_lookup_into_index_friendly_branches() -> None: + source = inspect.getsource(fetch_truth_chain) + helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows) + + assert "await _fetch_inbound_conversation_event_rows(" in source + assert "run_id::text = :source_id" in helper_source + assert "provider_event_id = :source_id" in helper_source + assert "source_refs,event_ids" in helper_source + assert "source_refs,incident_ids" in helper_source + assert "content_preview ILIKE :source_needle" in helper_source + assert "len(rows_by_id) < branch_limit" in helper_source + assert "OR content_preview ILIKE :source_needle" not in helper_source + assert "OR coalesce(source_envelope #>" not in helper_source def test_truth_status_marks_no_action_approval_as_manual_required() -> None: diff --git a/docs/LOGBOOK.md b/docs/LOGBOOK.md index 5034d073d..08dbdfcdd 100644 --- a/docs/LOGBOOK.md +++ b/docs/LOGBOOK.md @@ -51165,3 +51165,27 @@ production browser smoke: **下一步**： - rebase / push source 修法；110 控制面恢復後第一個 runtime 動作是套用新版 `systemd-units-textfile-exporter.py` 或暫停舊 exporter，再回讀 DBus/logind、systemd_units.prom 與 Gitea CPU。188 runner 只在 host pressure gate / queue readback 綠燈後恢復。 + +## 2026-07-01 — 08:44 P0 188 truth-chain source split 與 110 控制面讀回 + +**完成內容**： +- 188 DB CPU 再次讀回仍高：`k3s-postgres-recovery=609.72%`，active query 明確是 120 live API 發出的舊 `awooop_conversation_event` inbound lookup，大量 `run_id::text / provider_event_id / content_preview ILIKE / source_refs JSONB` 放在同一個 `OR` predicate，並產生 local parallel workers。 +- 已 live 套用 content preview fallback 索引：`idx_awooop_conv_event_content_preview_trgm` 與 `idx_awooop_conv_event_project_content_preview_trgm` 均讀回 `indisvalid=true / indisready=true`。 +- 新增 migration source：`apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql` / `_down.sql`，使用 `CREATE INDEX CONCURRENTLY IF NOT EXISTS`、`pg_trgm`、`btree_gin`，rollback 只 drop 本 migration indexes，不 drop extension。 +- 修正 `apps/api/src/services/awooop_truth_chain_service.py`：把 inbound conversation event lookup 拆成逐分支查詢，先查 `run_id`、`provider_event_id`、各 `source_refs` JSONB key，再有限度 fallback 到 `content_preview ILIKE` / fingerprint lookup；移除會讓 planner 放大掃描的大型 source-envelope `OR` 查詢。 +- 188 runtime 暫時降壓：`ALTER ROLE awoooi SET max_parallel_workers_per_gather=0`，取消 5 條舊 OR active query，並只斷開 `192.168.0.120` 到 `awoooi_prod` 的 30 條 app DB session 讓新設定生效。rollback：source split 部署收斂後執行 `ALTER ROLE awoooi RESET max_parallel_workers_per_gather`。 +- post-mitigation 讀回：`awoooi` role config 已是 `max_parallel_workers_per_gather=0`、舊 OR 查詢 local parallel workers 降為 `0`；但舊 OR query 仍有 `4` 條 active，`k3s-postgres-recovery=400.19%`，所以真正完成仍需 120 部署 source split。 +- 110 最新 node exporter 讀回：`node_load5=29.34`、`awoooi_host_load5_per_core=2.068333`、`node_procs_running=31`、`gitea=3.4019` cores；SSH / scp / stdin-over-ssh 嘗試仍 timeout，尚未能 live 套用新版 systemd exporter。 + +**本地驗證結果**： +- `DATABASE_URL=postgresql+asyncpg://test:test@localhost:5432/test PYTHONPATH=apps/api python3.11 -m pytest apps/api/tests/test_awooop_truth_chain_service.py::test_fetch_truth_chain_returns_inbound_redacted_envelope_fields apps/api/tests/test_awooop_truth_chain_service.py::test_fetch_truth_chain_splits_inbound_lookup_into_index_friendly_branches apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py -q`：`7 passed`。 +- `python3.11 -m py_compile apps/api/src/services/awooop_truth_chain_service.py`：通過。 +- 已更新 `docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json`，保留 188 temporary DB mitigation、source split、110 control-path blocker 與 rollback。 + +**仍維持**： +- 沒有讀 secret / token / `.env` / raw sessions / SQLite / auth；沒有讀 `.runner` 內容。 +- 沒有使用 GitHub / gh / GitHub API / GitHub Actions。 +- 沒有重啟主機，沒有 Docker / Nginx / K3s / DB restart，沒有 workflow_dispatch，沒有 DROP / TRUNCATE / restore / prune。 + +**下一步**： +- rebase / commit / push truth-chain source split；讀回 Gitea queue 與 188 DB active query。若 deploy 收斂，立即 reset `awoooi` role 的 temporary `max_parallel_workers_per_gather=0`；若 110 SSH 恢復，第一動作仍是套用新版 systemd exporter 或暫停舊 exporter，再回讀 systemd/logind、Gitea CPU 與 load5/core。 diff --git a/docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json b/docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json index a9a79d60d..9d6e22b9b 100644 --- a/docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json +++ b/docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json @@ -1,7 +1,7 @@ { "schema_version": "awoooi_host_cpu_pressure_drain_readback_v1", - "generated_at": "2026-07-01T08:19:00+08:00", - "status": "partial_110_control_path_blocked_188_drained", + "generated_at": "2026-07-01T08:44:00+08:00", + "status": "partial_188_bounded_source_fix_ready_110_control_path_blocked", "scope": { "hosts": ["188", "110"], "incident_family": "post_reboot_host_cpu_pressure", @@ -35,6 +35,14 @@ "reason": "active CD API tests were repeatedly spawning and driving production PostgreSQL CPU", "rollback": "restart the non-110 runner only after host pressure guard and source CD gate are green", "post_apply_readback": "runner inactive, no awoooi-cd-* containers, k3s-postgres-recovery docker stats 7.81 percent" + }, + { + "host": "188", + "target_selector": "awoooi PostgreSQL role sessions from 192.168.0.120 running old awooop_conversation_event OR lookup", + "action": "set max_parallel_workers_per_gather=0 for role, cancel active old OR queries, and terminate 192.168.0.120 app DB sessions so new sessions inherit the bound", + "reason": "the live 120 API was still repeatedly issuing the pre-fix OR query and spawning parallel workers after index apply", + "rollback": "ALTER ROLE awoooi RESET max_parallel_workers_per_gather; then let the app reconnect after source split deploys", + "post_apply_readback": "rolconfig shows max_parallel_workers_per_gather=0; local parallel workers for the old query dropped to 0" } ], "source_fixes": [ @@ -47,6 +55,16 @@ "path": "scripts/ops/systemd-units-textfile-exporter.py", "change": "shorten systemctl show timeout, compact error labels, stop probing after timeout budget, and add non-overlap lock", "evidence": "pytest covers timeout budget and busy lock" + }, + { + "path": "apps/api/src/services/awooop_truth_chain_service.py", + "change": "split the inbound conversation event lookup into index-friendly branches instead of one large OR predicate", + "evidence": "focused tests confirm the source_refs/content_preview branches are isolated and the old source_envelope OR pattern does not return" + }, + { + "path": "apps/api/migrations/awooop_conversation_event_content_preview_trgm_2026-07-01.sql", + "change": "add bounded pg_trgm / btree_gin indexes for content_preview fallback lookup", + "evidence": "live verifier read back both trigram indexes as valid and ready" } ], "readback": { @@ -64,13 +82,21 @@ "pg_dump_process": "none", "k3s_postgres_recovery_docker_stats_percent": "7.81", "docker_stats_textfile_k3s_postgres_recovery_cores": "0.406900" + }, + "post_bounded_db_mitigation_signals": { + "awoooi_role_config": "max_parallel_workers_per_gather=0", + "active_old_or_queries": "4", + "local_parallel_workers_for_old_or_query": "0", + "k3s_postgres_recovery_docker_stats_percent": "400.19", + "remaining_root_cause": "120 live API still runs the pre-fix OR query until the source split is deployed" } }, "host_110": { "signals": { "load5_initial": "39.14", - "load5_latest": "18.85", - "node_procs_running_latest": "6", + "load5_latest": "29.34", + "load5_per_core_latest": "2.068333", + "node_procs_running_latest": "31", "gitea_container_cpu_cores": "3.4019", "ssh_control_path": "timeout", "systemd_dbus_symptom": "systemd-logind pending replies exhausted; systemctl list/show timeout", @@ -78,22 +104,25 @@ }, "not_completed_live": [ "could not apply pkill/exporter drain on 110 because SSH command sessions timeout", + "scp and stdin-over-ssh attempts to place the new exporter file timed out", "could not restart or reexec systemd/logind because that would require a control path that is currently unavailable" ] } }, "verification": { - "py_compile": "passed: scripts/ops/systemd-units-textfile-exporter.py", + "py_compile": "passed: scripts/ops/systemd-units-textfile-exporter.py and apps/api/src/services/awooop_truth_chain_service.py", "bash_n": "passed: scripts/ci/wait-host-web-build-pressure.sh scripts/ops/systemd-units-textfile-exporter.py", "pytest": [ "scripts/ops/tests/test_systemd_units_textfile_exporter.py: 2 passed", - "scripts/ops/tests/test_systemd_units_textfile_exporter.py + ops/runner/test_cd_controlled_runtime_profile.py: 34 passed" + "scripts/ops/tests/test_systemd_units_textfile_exporter.py + ops/runner/test_cd_controlled_runtime_profile.py: 34 passed", + "truth-chain source split + content-preview migration focused suite: 7 passed" ], "diff_check": "passed" }, "next_actions": [ + "commit and push the truth-chain source split so 120 stops issuing the old OR query", + "after deploy convergence, reset the temporary awoooi role max_parallel_workers_per_gather override", "apply the systemd exporter source fix to 110 once SSH/control path is available", - "keep 188 non-110 runner drained until the pushed CD gate is in place and host pressure readback is green", "continue 110 Gitea queue / awoooi-host controlled lane recovery without generic runner restore" ] }