Retry lock-timed-out migration statements instead of failing the deploy

The production deploy of trialkind01 lost its single 8s lock window on the
busy trials table three runs in a row. Keep the short lock_timeout (an
uncapped ALTER queues all traffic behind it) but retry each contended
statement in its own autocommit transaction, and log the lock-holding
sessions on the first miss so a failed run names its blocker.

Author: kyle-compute

oddish/alembic/versions/clearstale01_clear_inflight_analysis_flags.py

@@ -18,24 +18,30 @@
 
 from alembic import op
 
+from oddish.db.migration_locks import run_with_lock_retry
+
 revision: str = "clearstale01"
 down_revision: Union[str, Sequence[str], None] = "retirejobs01"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.execute("SET lock_timeout = '8s'")
-    op.execute(
-        """
-        UPDATE trials
-        SET    analysis_status = NULL,
-               analysis_error = NULL,
-               analysis_started_at = NULL,
-               analysis_finished_at = NULL
-        WHERE  analysis_status::text IN ('PENDING', 'QUEUED', 'RUNNING')
-        """
-    )
+    # Row-lock waits on the busy ``trials`` table are bounded by a short
+    # lock_timeout and retried (see migration_locks).
+    def _clear_inflight_flags() -> None:
+        op.execute(
+            """
+            UPDATE trials
+            SET    analysis_status = NULL,
+                   analysis_error = NULL,
+                   analysis_started_at = NULL,
+                   analysis_finished_at = NULL
+            WHERE  analysis_status::text IN ('PENDING', 'QUEUED', 'RUNNING')
+            """
+        )
+
+    run_with_lock_retry(_clear_inflight_flags, table_name="trials")
 
 
 def downgrade() -> None:

oddish/alembic/versions/dropblocks01_drop_analyzer_blocks.py

@@ -22,56 +22,75 @@
 import sqlalchemy as sa
 from alembic import op
 
+from oddish.db.migration_locks import run_with_lock_retry
+
 revision: str = "dropblocks01"
 down_revision: Union[str, Sequence[str], None] = "analysisspend01"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.execute("SET lock_timeout = '8s'")
+    # Every statement here contends with live traffic (row locks on
+    # ``worker_jobs`` and ``trials``, the DROP's ACCESS EXCLUSIVE on
+    # ``analyzer_blocks``), so each is bounded by a short lock_timeout and
+    # retried (see migration_locks). All three are idempotent, so a retried
+    # or re-run statement redoes no work.
+    #
     # Between the cutover deploy (whose retirejobs01 cancelled the retired
     # kinds then in flight) and THIS deploy, the public summary route kept
     # enqueueing ANALYZER rows on every share-page summary miss. No handler
     # claims the kind, but a QUEUED row keeps its queue "active" to the
     # dispatcher's discovery query forever, spawning workers that claim
     # nothing and exit. Same terminal write retirejobs01 models.
-    op.execute(
-        """
-        UPDATE worker_jobs
-        SET    status = 'CANCELLED',
-               finished_at = NOW(),
-               error_message = 'block pipeline removed: summaries are read from trials.trajectory_summary',
-               current_worker_id = NULL,
-               current_queue_slot = NULL,
-               modal_function_call_id = NULL
-        WHERE  kind::text = 'ANALYZER'
-          AND  status::text IN ('QUEUED', 'RETRYING', 'RUNNING', 'BLOCKED')
-        """
-    )
-    result = op.get_bind().execute(
-        sa.text("SELECT to_regclass('analyzer_blocks') IS NOT NULL")
-    )
-    if result.scalar():
+    def _cancel_analyzer_jobs() -> None:
         op.execute(
             """
-            UPDATE trials t
-            SET    trajectory_summary = b.output
-            FROM (
-                SELECT DISTINCT ON (analyzer_id) analyzer_id, output
-                FROM analyzer_blocks
-                WHERE type = 'trajectory_summary'
-                  AND status::text = 'SUCCESS'
-                  AND output IS NOT NULL
-                  AND output != 'null'::jsonb
-                  AND deleted_at IS NULL
-                ORDER BY analyzer_id, created_at DESC, id
-            ) b
-            WHERE t.id = b.analyzer_id
-              AND t.trajectory_summary IS NULL
+            UPDATE worker_jobs
+            SET    status = 'CANCELLED',
+                   finished_at = NOW(),
+                   error_message = 'block pipeline removed: summaries are read from trials.trajectory_summary',
+                   current_worker_id = NULL,
+                   current_queue_slot = NULL,
+                   modal_function_call_id = NULL
+            WHERE  kind::text = 'ANALYZER'
+              AND  status::text IN ('QUEUED', 'RETRYING', 'RUNNING', 'BLOCKED')
             """
         )
