Re-run a stored economy or budget-window report, or a simulation, on the current .py version

changelog.d/c1-17-budget-window-reruns.added.md

@@ -0,0 +1 @@
+`POST /reports/{report_id}/runs` also re-runs a stored budget-window report, as a new budget-window run with a report run per year on each year's own simulations, its first years started as a new window's are.

changelog.d/c1-17-reruns.added.md

@@ -0,0 +1 @@
+The Simulation Entrypoint re-runs a stored economy report or simulation on the current .py version through `POST /reports/{report_id}/runs` and `POST /simulations/{simulation_id}/runs`, as a new run started by run ID, keeping every earlier run. A report's re-run reuses a run of each side's own simulation that succeeded on the same versions unless `force` is set; a record with a run pending or running is refused with 409 `run_in_progress`.

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

@@ -334,6 +334,18 @@ as for a report comparing it with itself, so a report whose plan needs more
 columns, for example a reform that turns on labour supply responses, can't
 reuse it and fails with `reused_simulation_output_incompatible`.
 
+**A re-run.** The service re-runs a stored report or simulation with the same
+two calls, by run ID. It creates the record's new run (a report run with
+trigger `manual`) and, for a report, a new `simulation_runs` row of each side's
+own simulation, or names that simulation's run that already succeeded on the
+same versions, unless the re-run is forced. The new runs record the catalog
+version, dataset and region of the current .py version's catalog; the
+`simulations` and `reports` rows are unchanged. A budget-window report's
+re-run is a new `budget_window_runs` row (trigger `manual`) with a report run
+per year, each on its year's own simulations, started as a new window's
+first years are. The coordinator and workers claim and calculate them as
+they would a new record's: nothing on the Modal side tells a re-run apart.
+
 **Each worker** claims its simulation run under its own Modal call ID: a pending
 or running run that records no call or its own. It records its output file, or
 its failure, only while it holds the run running. It returns the output's

libs/policyengine-simulation-persistence/README.md

@@ -125,6 +125,49 @@ commit fails after its rows were inserted raises
 `EconomyRecordOutcomeUnknownError`: PostgreSQL may have applied the commit,
 so the rows may exist.
 
+## Re-running a report or simulation
+
+`economy_rerun_records.py` holds a re-run's reads and writes. A re-run
+calculates a stored record again on the current versions, as a new run, and
+changes no `reports` or `simulations` row, which the service's role may not
+update: a report keeps its simulations, and each new run records the catalog
+version, dataset and region it calculates on.
+
+- `read_report_definition` and `read_simulation_definition` read what the
+  record calculates: its kind (`economy`, `budget_window`, `household` or
+  `unsupported`), country, year, its region's code and dataset's name, and
+  each simulation's policy and dynamic.
+- `find_reusable_run_of` searches one simulation's runs, by its ID, under
+  `find_reusable_run`'s conditions on a run (versions, the run's own catalog
+  version, dataset, region and year, the output file, a succeeded report run
+  with the same options, no refused reuse); it doesn't compare the
+  simulation row, which keeps the catalog rows it was created on.
+- `insert_report_rerun` creates a report's new run, trigger `manual`, and a
+  new run of each side's simulation that reuses none;
+  `insert_simulation_rerun` creates a simulation's new run. Each refuses with
+  `RunInProgressError`, naming the run, while the record has a run `pending`
+  or `running` (for a report, a report run; for a simulation, any of its
+  runs), and first takes `lock_reruns_of`, a transaction-level
+  advisory lock on the record's ID, so concurrent re-runs of one record
+  wait for each other. When the database refuses the new run's key to the
+  record itself, the record was deleted after it was read:
+  `RerunRecordDeletedError`.
+
+A budget window's definition also names its newest budget-window run's
+years (`WindowYearDefinition`): each year's sides are the simulations of the
+runs that year's report run names. `insert_window_rerun` creates the
+window's new budget-window run (trigger `manual`, `recalculate` set), a
+report run per year and each year's new runs of its own simulations. Every
+report re-run is refused while a budget-window run or a report run of the
+report is `pending` or `running`.
+
+`EconomyRunStore.read_report_rerun_inputs` reads a re-run's catalog rows by
+the stored region code and dataset name, and each side's values and
+reusable run of its own simulation, never giving both sides the same run.
+`read_window_rerun_inputs` does the same for each year of a window.
+`create_report_rerun`, `create_window_rerun` and `create_simulation_rerun`
+run the inserts in a transaction of their own.
+
 ## 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

