Start a run or household job again when its Modal call never started

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

@@ -1189,22 +1189,34 @@ jobs:
         run: ../../.github/scripts/stage12-v2-default-verify.sh
 
       # The scheduled sweep that fails report runs, simulation runs and
-      # household jobs nothing will finish. Its app is unversioned and its
-      # schedule starts when it deploys, so it deploys here, after every
-      # check, with the release's Stage 12 default; if it fails, the next
-      # step puts the default back.
+      # household jobs nothing will finish, and the restart check that starts
+      # again those whose Modal call never started. Their app is unversioned
+      # and its schedules start when it deploys, so it deploys here, after
+      # every check, with the release's Stage 12 default; if it fails, the
+      # next step puts the default back.
       # Only from main: a dispatch from another branch, which may be behind
-      # main, leaves the environment's sweep as it is. With no status
+      # main, leaves the environment's app as it is. With no status
       # function, the condition also requires every earlier step to pass.
-      # Its database URL comes from the Modal secret
+      # Their database URL comes from the Modal secret
       # policyengine-modal-worker-database, which the v2 workers' deploy in
-      # deploy_stage12_v2 already requires.
-      - name: Deploy the stuck-run sweep
+      # deploy_stage12_v2 already requires. The image bakes in the Modal
+      # environment the restart check starts calls in, and the v2 workers'
+      # telemetry settings, from the repository variables they use.
+      - name: Deploy the stuck-run sweep and the restart check
         if: ${{ github.ref == 'refs/heads/main' }}
         working-directory: projects/policyengine-simulation-executor
         env:
           MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }}
           MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }}
+          SIMULATION_V2_MODAL_ENVIRONMENT: ${{ inputs.modal_environment }}
+          OBSERVABILITY_SERVICE_NAMESPACE: ${{ vars.OBSERVABILITY_SERVICE_NAMESPACE }}
+          OBSERVABILITY_TRACE_PROJECT_ID: ${{ vars.OBSERVABILITY_TRACE_PROJECT_ID }}
+          OBSERVABILITY_LOGGING_PROJECT_ID: ${{ vars.OBSERVABILITY_LOGGING_PROJECT_ID }}
+          OBSERVABILITY_LOG_NAME: ${{ vars.OBSERVABILITY_LOG_NAME }}
+          OBSERVABILITY_GOOGLE_WORKLOAD_IDENTITY_PROVIDER: ${{ vars.OBSERVABILITY_GOOGLE_WORKLOAD_IDENTITY_PROVIDER }}
+          OBSERVABILITY_GOOGLE_SERVICE_ACCOUNT_EMAIL: ${{ vars.OBSERVABILITY_GOOGLE_SERVICE_ACCOUNT_EMAIL }}
+          OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OBSERVABILITY_OTLP_ENDPOINT }}
+          POLICYENGINE_OTEL_GOOGLE_AUDIENCE: ${{ vars.OBSERVABILITY_OTLP_GOOGLE_AUDIENCE }}
         run: uv run --frozen modal deploy --env="${{ inputs.modal_environment }}" src/modal/run_sweep_app.py
 
       # Writes back the manifest the move replaced, only while the stored

changelog.d/c1-19-restart-unstarted-runs.added.md

@@ -0,0 +1 @@
+A restart check in the stuck-run sweep's Modal application starts the Modal call of a household job, economy report run or standalone economy simulation run again when the run stayed pending with no call recorded 3 to 10 minutes after its creation, as when the Simulation Entrypoint stopped before starting its call; its telemetry counts every restart and how it ended. The sweep now fails a run still pending with no call at 15 minutes instead of 10.

docs/migration/stage-12-modal-worker-foundation.md

@@ -359,7 +359,7 @@ run's maximum age of 24 hours.
 anything: their `coordinate_report` requires an evaluation record, and their
 context model refuses `simulation_run_id`. Modal accepts the call, which fails
 at once, and the run stays pending with its call recorded until the sweep fails
-it (`modal_call_ended`), at least ten minutes after its creation. A call
+it (`modal_call_ended`), at least 15 minutes after its creation. A call
 reaches such an application only while the manifest's default, or
 `STAGE12_V2_WORKER_VERSION`, names one: during the deployment of this change,
 if it adds a newer .py version, until it moves the default to it, or while a
@@ -410,7 +410,7 @@ single-year report's runs, and the reverse.
 - **The stuck-run sweep.** It leaves a waiting year of an unfinished window to
   the window, counts a year started later from when its call was recorded,
   fails the window when it fails one of its years, and settles a window that