-    op.execute("DROP TABLE IF EXISTS analyzer_blocks")
+
+    run_with_lock_retry(_cancel_analyzer_jobs, table_name="worker_jobs")
+
+    result = op.get_bind().execute(
+        sa.text("SELECT to_regclass('analyzer_blocks') IS NOT NULL")
+    )
+    if result.scalar():
+
+        def _backfill_summaries() -> None:
+            op.execute(
+                """
+                UPDATE trials t
+                SET    trajectory_summary = b.output
+                FROM (
+                    SELECT DISTINCT ON (analyzer_id) analyzer_id, output
+                    FROM analyzer_blocks
+                    WHERE type = 'trajectory_summary'
+                      AND status::text = 'SUCCESS'
+                      AND output IS NOT NULL
+                      AND output != 'null'::jsonb
+                      AND deleted_at IS NULL
+                    ORDER BY analyzer_id, created_at DESC, id
+                ) b
+                WHERE t.id = b.analyzer_id
+                  AND t.trajectory_summary IS NULL
+                """
+            )
+
+        run_with_lock_retry(_backfill_summaries, table_name="trials")
+
+    run_with_lock_retry(
+        lambda: op.execute("DROP TABLE IF EXISTS analyzer_blocks"),
+        table_name="analyzer_blocks",
+    )
 
 
 def downgrade() -> None:

oddish/alembic/versions/retirejobs01_cancel_retired_kind_jobs.py

@@ -19,30 +19,35 @@
 
 from alembic import op
 
+from oddish.db.migration_locks import run_with_lock_retry
+
 revision: str = "retirejobs01"
 down_revision: Union[str, Sequence[str], None] = "shadowexp01"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    # Bound waiting on row locks a still-draining worker may briefly hold.
-    op.execute("SET lock_timeout = '8s'")
-    op.execute(
-        """
-        UPDATE worker_jobs
-        SET    status = 'CANCELLED',
-               finished_at = NOW(),
-               error_message = 'pipeline removed: analysis runs as trials now',
-               current_worker_id = NULL,
-               current_queue_slot = NULL,
-               modal_function_call_id = NULL
-        WHERE  kind::text IN
-               ('QA', 'VERDICT', 'ANALYSIS', 'QA_REVIEW',
-                'ANALYZER', 'ANALYZER_BLOCK')
-          AND  status::text IN ('QUEUED', 'RETRYING', 'RUNNING', 'BLOCKED')
-        """
-    )
+    # Waiting on row locks a still-draining worker may briefly hold is
+    # bounded by a short lock_timeout and retried (see migration_locks).
+    def _cancel_retired_kind_jobs() -> None:
+        op.execute(
+            """
+            UPDATE worker_jobs
+            SET    status = 'CANCELLED',
+                   finished_at = NOW(),
+                   error_message = 'pipeline removed: analysis runs as trials now',
+                   current_worker_id = NULL,
+                   current_queue_slot = NULL,
+                   modal_function_call_id = NULL
+            WHERE  kind::text IN
+                   ('QA', 'VERDICT', 'ANALYSIS', 'QA_REVIEW',
+                    'ANALYZER', 'ANALYZER_BLOCK')
+              AND  status::text IN ('QUEUED', 'RETRYING', 'RUNNING', 'BLOCKED')
+            """
+        )
+
+    run_with_lock_retry(_cancel_retired_kind_jobs, table_name="worker_jobs")
 
 
 def downgrade() -> None:

