UN-3972 [DEV] Cut dashboard cron DB time by indexing workflow_file_execution on (status, created_at)

backend/workflow_manager/file_execution/migrations/0007_wfe_status_created_idx.py

@@ -0,0 +1,97 @@
+"""Add a (status, created_at) index to workflow_file_execution.
+
+The dashboard metrics cron filters this table on status and a created_at window. No
+existing index leads with status or created_at — every secondary index is prefixed by
+the workflow_execution FK column — so the planner cannot drive from here and scans
+workflow_execution in full instead. Execution plan in UN-4045 (2026-08-31, which
+supersedes the earlier workflow_file_execution reading); cost measurements in UN-3883.
+
+Full rather than partial: get_failed_pages benefits at today's window (ERROR is 0.40%
+of rows), and get_documents_processed only once UN-3973 narrows the window to 2 days —
+at 52 days the COMPLETED slice is 20.4% of the table and the planner scans regardless.
+A partial index on ERROR would serve the first and never the second.
+
+Built CONCURRENTLY (atomic = False): a plain AddIndex holds a SHARE lock for the whole
+build and would block writes to a large, write-heavy table. Prefer building it out of
+band before the deploy, exactly this statement and no other:
+
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS wfe_status_created_idx
+        ON workflow_file_execution (status, created_at);
+
+The migration then no-ops via IF NOT EXISTS and asserts the existing index is valid
+and has the expected definition. Do not build the two-index variant from an older
+revision of UN-3972's description: wfe_created_at_desc_idx was struck as worse than
+nothing.
+"""
+
+from django.db import migrations, models
+
+INDEX_NAME = "wfe_status_created_idx"
+
+INDEX_DEF_SUFFIX = "USING btree (status, created_at)"
+
+# IF NOT EXISTS matches on name alone, so a hand-built index with different columns
+# would be kept while Django recorded (status, created_at) into model state. An
+# interrupted CONCURRENTLY build likewise leaves an INVALID index that costs on every
+# write and is never read. Fail loudly on both rather than diverge silently.
+_ASSERT_INDEX_MATCHES = f"""
+DO $$
+DECLARE
+    idx_def text;
+    idx_valid boolean;
+BEGIN
+    SELECT pg_get_indexdef(i.indexrelid), i.indisvalid INTO idx_def, idx_valid
+    FROM pg_class c
+    JOIN pg_namespace n ON n.oid = c.relnamespace
+    JOIN pg_index i ON i.indexrelid = c.oid
+    WHERE c.relname = '{INDEX_NAME}' AND n.nspname = current_schema();
+
+    IF idx_def IS NULL THEN
+        RAISE EXCEPTION 'Index {INDEX_NAME} is missing from schema % after CREATE INDEX.', current_schema();
+    END IF;
+
+    IF NOT idx_valid THEN
+        RAISE EXCEPTION 'Index {INDEX_NAME} exists but is INVALID (a prior CREATE INDEX CONCURRENTLY was interrupted). Drop it and re-run: DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME};';
+    END IF;
+
+    IF idx_def NOT LIKE '%{INDEX_DEF_SUFFIX}' THEN
+        RAISE EXCEPTION 'Index {INDEX_NAME} exists with an unexpected definition (%), expected {INDEX_DEF_SUFFIX}. Drop it and re-run: DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME};', idx_def;
+    END IF;
+END
+$$;
+"""
+
+
+class Migration(migrations.Migration):
+    # CREATE / DROP INDEX CONCURRENTLY cannot run inside a transaction block.
+    atomic = False
+
+    dependencies = [
+        (
+            "file_execution",
+            "0006_workflowfileexecution_wf_file_hash_path_status_idx_and_more",
+        ),
+    ]
+
+    operations = [
+        migrations.SeparateDatabaseAndState(
+            database_operations=[
+                migrations.RunSQL(
+                    sql=(
+                        f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {INDEX_NAME} "
+                        "ON workflow_file_execution (status, created_at);"
+                    ),
+                    reverse_sql=f"DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME};",
+                ),
+                migrations.RunSQL(
+                    sql=_ASSERT_INDEX_MATCHES, reverse_sql=migrations.RunSQL.noop
+                ),
+            ],
+            state_operations=[
+                migrations.AddIndex(
+                    model_name="workflowfileexecution",
+                    index=models.Index(fields=["status", "created_at"], name=INDEX_NAME),
+                ),
+            ],
+        ),
+    ]

backend/workflow_manager/file_execution/models.py

@@ -198,6 +198,14 @@ class Meta:
                 ],
                 name="wf_provider_uuid_path_stat_idx",
             ),
