Stamp saved UK outputs with the year file they were calculated from

changelog.d/606.fixed.md

@@ -0,0 +1 @@
+A UK `Simulation.run()` now stamps its output with the policyengine-core and policyengine-uk versions that calculated it, an output format revision, and the stamp of the year file its input data was cut into, if any. A UK dataset read from a year file, or returned by `create_datasets`, keeps the file's stamp in `metadata`. `Simulation.load()` refuses a saved UK output, and `Simulation.ensure()` runs it again, unless that stamp is the one a run of the simulation's dataset would record now: when other package versions are installed, or the simulation's dataset is read from another year file or another cut of it. Once `ensure_datasets` has cut a UK year file again, because the bundle re-pinned its dataset, its local source changed or a package was upgraded, a simulation over the new cut no longer reuses outputs calculated from the old one. The error names the years, sources or versions that differ. A UK output is now written beside its path and moved into place with its stamp, so a save that fails leaves the previous output whole, and two processes saving to one path cannot leave one run's tables under another's stamp. Loading a UK year file or output checks that the stamp and the tables were read from one unchanged file, and reads again if another process replaced it meanwhile. `PolicyEngineUKDataset.load()` now raises `ValueError` for a dataset with no filepath. UK outputs saved before the stamp existed are calculated again once.

docs/microsim.md

@@ -351,11 +351,123 @@ are moved into place whole. Three more apply:
 - a multi-year source is read as stored, without the projection.
 
 A year file opened directly, as `PolicyEngineUKDataset(filepath=...)`, is not
