From 43d2376d651be92688dd28b8f3c3069450074d4d Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 17 Jul 2026 03:52:34 +0800 Subject: [PATCH] fix(perf): recover invalid telegram receipt index --- agent99-contract-check.ps1 | 26 ++- agent99-db-bounded-executor.ps1 | 24 ++- ...m_receipt_hot_path_index_v2_2026-07-17.sql | 14 ++ apps/api/src/services/db_bounded_executor.py | 192 +++++++++++++++--- apps/api/tests/test_db_bounded_executor.py | 160 ++++++++++++++- ..._db_bounded_executor_runtime_entrypoint.py | 12 +- 6 files changed, 393 insertions(+), 35 deletions(-) create mode 100644 apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql diff --git a/agent99-contract-check.ps1 b/agent99-contract-check.ps1 index 7cb3d41c6..7504ef729 100644 --- a/agent99-contract-check.ps1 +++ b/agent99-contract-check.ps1 @@ -139,7 +139,31 @@ Add-Check "transport:dedicated_identity" ($control.Contains("sshIdentityFile") - Add-Check "transport:jump_first" ($control.Contains("preferJumpHost") -and $control.Contains("ssh_jump_fallback")) "preferred jump with bounded fallback is present" Add-Check "transport:stale_process_guard" ($control.Contains("Invoke-AgentStaleSshProcessGuard") -and $control.Contains("taskkill.exe /PID") -and $control.Contains("ssh_stale_process_guard")) "timed-out and stale SSH clients are bounded" Add-Check "transport:cross_process_serialization" ($control.Contains("Enter-AgentSshTransportLock") -and $control.Contains("[IO.FileShare]::None") -and $control.Contains("ssh_transport_lock_failed")) "cross-process SSH sessions are serialized" -Add-Check "database:fixed_bounded_executor" ($dbBoundedExecutor.Contains('$CommandId = "telegram_receipt_index_apply_v1"') -and $dbBoundedExecutor.Contains('$CanonicalAsset = "public.awooop_outbound_message"') -and $dbBoundedExecutor.Contains('$Kubectl = "/usr/local/bin/kubectl"') -and $dbBoundedExecutor.Contains('domain = "database"') -and $dbBoundedExecutor.Contains('dispatchOnly = $true') -and $dbBoundedExecutor.Contains('Invoke-Agent99ReadyApiPodReadback') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "check" $ExecutorPod') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "apply" $ExecutorPod') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "verify" $VerifierPod') -and $dbBoundedExecutor.Contains('sudo -n $Kubectl exec') -and -not $dbBoundedExecutor.Contains("Invoke-Expression") -and -not $dbBoundedExecutor.Contains("-EncodedCommand") -and -not $dbBoundedExecutor.Contains("/usr/local/sbin/awoooi-db-bounded-executor")) "Agent99 dispatches one fixed database command through the allowlisted host120 kubectl path and a second Ready API pod verifier" +$dbExecutorIdentityContract = [bool]( + $dbBoundedExecutor.Contains('$CommandId = "telegram_receipt_index_v2_apply_v1"') -and + $dbBoundedExecutor.Contains('$CanonicalAsset = "public.awooop_outbound_message"') -and + $dbBoundedExecutor.Contains('$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"') +) +$dbExecutorRecoveryContract = [bool]( + $dbBoundedExecutor.Contains('awoooi_db_bounded_executor_receipt_v2') -and + $dbBoundedExecutor.Contains('--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256') -and + $dbBoundedExecutor.Contains('"reindex_index_concurrently"') -and + $dbBoundedExecutor.Contains('Receipt.database_sqlstate') +) +$dbExecutorTransportContract = [bool]( + $dbBoundedExecutor.Contains('$Kubectl = "/usr/local/bin/kubectl"') -and + $dbBoundedExecutor.Contains('domain = "database"') -and + $dbBoundedExecutor.Contains('dispatchOnly = $true') -and + $dbBoundedExecutor.Contains('Invoke-Agent99ReadyApiPodReadback') -and + $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "check" $ExecutorPod') -and + $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "apply" $ExecutorPod') -and + $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "verify" $VerifierPod') -and + $dbBoundedExecutor.Contains('sudo -n $Kubectl exec') -and + -not $dbBoundedExecutor.Contains("Invoke-Expression") -and + -not $dbBoundedExecutor.Contains("-EncodedCommand") -and + -not $dbBoundedExecutor.Contains("/usr/local/sbin/awoooi-db-bounded-executor") +) +Add-Check "database:fixed_bounded_executor" ($dbExecutorIdentityContract -and $dbExecutorRecoveryContract -and $dbExecutorTransportContract) "Agent99 dispatches one digest-bound create/reindex command through the allowlisted host120 kubectl path, validates safe receipts, and requires a second Ready API pod verifier" Add-Check "recovery:auto_trigger" ($control.Contains("Get-AgentBootState") -and $control.Contains("Start-AgentRecoveryFromObservation") -and $control.Contains("recovery_already_queued")) "boot and host-down recovery use a single-flight queue" Add-Check "recovery:durable_boot_event" ($control.Contains("pendingRecovery") -and $control.Contains("host_reboot_recovery_pending") -and $control.Contains("Update-AgentBootRecoveryState") -and $control.Contains("recovered_late")) "boot event remains pending until verified recovery terminal" Add-Check "recovery:terminal_evidence_budget" ($taskRegistration.Contains('$recoverySettings') -and $taskRegistration.Contains('New-TimeSpan -Minutes 20')) "startup recovery outlives the ten-minute SLO long enough to write terminal evidence" diff --git a/agent99-db-bounded-executor.ps1 b/agent99-db-bounded-executor.ps1 index 9feb59171..b9b5e9051 100644 --- a/agent99-db-bounded-executor.ps1 +++ b/agent99-db-bounded-executor.ps1 @@ -14,10 +14,11 @@ param( $ErrorActionPreference = "Stop" -$CommandId = "telegram_receipt_index_apply_v1" +$CommandId = "telegram_receipt_index_v2_apply_v1" $CanonicalAsset = "public.awooop_outbound_message" -$MigrationSha256 = "50aa05dea1718475051bdf4d45b1361291ec2335512b4d5c7d0a7eb77f935f5e" -$NormalizedSqlSha256 = "7e2f18b618c5c97bf7b428e1b224d4ef117c5551d5dc66b84838b14e546bf63a" +$MigrationSha256 = "ca36ef38097de2f4a88e950e136ae175fbe1c292ad4a2c30f3a9547c08be1dc8" +$NormalizedSqlSha256 = "a10e88b55b66eaac7c09933acf33b13f3c9ada73f89c4a099e7eccaefaf1d0eb" +$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7" $JumpHost = "192.168.0.110" $ControlPlaneHost = "192.168.0.120" $RemoteUser = "wooo" @@ -65,6 +66,7 @@ function Write-Agent99DbTerminal { sourceRevision = $SourceRevision migrationSha256 = $MigrationSha256 normalizedSqlSha256 = $NormalizedSqlSha256 + normalizedReindexSqlSha256 = $NormalizedReindexSqlSha256 terminal = $Terminal errorCode = if ($ErrorCode) { $ErrorCode } else { $null } checkReceipt = $CheckReceipt @@ -111,6 +113,16 @@ function Test-Agent99DbReceipt { } else { [bool]($Receipt.write_attempted -eq $false -and $Receipt.writes_performed -eq $false) } + $operation = [string]$Receipt.operation + $operationContract = if ($ExpectedStage -ne "apply") { + [bool]($operation -eq "none") + } elseif ($operation -eq "no_write") { + [bool]($Receipt.write_attempted -eq $false -and $Receipt.writes_performed -eq $false) + } elseif ($operation -in @("create_index_concurrently", "reindex_index_concurrently")) { + [bool]($Receipt.write_attempted -eq $true) + } else { + [bool]($operation -eq "none" -and $Receipt.write_attempted -eq $false) + } $catalogContract = [bool]( ($terminal -ne "check_ready" -or $Receipt.catalog.apply_ready -eq $true) -and ($terminal -notin @("apply_verified_local", "verified_healthy") -or ( @@ -121,8 +133,9 @@ function Test-Agent99DbReceipt { return [bool]( $terminalAllowed -and $writeContract -and + $operationContract -and $catalogContract -and - [string]$Receipt.schema_version -eq "awoooi_db_bounded_executor_receipt_v1" -and + [string]$Receipt.schema_version -eq "awoooi_db_bounded_executor_receipt_v2" -and [string]$Receipt.command_id -eq $CommandId -and [string]$Receipt.canonical_asset -eq $CanonicalAsset -and [string]$Receipt.domain -eq "database" -and @@ -135,9 +148,11 @@ function Test-Agent99DbReceipt { [string]$Receipt.source_revision -eq $SourceRevision -and [string]$Receipt.migration_sha256 -eq $MigrationSha256 -and [string]$Receipt.normalized_sql_sha256 -eq $NormalizedSqlSha256 -and + [string]$Receipt.normalized_reindex_sql_sha256 -eq $NormalizedReindexSqlSha256 -and [string]$Receipt.pod_name -eq $ExpectedPod -and [string]$Receipt.pod_name -match $ApiPodPattern -and [string]$Receipt.process_identity_sha256 -match "^[0-9a-f]{64}$" -and + ($null -eq $Receipt.database_sqlstate -or [string]$Receipt.database_sqlstate -match "^[0-9A-Z]{5}$") -and $Receipt.raw_sql_emitted -eq $false -and $Receipt.secret_value_read -eq $false -and $Receipt.arbitrary_command_allowed -eq $false -and @@ -229,6 +244,7 @@ function Invoke-Agent99FixedDbStage { "python -B -m src.services.db_bounded_executor " + "--mode $Stage --command-id $CommandId --canonical-asset $CanonicalAsset " + "--migration-sha256 $MigrationSha256 --normalized-sql-sha256 $NormalizedSqlSha256 " + + "--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256 " + "--trace-id $TraceId --run-id $RunId --work-item-id $WorkItemId --source-revision $SourceRevision" ) $arguments = @( diff --git a/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql new file mode 100644 index 000000000..c67e89afc --- /dev/null +++ b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql @@ -0,0 +1,14 @@ +-- awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql +-- Add the corrected descending hot-path index under a new name so recovery is +-- additive and never drops or reindexes the drifted invalid predecessor. +-- +-- CONCURRENTLY must run outside a transaction block. + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest_v2 + ON awooop_outbound_message (project_id, queued_at DESC) + INCLUDE (send_status) + WHERE channel_type = 'telegram' + AND ( + source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' + OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' + ); diff --git a/apps/api/src/services/db_bounded_executor.py b/apps/api/src/services/db_bounded_executor.py index 1fc9ed8f0..32ce05bbd 100644 --- a/apps/api/src/services/db_bounded_executor.py +++ b/apps/api/src/services/db_bounded_executor.py @@ -24,20 +24,20 @@ from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine from src.db.base import get_engine -SCHEMA_VERSION = "awoooi_db_bounded_executor_receipt_v1" -COMMAND_ID = "telegram_receipt_index_apply_v1" +SCHEMA_VERSION = "awoooi_db_bounded_executor_receipt_v2" +COMMAND_ID = "telegram_receipt_index_v2_apply_v1" CANONICAL_ASSET = "public.awooop_outbound_message" DOMAIN = "database" EXECUTOR = "db_bounded_executor" VERIFIER = "db_independent_verifier" -INDEX_NAME = "idx_awooop_outbound_msg_telegram_receipt_latest" +INDEX_NAME = "idx_awooop_outbound_msg_telegram_receipt_latest_v2" MIGRATION_FILE = ( - "awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql" + "awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql" ) -MIGRATION_SHA256 = "50aa05dea1718475051bdf4d45b1361291ec2335512b4d5c7d0a7eb77f935f5e" +MIGRATION_SHA256 = "ca36ef38097de2f4a88e950e136ae175fbe1c292ad4a2c30f3a9547c08be1dc8" CREATE_INDEX_SQL = """\ -CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest_v2 ON awooop_outbound_message (project_id, queued_at DESC) INCLUDE (send_status) WHERE channel_type = 'telegram' @@ -48,7 +48,15 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt """ NORMALIZED_SQL = " ".join(CREATE_INDEX_SQL.split()) NORMALIZED_SQL_SHA256 = ( - "7e2f18b618c5c97bf7b428e1b224d4ef117c5551d5dc66b84838b14e546bf63a" + "a10e88b55b66eaac7c09933acf33b13f3c9ada73f89c4a099e7eccaefaf1d0eb" +) +REINDEX_INDEX_SQL = ( + "REINDEX INDEX CONCURRENTLY " + "public.idx_awooop_outbound_msg_telegram_receipt_latest_v2;" +) +NORMALIZED_REINDEX_SQL = " ".join(REINDEX_INDEX_SQL.split()) +NORMALIZED_REINDEX_SQL_SHA256 = ( + "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7" ) _SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$") @@ -56,17 +64,17 @@ _SOURCE_REVISION = re.compile(r"^[0-9a-f]{40}$") _API_POD = re.compile(r"^awoooi-api-[a-z0-9][a-z0-9-]{0,126}$") _LOCK_SQL = """ SELECT pg_try_advisory_lock( - hashtextextended('awoooi/db/telegram_receipt_index_apply_v1', 0) + hashtextextended('awoooi/db/telegram_receipt_index_v2_apply_v1', 0) ) AS acquired """ _UNLOCK_SQL = """ SELECT pg_advisory_unlock( - hashtextextended('awoooi/db/telegram_receipt_index_apply_v1', 0) + hashtextextended('awoooi/db/telegram_receipt_index_v2_apply_v1', 0) ) AS released """ _SESSION_GUARDS_SQL = ( "SET search_path TO pg_catalog, public", - "SET lock_timeout TO '5s'", + "SET lock_timeout TO '30s'", "SET statement_timeout TO '10min'", ) _CATALOG_SQL = f""" @@ -94,6 +102,18 @@ WITH target AS ( pg_get_indexdef(i.indexrelid, 1, true) AS key_1, pg_get_indexdef(i.indexrelid, 2, true) AS key_2, pg_get_indexdef(i.indexrelid, 3, true) AS include_1, + pg_index_column_has_property(i.indexrelid, 1, 'asc') + AS key_1_ascending, + pg_index_column_has_property(i.indexrelid, 1, 'nulls_last') + AS key_1_nulls_last, + pg_index_column_has_property(i.indexrelid, 2, 'asc') + AS key_2_ascending, + pg_index_column_has_property(i.indexrelid, 2, 'desc') + AS key_2_descending, + pg_index_column_has_property(i.indexrelid, 2, 'nulls_first') + AS key_2_nulls_first, + pg_index_column_has_property(i.indexrelid, 2, 'nulls_last') + AS key_2_nulls_last, pg_get_expr(i.indpred, i.indrelid) AS predicate FROM target t JOIN pg_index i ON i.indrelid = t.table_oid @@ -144,6 +164,18 @@ SELECT COALESCE((SELECT key_1 FROM index_catalog), '') AS key_1, COALESCE((SELECT key_2 FROM index_catalog), '') AS key_2, COALESCE((SELECT include_1 FROM index_catalog), '') AS include_1, + COALESCE((SELECT key_1_ascending FROM index_catalog), false) + AS key_1_ascending, + COALESCE((SELECT key_1_nulls_last FROM index_catalog), false) + AS key_1_nulls_last, + COALESCE((SELECT key_2_ascending FROM index_catalog), false) + AS key_2_ascending, + COALESCE((SELECT key_2_descending FROM index_catalog), false) + AS key_2_descending, + COALESCE((SELECT key_2_nulls_first FROM index_catalog), false) + AS key_2_nulls_first, + COALESCE((SELECT key_2_nulls_last FROM index_catalog), false) + AS key_2_nulls_last, COALESCE((SELECT predicate FROM index_catalog), '') AS predicate FROM target t """ @@ -177,6 +209,7 @@ def validate_execution_contract( canonical_asset: str, migration_sha256: str, normalized_sql_sha256: str, + normalized_reindex_sql_sha256: str, trace_id: str, run_id: str, work_item_id: str, @@ -194,6 +227,10 @@ def validate_execution_contract( normalized_sql_sha256, NORMALIZED_SQL_SHA256, ), + "normalized_reindex_sql_sha256": ( + normalized_reindex_sql_sha256, + NORMALIZED_REINDEX_SQL_SHA256, + ), } for field, (actual, expected) in exact.items(): if actual != expected: @@ -215,6 +252,11 @@ def validate_execution_contract( raise BoundedExecutorContractError("runtime_not_allowlisted_api_pod") if hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() != NORMALIZED_SQL_SHA256: raise BoundedExecutorContractError("embedded_sql_digest_mismatch") + if ( + hashlib.sha256(NORMALIZED_REINDEX_SQL.encode()).hexdigest() + != NORMALIZED_REINDEX_SQL_SHA256 + ): + raise BoundedExecutorContractError("embedded_reindex_sql_digest_mismatch") def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]: @@ -226,21 +268,56 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]: _normalize_catalog_expression(str(row.get("predicate") or "")) == _EXPECTED_PREDICATE ) - index_shape_exact = bool( + index_non_unique = row.get("index_non_unique") is True + key_count_exact = row.get("key_count_exact") is True + total_attribute_count_exact = row.get("total_attribute_count_exact") is True + access_method_exact = row.get("access_method_exact") is True + key_1_expression_exact = ( + _normalize_catalog_expression(str(row.get("key_1") or "")) == "project_id" + ) + key_1_order_exact = bool( + row.get("key_1_ascending") is True and row.get("key_1_nulls_last") is True + ) + key_1_exact = key_1_expression_exact and key_1_order_exact + key_2_expression_exact = ( + _normalize_catalog_expression(str(row.get("key_2") or "")) == "queued_at" + ) + key_2_ascending = row.get("key_2_ascending") is True + key_2_descending = row.get("key_2_descending") is True + key_2_nulls_first = row.get("key_2_nulls_first") is True + key_2_nulls_last = row.get("key_2_nulls_last") is True + if not key_2_expression_exact: + key_2_variant = "unknown_expression" + elif key_2_descending and key_2_nulls_first: + key_2_variant = "descending_nulls_first" + elif key_2_descending and key_2_nulls_last: + key_2_variant = "descending_nulls_last" + elif key_2_ascending and key_2_nulls_first: + key_2_variant = "ascending_nulls_first" + elif key_2_ascending and key_2_nulls_last: + key_2_variant = "ascending_nulls_last" + else: + key_2_variant = "unknown_order" + key_2_order_exact = key_2_descending and key_2_nulls_first + key_2_exact = key_2_expression_exact and key_2_order_exact + include_1_exact = ( + _normalize_catalog_expression(str(row.get("include_1") or "")) == "send_status" + ) + index_definition_exact_except_validity = bool( index_exists and row.get("index_ready") is True - and row.get("index_valid") is True - and row.get("index_non_unique") is True - and row.get("key_count_exact") is True - and row.get("total_attribute_count_exact") is True - and row.get("access_method_exact") is True - and _normalize_catalog_expression(str(row.get("key_1") or "")) == "project_id" - and _normalize_catalog_expression(str(row.get("key_2") or "")) - == "queued_atdesc" - and _normalize_catalog_expression(str(row.get("include_1") or "")) - == "send_status" + and index_non_unique + and key_count_exact + and total_attribute_count_exact + and access_method_exact + and key_1_exact + and key_2_exact + and include_1_exact and predicate_exact ) + index_shape_exact = bool( + index_definition_exact_except_validity and row.get("index_valid") is True + ) base_ready = bool( row.get("table_exists") is True and row.get("columns_complete") is True @@ -250,11 +327,22 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]: and row.get("schema_usage") is True and row.get("table_owner_privilege_inherited") is True ) - drift = bool(index_count > 1 or (index_exists and not index_shape_exact)) + reindex_ready = bool( + base_ready + and index_definition_exact_except_validity + and row.get("index_valid") is not True + ) + drift = bool( + index_count > 1 or (index_exists and not index_definition_exact_except_validity) + ) apply_ready = bool( base_ready and not drift - and (index_shape_exact or row.get("schema_create") is True) + and ( + index_shape_exact + or reindex_ready + or (not index_exists and row.get("schema_create") is True) + ) ) return { "table_exists": row.get("table_exists") is True, @@ -272,8 +360,28 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]: "index_exists": index_exists, "index_ready": row.get("index_ready") is True, "index_valid": row.get("index_valid") is True, + "index_non_unique": index_non_unique, + "key_count_exact": key_count_exact, + "total_attribute_count_exact": total_attribute_count_exact, + "access_method_exact": access_method_exact, + "key_1_expression_exact": key_1_expression_exact, + "key_1_order_exact": key_1_order_exact, + "key_1_exact": key_1_exact, + "key_2_expression_exact": key_2_expression_exact, + "key_2_order_exact": key_2_order_exact, + "key_2_ascending": key_2_ascending, + "key_2_descending": key_2_descending, + "key_2_nulls_first": key_2_nulls_first, + "key_2_nulls_last": key_2_nulls_last, + "key_2_exact": key_2_exact, + "key_2_variant": key_2_variant, + "include_1_exact": include_1_exact, + "index_definition_exact_except_validity": ( + index_definition_exact_except_validity + ), "index_shape_exact": index_shape_exact, "predicate_exact": predicate_exact, + "reindex_ready": reindex_ready, "catalog_drift": drift, "apply_ready": apply_ready, } @@ -297,7 +405,9 @@ def _receipt( catalog: Mapping[str, Any], writes_performed: bool | None, write_attempted: bool = False, + operation: str = "none", error_code: str | None = None, + database_sqlstate: str | None = None, ) -> dict[str, Any]: role = { "check": "precheck_executor", @@ -325,19 +435,31 @@ def _receipt( "migration_file": MIGRATION_FILE, "migration_sha256": MIGRATION_SHA256, "normalized_sql_sha256": NORMALIZED_SQL_SHA256, + "normalized_reindex_sql_sha256": NORMALIZED_REINDEX_SQL_SHA256, "terminal": terminal, "catalog": dict(catalog), "writes_performed": writes_performed, "write_attempted": write_attempted, + "operation": operation, "raw_sql_emitted": False, "secret_value_read": False, "arbitrary_command_allowed": False, "cross_domain_fallback_allowed": False, - "rollback": "no_write_or_create_index_concurrently_only", + "rollback": "no_write_or_fixed_concurrent_index_operation_only", "error_code": error_code, + "database_sqlstate": database_sqlstate, } +def _safe_database_sqlstate(exc: SQLAlchemyError) -> str | None: + """Return only a five-character SQLSTATE, never backend error text.""" + + original = getattr(exc, "orig", None) + value = getattr(original, "sqlstate", None) or getattr(original, "pgcode", None) + candidate = str(value or "").upper() + return candidate if re.fullmatch(r"[0-9A-Z]{5}", candidate) else None + + async def execute_bounded_db_command( *, mode: str, @@ -345,6 +467,7 @@ async def execute_bounded_db_command( canonical_asset: str, migration_sha256: str, normalized_sql_sha256: str, + normalized_reindex_sql_sha256: str, trace_id: str, run_id: str, work_item_id: str, @@ -361,6 +484,7 @@ async def execute_bounded_db_command( canonical_asset=canonical_asset, migration_sha256=migration_sha256, normalized_sql_sha256=normalized_sql_sha256, + normalized_reindex_sql_sha256=normalized_reindex_sql_sha256, trace_id=trace_id, run_id=run_id, work_item_id=work_item_id, @@ -371,6 +495,7 @@ async def execute_bounded_db_command( db_engine = engine writes_performed = False write_attempted = False + operation = "none" lock_acquired = False try: if db_engine is None: @@ -453,7 +578,15 @@ async def execute_bounded_db_command( writes_performed=False, error_code="locked_precheck_failed", ) - if not locked["index_shape_exact"]: + if locked["index_shape_exact"]: + operation = "no_write" + elif locked["reindex_ready"]: + operation = "reindex_index_concurrently" + write_attempted = True + await connection.execute(text(REINDEX_INDEX_SQL)) + writes_performed = True + else: + operation = "create_index_concurrently" write_attempted = True await connection.execute(text(CREATE_INDEX_SQL)) writes_performed = True @@ -472,6 +605,7 @@ async def execute_bounded_db_command( catalog=after, writes_performed=writes_performed, write_attempted=write_attempted, + operation=operation, error_code=( None if verified else "local_catalog_verification_failed" ), @@ -493,9 +627,10 @@ async def execute_bounded_db_command( catalog={}, writes_performed=None if write_attempted else writes_performed, write_attempted=write_attempted, + operation=operation, error_code="database_timeout", ) - except SQLAlchemyError: + except SQLAlchemyError as exc: return _receipt( mode=mode, trace_id=trace_id, @@ -507,7 +642,9 @@ async def execute_bounded_db_command( catalog={}, writes_performed=None if write_attempted else writes_performed, write_attempted=write_attempted, + operation=operation, error_code="database_operation_failed", + database_sqlstate=_safe_database_sqlstate(exc), ) except Exception: return _receipt( @@ -521,6 +658,7 @@ async def execute_bounded_db_command( catalog={}, writes_performed=None if write_attempted else writes_performed, write_attempted=write_attempted, + operation=operation, error_code="executor_internal_error", ) @@ -532,6 +670,7 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument("--canonical-asset", required=True) parser.add_argument("--migration-sha256", required=True) parser.add_argument("--normalized-sql-sha256", required=True) + parser.add_argument("--normalized-reindex-sql-sha256", required=True) parser.add_argument("--trace-id", required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--work-item-id", required=True) @@ -550,6 +689,7 @@ async def _async_main() -> int: canonical_asset=args.canonical_asset, migration_sha256=args.migration_sha256, normalized_sql_sha256=args.normalized_sql_sha256, + normalized_reindex_sql_sha256=args.normalized_reindex_sql_sha256, trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, diff --git a/apps/api/tests/test_db_bounded_executor.py b/apps/api/tests/test_db_bounded_executor.py index 71eea5519..286ed14dd 100644 --- a/apps/api/tests/test_db_bounded_executor.py +++ b/apps/api/tests/test_db_bounded_executor.py @@ -20,8 +20,11 @@ from src.services.db_bounded_executor import ( CREATE_INDEX_SQL, MIGRATION_FILE, MIGRATION_SHA256, + NORMALIZED_REINDEX_SQL, + NORMALIZED_REINDEX_SQL_SHA256, NORMALIZED_SQL, NORMALIZED_SQL_SHA256, + REINDEX_INDEX_SQL, BoundedExecutorContractError, evaluate_catalog_snapshot, execute_bounded_db_command, @@ -63,6 +66,12 @@ def _absent_catalog() -> dict: "key_count_exact": False, "total_attribute_count_exact": False, "access_method_exact": False, + "key_1_ascending": False, + "key_1_nulls_last": False, + "key_2_ascending": False, + "key_2_descending": False, + "key_2_nulls_first": False, + "key_2_nulls_last": False, "key_1": "", "key_2": "", "include_1": "", @@ -80,8 +89,12 @@ def _exact_catalog() -> dict: "key_count_exact": True, "total_attribute_count_exact": True, "access_method_exact": True, + "key_1_ascending": True, + "key_1_nulls_last": True, + "key_2_descending": True, + "key_2_nulls_first": True, "key_1": "project_id", - "key_2": "queued_at DESC", + "key_2": "queued_at", "include_1": "send_status", "predicate": """ ((channel_type = 'telegram'::text) AND @@ -93,6 +106,10 @@ def _exact_catalog() -> dict: } +def _invalid_exact_catalog() -> dict: + return {**_exact_catalog(), "index_valid": False} + + def _contract(mode: str = "check") -> dict: return { "mode": mode, @@ -100,6 +117,7 @@ def _contract(mode: str = "check") -> dict: "canonical_asset": CANONICAL_ASSET, "migration_sha256": MIGRATION_SHA256, "normalized_sql_sha256": NORMALIZED_SQL_SHA256, + "normalized_reindex_sql_sha256": NORMALIZED_REINDEX_SQL_SHA256, **IDENTITY, "source_revision": SOURCE_REVISION, "runtime_source_revision": SOURCE_REVISION, @@ -113,6 +131,10 @@ def test_fixed_migration_and_embedded_sql_digests_match() -> None: assert hashlib.sha256(migration).hexdigest() == MIGRATION_SHA256 assert _normalized_migration_sql() == NORMALIZED_SQL assert hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() == NORMALIZED_SQL_SHA256 + assert ( + hashlib.sha256(NORMALIZED_REINDEX_SQL.encode()).hexdigest() + == NORMALIZED_REINDEX_SQL_SHA256 + ) def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None: @@ -133,6 +155,25 @@ def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None: assert forbidden not in normalized +def test_embedded_reindex_is_one_fixed_transactionless_recovery_only() -> None: + normalized = REINDEX_INDEX_SQL.upper() + + assert normalized.count("REINDEX INDEX CONCURRENTLY") == 1 + assert normalized.count(";") == 1 + assert "PUBLIC.IDX_AWOOOP_OUTBOUND_MSG_TELEGRAM_RECEIPT_LATEST" in normalized + for forbidden in ( + "DROP ", + "TRUNCATE ", + "DELETE ", + "UPDATE ", + "INSERT ", + "ALTER ", + "BEGIN", + "COMMIT", + ): + assert forbidden not in normalized + + def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None: source = (ROOT / "src/services/db_bounded_executor.py").read_text() @@ -152,6 +193,11 @@ def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None "f" * 64, "normalized_sql_sha256_contract_mismatch", ), + ( + "normalized_reindex_sql_sha256", + "e" * 64, + "normalized_reindex_sql_sha256_contract_mismatch", + ), ("mode", "rollback", "mode_not_allowlisted"), ("pod_name", "awoooi-worker-1", "runtime_not_allowlisted_api_pod"), ( @@ -176,7 +222,24 @@ def test_contract_rejects_any_scope_or_runtime_drift( def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None: absent = evaluate_catalog_snapshot(_absent_catalog()) exact = evaluate_catalog_snapshot(_exact_catalog()) - drift_row = {**_exact_catalog(), "key_2": "queued_at"} + invalid_exact = evaluate_catalog_snapshot(_invalid_exact_catalog()) + wrong_null_order = evaluate_catalog_snapshot( + { + **_invalid_exact_catalog(), + "key_2_nulls_first": False, + "key_2_nulls_last": True, + } + ) + ascending_order = evaluate_catalog_snapshot( + { + **_invalid_exact_catalog(), + "key_2_ascending": True, + "key_2_descending": False, + "key_2_nulls_first": False, + "key_2_nulls_last": True, + } + ) + drift_row = {**_exact_catalog(), "key_2": "created_at"} drift = evaluate_catalog_snapshot(drift_row) non_owner = evaluate_catalog_snapshot( {**_absent_catalog(), "table_owner_privilege_inherited": False} @@ -192,6 +255,17 @@ def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None: assert absent["index_exists"] is False assert exact["index_shape_exact"] is True assert exact["apply_ready"] is True + assert invalid_exact["index_shape_exact"] is False + assert invalid_exact["index_definition_exact_except_validity"] is True + assert invalid_exact["reindex_ready"] is True + assert invalid_exact["catalog_drift"] is False + assert invalid_exact["apply_ready"] is True + assert invalid_exact["key_2_variant"] == "descending_nulls_first" + assert wrong_null_order["key_2_exact"] is False + assert wrong_null_order["key_2_variant"] == "descending_nulls_last" + assert ascending_order["key_2_exact"] is False + assert ascending_order["key_2_variant"] == "ascending_nulls_last" + assert wrong_null_order["catalog_drift"] is True assert drift["catalog_drift"] is True assert drift["apply_ready"] is False assert non_owner["apply_ready"] is False @@ -239,6 +313,8 @@ class _Connection: return _Result(scalar=True) if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql: return _Result() + if "REINDEX INDEX CONCURRENTLY" in sql: + return _Result() raise AssertionError(f"unexpected SQL: {sql}") @@ -284,6 +360,7 @@ async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() - assert receipt["terminal"] == "apply_verified_local" assert receipt["writes_performed"] is True + assert receipt["operation"] == "create_index_concurrently" assert receipt["domain"] == "database" assert receipt["cross_domain_fallback_allowed"] is False assert receipt["secret_value_read"] is False @@ -291,7 +368,7 @@ async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() - assert engine.connect_count == 1 assert connection.statements[:3] == [ "SET search_path TO pg_catalog, public", - "SET lock_timeout TO '5s'", + "SET lock_timeout TO '30s'", "SET statement_timeout TO '10min'", ] assert sum("CREATE INDEX CONCURRENTLY" in sql for sql in connection.statements) == 1 @@ -314,9 +391,55 @@ async def test_apply_is_idempotent_when_exact_index_already_exists() -> None: assert receipt["terminal"] == "apply_verified_local" assert receipt["writes_performed"] is False + assert receipt["operation"] == "no_write" assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements) +@pytest.mark.asyncio +async def test_apply_recovers_only_exact_invalid_index_with_concurrent_reindex() -> ( + None +): + connection = _Connection( + [_invalid_exact_catalog(), _invalid_exact_catalog(), _exact_catalog()] + ) + engine = _Engine(connection) + + receipt = await execute_bounded_db_command( + **_contract("apply"), + engine=engine, + ) + + assert receipt["terminal"] == "apply_verified_local" + assert receipt["writes_performed"] is True + assert receipt["operation"] == "reindex_index_concurrently" + assert connection.statements.count(REINDEX_INDEX_SQL) == 1 + assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements) + + +@pytest.mark.asyncio +async def test_apply_blocks_invalid_index_with_wrong_definition_without_write() -> None: + wrong = { + **_invalid_exact_catalog(), + "key_2_ascending": True, + "key_2_descending": False, + "key_2_nulls_first": False, + "key_2_nulls_last": True, + } + connection = _Connection([wrong]) + engine = _Engine(connection) + + receipt = await execute_bounded_db_command( + **_contract("apply"), + engine=engine, + ) + + assert receipt["terminal"] == "blocked" + assert receipt["writes_performed"] is False + assert receipt["operation"] == "none" + assert all("REINDEX INDEX" not in sql for sql in connection.statements) + assert all("CREATE INDEX" not in sql for sql in connection.statements) + + @pytest.mark.asyncio async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None: connection = _Connection([_exact_catalog()]) @@ -329,6 +452,7 @@ async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None: assert receipt["terminal"] == "verified_healthy" assert receipt["executor_role"] == "independent_catalog_verifier" + assert receipt["operation"] == "none" assert receipt["writes_performed"] is False assert all("advisory_lock" not in sql for sql in connection.statements) @@ -351,3 +475,33 @@ async def test_failed_create_reports_unknown_write_state_without_backend_detail( assert receipt["error_code"] == "database_operation_failed" assert "sensitive backend detail" not in json.dumps(receipt) assert any("pg_advisory_unlock" in sql for sql in connection.statements) + + +class _SafeSqlstateFailureConnection(_Connection): + async def execute(self, statement): + sql = str(statement) + if "REINDEX INDEX CONCURRENTLY" in sql: + self.statements.append(sql) + error = SQLAlchemyError("backend text must not escape") + error.orig = type("DriverError", (), {"sqlstate": "55P03"})() + raise error + return await super().execute(statement) + + +@pytest.mark.asyncio +async def test_failed_reindex_returns_safe_sqlstate_without_backend_detail() -> None: + connection = _SafeSqlstateFailureConnection( + [_invalid_exact_catalog(), _invalid_exact_catalog()] + ) + engine = _Engine(connection) + + receipt = await execute_bounded_db_command( + **_contract("apply"), + engine=engine, + ) + + assert receipt["terminal"] == "failed" + assert receipt["operation"] == "reindex_index_concurrently" + assert receipt["writes_performed"] is None + assert receipt["database_sqlstate"] == "55P03" + assert "backend text" not in json.dumps(receipt) diff --git a/scripts/ops/tests/test_db_bounded_executor_runtime_entrypoint.py b/scripts/ops/tests/test_db_bounded_executor_runtime_entrypoint.py index eaf160373..9c3b93fd3 100644 --- a/scripts/ops/tests/test_db_bounded_executor_runtime_entrypoint.py +++ b/scripts/ops/tests/test_db_bounded_executor_runtime_entrypoint.py @@ -30,8 +30,9 @@ def test_agent99_is_dispatch_only_and_apply_requires_independent_verifier() -> N source = AGENT99_ENTRYPOINT.read_text() for fixed in ( - '$CommandId = "telegram_receipt_index_apply_v1"', + '$CommandId = "telegram_receipt_index_v2_apply_v1"', '$CanonicalAsset = "public.awooop_outbound_message"', + '$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"', '$JumpHost = "192.168.0.110"', '$ControlPlaneHost = "192.168.0.120"', '$Kubectl = "/usr/local/bin/kubectl"', @@ -50,6 +51,11 @@ def test_agent99_is_dispatch_only_and_apply_requires_independent_verifier() -> N assert 'Invoke-Agent99FixedDbStage "check" $ExecutorPod' in source assert 'Invoke-Agent99FixedDbStage "apply" $ExecutorPod' in source assert 'Invoke-Agent99FixedDbStage "verify" $VerifierPod' in source + assert "--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256" in source + assert ( + '$operation -in @("create_index_concurrently", "reindex_index_concurrently")' + in source + ) apply_flow = source[source.index("# Apply is the only write path") :] assert apply_flow.index('Invoke-Agent99FixedDbStage "check"') < apply_flow.index( 'Invoke-Agent99FixedDbStage "apply"' @@ -112,6 +118,10 @@ def test_agent99_atomic_bundle_owns_the_dispatch_entrypoint() -> None: for source in (deploy, bootstrap, contract, transport, receiver): assert '"agent99-db-bounded-executor.ps1"' in source assert 'Add-Check "database:fixed_bounded_executor"' in contract + assert "$dbExecutorIdentityContract" in contract + assert "$dbExecutorRecoveryContract" in contract + assert "awoooi_db_bounded_executor_receipt_v2" in contract + assert "Receipt.database_sqlstate" in contract assert 'expectedRuntimeFileCount": 16' in transport assert "expectedRuntimeFileCount=16" in transport assert "expectedRuntimeFileCount -ne 16" in receiver