From 5e466efe1f59e9ca0ade8cb0637415de08843b73 Mon Sep 17 00:00:00 2001 From: cekees <613654+cekees@users.noreply.github.com> Date: Thu, 22 Oct 2026 23:06:42 +0200 Subject: [PATCH 1/3] Keep the .ymf current during the run, not only after a clean close The archive of record was assembled at close: the per-step grid metadata went into the HDF5 file as it was produced, and the document describing it was built from that only when close() ran. A run that was killed, hit a scheduler timeout or died on one rank therefore left an archive with the data in it and nothing describing the data -- recoverable, but only by a separate pass, which is what scripts/gatherArchives.py exists for. Master already holds this step's grids while allGatherIncremental runs, so the running document costs no extra read and no extra parse: appending is the same work the close-time assembly does once, done once per step. The file is written to .ymf.writing and renamed, so a reader sees either the previous step's document or this one, never a half-written file. A failed write is logged and swallowed. The authoritative document is still written at close from the HDF5, and a full disk at step 300 should not destroy the 300 steps already in the archive. It is not free, and the cost is measured rather than asserted: a current document is a whole document, so writing is O(steps) each step -- 0.6 ms at 10 steps, 6.8 ms at 100, 35 ms at 500, which is 8.6 s over a 500-step run and minutes over a 5,000-step one. Rather than guess a cadence policy, the cost is made visible: it is accumulated and reported at close, and a single write over half a second says so once and names the switch. Set writeIncrementalYmf = False to restore writing only at close. Nine tests. The one that matters runs the steps and never calls close(), then reads the document; its control does the same with the feature off and asserts there is no document at all, so the first is not a tautology. The structural one compares the running document against the one written at close: the first is appended in memory from master's grids, the second is re-assembled from the YAML in the HDF5, and they have to agree -- if they ever diverge, the archive of record depends on whether the run finished, which is the property this removes. Both layouts, one to three ranks. Co-Authored-By: Claude Opus 5 --- proteus/Archiver.py | 109 +++++++++++++++++++++++++ test/test_archive_parallel_metadata.py | 93 ++++++++++++++++++++- 2 files changed, 199 insertions(+), 3 deletions(-) diff --git a/proteus/Archiver.py b/proteus/Archiver.py index ced92758b0..ef04c98745 100644 --- a/proteus/Archiver.py +++ b/proteus/Archiver.py @@ -70,6 +70,11 @@ def __init__(self,dataDir,filename, self.readOnly = readOnly self.n_datasets = 0 self.archived_domain = None + #the running document, built step by step on master + self._live_domain = None + self._live_collections = 0 + self._liveYmfSeconds = 0.0 + self._liveYmfWarned = False self.global_sync = global_sync comm_world = self.comm.comm.tompi4py() self.tree=ElementTree( @@ -427,6 +432,90 @@ def _metadata_dataset_name(self, collection_name, index): """Name of the HDF5 dataset holding one collection's step metadata.""" return (collection_name + "_" + str(index)).replace(" ", "_") + #: Whether the ``.ymf`` is rewritten after every step, so that the + #: archive of record is current during the run rather than only after a + #: clean close. Set False to restore the write-at-close-only behaviour. + writeIncrementalYmf = True + + def _appendLiveStep(self, live_steps): + """Add one timestep to the running document, on master. + + Costs no HDF5 read and no YAML parse: :meth:`allGatherIncremental` + has the grids in hand already, so appending is the same work the + close-time assembly would do once, done once per step instead. + """ + from ymf.archive import (add_collection, add_spatial_step, + add_uniform_step, new_domain) + + for ci, name, t, grids in live_steps: + if self._live_domain is None: + self._live_domain = new_domain(name) + self._live_collections = 1 + elif ci >= self._live_collections: + add_collection(self._live_domain, name) + self._live_collections = ci + 1 + if self.global_sync: + g = grids[0] + add_uniform_step(self._live_domain, t, g["topology"], + g["geometry"], g.get("attributes", []), + collection=ci) + else: + add_spatial_step(self._live_domain, t, grids, collection=ci) + + def _writeLiveYmf(self): + """Replace the ``.ymf`` with the document as it stands. + + Written to a temporary file and renamed, so the archive of record is + never a partially written file: a reader either sees the previous + step's document or this one. ``os.replace`` is atomic within a + directory, which is why the temporary sits beside the target rather + than in the system temp directory. + + A failure here is logged and swallowed. The incremental write is a + convenience -- the authoritative document is still written at close, + from the HDF5 file -- and a full disk at step 300 should not destroy + the 300 steps already in the archive. + + **It costs O(steps) per step**, since a current document is a whole + document. Measured on a one-collection archive, writing is 0.6 ms at + 10 steps, 6.8 ms at 100 and 35 ms at 500, so a 500-step run spends + about 8.6 s in total here and a 5,000-step run would spend minutes. + That is bought rather than assumed: the time is accumulated and + reported at close, and a single write over half a second says so once + and names the switch. Set :attr:`writeIncrementalYmf` to False if the + trade is wrong for a particular run. + """ + import time + + from ymf.archive import write_ymf + + if self._live_domain is None: + return + path = os.path.join(self.dataDir, self.filename + ".ymf") + tmp = path + ".writing" + try: + started = time.perf_counter() + write_ymf(self._live_domain, tmp) + os.replace(tmp, path) + elapsed = time.perf_counter() - started + self._liveYmfSeconds += elapsed + if elapsed > 0.5 and not self._liveYmfWarned: + self._liveYmfWarned = True + logEvent("Updating %s after each step took %.2f s at step %d; " + "the cost grows with the number of steps. Set " + "Archiver.AR_base.writeIncrementalYmf = False to " + "write it only at close." + % (os.path.basename(path), elapsed, self.n_datasets), + level=1) + except Exception as error: + logEvent("Could not update %s during the run (%s); the archive " + "is unaffected and the document will be written at close" + % (os.path.basename(path), error), level=1) + try: + os.remove(tmp) + except OSError: + pass + def gatherAndWriteTimes(self): """Assemble the whole archive document and write it out. @@ -452,6 +541,11 @@ def gatherAndWriteTimes(self): write_ymf(domain, ymf_path) logEvent("Wrote YMF archive %s (ymf2xmf %s writes an .xmf for viewers)" % (ymf_path, os.path.basename(ymf_path))) + if self._liveYmfSeconds: + logEvent("Keeping %s current during the run cost %.2f s over %d " + "steps" % (os.path.basename(ymf_path), + self._liveYmfSeconds, self.n_datasets), + level=2) def close(self): logEvent("Closing Archive") if self.comm.isMaster() and not self.readOnly: @@ -509,6 +603,7 @@ def allGatherIncremental(self): self.hdfFile.attrs[self.GLOBAL_SYNC_ATTR] = \ 1 if self.global_sync else 0 + live_steps = [] for i, TemporalGridCollection in enumerate(Domain): GridLocal = TemporalGridCollection[-1] time_elem = GridLocal.find("Time") @@ -545,7 +640,17 @@ def allGatherIncremental(self): if self.comm.isMaster(): for j, payload in enumerate(payloads): grid_data[j] = payload + if self.comm.isMaster() and self.hdfFile is not None: + #master already holds this step's grids, so the running + #document costs no extra read and no extra parse + live_grids = (grid_dicts if not self.global_sync + else [local_grid]) + live_steps.append((i, TemporalGridCollection.attrib['Name'], + float(TimeAttrib), live_grids)) self.n_datasets += 1 + if live_steps and self.writeIncrementalYmf: + self._appendLiveStep(live_steps) + self._writeLiveYmf() logEvent("Done Gathering Archive Time Step") def sync(self): logEvent("Syncing Archive",level=3) @@ -789,6 +894,10 @@ def readArchiveDomain(filename, dataDir='.'): archive.hdfFilename = filename + ".h5" archive.global_sync = True # overridden by the recorded value archive.archived_domain = None + archive._live_domain = None + archive._live_collections = 0 + archive._liveYmfSeconds = 0.0 + archive._liveYmfWarned = False with h5py.File(base + ".h5", "r") as hdfFile: archive.hdfFile = hdfFile domain = archive.assemble_domain() diff --git a/test/test_archive_parallel_metadata.py b/test/test_archive_parallel_metadata.py index de6abb4330..8272474b8e 100644 --- a/test/test_archive_parallel_metadata.py +++ b/test/test_archive_parallel_metadata.py @@ -74,6 +74,8 @@ def _find_mpiexec(): hot = os.environ.get("HOT_START") == "1" ar = Archiver.XdmfArchive(datadir, "mpitest", global_sync=global_sync, hotStart=hot) + if os.environ.get("NO_INCREMENTAL") == "1": + ar.writeIncrementalYmf = False if hot: # what NumericalSolution does: continue after the archive's last step ar.n_datasets = len(ar.archived_times) @@ -127,7 +129,13 @@ def _find_mpiexec(): dimensions=[n_nodes]) ar.sync() - ar.close() + if os.environ.get("SKIP_CLOSE") == "1": + # what a killed run leaves behind: the steps are synced, close() + # never happens. Flush so the HDF5 is as it would be, then stop. + if ar.hdfFile is not None: + ar.hdfFile.flush() + else: + ar.close() comm.barrier() if rank == 0: print("OBSERVED_SIZE=%d" % size) @@ -135,7 +143,8 @@ def _find_mpiexec(): ) -def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False): +def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False, + skip_close=False, incremental=True): """Run the rank script under mpiexec (or directly for one rank). ``hot_start`` continues the archive an earlier call wrote, for one @@ -143,7 +152,9 @@ def run_ranks(tmp_path, n_ranks, global_sync, hot_start=False): """ datadir = str(tmp_path / "archive") env = dict(os.environ, GLOBAL_SYNC="1" if global_sync else "0", - DATADIR=datadir, HOT_START="1" if hot_start else "0") + DATADIR=datadir, HOT_START="1" if hot_start else "0", + SKIP_CLOSE="1" if skip_close else "0", + NO_INCREMENTAL="0" if incremental else "1") # More ranks than cores (3 on a 2-core CI runner): Open MPI refuses # unless told it may oversubscribe -- OMPI_MCA_* for Open MPI 4, # PRTE_MCA_* for Open MPI 5's PRRTE launcher. MPICH oversubscribes by @@ -637,3 +648,79 @@ def test_a_parallel_hot_start_continues_the_archive(tmp_path): run_ranks(tmp_path, 2, global_sync=False, hot_start=True) (collection,) = read_domain(datadir)["time_collections"] assert [step["time"] for step in collection["data"]] == [0.0, 0.5, 1.0] + + +# -------------------------------------------------------------------------- +# the document is current during the run, not only after a clean close +# -------------------------------------------------------------------------- + + +def test_an_interrupted_run_still_leaves_a_readable_document(tmp_path): + """The point of the incremental write. + + A run that is killed, times out on a scheduler or dies on one rank never + reaches close(), and used to leave an archive with no .ymf at all: the + metadata was in the HDF5 file but the document that describes it was + assembled only at close. Recovering it meant a separate pass. Now the + document is already there and already correct. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + (collection,) = read_domain(datadir)["time_collections"] + assert [s["time"] for s in collection["data"]] == [0.0, 0.5] + + +def test_what_it_looked_like_before(tmp_path): + """The same interrupted run with the incremental write switched off. + + Kept as the control: without it there is no document, which is what + makes the test above worth having rather than a tautology. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True, + incremental=False) + assert not os.path.exists(os.path.join(datadir, "mpitest.ymf")) + assert os.path.exists(os.path.join(datadir, "mpitest.h5")) + + +def test_the_running_document_agrees_with_the_one_written_at_close(tmp_path): + """Two independent paths to the same document, so they can be compared. + + The running document is appended to in memory from the grids master + already holds; the one at close is re-assembled from the YAML in the + HDF5 file. They are built by different code from different sources, and + they have to agree -- if they ever diverge, the archive of record + depends on whether the run finished, which is exactly the property this + campaign exists to remove. + """ + running = read_domain(run_ranks(tmp_path / "a", 1, global_sync=False, + skip_close=True)) + closed = read_domain(run_ranks(tmp_path / "b", 1, global_sync=False)) + assert running == closed + + +@pytest.mark.parametrize("global_sync", [False, True]) +def test_the_running_document_agrees_for_both_layouts(tmp_path, global_sync): + running = read_domain(run_ranks(tmp_path / "a", 1, global_sync=global_sync, + skip_close=True)) + closed = read_domain(run_ranks(tmp_path / "b", 1, global_sync=global_sync)) + assert running == closed + + +@pytest.mark.skipif(MPIEXEC is None, reason="no MPI launcher available") +@pytest.mark.parametrize("n_ranks", [2, 3]) +def test_the_running_document_holds_every_ranks_grid(tmp_path, n_ranks): + """Only master writes it, but it must describe the whole decomposition.""" + datadir = run_ranks(tmp_path, n_ranks, global_sync=False, skip_close=True) + (collection,) = read_domain(datadir)["time_collections"] + assert len(collection["data"]) == 2 + for step in collection["data"]: + subs = step["spatial_collection"] + assert len(subs) == n_ranks + assert [g["name"] for g in subs] == [ + "Grid_p%d" % r for r in range(n_ranks)] + + +def test_no_leftover_temporary_file(tmp_path): + """The write is rename-based, so nothing may be left beside the target.""" + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + leftovers = [f for f in os.listdir(datadir) if f.endswith(".writing")] + assert leftovers == [] From d327be2c7b0b8024ae72a6ca90afabd504070ab4 Mon Sep 17 00:00:00 2001 From: cekees <613654+cekees@users.noreply.github.com> Date: Fri, 23 Oct 2026 05:16:48 +0200 Subject: [PATCH 2/3] Append each step to the .ymf instead of rewriting the document Review of the first version: rewriting the whole document to record a step is O(steps) per step, so O(steps^2) over a run, and many proteus runs have very large numbers of steps. That requirement is inherited from XML, where a document has to be closed to be well formed. YAML has no such rule, so a step can be appended and flushed. Measured on a one-collection archive, per-step write and cumulative: steps rewrite append 10 0.6 ms 0.10 ms 100 6.8 ms 0.11 ms 500 35.0 ms 0.09 ms 5000 ~350 ms 0.11 ms 5000 steps: ~15 min (extrapolated) vs 0.49 s Flat per step, as it should be: the write no longer depends on what the archive already holds. This needs ymf's append_step and its stream-aware read_ymf (cekees/ymf#18), which must merge first -- proteus CI installs ymf from @main. There the document grows as a stream: a header, then one document per step naming its collection. Naming rather than indexing matters because proteus creates collections lazily, one per archived FE space, so a model that first archives at step 40 adds one then. close() still writes the canonical single document from the HDF5, so a run that finishes leaves exactly what it left before; only an interrupted run is in the live shape, and read_ymf takes both. The cost accounting and the slow-write warning the first version added are gone with the cost they measured. Two tests pin the mechanism rather than the outcome: that the mid-run file really is a stream, one record per step with each one completed, and that close collapses it so nothing downstream needs to know the live shape exists. Co-Authored-By: Claude Opus 5 --- proteus/Archiver.py | 135 +++++++++---------------- test/test_archive_parallel_metadata.py | 23 +++++ 2 files changed, 70 insertions(+), 88 deletions(-) diff --git a/proteus/Archiver.py b/proteus/Archiver.py index ef04c98745..1bae0b12b6 100644 --- a/proteus/Archiver.py +++ b/proteus/Archiver.py @@ -70,10 +70,8 @@ def __init__(self,dataDir,filename, self.readOnly = readOnly self.n_datasets = 0 self.archived_domain = None - #the running document, built step by step on master - self._live_domain = None - self._live_collections = 0 - self._liveYmfSeconds = 0.0 + #the running document: appended to a step at a time, never rewritten + self._liveYmfStarted = False self._liveYmfWarned = False self.global_sync = global_sync comm_world = self.comm.comm.tompi4py() @@ -432,89 +430,57 @@ def _metadata_dataset_name(self, collection_name, index): """Name of the HDF5 dataset holding one collection's step metadata.""" return (collection_name + "_" + str(index)).replace(" ", "_") - #: Whether the ``.ymf`` is rewritten after every step, so that the - #: archive of record is current during the run rather than only after a - #: clean close. Set False to restore the write-at-close-only behaviour. + #: Whether each step is appended to the ``.ymf`` as it is produced, so + #: that the archive of record is current during the run rather than only + #: after a clean close. Appending is constant work per step -- see + #: :meth:`_appendLiveStep` -- so this is on by default. Set False to + #: restore the write-at-close-only behaviour. writeIncrementalYmf = True def _appendLiveStep(self, live_steps): - """Add one timestep to the running document, on master. - - Costs no HDF5 read and no YAML parse: :meth:`allGatherIncremental` - has the grids in hand already, so appending is the same work the - close-time assembly would do once, done once per step instead. + """Append this timestep to the running document. + + One record per collection, appended and flushed. Master already + holds the grids, so this costs no HDF5 read and no YAML parse, and + because it appends rather than rewrites it costs the same whether + the archive holds ten steps or ten thousand. + + The first call writes the header document -- the archive version and + an empty domain -- which the records are then appended to. The + header is written to a temporary file and renamed so that a reader + arriving mid-write never sees half of it; the records after it need + no such care, since each ends with the document-end marker that + tells a reader it is complete. + + Failures are logged and swallowed. The authoritative document is + still written at close from the HDF5 file, and the steps already in + the archive matter more than this one. """ - from ymf.archive import (add_collection, add_spatial_step, - add_uniform_step, new_domain) - - for ci, name, t, grids in live_steps: - if self._live_domain is None: - self._live_domain = new_domain(name) - self._live_collections = 1 - elif ci >= self._live_collections: - add_collection(self._live_domain, name) - self._live_collections = ci + 1 - if self.global_sync: - g = grids[0] - add_uniform_step(self._live_domain, t, g["topology"], - g["geometry"], g.get("attributes", []), - collection=ci) - else: - add_spatial_step(self._live_domain, t, grids, collection=ci) - - def _writeLiveYmf(self): - """Replace the ``.ymf`` with the document as it stands. - - Written to a temporary file and renamed, so the archive of record is - never a partially written file: a reader either sees the previous - step's document or this one. ``os.replace`` is atomic within a - directory, which is why the temporary sits beside the target rather - than in the system temp directory. - - A failure here is logged and swallowed. The incremental write is a - convenience -- the authoritative document is still written at close, - from the HDF5 file -- and a full disk at step 300 should not destroy - the 300 steps already in the archive. - - **It costs O(steps) per step**, since a current document is a whole - document. Measured on a one-collection archive, writing is 0.6 ms at - 10 steps, 6.8 ms at 100 and 35 ms at 500, so a 500-step run spends - about 8.6 s in total here and a 5,000-step run would spend minutes. - That is bought rather than assumed: the time is accumulated and - reported at close, and a single write over half a second says so once - and names the switch. Set :attr:`writeIncrementalYmf` to False if the - trade is wrong for a particular run. - """ - import time - - from ymf.archive import write_ymf + from ymf.archive import append_step, new_domain, write_ymf - if self._live_domain is None: - return path = os.path.join(self.dataDir, self.filename + ".ymf") - tmp = path + ".writing" try: - started = time.perf_counter() - write_ymf(self._live_domain, tmp) - os.replace(tmp, path) - elapsed = time.perf_counter() - started - self._liveYmfSeconds += elapsed - if elapsed > 0.5 and not self._liveYmfWarned: - self._liveYmfWarned = True - logEvent("Updating %s after each step took %.2f s at step %d; " - "the cost grows with the number of steps. Set " - "Archiver.AR_base.writeIncrementalYmf = False to " - "write it only at close." - % (os.path.basename(path), elapsed, self.n_datasets), - level=1) + if not self._liveYmfStarted: + tmp = path + ".writing" + write_ymf(new_domain(live_steps[0][1]), tmp) + os.replace(tmp, path) + self._liveYmfStarted = True + for _, name, t, grids in live_steps: + if self.global_sync: + g = grids[0] + step = {"time": t, "topology": g["topology"], + "geometry": g["geometry"], + "attributes": g.get("attributes", [])} + else: + step = {"time": t, "spatial_collection": grids} + append_step(path, name, step) except Exception as error: - logEvent("Could not update %s during the run (%s); the archive " - "is unaffected and the document will be written at close" - % (os.path.basename(path), error), level=1) - try: - os.remove(tmp) - except OSError: - pass + if not self._liveYmfWarned: + self._liveYmfWarned = True + logEvent("Could not update %s during the run (%s); the " + "archive is unaffected and the document will be " + "written at close" + % (os.path.basename(path), error), level=1) def gatherAndWriteTimes(self): """Assemble the whole archive document and write it out. @@ -541,11 +507,7 @@ def gatherAndWriteTimes(self): write_ymf(domain, ymf_path) logEvent("Wrote YMF archive %s (ymf2xmf %s writes an .xmf for viewers)" % (ymf_path, os.path.basename(ymf_path))) - if self._liveYmfSeconds: - logEvent("Keeping %s current during the run cost %.2f s over %d " - "steps" % (os.path.basename(ymf_path), - self._liveYmfSeconds, self.n_datasets), - level=2) + def close(self): logEvent("Closing Archive") if self.comm.isMaster() and not self.readOnly: @@ -650,7 +612,6 @@ def allGatherIncremental(self): self.n_datasets += 1 if live_steps and self.writeIncrementalYmf: self._appendLiveStep(live_steps) - self._writeLiveYmf() logEvent("Done Gathering Archive Time Step") def sync(self): logEvent("Syncing Archive",level=3) @@ -894,9 +855,7 @@ def readArchiveDomain(filename, dataDir='.'): archive.hdfFilename = filename + ".h5" archive.global_sync = True # overridden by the recorded value archive.archived_domain = None - archive._live_domain = None - archive._live_collections = 0 - archive._liveYmfSeconds = 0.0 + archive._liveYmfStarted = False archive._liveYmfWarned = False with h5py.File(base + ".h5", "r") as hdfFile: archive.hdfFile = hdfFile diff --git a/test/test_archive_parallel_metadata.py b/test/test_archive_parallel_metadata.py index 8272474b8e..1527a6f391 100644 --- a/test/test_archive_parallel_metadata.py +++ b/test/test_archive_parallel_metadata.py @@ -724,3 +724,26 @@ def test_no_leftover_temporary_file(tmp_path): datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) leftovers = [f for f in os.listdir(datadir) if f.endswith(".writing")] assert leftovers == [] + + +def test_the_running_document_is_appended_not_rewritten(tmp_path): + """The file is a stream mid-run, which is what makes the write O(1). + + Rewriting the whole document to record a step is O(steps) per step and + so quadratic over a run -- 35 ms a step at 500 steps, minutes over a + long one. Appending writes one record whatever the archive holds. The + observable difference is the shape of the file: a stream of documents + while the run is going, collapsed to a single document at close. + """ + datadir = run_ranks(tmp_path, 1, global_sync=False, skip_close=True) + text = open(os.path.join(datadir, "mpitest.ymf"), encoding="utf-8").read() + assert text.count("\n---\n") == 2 # one record per step + assert text.count("\n...\n") == 2 # each one completed + + +def test_close_leaves_a_single_document(tmp_path): + """A finished run is in the canonical shape, so nothing downstream has + to care that the live form exists.""" + datadir = run_ranks(tmp_path, 1, global_sync=False) + text = open(os.path.join(datadir, "mpitest.ymf"), encoding="utf-8").read() + assert "\n---\n" not in text From daa82d7cfd19100ab6f076086353e6f147cf85b3 Mon Sep 17 00:00:00 2001 From: cekees <613654+cekees@users.noreply.github.com> Date: Fri, 23 Oct 2026 09:34:06 +0200 Subject: [PATCH 3/3] Survive an ymf that predates append_step The import sat outside the guard, so a proteus newer than its ymf raised out of the middle of a timestep instead of losing its live document -- the one failure mode the writer was built to absorb. CI found it the honest way. Three legs of run 39894170430 resolved ymf@main at 03:18, five minutes before cekees/ymf#18 merged at 03:23; the macOS legs resolved it at 04:25 and passed. The provenance probe this workflow prints made that a two-minute diagnosis rather than a hunt, but the skew it exposed is real for anyone installing the two repositories separately. Co-Authored-By: Claude Opus 5 --- proteus/Archiver.py | 11 +++++----- test/test_archive_parallel_metadata.py | 29 ++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 5 deletions(-) diff --git a/proteus/Archiver.py b/proteus/Archiver.py index 1bae0b12b6..1b0187f18a 100644 --- a/proteus/Archiver.py +++ b/proteus/Archiver.py @@ -452,14 +452,15 @@ def _appendLiveStep(self, live_steps): no such care, since each ends with the document-end marker that tells a reader it is complete. - Failures are logged and swallowed. The authoritative document is - still written at close from the HDF5 file, and the steps already in - the archive matter more than this one. + Failures are logged and swallowed, the import among them: an ymf + too old to have ``append_step`` should cost the run its live + document, not the run. The authoritative document is still written + at close from the HDF5 file, and the steps already in the archive + matter more than this one. """ - from ymf.archive import append_step, new_domain, write_ymf - path = os.path.join(self.dataDir, self.filename + ".ymf") try: + from ymf.archive import append_step, new_domain, write_ymf if not self._liveYmfStarted: tmp = path + ".writing" write_ymf(new_domain(live_steps[0][1]), tmp) diff --git a/test/test_archive_parallel_metadata.py b/test/test_archive_parallel_metadata.py index 1527a6f391..e0d5a24fde 100644 --- a/test/test_archive_parallel_metadata.py +++ b/test/test_archive_parallel_metadata.py @@ -747,3 +747,32 @@ def test_close_leaves_a_single_document(tmp_path): datadir = run_ranks(tmp_path, 1, global_sync=False) text = open(os.path.join(datadir, "mpitest.ymf"), encoding="utf-8").read() assert "\n---\n" not in text + + +def test_an_ymf_without_append_step_costs_the_document_not_the_run(tmp_path): + """proteus and ymf move independently, so the skew has to be survivable. + + ``append_step`` arrived in ymf after this writer was written, and the + import of it used to sit outside the guard -- which turned "your live + document is stale" into an exception out of the middle of a timestep. + CI caught it honestly: three legs of one run resolved ``ymf@main`` five + minutes before the merge that added the function. Any user whose ymf + predates it would have hit the same thing. + """ + import types + import proteus.Archiver as AR + import ymf.archive + + live = types.SimpleNamespace(dataDir=str(tmp_path), filename="skew", + global_sync=True, _liveYmfStarted=False, + _liveYmfWarned=False) + step = [(0, "Grid", 0.0, [{"topology": {}, "geometry": {}}])] + + saved = ymf.archive.append_step + del ymf.archive.append_step + try: + AR.AR_base._appendLiveStep(live, step) # must not raise + finally: + ymf.archive.append_step = saved + + assert live._liveYmfWarned, "the failure has to be reported, not hidden"