-checked, and saving it again with `save()` drops its stamp. A saved UK
-simulation output records the data year its run anchored on;
-`Simulation.load()` refuses an output saved without one, and
-`Simulation.ensure()` runs it again. A saved UK output does not record the
-stamp of the year file it was calculated from.
+checked, and saving it again with `save()` drops its stamp. A dataset read
+from a year file, or returned by `create_datasets`, keeps the file's stamp as
+`metadata["year_file_stamp"]`, so an output calculated from it records the
+stamp (see below).
+
+Loading reads the stamp and the tables through separate opens of the file,
+and another process can move another cut onto the path in between. So
+loading holds the file open while it reads, and afterwards checks that the
+path still names that file, with the size and modification time it had
+before. If not, it reads again, and raises `RuntimeError` after three
+attempts. The stamp a dataset keeps is therefore the stamp of the cut its
+tables came from.
+
+### Saved UK outputs
+
+A UK `Simulation.run()` records a stamp in its output's
+`metadata["output_stamp"]`:
+
+```json
+{
+  "packages": {"policyengine-core": "3.33.0", "policyengine-uk": "2.126.12"},
+  "revision": 1,
+  "year_file": {
+    "packages": {"policyengine-core": "3.33.0", "policyengine-uk": "2.126.12"},
+    "revision": 1,
+    "source": {
+      "sha256": "5ba399ea5ab9a6c7d4e9aaf6c623080beb34d9ad5f330f6244dbad72b8d960b6",
+      "uri": "hf://policyengine/policyengine-uk-data-private/enhanced_frs_2024_25.h5@1.58.0"
+    },
+    "year": 2026
+  }
+}
+```
+
+- `packages`: the policyengine-core and policyengine-uk versions installed
+  when the run started, which calculated it;
+- `revision`: the UK output format revision;
+- `year_file`: the stamp the input dataset keeps of the year file it was cut
+  into, or `null` when it keeps none (a dataset built in memory, for
+  example). It names the tables the run read, and the versions that cut
+  them. Those need not be the versions that calculated the run: a year file
+  opened directly after an upgrade is run with the new ones.
+
+`save()` writes the tables, the data year and then the stamp to a temporary
+file beside the output's path (`.{id}.…partial`), and moves that onto the
+path in one step, as `create_datasets` writes a year file. The tables and the
+stamp at the path therefore always come from one save, even when two
+processes save an output to the same path, and a save that fails leaves the
+output already there as it was. The stamp is stored as UTF-8 JSON in
+`uk.datasets.OUTPUT_STAMP_H5_DATASET` (`policyengine_uk_output_stamp`), and
+`load()` restores it. UK year files and US outputs store their stamps under
+other names, so none passes as another.
+
+`Simulation.load()` refuses a saved UK output unless its stamp is the one a
+run of the simulation's dataset would record now,
+`uk.datasets.output_stamp(uk.datasets.input_year_file_stamp(simulation.dataset))`.
+`Simulation.ensure()` then runs the simulation again and saves the new
+output. `uk.datasets.output_stamp_problem(path, year_file)` returns the
+reason, or `None` for a current output. It reads only the output's stamp and
+whether the output has a data year, never its tables or the year file. A
+saved UK output is out of date when:
+
+- it has no stamp: it was saved before the stamp existed, and is calculated
+  again once;
+- it has no data year. It was saved before runs anchored on the observed
+  data year, or not saved completely, and may have taken a projected year as
+  observed data and uprated the State Pension by CPI;
+- it was calculated with other policyengine-core or policyengine-uk versions
+  than are installed, whatever its input was;
+- it was calculated from another year file than the simulation's dataset was
+  read from, or from another cut of it: the two year-file stamps differ.
+  `ensure_datasets` cuts a year file again, with another stamp, when the
+  bundle re-pins its dataset, a local source is edited or a package is
+  upgraded, so an output calculated from the old cut is not reused for a
+  simulation over the new one. Nor is an output reused for another year's
+  file in the same folder, or between a dataset that keeps a stamp and one
+  that keeps none;
+- the stamp its dataset kept of the year file is unreadable or malformed.
+  Such a stamp names no cut, so the output is calculated again every time;
+- it has another output format revision, or its stamp cannot be read.
+
+The reason names the years, sources and package versions that differ. A
+missing output file raises `FileNotFoundError`, as before, so `ensure()` runs
+the simulation. An output file that cannot be opened for reasons unrelated
+to its contents (no permission, a directory, a lock held by another process)
+raises that error rather than counting as out of date.
+
+Loading checks the stamp and reads the tables through separate opens of the
+output, so it holds the file open and checks afterwards that the path still
+names it unchanged, as loading a year file does. If another process moved
+another output onto the path meanwhile, the check is made again on that
+output.
+
+The check has these limits:
+
+- it compares the output with the simulation's dataset, not with the source
+  requested now. A year file that is out of date but not yet cut again, and
+  opened directly, is run and its outputs are reused: they are what a run of
+  that file gives. `load_datasets` refuses such a year file, and
+  `ensure_datasets` cuts it again;
+- the stamp records policyengine-core and policyengine-uk, not
+  policyengine.py's own version. A change to how policyengine.py runs a UK
+  simulation is marked by raising the output format revision;
+- the versions are those installed when the run starts, read from the
+  installed distributions' metadata. As for year files, restart the process
+  after upgrading a package;
+- a dataset built in memory, and a year file cut before the stamp existed
+  and opened directly, keep no stamp. Outputs calculated from them record
+  `null`, so two such datasets saved under one simulation id in one folder
+  are not told apart;
+- editing a dataset's tables in memory does not change the stamp it keeps;
+- within one process, `ensure()` returns the output it already holds in
+  memory for a simulation id without checking its stamp;
+- as for year files, moving an output into place replaces the file at its
+  path rather than rewriting it, so the save needs write access to the
+  folder, and a process killed mid-save can leave a temporary file behind,
+  which is never read and can be deleted.
 
 ## Simulations
 

src/policyengine/provenance/year_file_stamp.py

@@ -27,18 +27,23 @@
 - each requested dataset's year-file stem and source identity, refusing two
   sources that would share year files (:func:`year_file_sources`);
 - writing a year file beside its path and moving it into place only once it