-  nothing has changed for ten minutes with no year under way: from its years
+  nothing has changed for 15 minutes with no year under way: from its years
   when they have all ended, or as stalled (`budget_window_stalled`) when years
   are still waiting.
 
@@ -762,7 +762,7 @@ call that names no job, as deployment validation makes, touches no database.
 The stuck-run sweep (`src/modal/run_sweep_app.py`) fails household jobs
 nothing will finish by the rule it applies to report runs and simulation
 runs, counting a job's age from its creation: a pending job with no call
-recorded ten minutes after its creation, a pending or running job once
+recorded 15 minutes after its creation, a pending or running job once
 Modal reports its recorded call ended, and any job still unfinished 24 hours
 after its creation, whatever Modal reports.
 
@@ -792,6 +792,40 @@ validation like any other validation failure: the manifest is not published,
 and, as described under "Release and deployment constraints", the deployment
 does not promote the new Simulation Entrypoint revision.
 
+**Restarting a run whose call never started.** The stuck-run sweep's Modal
+application also runs a restart check every two minutes, as the Modal worker
+role. It lists, at most 20 a check, the household jobs, report runs and
+standalone simulation runs created 3 to 10 minutes ago that are still pending
+with no call recorded and that nothing has changed since their insert, as
+when the Simulation Entrypoint stopped between creating a run and starting
+its call. For each, on a thread, it reads back what the entrypoint built the
+call from and builds the same call with the entrypoint's own builders, on the
+worker of the .py version the row records. It then sets the row's
+`updated_at`, so the row is never restarted twice, starts the call and
+records its ID while the row is still pending with none. A row it can't start
+again (its worker left the manifest, its stored rows changed, or Modal
+refuses the call) is failed with `run_not_started`, a report run with its
+pending simulation runs. A row is left unmarked for the next check when its
+rows or manifest can't be read now, its rebuild raises an unexpected error
+(logged with its traceback), its mark can't be written, its call can't be
+rebuilt and its failure can't be written, or the check's time is up. After
+three minutes a check marks no more rows, so it ends within its five-minute
+timeout without leaving a marked row with no call. The sweep's 15 minutes come after the restart window, so a second miss
+is still failed with `run_not_started`. The restart code is the simulation
+service's (`policyengine_simulation_entry.run_restarts`, `run_restart_calls`
+and `run_restart_check`, on `policyengine_simulation_persistence`), which the
+image adds as source.
+
+It leaves alone a budget window's report runs and household simulations'
+runs: it can't rebuild their calls yet. Each check is a
+`v2_run_restart_check` operation and each row it takes up a
+`v2_run_restart` operation, with the row's kind, ID, age and outcome, and
+each outcome is also logged as a `v2_run_restart_outcome` event. A deferred
+row is taken up again by the next check, so the distinct run IDs are the
+runs that needed a restart that a check took up; a row no check took up, in
+a burst larger than the checks in its window had time for, is failed by the
+sweep with `run_not_started` instead. The events say how each ended.
+
 ## Temporary direct runner endpoint
 
 > **Temporary Stage 12 interface:** The authenticated routes in this section

libs/policyengine-simulation-observability/src/policyengine_simulation_observability/stages.py

@@ -27,6 +27,7 @@ class RunConfiguration(StrEnum):
     STAGE12_SIMULATION = "stage12_simulation"
     STAGE12_SEGMENT = "stage12_segment"
     V2_HOUSEHOLD = "v2_household"
+    V2_RUN_RESTART = "v2_run_restart"
 
 
 class Stage(StrEnum):
@@ -116,6 +117,8 @@ class Stage(StrEnum):
     V2_HOUSEHOLD_COUNTRY_MODEL_LOAD = "v2_household_country_model_load"
     V2_HOUSEHOLD_INPUT_NORMALIZATION = "v2_household_input_normalization"
     V2_HOUSEHOLD_CALCULATION = "v2_household_calculation"
+    V2_RUN_RESTART_CHECK = "v2_run_restart_check"
+    V2_RUN_RESTART = "v2_run_restart"
 
 
 @dataclass(frozen=True)
@@ -287,6 +290,12 @@ def name(self, stage: Stage) -> str:
                 Stage.V2_HOUSEHOLD_CALCULATION,
             ),
         ),
