Recover invalid concurrent indexes on migration retry

An interrupted CREATE INDEX CONCURRENTLY leaves a same-name INVALID index
behind; IF NOT EXISTS then sees the relation and skips creation, so a
retried migration completes with an index that serves no queries. For
uq_experiments_shadow_of_live that silently breaks the shadow creator's
INSERT .. ON CONFLICT arbiter inference, which only considers valid
indexes. Before each concurrent CREATE, drop a same-name index whose
pg_index.indisvalid is false so the CREATE rebuilds it.

Drilled against Postgres 16: corrupting both indexes and rerunning the
unfixed migrations left indisvalid=f and the ON CONFLICT insert failing
with "no unique or exclusion constraint matching"; with this change the
rerun rebuilds both (indisvalid=t) and the conflict path dedupes again.

Claude-Session: https://claude.ai/code/session_01ACF6SUXdbLFarpj3qwF1Ki

Author: kyle-compute

oddish/alembic/versions/shadowexp01_add_experiment_shadow_of.py

@@ -29,6 +29,34 @@
 depends_on: Union[str, Sequence[str], None] = None
 
 
+def _recover_invalid_index(index_name: str) -> None:
+    """An interrupted ``CREATE INDEX CONCURRENTLY`` leaves a same-name INVALID
+    index behind (``pg_index.indisvalid = false``). ``IF NOT EXISTS`` sees that
+    relation and skips the CREATE, so a retried migration would complete with
+    an index that enforces nothing -- for this unique index that silently
+    breaks the shadow creator's ``INSERT .. ON CONFLICT`` arbiter inference,
+    which only considers valid indexes. Drop the invalid leftover
+    (concurrently, we are in the autocommit block) so the CREATE below
+    rebuilds it."""
+    result = op.get_bind().execute(
+        sa.text(
+            """
+            SELECT 1
+            FROM pg_index i
+            JOIN pg_class c ON c.oid = i.indexrelid
+            JOIN pg_namespace n ON n.oid = c.relnamespace
+            WHERE c.relname = :index_name
+              AND n.nspname = current_schema()
+              AND NOT i.indisvalid
+            """
+        ),
+        {"index_name": index_name},
+    )
+    invalid = result.first()
+    if invalid is not None:
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {index_name}")
+
+
 def upgrade() -> None:
     # Bound the ALTER's brief ACCESS EXCLUSIVE lock: without a timeout it
     # queues behind any long-running query, and all traffic queues behind it.
@@ -41,6 +69,7 @@ def upgrade() -> None:
     # Index name matches the model's ``__table_args__`` declaration so the
     # create_all() index and this one are the same object.
     with op.get_context().autocommit_block():
+        _recover_invalid_index("uq_experiments_shadow_of_live")
         op.execute(
             "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "
             "uq_experiments_shadow_of_live ON experiments (shadow_of) "

oddish/alembic/versions/trialkind01_add_trial_kind.py

@@ -30,6 +30,31 @@
 depends_on: Union[str, Sequence[str], None] = None
 
 
+def _recover_invalid_index(index_name: str) -> None:
+    """An interrupted ``CREATE INDEX CONCURRENTLY`` leaves a same-name INVALID
+    index behind (``pg_index.indisvalid = false``). ``IF NOT EXISTS`` sees that
+    relation and skips the CREATE, so a retried migration would complete with
+    an index that serves no queries. Drop the invalid leftover (concurrently,
+    we are in the autocommit block) so the CREATE below rebuilds it."""
+    result = op.get_bind().execute(
+        sa.text(
+            """
+            SELECT 1
+            FROM pg_index i
+            JOIN pg_class c ON c.oid = i.indexrelid
+            JOIN pg_namespace n ON n.oid = c.relnamespace
+            WHERE c.relname = :index_name
+              AND n.nspname = current_schema()
+              AND NOT i.indisvalid
+            """
+        ),
+        {"index_name": index_name},
+    )
+    invalid = result.first()
+    if invalid is not None:
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {index_name}")
+
+
 def upgrade() -> None:
     # Bound the ALTER's brief ACCESS EXCLUSIVE lock: without a timeout it
     # queues behind any long-running query, and all traffic queues behind it.
@@ -47,6 +72,7 @@ def upgrade() -> None:
     # Index name matches the model's ``__table_args__`` declaration so the
     # create_all() index and this one are the same object.
     with op.get_context().autocommit_block():
+        _recover_invalid_index("ix_trials_kind_non_agent")
         op.execute(
             "CREATE INDEX CONCURRENTLY IF NOT EXISTS "
             "ix_trials_kind_non_agent ON trials (kind) WHERE kind != 'agent'"