fix(api): make provider enum-to-varchar migration deploy-safe

- Backfill provider_str synchronously in 0095 (single UPDATE) so the column
  is populated before 0096 sets it NOT NULL, removing the Celery race
- Recreate the unique index inside 0096's transaction with a lock_timeout,
  closing the duplicate-provider window left by the concurrent rebuild
- Drop migration 0097 and the now-unused backfill_provider_str task

api/src/backend/api/migrations/0095_backfill_provider_str.py

@@ -1,31 +1,25 @@
 from django.db import migrations
-from tasks.tasks import backfill_provider_str_task
-
-from api.db_router import MainRouter
-
-
-def trigger_provider_str_backfill(apps, _schema_editor):
-    """Dispatch a per-tenant Celery task to populate the transitional
-    `provider_str` shadow column for rows created before the sync trigger
-    from 0094 existed.
-
-    New writes are already covered by the trigger, so this only fills the gap
-    left by pre-existing rows. The work runs in the background, batched per
-    tenant, so the migration itself finishes in seconds regardless of table
-    size.
-    """
-    Tenant = apps.get_model("api", "Tenant")
-    tenant_ids = Tenant.objects.using(MainRouter.admin_db).values_list("id", flat=True)
-
-    for tenant_id in tenant_ids:
-        backfill_provider_str_task.delay(tenant_id=str(tenant_id))
 
 
 class Migration(migrations.Migration):
+    """Synchronous backfill of the `provider_str` shadow column.
+
+    A single UPDATE fills rows that predate the 0094 trigger. The providers
+    table is small, so this is safe inline and guarantees the column is fully
+    populated before 0096 sets it NOT NULL (no race with an async backfill).
+    Runs on the migration connection, which is exempt from RLS.
+    """
+
     dependencies = [
         ("api", "0094_provider_str_shadow_column"),
     ]
 
     operations = [
-        migrations.RunPython(trigger_provider_str_backfill, migrations.RunPython.noop),
+        migrations.RunSQL(
+            sql=(
+                "UPDATE providers SET provider_str = provider::text "
+                "WHERE provider_str IS NULL;"
+            ),
+            reverse_sql=migrations.RunSQL.noop,
+        ),
     ]

api/src/backend/api/migrations/0096_provider_enum_to_varchar_contract.py

@@ -2,20 +2,14 @@ from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
-    """Contract step of the zero-downtime migration of Provider.provider from a
-    native PostgreSQL enum to varchar.
+    """Contract step: promote `provider_str` into `provider`.
 