+        # One scheduled restart check, then each run or household job it
+        # lists because its Modal call never started.
+        RunConfiguration.V2_RUN_RESTART: StagePlan(
+            RunConfiguration.V2_RUN_RESTART,
+            (Stage.V2_RUN_RESTART_CHECK, Stage.V2_RUN_RESTART),
+        ),
     }
 )
 
@@ -302,3 +311,4 @@ def name(self, stage: Stage) -> str:
 STAGE12_SIMULATION_STAGES = RUN_STAGE_REGISTRY[RunConfiguration.STAGE12_SIMULATION]
 STAGE12_SEGMENT_STAGES = RUN_STAGE_REGISTRY[RunConfiguration.STAGE12_SEGMENT]
 V2_HOUSEHOLD_STAGES = RUN_STAGE_REGISTRY[RunConfiguration.V2_HOUSEHOLD]
+V2_RUN_RESTART_STAGES = RUN_STAGE_REGISTRY[RunConfiguration.V2_RUN_RESTART]

libs/policyengine-simulation-observability/tests/test_stages.py

@@ -1,6 +1,7 @@
 import pytest
 from policyengine_simulation_observability.stages import (
     RUN_STAGE_REGISTRY,
+    V2_RUN_RESTART_STAGES,
     RunConfiguration,
     Stage,
 )
@@ -66,3 +67,13 @@ def test_v2_household_plan_measures_model_load_inputs_and_each_policy() -> None:
         Stage.V2_HOUSEHOLD_CALCULATION,
     )
     assert household.name(Stage.V2_HOUSEHOLD_CALCULATION) == "v2_household_calculation"
+
+
+def test_v2_run_restart_plan_measures_each_check_and_each_restart() -> None:
+    """The scheduled check is one operation; each row it lists is one more."""
+    restart = RUN_STAGE_REGISTRY[RunConfiguration.V2_RUN_RESTART]
+
+    assert restart is V2_RUN_RESTART_STAGES
+    assert restart.stages == (Stage.V2_RUN_RESTART_CHECK, Stage.V2_RUN_RESTART)
+    assert restart.name(Stage.V2_RUN_RESTART_CHECK) == "v2_run_restart_check"
+    assert restart.name(Stage.V2_RUN_RESTART) == "v2_run_restart"

libs/policyengine-simulation-persistence/README.md

@@ -125,6 +125,37 @@ commit fails after its rows were inserted raises
 `EconomyRecordOutcomeUnknownError`: PostgreSQL may have applied the commit,
 so the rows may exist.
 
+## Restarting runs whose call never started
+
+`run_restart_records.py` holds the restart check's reads and writes. The
+check runs as the Modal worker role, whose grants cover them: `SELECT` on
+the run tables and their inputs, and `UPDATE` of `updated_at`,
+`modal_call_id`, `status`, `error_message` and `completed_at`.
+
+- `list_unstarted_runs` returns the oldest rows, up to a limit, created in a
+  window that are still `pending` with no Modal call and whose `updated_at`
+  still equals `created_at`: household jobs, report runs outside a budget
+  window whose baseline is an economy simulation, and runs of economy
+  simulations that no report run names (standalone runs).
+- `mark_restart` sets a listed row's `updated_at` to the statement's clock
+  while it is still unstarted, so it is listed and marked only once.
+- `read_unstarted_report_run` and `read_unstarted_simulation_run` read back
+  what the Simulation Entrypoint built a call from: each side's simulation
+  run, policy and dynamic, with the values read against the run's own
+  catalog version, the region, dataset and year the run records, the
+  report's options and the run's versions. `StoredRunUnreadableError` says
+  the rows no longer describe one calculation: a side's region or dataset
+  is gone, or the two sides differ. The values are read with
+  `read_simulation_reform`, the reader a submission uses.
+- `read_report_run_simulation_run_ids` returns the simulation runs a report
+  run names, so one that can't be rebuilt is failed with them, and
+  `read_run_call` returns a row's status and recorded call, which says whose
+  call holds it after a restart.
+
+`RunRestartStore` runs each in a transaction of its own. The check records
+a call, or fails a row, with `EconomyRunStore`'s and `HouseholdJobStore`'s
+own conditional writes.
+
 ## Reading a policy
 
 `read_policy_reform` (`policy_reform.py`) reads a stored policy as the reform

libs/policyengine-simulation-persistence/src/policyengine_simulation_persistence/__init__.py

@@ -57,6 +57,7 @@
     SimulationInputs,
     UnknownDatasetError,
     UnknownRegionError,
