diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index b720bed5e..f6890ae93 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -1958,6 +1958,197 @@ jobs: asyncio.run(main()) PY + # A valid unique index can still hide heap/index drift introduced by an + # older restore or repair. Quarantine only the non-canonical rows, rebuild + # the index transactionally, and prove the conflict-update path before + # rolling out the scanner. + - name: Repair Asset Inventory Canonical Integrity + env: + DATABASE_URL: ${{ secrets.DATABASE_URL }} + MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }} + run: | + set -euo pipefail + if [ -z "${MIGRATION_DATABASE_URL:-}" ]; then + echo "::error::MIGRATION_DATABASE_URL secret not set in Gitea" + exit 1 + fi + docker run --rm --network bridge -i \ + -e DATABASE_URL \ + -e MIGRATION_DATABASE_URL \ + -v "$PWD/apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql:/tmp/asset_inventory_integrity.sql:ro" \ + "${HARBOR}/awoooi/api:${{ github.sha }}" python - <<'PY' + import asyncio + import json + import os + from pathlib import Path + + import asyncpg + + MIGRATION_NAME = "asset_inventory_canonical_integrity_2026-07-14" + MIGRATION = Path("/tmp/asset_inventory_integrity.sql").read_text( + encoding="utf-8" + ) + + + def normalize_url(value: str) -> str: + return value.replace("postgresql+asyncpg://", "postgresql://", 1) + + + def receipt_valid(receipt: dict[str, object] | None) -> bool: + return bool( + receipt + and receipt.get("duplicate_group_count_after") == 0 + and receipt.get("index_rebuilt") is True + and receipt.get("canonical_upsert_verified") is True + and receipt.get("data_rows_deleted") == 0 + ) + + + async def integrity_status( + connection: asyncpg.Connection, + ) -> tuple[int, int, bool, dict[str, object] | None]: + row = await connection.fetchrow( + """ + WITH duplicate_groups AS ( + SELECT COUNT(*) AS count + FROM ( + SELECT asset_key + FROM asset_inventory + GROUP BY asset_key + HAVING COUNT(*) > 1 + ) AS grouped_duplicates + ), latest_receipt AS ( + SELECT output::text AS output + FROM automation_operation_log + WHERE actor = 'gitea_cd_asset_integrity_reconciler' + AND operation_type = 'asset_discovered' + AND status = 'success' + AND input ->> 'migration' = $1 + ORDER BY created_at DESC + LIMIT 1 + ) + SELECT + duplicate_groups.count AS duplicate_group_count, + ( + SELECT COUNT(*) + FROM asset_inventory + WHERE lifecycle_state = 'deprecated' + AND metadata ->> 'integrity_state' = + 'quarantined_duplicate' + ) AS quarantined_row_count, + COALESCE(( + SELECT + index_row.indisvalid + AND index_row.indisready + AND index_row.indislive + AND index_row.indisunique + FROM pg_index AS index_row + WHERE index_row.indexrelid = + 'asset_inventory_asset_key_key'::regclass + ), FALSE) AS unique_index_healthy, + (SELECT output FROM latest_receipt) AS receipt_output + FROM duplicate_groups + """, + MIGRATION_NAME, + ) + if row is None: + raise RuntimeError("asset_inventory_integrity_status_missing") + receipt_text = row["receipt_output"] + receipt = json.loads(receipt_text) if receipt_text else None + return ( + int(row["duplicate_group_count"] or 0), + int(row["quarantined_row_count"] or 0), + bool(row["unique_index_healthy"]), + receipt, + ) + + + async def reconcile( + url: str, + ) -> tuple[int, int, bool, dict[str, object] | None, bool]: + connection = await asyncpg.connect(normalize_url(url), timeout=10) + applied = False + try: + # Do not let a corrupt-but-valid unique index hide duplicate + # heap rows from either the preflight or post-verifier. + await connection.execute("SET enable_indexscan TO off") + await connection.execute("SET enable_indexonlyscan TO off") + await connection.execute("SET enable_bitmapscan TO off") + duplicate_groups, _, index_healthy, receipt = ( + await integrity_status(connection) + ) + if ( + duplicate_groups > 0 + or not index_healthy + or not receipt_valid(receipt) + ): + async with connection.transaction(): + await connection.execute(MIGRATION) + applied = True + status = await integrity_status(connection) + return (*status, applied) + finally: + await connection.close() + + + async def main() -> None: + candidates = [ + ("migration_role", os.environ["MIGRATION_DATABASE_URL"]), + ("owner_role", os.environ.get("DATABASE_URL", "")), + ] + for index, (role, url) in enumerate(candidates): + if not url: + continue + try: + ( + duplicate_groups, + quarantined_rows, + index_healthy, + receipt, + applied, + ) = await reconcile(url) + verified = ( + duplicate_groups == 0 + and index_healthy + and receipt_valid(receipt) + ) + print(f"asset_inventory_integrity_migration_role={role}") + print( + "asset_inventory_integrity_migration_applied=" + f"{str(applied).lower()}" + ) + print(f"duplicate_group_count={duplicate_groups}") + print(f"quarantined_row_count={quarantined_rows}") + print(f"unique_index_healthy={str(index_healthy).lower()}") + print( + "durable_integrity_receipt_present=" + f"{str(receipt is not None).lower()}" + ) + print(f"canonical_integrity_verified={str(verified).lower()}") + if not verified: + raise RuntimeError( + "asset_inventory_canonical_integrity_verifier_failed" + ) + return + except asyncpg.PostgresError as exc: + permission_denied = getattr(exc, "sqlstate", "") == "42501" + has_owner_fallback = index == 0 and bool(candidates[1][1]) + if permission_denied and has_owner_fallback: + print("asset_inventory_integrity_owner_fallback=true") + continue + print( + "asset_inventory_integrity_error_class=" + f"{type(exc).__name__}" + ) + raise + raise RuntimeError( + "asset_inventory_integrity_migration_no_usable_connection" + ) + + + asyncio.run(main()) + PY + # 2026-03-31 ogt: 移除中間通知 # 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制) diff --git a/apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql b/apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql new file mode 100644 index 000000000..80602ae97 --- /dev/null +++ b/apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql @@ -0,0 +1,215 @@ +-- Repair duplicate canonical asset keys without deleting historical rows. +-- Safety: bounded quarantine, transactional index rebuild, independent verifier. + +SET lock_timeout = '10s'; +SET statement_timeout = '90s'; +SET enable_indexscan = 'off'; +SET enable_indexonlyscan = 'off'; +SET enable_bitmapscan = 'off'; + +CREATE TEMP TABLE asset_inventory_integrity_repair_map +ON COMMIT DROP +AS +WITH ranked_assets AS ( + SELECT + asset_id, + asset_key AS original_asset_key, + FIRST_VALUE(asset_id) OVER ( + PARTITION BY asset_key + ORDER BY first_seen_at ASC, asset_id ASC + ) AS canonical_asset_id, + ROW_NUMBER() OVER ( + PARTITION BY asset_key + ORDER BY first_seen_at ASC, asset_id ASC + ) AS duplicate_rank + FROM asset_inventory +) +SELECT + asset_id, + original_asset_key, + canonical_asset_id +FROM ranked_assets +WHERE duplicate_rank > 1; + +DO $$ +DECLARE + repair_row_count BIGINT; +BEGIN + SELECT COUNT(*) INTO repair_row_count + FROM asset_inventory_integrity_repair_map; + + IF repair_row_count > 100 THEN + RAISE EXCEPTION + 'asset_inventory_integrity_repair_bound_exceeded: % rows', + repair_row_count; + END IF; +END +$$; + +UPDATE asset_inventory AS duplicate_asset +SET + asset_key = duplicate_asset.asset_key + || '/__integrity_quarantine__/' + || duplicate_asset.asset_id::TEXT, + lifecycle_state = 'deprecated', + decommissioned_at = COALESCE(duplicate_asset.decommissioned_at, NOW()), + metadata = COALESCE(duplicate_asset.metadata, '{}'::JSONB) + || JSONB_BUILD_OBJECT( + 'integrity_state', 'quarantined_duplicate', + 'integrity_canonical_asset_id', repair.canonical_asset_id, + 'integrity_repair', 'asset_inventory_canonical_integrity_2026-07-14' + ), + tags = CASE + WHEN 'integrity:quarantined_duplicate' = ANY(duplicate_asset.tags) + THEN duplicate_asset.tags + ELSE ARRAY_APPEND( + duplicate_asset.tags, + 'integrity:quarantined_duplicate' + ) + END, + updated_at = NOW() +FROM asset_inventory_integrity_repair_map AS repair +WHERE duplicate_asset.asset_id = repair.asset_id; + +-- Rebuild the unique index after every formerly unindexed duplicate has a +-- distinct quarantine key. The outer CD transaction rolls back on any error. +REINDEX INDEX asset_inventory_asset_key_key; + +DO $$ +DECLARE + duplicate_group_count BIGINT; + repaired_row_count BIGINT; + expected_repair_count BIGINT; + index_healthy BOOLEAN; +BEGIN + SELECT COUNT(*) INTO duplicate_group_count + FROM ( + SELECT asset_key + FROM asset_inventory + GROUP BY asset_key + HAVING COUNT(*) > 1 + ) AS duplicate_groups; + + SELECT COUNT(*) INTO expected_repair_count + FROM asset_inventory_integrity_repair_map; + + SELECT COUNT(*) INTO repaired_row_count + FROM asset_inventory AS repaired + JOIN asset_inventory_integrity_repair_map AS repair + ON repair.asset_id = repaired.asset_id + WHERE repaired.lifecycle_state = 'deprecated' + AND repaired.asset_key = repair.original_asset_key + || '/__integrity_quarantine__/' + || repaired.asset_id::TEXT + AND repaired.metadata ->> 'integrity_state' = 'quarantined_duplicate'; + + SELECT + index_row.indisvalid + AND index_row.indisready + AND index_row.indislive + AND index_row.indisunique + INTO index_healthy + FROM pg_index AS index_row + WHERE index_row.indexrelid = 'asset_inventory_asset_key_key'::REGCLASS; + + IF duplicate_group_count <> 0 THEN + RAISE EXCEPTION + 'asset_inventory_duplicate_groups_remain: %', + duplicate_group_count; + END IF; + + IF repaired_row_count <> expected_repair_count THEN + RAISE EXCEPTION + 'asset_inventory_quarantine_verifier_failed: expected %, got %', + expected_repair_count, + repaired_row_count; + END IF; + + IF index_healthy IS DISTINCT FROM TRUE THEN + RAISE EXCEPTION 'asset_inventory_unique_index_not_healthy'; + END IF; +END +$$; + +-- Exercise the exact conflict-update path that failed before the repair while +-- preserving every application value. +WITH probe AS ( + SELECT + asset_key, + asset_type, + host, + namespace, + name, + metadata, + tags, + environment, + lifecycle_state, + first_seen_at, + last_seen_at + FROM asset_inventory + ORDER BY asset_id + LIMIT 1 +) +INSERT INTO asset_inventory ( + asset_key, + asset_type, + host, + namespace, + name, + metadata, + tags, + environment, + lifecycle_state, + first_seen_at, + last_seen_at +) +SELECT + asset_key, + asset_type, + host, + namespace, + name, + metadata, + tags, + environment, + lifecycle_state, + first_seen_at, + last_seen_at +FROM probe +ON CONFLICT (asset_key) DO UPDATE +SET + last_seen_at = asset_inventory.last_seen_at, + updated_at = asset_inventory.updated_at; + +INSERT INTO automation_operation_log ( + operation_type, + actor, + status, + input, + output, + tags +) +SELECT + 'asset_discovered', + 'gitea_cd_asset_integrity_reconciler', + 'success', + '{"work_item_id":"AIA-P0-006","migration":"asset_inventory_canonical_integrity_2026-07-14"}'::JSONB, + JSONB_BUILD_OBJECT( + 'duplicate_group_count_before', COUNT(DISTINCT original_asset_key), + 'rows_quarantined', COUNT(*), + 'duplicate_group_count_after', 0, + 'index_rebuilt', TRUE, + 'canonical_upsert_verified', TRUE, + 'data_rows_deleted', 0 + ), + ARRAY[ + 'asset_control_plane', + 'integrity_repair', + 'controlled_apply', + 'post_verified', + 'no_delete' + ] +FROM asset_inventory_integrity_repair_map; + +COMMENT ON INDEX asset_inventory_asset_key_key IS + 'Canonical asset identity unique index; transactionally rebuilt and verified by Gitea CD.'; diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 7f15c2c61..7875c94ea 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -114,6 +114,8 @@ async def scan_once( error_msg: str | None = None try: + await _assert_asset_key_integrity() + # 2026-04-19 v3 擴充: 多資源類型掃描 + relationship 提取 # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload), # services (k8s_resource), nodes (host), configmaps (k8s_resource) @@ -917,6 +919,61 @@ async def _finalize_discovery_run( "err": (error or "")[:2000] if error else None, "tags": ["asset_scanner", "discovery", *scope], }, + ) + + +async def _assert_asset_key_integrity() -> None: + """Fail closed before discovery when canonical asset identity is corrupt.""" + from sqlalchemy import text as _sql + + from src.db.base import get_db_context + + async with get_db_context(_PROJECT_ID) as db: + # A corrupt-but-valid unique index can make the planner incorrectly + # assume the heap is unique. Inspect the heap before trusting it. + await db.execute(_sql("SET LOCAL enable_indexscan = 'off'")) + await db.execute(_sql("SET LOCAL enable_indexonlyscan = 'off'")) + await db.execute(_sql("SET LOCAL enable_bitmapscan = 'off'")) + row = ( + await db.execute( + _sql( + """ + SELECT + ( + SELECT COUNT(*) + FROM ( + SELECT asset_key + FROM asset_inventory + GROUP BY asset_key + HAVING COUNT(*) > 1 + ) AS duplicate_groups + ) AS duplicate_group_count, + COALESCE(( + SELECT + index_row.indisvalid + AND index_row.indisready + AND index_row.indislive + AND index_row.indisunique + FROM pg_constraint AS constraint_row + JOIN pg_index AS index_row + ON index_row.indexrelid = constraint_row.conindid + WHERE constraint_row.conrelid = + 'asset_inventory'::regclass + AND constraint_row.conname = + 'asset_inventory_asset_key_key' + ), FALSE) AS unique_index_healthy + """ + ) + ) + ).one() + + duplicate_group_count = int(row[0] or 0) + unique_index_healthy = bool(row[1]) + if duplicate_group_count or not unique_index_healthy: + raise RuntimeError( + "asset_inventory_key_integrity_failed " + f"duplicate_groups={duplicate_group_count} " + f"unique_index_healthy={str(unique_index_healthy).lower()}" ) diff --git a/apps/api/tests/test_asset_inventory_database_type_migration.py b/apps/api/tests/test_asset_inventory_database_type_migration.py index fb1dd47e2..1aae92698 100644 --- a/apps/api/tests/test_asset_inventory_database_type_migration.py +++ b/apps/api/tests/test_asset_inventory_database_type_migration.py @@ -52,7 +52,7 @@ def test_cd_applies_and_independently_verifies_the_bounded_migration() -> None: workflow = CD_WORKFLOW.read_text(encoding="utf-8") step = workflow.split( "- name: Reconcile Asset Inventory Type Constraint", 1 - )[1].split("# 2026-03-31 ogt: 移除中間通知", 1)[0] + )[1].split("- name: Repair Asset Inventory Canonical Integrity", 1)[0] script = step.split("run: |", 1)[1] assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step diff --git a/apps/api/tests/test_asset_inventory_integrity_migration.py b/apps/api/tests/test_asset_inventory_integrity_migration.py new file mode 100644 index 000000000..36fb403f3 --- /dev/null +++ b/apps/api/tests/test_asset_inventory_integrity_migration.py @@ -0,0 +1,71 @@ +import textwrap +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[3] +MIGRATION = ( + ROOT + / "apps" + / "api" + / "migrations" + / "asset_inventory_canonical_integrity_2026-07-14.sql" +) +CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.yaml" + + +def test_migration_quarantines_duplicates_without_deleting_history() -> None: + sql = MIGRATION.read_text(encoding="utf-8") + upper = sql.upper() + + assert "SET lock_timeout = '10s'" in sql + assert "SET statement_timeout = '90s'" in sql + assert "SET enable_indexscan = 'off'" in sql + assert "SET enable_indexonlyscan = 'off'" in sql + assert "SET enable_bitmapscan = 'off'" in sql + assert "ROW_NUMBER() OVER" in sql + assert "repair_row_count > 100" in sql + assert "__integrity_quarantine__" in sql + assert "lifecycle_state = 'deprecated'" in sql + assert "REINDEX INDEX asset_inventory_asset_key_key" in sql + assert "HAVING COUNT(*) > 1" in sql + assert "ON CONFLICT (asset_key) DO UPDATE" in sql + assert "gitea_cd_asset_integrity_reconciler" in sql + assert '"work_item_id":"AIA-P0-006"' in sql + assert "'data_rows_deleted', 0" in sql + assert "DELETE FROM" not in upper + assert "DROP TABLE" not in upper + assert "TRUNCATE" not in upper + + +def test_cd_runs_bounded_integrity_repair_and_independent_verifier() -> None: + workflow = CD_WORKFLOW.read_text(encoding="utf-8") + step = workflow.split( + "- name: Repair Asset Inventory Canonical Integrity", 1 + )[1].split("# 2026-03-31 ogt: 移除中間通知", 1)[0] + script = step.split("run: |", 1)[1] + + assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step + assert "DATABASE_URL: ${{ secrets.DATABASE_URL }}" in step + assert "${{ secrets." not in script + assert "--network bridge" in script + assert "--network host" not in script + assert "asset_inventory_canonical_integrity_2026-07-14.sql" in script + assert "SET enable_indexscan TO off" in script + assert "SET enable_indexonlyscan TO off" in script + assert "SET enable_bitmapscan TO off" in script + assert "asset_inventory_integrity_migration_applied=" in script + assert "duplicate_group_count=" in script + assert "quarantined_row_count=" in script + assert "unique_index_healthy=" in script + assert "durable_integrity_receipt_present=" in script + assert "canonical_integrity_verified=" in script + assert "asset_inventory_canonical_integrity_verifier_failed" in script + + +def test_cd_embedded_integrity_migration_python_compiles() -> None: + workflow = CD_WORKFLOW.read_text(encoding="utf-8") + step = workflow.split( + "- name: Repair Asset Inventory Canonical Integrity", 1 + )[1] + source = step.split("python - <<'PY'", 1)[1].split(" PY", 1)[0] + + compile(textwrap.dedent(source), "", "exec") diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index 7182b859c..c3ae72430 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -142,6 +142,81 @@ def test_asset_upsert_propagates_database_failure(monkeypatch) -> None: asyncio.run(exercise()) +def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> None: + requested_projects: list[str | None] = [] + statements: list[str] = [] + + class IntegrityResult: + def one(self): + return 0, True + + class IntegrityDatabase: + async def execute(self, statement, parameters=None): + statements.append(str(statement)) + return IntegrityResult() + + class IntegrityContext: + async def __aenter__(self): + return IntegrityDatabase() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + def fake_db_context(project_id=None): + requested_projects.append(project_id) + return IntegrityContext() + + monkeypatch.setattr("src.db.base.get_db_context", fake_db_context) + + asyncio.run(asset_scanner_job._assert_asset_key_integrity()) + + assert requested_projects == ["awoooi"] + assert any("enable_indexscan" in statement for statement in statements) + assert any("enable_indexonlyscan" in statement for statement in statements) + assert any("enable_bitmapscan" in statement for statement in statements) + assert any("HAVING COUNT(*) > 1" in statement for statement in statements) + assert any( + "asset_inventory_asset_key_key" in statement for statement in statements + ) + + +def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key( + monkeypatch, +) -> None: + class IntegrityResult: + def one(self): + return 1, True + + class IntegrityDatabase: + async def execute(self, statement, parameters=None): + return IntegrityResult() + + class IntegrityContext: + async def __aenter__(self): + return IntegrityDatabase() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + monkeypatch.setattr( + "src.db.base.get_db_context", + lambda project_id=None: IntegrityContext(), + ) + + async def exercise() -> None: + try: + await asset_scanner_job._assert_asset_key_integrity() + except RuntimeError as exc: + assert str(exc) == ( + "asset_inventory_key_integrity_failed " + "duplicate_groups=1 unique_index_healthy=true" + ) + else: + raise AssertionError("duplicate canonical keys must fail closed") + + asyncio.run(exercise()) + + def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(monkeypatch) -> None: captured: dict[str, object] = {} @@ -230,6 +305,9 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) - async def fake_collect(): return [], [], (), ("prometheus_targets",) + async def fake_integrity(): + return None + async def fake_upsert(*args, **kwargs): return 0, 0 @@ -246,6 +324,11 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) - terminal.update(kwargs) monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start) + monkeypatch.setattr( + asset_scanner_job, + "_assert_asset_key_integrity", + fake_integrity, + ) monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect) monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert) monkeypatch.setattr(