oddish/alembic/versions/shadowexp01_add_experiment_shadow_of.py

@@ -23,6 +23,8 @@
 import sqlalchemy as sa
 from alembic import op
 
+from oddish.db.migration_locks import run_with_lock_retry
+
 revision: str = "shadowexp01"
 down_revision: Union[str, Sequence[str], None] = "trialkind01"
 branch_labels: Union[str, Sequence[str], None] = None
@@ -58,24 +60,32 @@ def _recover_invalid_index(index_name: str) -> None:
 
 
 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.
-    op.execute("SET lock_timeout = '8s'")
-    op.add_column(
-        "experiments",
-        sa.Column("shadow_of", sa.String(64), nullable=True),
-        if_not_exists=True,
+    # The ALTER's ACCESS EXCLUSIVE lock is bounded by a short lock_timeout
+    # (an uncapped wait queues all traffic on ``experiments`` behind it) and
+    # retried, since one short window can lose to in-flight queries.
+    run_with_lock_retry(
+        lambda: op.add_column(
+            "experiments",
+            sa.Column("shadow_of", sa.String(64), nullable=True),
+            if_not_exists=True,
+        ),
+        table_name="experiments",
     )
+
     # 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():
+    # create_all() index and this one are the same object. The invalid-index
+    # recovery runs inside the retried step: a lock-timed-out CREATE INDEX
+    # CONCURRENTLY leaves an INVALID index the next attempt must drop first.
+    def _create_index() -> None:
         _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) "
             "WHERE deleted_at IS NULL"
         )
 
+    run_with_lock_retry(_create_index, table_name="experiments")
+
 
 def downgrade() -> None:
     with op.get_context().autocommit_block():

oddish/alembic/versions/trialkind01_add_trial_kind.py

@@ -24,6 +24,8 @@
 import sqlalchemy as sa
 from alembic import op
 
+from oddish.db.migration_locks import run_with_lock_retry
+
 revision: str = "trialkind01"
 down_revision: Union[str, Sequence[str], None] = "dropanalyzers01"
 branch_labels: Union[str, Sequence[str], None] = None
@@ -56,28 +58,38 @@ def _recover_invalid_index(index_name: str) -> None:
 
 
 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.
-    op.execute("SET lock_timeout = '8s'")
-    op.add_column(
-        "trials",
-        sa.Column(
-            "kind",
-            sa.String(32),
-            nullable=False,
-            server_default="agent",
+    # The ALTER's ACCESS EXCLUSIVE lock is bounded by a short lock_timeout
+    # (an uncapped wait queues all traffic on ``trials`` behind it) and
+    # retried: one short window regularly loses to in-flight queries on this
+    # busy table — the 2026-08-19 production deploy lost its single 8s
+    # window three runs in a row.
+    run_with_lock_retry(
+        lambda: op.add_column(
+            "trials",
+            sa.Column(
+                "kind",
+                sa.String(32),
+                nullable=False,
+                server_default="agent",
+            ),
+            if_not_exists=True,
         ),
-        if_not_exists=True,
+        table_name="trials",
     )
+
     # 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():
+    # create_all() index and this one are the same object. The invalid-index
+    # recovery runs inside the retried step: a lock-timed-out CREATE INDEX
+    # CONCURRENTLY leaves an INVALID index the next attempt must drop first.
+    def _create_index() -> None:
         _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'"
         )
 
+    run_with_lock_retry(_create_index, table_name="trials")
+
 
 def downgrade() -> None:
     with op.get_context().autocommit_block():