Fail report runs and simulation runs left running after their Modal call ended

.github/workflows/simulation-deploy.reusable.yml

@@ -377,6 +377,17 @@ jobs:
           POLICYENGINE_OTEL_GOOGLE_AUDIENCE: ${{ vars.OBSERVABILITY_OTLP_GOOGLE_AUDIENCE }}
         run: uv run --frozen modal deploy --env="${{ inputs.modal_environment }}" src/modal/app.py
 
+      # The scheduled sweep that fails report runs and simulation runs nothing
+      # will finish. Its database URL comes from the Modal secret
+      # policyengine-modal-worker-database, so this step fails until an
+      # operator has created that secret in the Modal environment.
+      - name: Deploy the stuck-run sweep
+        working-directory: projects/policyengine-simulation-executor
+        env:
+          MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }}
+          MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }}
+        run: uv run --frozen modal deploy --env="${{ inputs.modal_environment }}" src/modal/run_sweep_app.py
+
       - name: Run executor unit tests
         working-directory: projects/policyengine-simulation-executor
         run: uv run --frozen pytest tests/ -v

changelog.d/stuck-run-sweep.added.md

@@ -0,0 +1 @@
+A scheduled Modal function, in the new Modal application `policyengine-simulation-run-sweep`, marks the Public API's report runs and simulation runs failed when nothing will finish them: a pending or running run whose recorded Modal call has ended, or a pending run that still has no Modal call recorded ten minutes after its creation. It checks every ten minutes and is safe to run twice at once. It leaves a run alone while Modal reports its call running, or can't report it, until the run is 24 hours old, counted from its first claim or, if it was never claimed, from its creation, longer than any of its calls can run; then it fails the run whatever Modal reports. The deployment workflow deploys it with the executor, and fails until the Modal secret `policyengine-modal-worker-database` exists in the Modal environment.

libs/policyengine-stage12-persistence/README.md

@@ -11,9 +11,10 @@ revisions and live schema. This package never changes a schema.
   SPM receipts and data version, `simulation_runs`, and the result tables,
   the two labour supply tables included. The v2 Modal coordinator and workers
   write them as the Modal worker role (`policyengine_v2_modal_worker`), for a
-  call that names a report run. The mappings declare only the columns the
-  coordinator's and workers' statements read or write, and each `UPDATE`
-  sets only columns the role may update. Every status write is one `UPDATE`
+  call that names a report run, and the scheduled stuck-run sweep fails the
+  runs nothing will finish, as the same role. The mappings declare only the
+  columns those statements read or write, and each `UPDATE` sets only
+  columns the role may update. Every status write is one `UPDATE`
   conditional on the run's status; a report run's completion sets its own
   result columns and inserts its result rows in the same transaction.
 

libs/policyengine-stage12-persistence/src/policyengine_stage12_persistence/__init__.py

@@ -6,7 +6,9 @@
     report_run_result,
 )
 from policyengine_stage12_persistence.report_run_store import (
+    ActiveRun,
     ReportRunStore,
+    RunKind,
     RunStatus,
     RunWrite,
 )