-    The shadow column added in 0094 has been kept in sync by the trigger and
-    backfilled in 0095, so it now holds the value for every row. This migration
-    promotes it into place: drop the trigger and the enum column, rename the
-    shadow column to `provider`, and drop the orphaned enum type. The column
-    name is preserved throughout, and varchar accepts the same string values
-    the enum held, so app instances running the previous release keep working
-    against the swapped column.
-
-    The drop/rename runs in this migration's transaction so `provider` never
-    disappears for readers. The partial unique index is dropped here and
-    rebuilt concurrently in the next migration to avoid a long write lock.
+    Drops the trigger and enum column, renames the shadow column, sets it NOT
+    NULL, and drops the enum type. The unique index is dropped and recreated in
+    the same transaction, so there is no window for duplicate active providers;
+    recreated non-concurrently since the table is small, with a short
+    lock_timeout so the migration fails fast instead of queueing behind a
+    long-running transaction.
     """
 
     dependencies = [
@@ -38,6 +32,7 @@ class Migration(migrations.Migration):
             database_operations=[
                 migrations.RunSQL(
                     sql=(
+                        "SET LOCAL lock_timeout = '10s';\n"
                         "DROP TRIGGER IF EXISTS providers_sync_provider_str ON providers;\n"
                         "DROP FUNCTION IF EXISTS sync_provider_str();\n"
                         "DROP INDEX IF EXISTS unique_provider_uids;\n"
@@ -45,7 +40,9 @@ class Migration(migrations.Migration):
                         "ALTER TABLE providers RENAME COLUMN provider_str TO provider;\n"
                         "ALTER TABLE providers ALTER COLUMN provider SET DEFAULT 'aws';\n"
                         "ALTER TABLE providers ALTER COLUMN provider SET NOT NULL;\n"
-                        "DROP TYPE provider;"
+                        "DROP TYPE provider;\n"
+                        "CREATE UNIQUE INDEX unique_provider_uids ON providers "
+                        "(tenant_id, provider, uid) WHERE NOT is_deleted;"
                     ),
                     reverse_sql=migrations.RunSQL.noop,
                 ),

api/src/backend/api/migrations/0097_provider_unique_index_concurrent.py

@@ -1,28 +0,0 @@
-from django.db import migrations
-
-
-class Migration(migrations.Migration):
-    """Rebuild the partial unique index on `providers` after the enum-to-varchar
-    contract in 0096 dropped it along with the old enum column.
-
-    Built with CREATE INDEX CONCURRENTLY (hence `atomic = False`) so the rebuild
-    holds no long write lock on a large table. The index keeps the name and
-    predicate Django expects for the existing `unique_provider_uids` constraint,
-    which stays in the model state untouched, so no state operation is needed.
-    """
-
-    atomic = False
-
-    dependencies = [
-        ("api", "0096_provider_enum_to_varchar_contract"),
-    ]
-
-    operations = [
-        migrations.RunSQL(
-            sql=(
-                "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS unique_provider_uids "
-                "ON providers (tenant_id, provider, uid) WHERE NOT is_deleted;"
-            ),
-            reverse_sql="DROP INDEX CONCURRENTLY IF EXISTS unique_provider_uids;",
-        ),
-    ]

api/src/backend/tasks/jobs/backfill.py

@@ -40,35 +40,6 @@ from api.models import (
 logger = get_task_logger(__name__)
 
 
-def backfill_provider_str(tenant_id: str, batch_size: int = 1000):
-    """Populate the transitional `provider_str` shadow column for rows that
-    predate the sync trigger, copying `provider::text` in bounded batches.
-
-    Each batch runs in its own RLS transaction so the lock is held only for the
-    rows in that batch, keeping the operation safe on a large table. Idempotent:
-    only rows where `provider_str IS NULL` are touched, so an interrupted run
-    resumes cleanly and a completed column is a no-op on retry.
-    """
-    total_updated = 0
-    while True:
-        with rls_transaction(tenant_id) as cursor:
-            cursor.execute(
-                "UPDATE providers SET provider_str = provider::text "
-                "WHERE id IN ("
-                "    SELECT id FROM providers WHERE provider_str IS NULL LIMIT %s"
-                ")",
-                [batch_size],
-            )
-            updated = cursor.rowcount
-        total_updated += updated
-        if updated < batch_size:
-            break
-    logger.info(
-        "Backfilled provider_str for tenant %s: %d rows", tenant_id, total_updated
-    )
-    return {"tenant_id": tenant_id, "updated": total_updated}
-
-
 def backfill_resource_scan_summaries(tenant_id: str, scan_id: str):
     with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
         if ResourceScanSummary.objects.filter(

api/src/backend/tasks/tasks.py

@@ -19,7 +19,6 @@ from tasks.jobs.backfill import (
     backfill_daily_severity_summaries,
     backfill_finding_group_summaries,
     backfill_provider_compliance_scores,
-    backfill_provider_str,
     backfill_resource_scan_summaries,
     aggregate_scan_category_summaries,
     aggregate_scan_resource_group_summaries,
@@ -723,12 +722,6 @@ def backfill_finding_group_summaries_task(tenant_id: str, days: int = None):
     return backfill_finding_group_summaries(tenant_id=tenant_id, days=days)
 
 
-@shared_task(name="backfill-provider-str", queue="backfill")
-def backfill_provider_str_task(tenant_id: str):
-    """Backfill the transitional provider_str shadow column for a tenant."""
-    return backfill_provider_str(tenant_id=tenant_id)
-
-
 @shared_task(name="scan-category-summaries", queue="overview")
 @handle_provider_deletion
 def aggregate_scan_category_summaries_task(tenant_id: str, scan_id: str):