@@ -9,8 +9,30 @@
 from policyengine_simulation_persistence.budget_window_records import (
     BudgetWindowYear,
     NewBudgetWindowReport,
+    WindowSide,
+    WindowYearRuns,
     insert_budget_window_report,
 )
+from policyengine_simulation_persistence.economy_rerun_records import (
+    NewReportRerun,
+    NewSimulationRerun,
+    NewWindowRerun,
+    RecordKind,
+    ReportDefinition,
+    RerunRecordDeletedError,
+    RerunSide,
+    RunInProgressError,
+    SideDefinition,
+    SimulationDefinition,
+    WindowRerunYear,
+    WindowYearDefinition,
+    insert_report_rerun,
+    insert_simulation_rerun,
+    insert_window_rerun,
+    lock_reruns_of,
+    read_report_definition,
+    read_simulation_definition,
+)
 from policyengine_simulation_persistence.economy_run_reads import (
     EconomyReportState,
     EconomySimulationState,
@@ -33,6 +55,7 @@
     fail_unstarted_report_run,
     fail_unstarted_simulation_run,
     find_reusable_run,
+    find_reusable_run_of,
     insert_economy_report,
     insert_economy_simulation,
     read_dataset,
@@ -139,19 +162,29 @@
     "NewEconomyReport",
     "NewEconomySimulation",
     "NewHouseholdJob",
+    "NewReportRerun",
+    "NewSimulationRerun",
+    "NewWindowRerun",
     "OverlappingReformError",
     "PolicyReform",
+    "RecordKind",
+    "ReportDefinition",
     "ReportRunState",
     "ReportRunStatus",
     "ReportRunTrigger",
     "ReportSide",
     "ReportSideError",
     "ReportSimulation",
+    "RerunRecordDeletedError",
+    "RerunSide",
     "ReusableRun",
+    "RunInProgressError",
     "RunReuse",
     "RunVersions",
+    "SideDefinition",
     "SideRefusal",
     "SimulationChoice",
+    "SimulationDefinition",
     "SimulationInputs",
     "SimulationRunState",
     "SimulationStatus",
@@ -163,6 +196,10 @@
     "UnknownHouseholdError",
     "UnknownPolicyError",
     "UnknownRegionError",
+    "WindowRerunYear",
+    "WindowSide",
+    "WindowYearDefinition",
+    "WindowYearRuns",
     "bounded_transaction",
     "budget_window_runs",
     "combine_reforms",
@@ -171,12 +208,17 @@
     "fail_unstarted_report_run",
     "fail_unstarted_simulation_run",
     "find_reusable_run",
+    "find_reusable_run_of",
     "household_jobs",
     "input_metadata",
     "insert_budget_window_report",
     "insert_economy_report",
     "insert_economy_simulation",
     "insert_household_job",
+    "insert_report_rerun",
+    "insert_simulation_rerun",
+    "insert_window_rerun",
+    "lock_reruns_of",
     "missing_table_privileges",
     "normalized_database_url",
     "read_catalog_version_id",
@@ -190,6 +232,8 @@
     "read_model_id",
     "read_policy_reform",
     "read_region",
+    "read_report_definition",
+    "read_simulation_definition",
     "read_stored_household",
     "record_household_call",
     "record_report_run_call",

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

@@ -17,7 +17,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, replace
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol
 
 from sqlalchemy import insert
 
@@ -40,11 +40,59 @@
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
     from uuid import UUID
 
     from sqlalchemy import Connection
 
 
+class WindowSide(Protocol):
+    """A window year's side: its simulation and the run the year names."""
+
+    @property
+    def simulation_id(self) -> UUID:
+        """The side's simulation."""
+        ...
+
+    @property
+    def simulation_run_id(self) -> UUID:
+        """The run of it the year names."""
+        ...
+
+    @property
+    def reused(self) -> bool:
+        """Whether the run already succeeded, rather than being created."""
+        ...
+
+
+class WindowYearRuns(Protocol):
+    """A year of a window run: its year, its report run and each side.
+
+    A new window's ``BudgetWindowYear`` and a re-run's ``WindowRerunYear``
+    are both one.
+    """
+
+    @property
+    def year(self) -> int:
+        """The year."""
+        ...
+
+    @property
+    def report_run_id(self) -> UUID:
+        """The year's report run."""
+        ...
+
+    @property
+    def baseline(self) -> WindowSide:
+        """The baseline side."""
+        ...
+
+    @property
+    def reform(self) -> WindowSide:
+        """The reform side."""
+        ...
+
+
 @dataclass(frozen=True)
 class BudgetWindowYear:
     """One year of a window: its report run and its two sides' runs."""
@@ -78,20 +126,7 @@ class NewBudgetWindowReport:
 
     def __post_init__(self) -> None:
         """Refuse years that aren't 1 to 75 consecutive ones from the first."""
-        first_year = self.population.year
-        expected = tuple(range(first_year, first_year + len(self.years)))
-        if (
-            not self.years
-            or len(self.years) > MAX_BUDGET_WINDOW_YEARS
-            or tuple(year.year for year in self.years) != expected
-            or expected[-1] > LATEST_YEAR
-        ):
-            msg = (
-                "a budget window's years must be 1 to "
-                f"{MAX_BUDGET_WINDOW_YEARS} consecutive years from its first, "
-                f"ending by {LATEST_YEAR}"
-            )
-            raise ValueError(msg)
+        check_window_years(self.population.year, [year.year for year in self.years])
 
     @property
     def first_year(self) -> BudgetWindowYear:
@@ -103,6 +138,26 @@ def population_in(self, year: int) -> EconomyPopulation:
         return replace(self.population, year=year)
 
 
+def check_window_years(first_year: int, years: Sequence[int]) -> None:
+    """Raise ValueError unless ``years`` are 1 to 75 in a row from ``first_year``.
+
+    The last of them must also be no later than the tables' latest year.
+    """
+    expected = tuple(range(first_year, first_year + len(years)))
+    if (
+        not years
+        or len(years) > MAX_BUDGET_WINDOW_YEARS
+        or tuple(years) != expected
+        or expected[-1] > LATEST_YEAR
+    ):
+        msg = (
+            "a budget window's years must be 1 to "
+            f"{MAX_BUDGET_WINDOW_YEARS} consecutive years from its first, "
+            f"ending by {LATEST_YEAR}"
+        )
+        raise ValueError(msg)
+
+
 def insert_budget_window_report(
     connection: Connection, report: NewBudgetWindowReport
 ) -> None:
@@ -159,24 +214,42 @@ def insert_budget_window_report(
         )
         connection.execute(
             insert(report_runs),
-            [_year_run_row(report, window_year) for window_year in report.years],
+            [
+                year_run_row(
+                    window_year,
+                    report_id=report.id,
+                    window_run_id=report.window_run_id,
+                    versions=report.versions,
+                    trigger=ReportRunTrigger.INITIAL,
+                )
+                for window_year in report.years
+            ],
         )
 
 
-def _year_run_row(
-    report: NewBudgetWindowReport, window_year: BudgetWindowYear
+def year_run_row(
+    window_year: WindowYearRuns,
+    *,
+    report_id: UUID,
+    window_run_id: UUID,
+    versions: RunVersions,
+    trigger: ReportRunTrigger,
 ) -> dict[str, Any]:
-    """Return a year's report run, pending with no call, as inserted."""
+    """Return a year's report run in a window run, pending with no call, as inserted.
+
+    ``window_year`` is a new window's year or a re-run's; ``trigger`` says
+    which.
+    """
     return {
         "id": window_year.report_run_id,
-        "report_id": report.id,
+        "report_id": report_id,
         "baseline_simulation_run_id": window_year.baseline.simulation_run_id,
         "reform_simulation_run_id": window_year.reform.simulation_run_id,
-        "budget_window_run_id": report.window_run_id,
+        "budget_window_run_id": window_run_id,
         "year": window_year.year,
-        "country_package_version": report.versions.country_package_version,
-        "policyengine_version": report.versions.policyengine_version,
+        "country_package_version": versions.country_package_version,
+        "policyengine_version": versions.policyengine_version,
         "status": ReportRunStatus.PENDING,
-        "trigger": ReportRunTrigger.INITIAL,
+        "trigger": trigger,
         "modal_call_id": None,
     }