fix(cd): skip converged MCP receipt DDL

.gitea/workflows/cd.yaml

@@ -2351,44 +2351,68 @@ jobs:
               return value.replace("postgresql+asyncpg://", "postgresql://", 1)
 
 
+          async def read_schema_state(connection):
+              rows = await connection.fetch(
+                  """
+                  SELECT relname, relrowsecurity, relforcerowsecurity
+                  FROM pg_class
+                  JOIN pg_namespace ON pg_namespace.oid = pg_class.relnamespace
+                  WHERE pg_namespace.nspname = 'public'
+                    AND relname = ANY($1::text[])
+                  """,
+                  sorted(REQUIRED_TABLES),
+              )
+              table_state = {
+                  str(row["relname"]): (
+                      bool(row["relrowsecurity"]),
+                      bool(row["relforcerowsecurity"]),
+                  )
+                  for row in rows
+              }
+              policy_count = await connection.fetchval(
+                  """
+                  SELECT COUNT(*)::int
+                  FROM pg_policies
+                  WHERE tablename = ANY($1::text[])
+                  """,
+                  sorted(REQUIRED_TABLES),
+              )
+              schema_verified = (
+                  set(table_state) == REQUIRED_TABLES
+                  and all(state == (True, True) for state in table_state.values())
+                  and int(policy_count or 0) == 2
+              )
+              return schema_verified, int(policy_count or 0)
+
+
           async def apply_and_verify_schema() -> None:
               connection = await asyncpg.connect(
                   normalize_url(os.environ["MIGRATION_DATABASE_URL"]),
                   timeout=10,
               )
               try:
-                  await connection.execute(MIGRATION)
+                  # Verified schema is already converged; avoid needless
-                  rows = await connection.fetch(
+                  # ACCESS EXCLUSIVE policy DDL during a production pg_dump.
-                      """
+                  schema_verified, policy_count = await read_schema_state(connection)
-                      SELECT relname, relrowsecurity, relforcerowsecurity
+                  migration_applied = False
-                      FROM pg_class
+                  if not schema_verified:
-                      WHERE relname = ANY($1::text[])
+                      await connection.execute("SET lock_timeout = '10s'")
-                      """,
+                      await connection.execute("SET statement_timeout = '120s'")
-                      sorted(REQUIRED_TABLES),
+                      await connection.execute(MIGRATION)
-                  )
+                      migration_applied = True
-                  table_state = {
+                      schema_verified, policy_count = await read_schema_state(
-                      str(row["relname"]): (
+                          connection
-                          bool(row["relrowsecurity"]),
-                          bool(row["relforcerowsecurity"]),
                       )
-                      for row in rows
+                  print(
-                  }
+                      "mcp_version_lifecycle_migration_applied="
-                  policy_count = await connection.fetchval(
+                      f"{str(migration_applied).lower()}"
-                      """
-                      SELECT COUNT(*)::int
-                      FROM pg_policies
-                      WHERE tablename = ANY($1::text[])
-                      """,
-                      sorted(REQUIRED_TABLES),
                   )
-                  schema_verified = (
+                  print(
-                      set(table_state) == REQUIRED_TABLES
+                      "mcp_version_lifecycle_migration_skipped_existing_verified="
-                      and all(state == (True, True) for state in table_state.values())
+                      f"{str(not migration_applied and schema_verified).lower()}"
-                      and int(policy_count or 0) == 2
                   )
-                  print("mcp_version_lifecycle_migration_applied=true")
                   print(f"mcp_version_lifecycle_schema_verified={str(schema_verified).lower()}")
-                  print(f"mcp_version_lifecycle_rls_policy_count={int(policy_count or 0)}")
+                  print(f"mcp_version_lifecycle_rls_policy_count={policy_count}")
                   if not schema_verified:
                       raise RuntimeError("mcp_version_lifecycle_schema_verifier_failed")
               finally:

apps/api/tests/test_mcp_version_lifecycle_service.py

@@ -261,3 +261,6 @@ def test_migration_is_additive_rls_and_does_not_store_raw_content() -> None:
     assert "Apply MCP Version Lifecycle Receipt Schema" in workflow
     assert "mcp_version_lifecycle_schema_verified" in workflow
     assert "mcp_version_lifecycle_runtime_role_read_verified" in workflow
+    assert "mcp_version_lifecycle_migration_skipped_existing_verified" in workflow
+    assert "SET lock_timeout = '10s'" in workflow
+    assert "SET statement_timeout = '120s'" in workflow