@@ -22,10 +24,12 @@
 )
 
 __all__ = [
+    "ActiveRun",
     "ConnectionOpener",
     "DistrictCodes",
     "ReportRunResult",
     "ReportRunStore",
+    "RunKind",
     "RunStatus",
     "RunWrite",
     "Stage12PersistenceStore",

libs/policyengine-stage12-persistence/src/policyengine_stage12_persistence/report_run_store.py

@@ -14,7 +14,7 @@
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any
 
-from sqlalchemy import func, insert, or_, select, update
+from sqlalchemy import exists, func, insert, or_, select, update
 from sqlalchemy.dialects.postgresql import insert as postgresql_insert
 
 from policyengine_stage12_persistence.report_run_tables import (
@@ -33,7 +33,7 @@
     from uuid import UUID
 
     from policyengine_simulation_contract.stage12_execution import ReportRunIds
-    from sqlalchemy import Engine, Table, Update
+    from sqlalchemy import ColumnElement, Engine, Table, Update
     from sqlalchemy.engine import Connection
 
     from policyengine_stage12_persistence.report_run_results import ReportRunResult
@@ -51,6 +51,29 @@ class RunStatus(StrEnum):
 ACTIVE_STATUSES = (RunStatus.PENDING.value, RunStatus.RUNNING.value)
 
 
+class RunKind(StrEnum):
+    """Which run table a run is in."""
+
+    REPORT_RUN = "report_run"
+    SIMULATION_RUN = "simulation_run"
+
+
+RUN_TABLES = {RunKind.REPORT_RUN: report_runs, RunKind.SIMULATION_RUN: simulation_runs}
+
+
+@dataclass(frozen=True)
+class ActiveRun:
+    """A pending or running run, and the Modal call recorded on it.
+
+    ``modal_call_id`` is None for a run with no call recorded.
+    """
+
+    kind: RunKind
+    run_id: UUID
+    status: RunStatus
+    modal_call_id: str | None
+
+
 @dataclass(frozen=True)
 class RunWrite:
     """One conditional write's outcome: whether it changed the run, and its status."""
@@ -298,6 +321,92 @@ def fail_simulation_run(
             simulation_runs, simulation_run_id, statement, "fail_simulation_run"
         )
 
+    def list_active_runs(
+        self,
+        kind: RunKind,
+        *,
+        active_since_before: datetime,
+    ) -> tuple[ActiveRun, ...]:
+        """Return the pending and running runs of ``kind`` active since before a time.
+
+        A run counts from its first claim, or from its creation while it is
+        pending. Runs are oldest first, with or without a recorded Modal
+        call. A simulation run is listed only while no pending or running
+        report run names it as its baseline or reform: that report run's
+        coordinator starts its children's calls, records their outcomes, and
+        stops and starts them again when Modal restarts it, so they are its
+        to judge until it has finished.
+        """
+        table = RUN_TABLES[kind]
+        active_since = func.coalesce(table.c.started_at, table.c.created_at)
+        conditions = [
+            table.c.status.in_(ACTIVE_STATUSES),
+            active_since < active_since_before,
+        ]
+        if kind is RunKind.SIMULATION_RUN:
+            conditions.append(_no_active_report_run_names(table))
+        statement = (
+            select(table.c.id, table.c.status, table.c.modal_call_id)
+            .where(*conditions)
+            .order_by(active_since, table.c.id)
+        )
+        with self._transaction("list_active_runs") as connection:
+            return tuple(
+                ActiveRun(
+                    kind=kind,
+                    run_id=row.id,
+                    status=RunStatus(row.status),
+                    modal_call_id=row.modal_call_id,
+                )
+                for row in connection.execute(statement)
+            )
+
+    def fail_stuck_run(
+        self,
+        run: ActiveRun,
+        *,
+        error_message: str,
+        failed_at: datetime,
+    ) -> RunWrite:
+        """Mark a stuck ``run`` failed, while it is as it was listed.
+
+        It must still have its listed status and ``run.modal_call_id``, or no
+        call if it was listed with none, and a simulation run must still have
+        no pending or running report run naming it. A run that was claimed,
+        finished, or moved to another call, that the simulation service
+        recorded a call on, or that a pending or running report run has named
+        since it was listed, is left as it is, so concurrent sweeps change it
+        once. A report run naming it counts once its insert has committed,
+        before this update starts: the database doesn't make an insert in
+        progress wait for this update, or this update for it.
+        """
+        table = RUN_TABLES[run.kind]
+        same_call = (
+            table.c.modal_call_id.is_(None)
+            if run.modal_call_id is None
+            else table.c.modal_call_id == run.modal_call_id
+        )
+        statement = (
+            update(table)
+            .where(
+                table.c.id == run.run_id,
+                table.c.status == run.status.value,
+                same_call,
+                *(
+                    (_no_active_report_run_names(table),)
+                    if run.kind is RunKind.SIMULATION_RUN
+                    else ()
+                ),
+            )
+            .values(
+                status=RunStatus.FAILED.value,
+                error_message=error_message,
+                completed_at=failed_at,
+                updated_at=failed_at,
+            )
+        )
+        return self._write(table, run.run_id, statement, "fail_stuck_run")
+
     def _write(
         self,
         table: Table,
@@ -351,3 +460,17 @@ def _uniform_rows(rows: tuple[dict[str, Any], ...]) -> list[dict[str, Any]]:
     """Give every row the same columns, unset ones null, for one batched insert."""
     columns = {name for row in rows for name in row}
     return [dict.fromkeys(columns) | row for row in rows]
+
+
+def _no_active_report_run_names(table: Table) -> ColumnElement[bool]:
+    """Return the condition that no pending or running report run names the run.
+
+    A report run names its simulation runs as its baseline and its reform.
+    """
+    return ~exists().where(
+        or_(
+            report_runs.c.baseline_simulation_run_id == table.c.id,
+            report_runs.c.reform_simulation_run_id == table.c.id,
+        ),
+        report_runs.c.status.in_(ACTIVE_STATUSES),
+    )

libs/policyengine-stage12-persistence/src/policyengine_stage12_persistence/report_run_tables.py

@@ -1,9 +1,9 @@
 """SQLAlchemy mappings of the Public API's run tables that report calls update.
 
 ``policyengine-api`` owns these tables and every schema migration. The
-mappings declare only the columns the coordinator and workers read or write
-as the Modal worker role, so they can issue typed DML; runtime code never
-emits DDL.
+mappings declare only the columns the coordinator, the workers and the
+stuck-run sweep read or write as the Modal worker role, so they can issue
+typed DML; runtime code never emits DDL.
 """
 
 from __future__ import annotations
@@ -33,6 +33,7 @@
     "report_runs",
     metadata,
     Column("id", UUID(as_uuid=True), primary_key=True),
+    Column("created_at", DateTime(timezone=True), nullable=False),
     Column("baseline_simulation_run_id", UUID(as_uuid=True), nullable=False),
     Column("reform_simulation_run_id", UUID(as_uuid=True)),
     Column("status", report_run_status_type, nullable=False),
@@ -53,6 +54,7 @@
     "simulation_runs",
     metadata,
     Column("id", UUID(as_uuid=True), primary_key=True),
+    Column("created_at", DateTime(timezone=True), nullable=False),
     Column("status", simulation_run_status_type, nullable=False),
     Column("modal_call_id", String(255)),
     Column("error_message", String),