+    read_simulation_reform,
 )
 from policyengine_simulation_persistence.engine import (
     bounded_transaction,
@@ -92,6 +93,24 @@
     read_dynamic_reform,
     read_policy_reform,
 )
+from policyengine_simulation_persistence.run_restart_records import (
+    RESTART_TABLES,
+    EconomyScope,
+    RestartKind,
+    RunCall,
+    StoredEconomySide,
+    StoredReportRun,
+    StoredRunUnreadableError,
+    StoredSimulationRun,
+    UnstartedRun,
+    list_unstarted_runs,
+    mark_restart,
+    read_report_run_simulation_run_ids,
+    read_run_call,
+    read_unstarted_report_run,
+    read_unstarted_simulation_run,
+)
+from policyengine_simulation_persistence.run_restart_store import RunRestartStore
 from policyengine_simulation_persistence.tables import (
     LATEST_YEAR,
     MAX_BUDGET_WINDOW_YEARS,
@@ -113,6 +132,7 @@
 __all__ = [
     "LATEST_YEAR",
     "MAX_BUDGET_WINDOW_YEARS",
+    "RESTART_TABLES",
     "SERVICE_TABLE_PRIVILEGES",
     "SERVICE_UPDATABLE_COLUMNS",
     "BudgetWindowInputs",
@@ -127,6 +147,7 @@
     "EconomyReportState",
     "EconomyRequest",
     "EconomyRunStore",
+    "EconomyScope",
     "EconomySimulationInputs",
     "EconomySimulationState",
     "HouseholdCalculationInputs",
@@ -147,7 +168,10 @@
     "ReportSide",
     "ReportSideError",
     "ReportSimulation",
+    "RestartKind",
     "ReusableRun",
+    "RunCall",
+    "RunRestartStore",
     "RunReuse",
     "RunVersions",
     "SideRefusal",
@@ -156,13 +180,18 @@
     "SimulationRunState",
     "SimulationStatus",
     "SimulationType",
+    "StoredEconomySide",
     "StoredHousehold",
+    "StoredReportRun",
+    "StoredRunUnreadableError",
+    "StoredSimulationRun",
     "UnknownCatalogVersionError",
     "UnknownDatasetError",
     "UnknownDynamicError",
     "UnknownHouseholdError",
     "UnknownPolicyError",
     "UnknownRegionError",
+    "UnstartedRun",
     "bounded_transaction",
     "budget_window_runs",
     "combine_reforms",
@@ -177,6 +206,8 @@
     "insert_economy_report",
     "insert_economy_simulation",
     "insert_household_job",
+    "list_unstarted_runs",
+    "mark_restart",
     "missing_table_privileges",
     "normalized_database_url",
     "read_catalog_version_id",
@@ -190,7 +221,12 @@
     "read_model_id",
     "read_policy_reform",
     "read_region",
+    "read_report_run_simulation_run_ids",
+    "read_run_call",
+    "read_simulation_reform",
     "read_stored_household",
+    "read_unstarted_report_run",
+    "read_unstarted_simulation_run",
     "record_household_call",
     "record_report_run_call",
     "record_simulation_run_call",

libs/policyengine-simulation-persistence/src/policyengine_simulation_persistence/economy_run_store.py

@@ -529,6 +529,29 @@ def _simulation_inputs(
     choice: SimulationChoice,
 ) -> SimulationInputs:
     """Read one simulation's policy and dynamic values, combined."""
+    return SimulationInputs(
+        choice=choice,
+        reform=read_simulation_reform(
+            connection,
+            country=country,
+            catalog_version_id=catalog_version_id,
+            choice=choice,
+        ),
+    )
+
+
+def read_simulation_reform(
+    connection: Connection,
+    *,
+    country: str,
+    catalog_version_id: UUID,
+    choice: SimulationChoice,
+) -> PolicyReform:
+    """Read a simulation's policy and dynamic values on a catalog version, combined.
+
+    A submission and a restart of its run read them this way. Raise the
+    policy reader's errors, or OverlappingReformError.
+    """
     policy_reform = (
         {}
         if choice.policy_id is None
@@ -548,9 +571,7 @@ def _simulation_inputs(
             catalog_version_id=catalog_version_id,
         )
     )
-    return SimulationInputs(
-        choice=choice, reform=combine_reforms(policy_reform, dynamic_reform)
-    )
+    return combine_reforms(policy_reform, dynamic_reform)
 
 
 def _side_inputs(