-  is stamped (:func:`temporary_year_file`, :func:`publish_year_file`).
+  is stamped (:func:`temporary_year_file`, :func:`publish_year_file`);
+- reading a file's stamp and its tables from one file, though each is read
+  through its own open of the path (:func:`read_one_file`).
+
+A saved UK simulation output is written, stamped and read the same way (see
+``tax_benefit_models.uk.datasets.OUTPUT_STAMP_H5_DATASET``).
 """
 
 from __future__ import annotations
 
 import json
 import os
 import secrets
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from importlib import metadata as importlib_metadata
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Optional, TypeVar
 
 from .dataset_materialization import (
     describe_dataset_source_identity,
@@ -180,6 +185,57 @@ def publish_year_file(
     os.replace(temporary, path)
 
 
+_Read = TypeVar("_Read")
+
+
+def read_one_file(
+    path: str | Path, read: Callable[[str], _Read], attempts: int = 3
+) -> _Read:
+    """Return ``read(path)``, having checked that it read one unchanged file.
+
+    ``read`` may open ``path`` several times, for example once for a stamp
+    and once for the tables. Another process can move another file onto the
+    path in between (:func:`publish_year_file` does), so what one open reads
+    need not belong with what the next reads.
+
+    The file at ``path`` is therefore held open while ``read`` runs. After it
+    returns, ``path`` must still name that file (the same device and inode,
+    which the file system does not give to another file while this one is
+    held open), and the file must have the size and modification time it had
+    before. If not, ``read`` runs again, up to ``attempts`` times in all, and
+    then ``RuntimeError`` is raised. A file moved onto the path is another
+    file, even when its contents are those of an earlier one.
+
+    Two changes are not noticed: a file rewritten in place to the same size
+    within one tick of the file system's modification time, and a file
+    replaced and then put back by moving another hard link to it onto the
+    path. Neither ``create_datasets`` nor ``save_output`` writes that way:
+    each moves a newly written file onto the path.
+
+    Whatever ``read`` raises is raised, as is whatever opening ``path``
+    raises (``FileNotFoundError`` for a missing file).
+    """
+    path = str(path)
+    for _ in range(attempts):
+        with open(path, "rb") as held:
+            before = os.fstat(held.fileno())
+            result = read(path)
+            after = os.fstat(held.fileno())
+            try:
+                named = os.stat(path)
+            except FileNotFoundError:
+                continue
+        if os.path.samestat(named, before) and (
+            after.st_size,
+            after.st_mtime_ns,
+        ) == (before.st_size, before.st_mtime_ns):
+            return result
+    raise RuntimeError(
+        f"{path} was replaced or modified while it was read, {attempts} times; "
+        "read it again once it is no longer being written"
+    )
+
+
 def is_version_map(value: Any) -> bool:
     """Whether ``value`` maps package names to versions, ``None`` for absent."""
     return isinstance(value, dict) and all(
@@ -213,27 +269,48 @@ def source_and_package_differences(
             "requested source is "
             f"{describe_dataset_source_identity(current['source'])}"
         )
-    stamped_packages, packages = stamp["packages"], current["packages"]
-    for name in sorted(stamped_packages.keys() | packages.keys()):
+    reasons.extend(package_differences(stamp["packages"], current["packages"]))
+    return reasons
+
+
+def package_differences(
+    stamped: Mapping[str, Optional[str]],
+    packages: Mapping[str, Optional[str]],
+    done: str = "cut",
+) -> list[str]:
+    """Return the clauses naming the package versions a stamp differs in.
+
+    ``stamped`` is the version map a stamp records and ``packages`` the one a
+    stamp made now would (each as :func:`is_version_map` accepts). ``done``
+    says what the stamped file had done to it with those versions: a year
+    file was "cut" with them, an output "calculated". One clause, starting
+    "it", names each package whose version differs, in name order. A package
+    ``stamped`` records that ``packages`` does not is named too, and a
+    package recorded in both with one version is not.
+
+    Returns no clause exactly when the two maps are equal.
+    """
+    reasons = []
+    for name in sorted(stamped.keys() | packages.keys()):
         if name not in packages:
             reasons.append(
-                f"it records the {name} version it was cut with, which this "
+                f"it records the {name} version it was {done} with, which this "
                 "version of policyengine.py does not check"
             )
             continue
-        if name in stamped_packages and stamped_packages[name] == packages[name]:
+        if name in stamped and stamped[name] == packages[name]:
             continue
-        cut_with = (
-            f"{name} {stamped_packages[name]}"
-            if stamped_packages.get(name) is not None
+        done_with = (
+            f"{name} {stamped[name]}"
+            if stamped.get(name) is not None
             else f"no recorded {name} version"
         )
         installed = (
             f"{name} {packages[name]} is installed"
             if packages[name] is not None
             else f"{name} is not installed"
         )
-        reasons.append(f"it was cut with {cut_with}, and {installed}")
+        reasons.append(f"it was {done} with {done_with}, and {installed}")
     return reasons
 
 

src/policyengine/tax_benefit_models/common/model_version.py

@@ -388,6 +388,28 @@ def save(self, simulation: Simulation) -> None:
                     "inputs were mapped when it was calculated; run again "
                     "before saving"
                 )
+        if self.country_code == "uk":
+            from policyengine.tax_benefit_models.uk.datasets import (
+                OUTPUT_STAMP_KEY,
+                save_output,
+            )
+
+            # The stamp the run recorded, not one made now: the year file the
+            # output was calculated from, and the package versions it was
+            # calculated with.
+            output: Any = simulation.output_dataset
+            uk_stamp = (getattr(output, "metadata", None) or {}).get(OUTPUT_STAMP_KEY)
+            if uk_stamp is None:
+                raise ValueError(
+                    "This UK output does not record what it was calculated "
+                    "from and with; run again before saving"
+                )
+            # The tables and the stamp reach the output's path in one step,
+            # so neither a save that stops part-way nor another process
+            # saving to the same path leaves one run's tables under another
+            # run's stamp.
+            save_output(output, uk_stamp)
+            return
         simulation.output_dataset.save()
         if serialized_spm is not None:
             # Store UTF-8 JSON in a dataset rather than an attribute: the
@@ -443,29 +465,63 @@ def load(self, simulation: Simulation) -> None:
                     "stored inputs (it has no record of them); run it again"
                 )
 
+        def hydrate(path: str, **kept: Any) -> Any:
+            return self._dataset_class(
+                id=simulation.id,
+                name=simulation.dataset.name,
+                description=simulation.dataset.description,
+                filepath=path,
+                year=simulation.dataset.year,
+                is_output_dataset=True,
+                **kept,
+            )
+
         if self.country_code == "uk":
+            from policyengine.provenance.year_file_stamp import read_one_file
             from policyengine.tax_benefit_models.uk.datasets import (
-                _year_file_records_data_year,
+                OUTPUT_STAMP_KEY,
+                input_year_file_stamp,
+                output_stamp,
+                output_stamp_problem,
             )
 
-            # UK outputs saved before runs anchored on the observed data year
-            # (PolicyEngine/policyengine.py#556) took a projected year as
-            # observed and uprated the State Pension by CPI, so they are not
-            # reused. ``Simulation.ensure()`` runs such a simulation again.
-            if not _year_file_records_data_year(Path(filepath)):
-                raise ValueError(
-                    "Saved UK simulation predates anchoring on the observed "
-                    "data year (it records none); run it again"
+            # The stamp a run of the simulation's dataset would record now:
+            # the installed package versions and the year-file stamp the
+            # dataset keeps.
+            expected = output_stamp(input_year_file_stamp(simulation.dataset))
+
+            def read(path: str) -> Any:
+                # Outputs calculated with other package versions, from
+                # another year file than the simulation's dataset is read
+                # from, or from another cut of it (cut again after the
+                # bundle re-pinned its dataset, its local source changed or
+                # a package was upgraded), hold other results. So do outputs
+                # saved before runs anchored on the observed data year
+                # (PolicyEngine/policyengine.py#556), which took a projected
+                # year as observed and uprated the State Pension by CPI.
+                # None is reused: ``Simulation.ensure()`` runs such a
+                # simulation again. A missing file raises
+                # ``FileNotFoundError`` here.
+                problem = output_stamp_problem(
+                    path, expected["year_file"], expected["packages"]
                 )
-
-        simulation.output_dataset = self._dataset_class(
-            id=simulation.id,
-            name=simulation.dataset.name,
-            description=simulation.dataset.description,
-            filepath=filepath,
-            year=simulation.dataset.year,
-            is_output_dataset=True,
-        )
+                if problem is not None:
+                    raise ValueError(
+                        f"Saved UK simulation output {path} is out of date: "
+                        f"{problem}. Run the simulation again "
+                        "(Simulation.ensure() does so)."
+                    )
+                # The stored stamp equals ``expected``; kept so that saving
+                # the loaded output again writes it.
+                return hydrate(path, metadata={OUTPUT_STAMP_KEY: expected})
+
+            # The stamp is checked and the tables are read through separate
+            # opens of the file, and another process can move another output
+            # onto the path in between. ``read_one_file`` checks that both
+            # read one unchanged file, and reads again if not.
+            simulation.output_dataset = read_one_file(filepath, read)
+        else:
+            simulation.output_dataset = hydrate(filepath)
         if self.country_code == "us":
             from policyengine.core.spm import SPMSelection