+            # Every other index on this table is prefixed by the workflow_execution
+            # FK column, so none can serve a status + created_at filter. Effective for
+            # get_failed_pages today; the COMPLETED path stays inert until UN-3973
+            # narrows the window. See migration 0007.
+            models.Index(
+                fields=["status", "created_at"],
+                name="wfe_status_created_idx",
+            ),
         ]
         constraints = [
             models.UniqueConstraint(

backend/workflow_manager/file_execution/tests/test_wfe_status_created_idx.py

@@ -0,0 +1,121 @@
+"""Shape guard for the ``(status, created_at)`` index migration (UN-3972).
+
+``workflow_file_execution`` is ~3.4 GB in production and takes live inserts. A plain
+``AddIndex`` — which is what ``makemigrations`` emits from ``Meta.indexes`` — holds a
+``SHARE`` lock for the whole build and stalls file processing. ``0007`` is therefore
+hand-written: non-atomic, ``CONCURRENTLY``, and split so the ``AddIndex`` updates model
+state only.
+
+Nothing else guards that. The suite runs with ``--no-migrations``, so this migration is
+never executed in CI; regenerating or "tidying" it would land the locking version with
+every test still green. These assertions are what fails instead.
+
+DB-free by design: the migration module is imported and inspected directly.
+"""
+
+from __future__ import annotations
+
+import importlib
+
+from django.db import migrations
+from django.test import SimpleTestCase
+
+from workflow_manager.file_execution.models import WorkflowFileExecution
+
+_MIGRATION = "workflow_manager.file_execution.migrations.0007_wfe_status_created_idx"
+
+INDEX_NAME = "wfe_status_created_idx"
+INDEX_FIELDS = ["status", "created_at"]
+TABLE = "workflow_file_execution"
+
+
+class MigrationShapeTests(SimpleTestCase):
+    """The properties that keep the build off the write path."""
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+        cls.migration = importlib.import_module(_MIGRATION).Migration
+        cls.operation = cls.migration.operations[0]
+
+    def test_the_migration_has_exactly_one_operation(self) -> None:
+        """Every other assertion reads operations[0], so anything appended after the
+        SeparateDatabaseAndState is invisible — including a bare AddIndex, which is a
+        real lock-taking build on a 3.4 GB table.
+        """
+        self.assertEqual(len(self.migration.operations), 1)
+
+    def test_no_operation_builds_an_index_against_the_database(self) -> None:
+        """The same failure stated directly, so it survives the count changing."""
+        for op in self.migration.operations:
+            self.assertNotIsInstance(op, migrations.AddIndex)
+
+    def test_migration_is_non_atomic(self) -> None:
+        """CREATE/DROP INDEX CONCURRENTLY is rejected inside a transaction block."""
+        self.assertIs(self.migration.atomic, False)
+
+    def test_index_is_built_and_dropped_concurrently(self) -> None:
+        """Both directions must stay off the write-blocking lock path."""
+        create = self.operation.database_operations[0]
+        self.assertIn("CREATE INDEX CONCURRENTLY IF NOT EXISTS", create.sql)
+        self.assertIn(INDEX_NAME, create.sql)
+        # Column order is the point — (created_at, status) cannot serve an equality
+        # plus range filter. Whitespace and case are not, and a red test on
+        # semantically identical SQL only teaches people to loosen the assertion.
+        self.assertRegex(
+            create.sql,
+            rf"ON\s+{TABLE}\s*\(\s*{INDEX_FIELDS[0]}\s*,\s*{INDEX_FIELDS[1]}\s*\)",
+        )
+        self.assertIn("DROP INDEX CONCURRENTLY IF EXISTS", create.reverse_sql)
+        # A typo here makes rollback a silent no-op through IF EXISTS: Django unapplies
+        # the migration while the index stays on the table.
+        self.assertIn(INDEX_NAME, create.reverse_sql)
+
+    def test_every_database_operation_is_reversible(self) -> None:
+        """One irreversible operation kills the whole rollback, DROP INDEX included."""
+        self.assertTrue(
+            all(op.reversible for op in self.operation.database_operations)
+        )
+
+    def test_pre_existing_index_guard_is_present(self) -> None:
+        """``IF NOT EXISTS`` matches on name alone, so the guard carries the rest.
+
+        An interrupted concurrent build leaves an INVALID index, and a hand-built one
+        may have different columns; either would be kept while Django recorded the
+        migration as applied. The guard turns both into a loud failure, and looks the
+        index up in ``current_schema()`` because app tables do not live in ``public``.
+        """
+        guard = self.operation.database_operations[1].sql
+        # Polarity, not presence: `NOT idx_valid` raises on a broken index, `idx_valid`
+        # raises on every healthy deploy, and both contain "indisvalid".
+        self.assertIn("i.indisvalid", guard)
+        self.assertIn("IF NOT idx_valid THEN", guard)
+        # Scoped to this index, in this schema.
+        self.assertIn(f"c.relname = '{INDEX_NAME}'", guard)
+        self.assertIn("n.nspname = current_schema()", guard)
+        # Definition, not just validity.
+        self.assertIn("pg_get_indexdef", guard)
+        self.assertIn(f"USING btree ({', '.join(INDEX_FIELDS)})", guard)
+        self.assertIn("NOT LIKE", guard)
+        self.assertIn("RAISE EXCEPTION", guard)
+        self.assertIn(f"DROP INDEX CONCURRENTLY IF EXISTS {INDEX_NAME}", guard)
+
+    def test_add_index_updates_state_only(self) -> None:
+        """``AddIndex`` must not reach the database, or it builds a second time."""
+        self.assertIsInstance(self.operation, migrations.SeparateDatabaseAndState)
+        self.assertTrue(
+            all(
+                isinstance(op, migrations.RunSQL)
+                for op in self.operation.database_operations
+            )
+        )
+        self.assertEqual(len(self.operation.state_operations), 1)
+        state_op = self.operation.state_operations[0]
+        self.assertIsInstance(state_op, migrations.AddIndex)
+        self.assertEqual(state_op.index.name, INDEX_NAME)
+        self.assertEqual(state_op.index.fields, INDEX_FIELDS)
+
+    def test_model_meta_matches_the_migration(self) -> None:
+        """Model state and migration state drift silently otherwise."""
+        declared = {idx.name: idx.fields for idx in WorkflowFileExecution._meta.indexes}
+        self.assertEqual(declared.get(INDEX_NAME), INDEX_FIELDS)