diff --git a/backend/workflow_manager/file_execution/migrations/0007_wfe_status_created_idx.py b/backend/workflow_manager/file_execution/migrations/0007_wfe_status_created_idx.py new file mode 100644 index 0000000000..06746771fd --- /dev/null +++ b/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), + ), + ], + ), + ] diff --git a/backend/workflow_manager/file_execution/models.py b/backend/workflow_manager/file_execution/models.py index 105b3875a9..9578f8b1be 100644 --- a/backend/workflow_manager/file_execution/models.py +++ b/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( diff --git a/backend/workflow_manager/file_execution/tests/__init__.py b/backend/workflow_manager/file_execution/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/workflow_manager/file_execution/tests/test_wfe_status_created_idx.py b/backend/workflow_manager/file_execution/tests/test_wfe_status_created_idx.py new file mode 100644 index 0000000000..47709bce1b --- /dev/null +++ b/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)