From 7124233569a020e02ab3fdbf03909d650f68275c Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Tue, 1 Sep 2026 10:05:11 -0400 Subject: [PATCH 1/5] feat(orchestration): the keep list, and the script that acts on it persist_exp.py copies one exposure's named PSF products off /scratch onto the persistent root and records what went, with sizes. The threat it answers is the 60-day purge, not clean_exposure: run_dir is scratch and products_dir is /project, so the only way a per-exposure product outlives its campaign is to leave the filesystem. Exempting files from reclamation would not have done it. The search is recursive beneath the PSF chain's four module output dirs, because setools writes into mask/, rand_split/, new_cat/, plot/ and stat/ rather than flat -- so the config's patterns stay plain file names and the layout stays ours. A pattern that matches nothing is a recorded warning (setools rejects sparse CCDs); nothing matching at all is a failure, since a green manifest over an empty copy is what would let reclamation delete an unsaved exposure. config.yaml's persist_exp: defaults to validation_psf-*.fits -- the psfex_interp VALIDATION catalogue, the rho/tau statistics input, the minimum. The opt-in candidates are documented there with what each buys; sizes are still to be measured. PSFEx residuals and XML are not candidates as the chain stands: the committed default.psfex sets CHECKIMAGE_TYPE NONE and WRITE_XML N. Co-Authored-By: Claude Opus 5 --- workflow/config.yaml | 59 ++++++++++++ workflow/scripts/persist_exp.py | 156 ++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+) create mode 100644 workflow/scripts/persist_exp.py diff --git a/workflow/config.yaml b/workflow/config.yaml index 8e4de8506..e55714021 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -83,6 +83,65 @@ star_cats: /project/def-mjhudson/cdaley/sp-products/star-cat-cache # would otherwise have to rebuild from tile headers. index_db: /project/def-mjhudson/cdaley/sp-products/smk-g6/index/run_index.sqlite +# Per-exposure PSF products to COPY onto the persistent root before the scratch +# store goes (`exp_persist`, exposure.smk). A list of plain file-name globs, +# matched recursively under the PSF chain's four module output dirs +# (/exp///output/run_sp_exp_SxSePsfPi/*/output/ — +# sextractor_runner, setools_runner, psfex_runner, psfex_interp_runner). +# Matches land flat in /exp///psf/. +# +# WHY COPY RATHER THAN EXEMPT THESE FROM CLEANUP. Reclamation is not the threat. +# run_dir is /scratch and is PURGED on a 60-day window whether or not +# clean_exposure ever ran; products_dir is /project, backed up and not purged. +# The only way a per-exposure product outlives its campaign is to leave the +# filesystem. (Ordering is free: clean_exposure takes the exp_persist manifest +# as an input, so a store is never reclaimed before its keepers are written.) +# +# EDITING THIS LIST IS CHEAP. It rides on exp_persist's `params`, so a change +# reruns the copy (seconds) and NOT exp_psf (four hours per exposure). That +# separation is the whole reason exp_persist is a rule of its own. +# +# The default is the minimum: the psfex_interp VALIDATION catalogue, one per +# CCD, which is the input to the rho/tau statistics. Without it the PSF +# diagnostics cannot be recomputed after a purge without rebuilding the exposure +# chain from VOS. +# +# OPT-IN CANDIDATES, and what each buys (sizes to be measured): +# star_split_ratio_80-*.fits setools' 80% TRAINING star sample, the set PSFEx +# actually fitted. Refit or perturb the model. +# +# star_split_ratio_20-*.fits the 20% VALIDATION sample — the positions the +# validation_psf rows correspond to, with the +# measured star shapes beside them. +# star_selection-*.fits the PRE-SPLIT selection (setools writes it under +# mask/). The only file that can answer "which +# stars were rejected, and why" — the split +# samples have already lost the rejects. +# +# star_stat-*.txt setools' per-CCD STAT block (star counts, +# stars/deg^2, FWHM mode and cuts, under stat/): +# the selection's summary without its catalogue. +# +# *.psf the PSFEx model itself. Keeping it means the PSF +# can be re-interpolated at ANY position later +# without rebuilding the exposure chain — the +# single most capability-adding entry here. +# +# psfex_cat*.cat PSFEx's own output catalogue (FITS_LDAC). +# +# PSFEx residual/check images and its XML diagnostics are NOT candidates as the +# chain stands: the committed default.psfex sets CHECKIMAGE_TYPE NONE and +# WRITE_XML N, so nothing is emitted to match. They are a config change first, +# a pattern second. +# +# NOTE ON products_dir DEFAULTING TO run_dir (a fixture or smoke test): the copy +# then lands beside the store on the same filesystem and buys nothing, and the +# manifest sits in the exposure's own manifests/ dir, which clean_exposure +# deletes wholesale — so a one-root run re-persists after every reclamation. +# Harmless, and exactly the pre-D5 behaviour a one-root run asks for. +persist_exp: + - validation_psf-*.fits + # Rolling exposure-store reclamation (D5). When true, the COMPUTE DAG grows one # `clean_exposure` job per exposure. It fires once every campaign tile that reads # that exposure has its vignets, deletes the exposure's store AND its manifests, diff --git a/workflow/scripts/persist_exp.py b/workflow/scripts/persist_exp.py new file mode 100644 index 000000000..1a17c40cb --- /dev/null +++ b/workflow/scripts/persist_exp.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""Copy ONE exposure's keepable PSF products off scratch, and record what went. + +Run as the shell of the in-DAG ``exp_persist`` rule, never by hand. + +WHY A COPY AND NOT AN EXEMPTION FROM CLEANUP. The obvious alternative — teach +``clean_exposure`` to spare these files — does not work, because reclamation is +not what threatens them. The exposure store lives on ``run_dir``, which is +/scratch: a 60-day purge takes everything there whether or not this workflow +ever cleaned it. ``products_dir`` is /project, backed up and not purged. So the +only way a per-exposure product outlives its campaign is to LEAVE THE +FILESYSTEM, and that is a copy. Reclamation ordering then falls out for free: +``clean_exposure`` takes this rule's manifest as an input, so the store is never +deleted before its keepers have been written elsewhere. + +WHY A SEPARATE RULE AND NOT A ``cp`` APPENDED TO ``exp_psf``. The list of what +to keep is a decision that will be revisited — rho statistics want one file +today, a residual study may want three tomorrow — and ``exp_psf`` is four hours +per exposure. The list rides on this rule's ``params``, so editing it makes +snakemake rerun THIS rule (seconds of cp) and leaves the PSF chain alone. Folded +into ``exp_psf``, the same edit would re-derive every PSF model in the campaign. + +WHAT IT SEARCHES. ``/output/run_sp_exp_SxSePsfPi/*/output/`` — the four +module output dirs of the PSF config (sextractor, setools, psfex, psfex_interp) +— RECURSIVELY. The recursion is not laziness: setools does not write flat, it +writes into ``mask/``, ``rand_split/``, ``new_cat/``, ``plot/`` and ``stat/`` +beneath its own output dir, so a caller who wrote ``star_split_ratio_80-*.fits`` +meaning "the training star sample" would match nothing under a non-recursive +glob. Patterns are therefore plain FILE names and the layout is ours to know, +not the config author's. + +ZERO MATCHES FOR ONE PATTERN IS A WARNING, NOT A FAILURE. setools rejects sparse +CCDs (~0.2% attrition, tolerated by exp_psf's own count floor), so per-CCD +counts are not fixed, and a pattern naming an optional diagnostic may legitimately +find nothing. ZERO FILES IN TOTAL IS A FAILURE: it means the store was not what +we think it is, and writing a green manifest over that would let +``clean_exposure`` delete an exposure whose products were never saved. + +The destination is FLAT — one ``psf/`` dir per exposure, no module subtree — +because the module a file came from is already in its name and the consumer +(rho/tau statistics) globs the directory. A name collision between two modules +is therefore a hard error rather than a silent overwrite; nothing in the current +config can produce one, and if a future one can we want to hear about it. + +The manifest is the rule's ONLY declared output, and it lives on the persistent +root beside the copies (``/exp///manifests/``), NOT in +the exposure's scratch ``manifests/`` dir which ``clean_exposure`` deletes +wholesale. It is deliberately NOT a ``directory()`` output: what was copied, and +how big each file was, is provenance we want written down, and a directory +output attests only that some directory exists. + +It carries no timestamp and is written tmp-then-``cmp``-then-``mv`` (the pattern +``exp_star_cat`` uses), so a rerun that copies the same files leaves the mtime +alone — mtime is a rerun trigger, and an unconditional rewrite would make every +downstream ``clean_exposure`` look out of date once per invocation. +""" + +import argparse +import filecmp +import json +import shutil +import sys +from pathlib import Path + +# The PSF chain's run dir (RUN_NAME in config_exp_psfex.ini). Hardcoded rather +# than passed: this rule persists the PSF stage's products and nothing else, and +# a knob here would be a knob for "persist some other stage", which is a +# different rule. +RUN_NAME = "run_sp_exp_SxSePsfPi" + + +def collect(exp_dir: Path, patterns: list) -> tuple: + """Matched files per pattern, in a stable order, plus the empty patterns.""" + root = exp_dir / "output" / RUN_NAME + found, empty = {}, [] + for pat in patterns: + # One glob per module output dir, recursive beneath it (see the module + # docstring on setools' subdirectories). sorted() over the union keeps + # the manifest byte-stable across filesystem readdir order. + hits = sorted({p for mod in sorted(root.glob("*/output")) + for p in mod.rglob(pat) if p.is_file()}) + if hits: + found[pat] = hits + else: + empty.append(pat) + return found, empty + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--exp-dir", required=True, type=Path, + help="the exposure's scratch store") + p.add_argument("--exp", required=True) + p.add_argument("--dest", required=True, type=Path, + help="/exp///psf") + p.add_argument("--manifest", required=True, type=Path) + p.add_argument("--pattern", action="append", default=[], + help="repeatable; a plain file-name glob") + args = p.parse_args() + + if not args.pattern: + sys.exit("persist_exp: no --pattern given (config persist_exp is empty)") + + found, empty = collect(args.exp_dir, args.pattern) + if not found: + sys.exit(f"persist_exp: {args.exp}: no file matched any of " + f"{args.pattern} under {args.exp_dir}/output/{RUN_NAME}") + + args.dest.mkdir(parents=True, exist_ok=True) + seen, files = {}, [] + for pat, hits in found.items(): + for src in hits: + dst = args.dest / src.name + if src.name in seen: + sys.exit(f"persist_exp: {args.exp}: two source files are both " + f"named {src.name} ({seen[src.name]} and {src}); the " + f"destination is flat, so this would silently overwrite") + seen[src.name] = src + # Skip a byte-identical copy: it is not just an I/O saving, it keeps + # the destination's mtimes still for anything downstream that reads + # them. + if not (dst.exists() and filecmp.cmp(src, dst, shallow=False)): + tmp = dst.with_name(dst.name + ".tmp") + shutil.copy2(src, tmp) + tmp.replace(dst) # atomic: no half-copied product + files.append({"name": src.name, "pattern": pat, + "src": str(src), "bytes": dst.stat().st_size}) + + body = { + "stage": "exp_persist", "level": "exp", "unit": args.exp, + "status": "complete", + "dest": str(args.dest), + "patterns": list(args.pattern), + # The warning the docstring argues for: named patterns that matched + # nothing. Present as a key even when empty, so a reader never has to + # wonder whether an old manifest predates the field. + "patterns_unmatched": empty, + "n_files": len(files), + "bytes": sum(f["bytes"] for f in files), + "files": sorted(files, key=lambda f: f["name"]), + } + args.manifest.parent.mkdir(parents=True, exist_ok=True) + tmp = args.manifest.with_name(args.manifest.name + ".tmp") + tmp.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n") + if args.manifest.exists() and filecmp.cmp(tmp, args.manifest, shallow=False): + tmp.unlink() # unchanged: leave the mtime alone + else: + tmp.replace(args.manifest) + + warn = f" ({len(empty)} pattern(s) matched nothing: {empty})" if empty else "" + print(f"[persist_exp] {args.exp}: {len(files)} file(s), " + f"{body['bytes'] / 1e6:.1f} MB -> {args.dest}{warn}") + + +if __name__ == "__main__": + main() From f59ab5188a833130ad6fe197c59a59b08e013691 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Tue, 1 Sep 2026 10:05:25 -0400 Subject: [PATCH 2/5] feat(orchestration): exp_persist, between exp_psf and clean_exposure One rule per exposure, output = ONE manifest on the persistent root at /exp///manifests/exp_persist.json. Not a directory() output: what we want written down is which files were copied and how big each was, and a directory attests only that a directory exists. Byte-stable, so a no-op rerun does not move an mtime clean_exposure reads. Its own rule rather than a cp on the end of exp_psf, and that is the whole point: the keep list rides on params, so adding a pattern reruns seconds of copying instead of four hours of PSF fitting per exposure. A localrule, by the arithmetic that made exp_star_cat one -- a few MB of cp, ~20k of them at DR6 scale, each shorter than the scheduling latency that would submit it. The mid-chain grouping constraint does not bite: its neighbours are exp_psf (too heavy to fuse) and clean_exposure (local already). clean_exposure gains the manifest as an input, so a store is never reclaimed before its keepers have left scratch -- conditional only on there being a keep list, since "keep nothing" must not become a dependency on a rule that would fail for having nothing to copy. rule all requests the persist manifests DIRECTLY, not only through clean_exposure: the purge takes the store whether or not clean: is on, so hanging the copy off reclamation alone would lose everything in a clean:false campaign. Cleaned exposures are excluded -- their exp_psf manifest is gone, so asking would rebuild the chain from VOS, and a tombstone already means the copy happened. Co-Authored-By: Claude Opus 5 --- workflow/Snakefile | 63 +++++++++++++++++++++++++++++--- workflow/rules/exposure.smk | 72 +++++++++++++++++++++++++++++++++++-- 2 files changed, 129 insertions(+), 6 deletions(-) diff --git a/workflow/Snakefile b/workflow/Snakefile index b925adc06..7d2d222c6 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -258,6 +258,7 @@ EXP_DIR = str(RUN_DIR / "exp" / "{shard}" / "{exp}") # The persistent root mirrors the scratch one, shard for shard, so the two trees # read as the same campaign seen from two filesystems. PROD_TILE_DIR = str(PRODUCTS_DIR / "tiles" / "{shard}" / "{tile}") +PROD_EXP_DIR = str(PRODUCTS_DIR / "exp" / "{shard}" / "{exp}") def tile_dir(tile): return f"{RUN_DIR}/tiles/{tile[:2]}/{tile}" @@ -271,6 +272,18 @@ def tile_manifest(tile, stage): def exp_manifest(exp, stage): return f"{exp_dir(exp)}/manifests/{stage}.json" +def prod_exp_dir(exp): + """The exposure's dir on the PERSISTENT root — where exp_persist writes. + + Sharded identically to the scratch one, so the two trees read as the same + campaign seen from two filesystems, exposure side as well as tile side.""" + return f"{PRODUCTS_DIR}/exp/{exp[:2]}/{exp}" + +def prod_exp_manifest(exp, stage): + """A manifest that must SURVIVE reclamation, so it is not in the exposure's + scratch manifests/ dir (clean_exposure deletes that wholesale).""" + return f"{prod_exp_dir(exp)}/manifests/{stage}.json" + def forest_dir(tile): return f"{tile_dir(tile)}/exp_forest" @@ -309,6 +322,7 @@ SCRIPT_HASH = script_hash("completeness.py") FOREST_HASH = script_hash("build_forest.py") CLEAN_HASH = script_hash("clean_exposure.py") CLEAN_TILE_HASH = script_hash("clean_tile.py") +PERSIST_HASH = script_hash("persist_exp.py") # Same argument for star_cats.py, which both star-cat rules call: their params # otherwise fingerprint nothing but paths, so an edit to the chunking or the cut # would never rerun them. ONE hash for both rules because it is one script — and @@ -434,6 +448,40 @@ def clean_targets(): out.append(tombstone(exp)) return sorted(out) +# --- persisted exposure products (D5) -------------------------------------- +# The keep list is config, not a rule input, and it is READ HERE so that exactly +# one place converts it into the form the rule carries. An empty list is a +# deliberate "keep nothing" and produces no jobs at all. +PERSIST_EXP = list(config.get("persist_exp") or []) + + +def persist_targets(): + """Which exposures this invocation must copy PSF products off scratch for. + + `rule all` requests these DIRECTLY rather than reaching them only through + clean_exposure. Persistence and reclamation are different concerns — the + /scratch purge takes the store whether or not `clean:` is on — and hanging + the copy off the clean rule alone would mean a campaign run with clean:false + persists nothing and loses everything at the purge. + + Scope is the ready tiles' exposures, which `all` already builds through the + tile chain, so nothing new is pulled into the DAG by asking. + + EXCEPT A CLEANED EXPOSURE. Its exp_psf manifest was deleted by + clean_exposure, so requesting its persist manifest would make the DAG + rebuild the whole exposure chain from VOS — the avalanche tile.smk's + reclaimed-edge cut exists to prevent, arriving through a new target instead. + A tombstone means the copy already happened (clean_exposure cannot run + before exp_persist), so there is nothing to ask for. + + HEAD PROCESS ONLY, for the same reason as clean_targets() above. + """ + if not PERSIST_EXP or not workflow.is_main_process: + return [] + exps = {e for t in TILES_READY for e in tile_exposures(t)} + return sorted(prod_exp_manifest(e, "exp_persist") for e in exps + if not Path(tombstone(e)).exists()) + # --- tile reclamation (D5) -------------------------------------------------- # A SEPARATE FLAG from `clean:`, deliberately (config.yaml carries the full # argument): exposure reclamation costs nothing but a rebuild if a tile is @@ -589,22 +637,29 @@ include: "rules/exposure.smk" include: "rules/tile.smk" # --- top-level targets ------------------------------------------------------ -# The aggregation targets, clean_exposure, clean_tile, star_catalogue and -# exp_star_cat run in the head process. The two clean rules are seconds of rmtree +# The aggregation targets, clean_exposure, clean_tile, star_catalogue, +# exp_star_cat and exp_persist run in the head process. The two clean rules are seconds of rmtree # and hang off `all`; exp_star_cat is seconds of local FITS work; all three would # otherwise be ~20k (clean_exposure, exp_star_cat) or ~23k (clean_tile) sbatch # submissions at DR6 scale for work shorter than the scheduling latency. # star_catalogue is one job either way, and local keeps its CDS concurrency the # explicit number its thread pool sets (see exposure.smk). # +# exp_persist joins them for the same arithmetic — a few MB of `cp` per exposure, +# ~20k of them at DR6 scale, each far shorter than the scheduling latency that +# would submit it (exposure.smk argues the placement in full). +# # star_catalogue and exp_star_cat are MID-CHAIN localrules, so they must stay out # of any future `group:` label: a local job cannot be fused into a submitted group. -# The two clean rules are DAG leaves and have no such constraint. -localrules: all, prepare_all_tiles, clean_exposure, clean_tile, star_catalogue, exp_star_cat +# exp_persist sits between exp_psf and clean_exposure, both of which are outside +# every group already (exp_psf is heavy, clean_exposure is local), so it adds no +# new constraint. The two clean rules are DAG leaves and have none either. +localrules: all, prepare_all_tiles, clean_exposure, clean_tile, star_catalogue, exp_star_cat, exp_persist rule all: input: [final_cat(t) for t in TILES_READY], + persist_targets(), clean_targets(), clean_tile_targets(), diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk index 9fe3e7b29..c2a537891 100644 --- a/workflow/rules/exposure.smk +++ b/workflow/rules/exposure.smk @@ -1,6 +1,6 @@ """Exposure chain — per exposure, keyed by exp base id (dedup is structural). - exp_get_images -> exp_split -----> exp_mask -> exp_psf + exp_get_images -> exp_split -----> exp_mask -> exp_psf -> exp_persist -> exp_star_cat --/ star_catalogue ---------------/ @@ -12,6 +12,12 @@ reads fixed ``$SP_RUN/output/run_sp_exp_*`` INPUT_DIRs, so nothing resolves a run log. There is no `prepare_exposures` aggregation target: these chains hang off the compute DAG (`all` <- final_cat <- tile chain <- exposure manifests). +``exp_persist`` is the one rule here that writes to the PERSISTENT root: it +copies the PSF products named by `persist_exp:` off /scratch before the purge +(or clean_exposure) can take them. It is a separate rule from exp_psf precisely +so that editing that list costs a `cp` and not a four-hour refit; the full +argument is in workflow/scripts/persist_exp.py. + NO temp() anywhere in this file, ever (D5). Exposures overlap tiles by construction (~7-10 tiles each), so their consumer set closes over the CAMPAIGN, not over one invocation — reclamation here is clean_exposure's job (S5), driven @@ -315,6 +321,59 @@ rule exp_psf: sp_shell("exp_psf", "config_exp_psfex.ini") +# --- persistence (D5) ------------------------------------------------------- +# The counterpart of reclamation, and it must come first in the DAG: this copies +# the exposure's keepable PSF products onto the persistent root, and +# clean_exposure below takes its manifest as an input so the store is never +# reclaimed before the keepers have left /scratch. The purge would take them +# anyway — that, not clean_exposure, is what this rule exists for +# (persist_exp.py's docstring argues both halves, and config.yaml's +# `persist_exp:` block carries the keep list and its candidates). +# +# A LOCALRULE (declared in the Snakefile), by exactly the arithmetic that made +# exp_star_cat one: the body is `cp` of a few MB from one shared filesystem to +# another, seconds of work, and one sbatch per exposure would be ~20k +# submissions at DR6 scale for jobs shorter than the scheduling latency. The +# grouping constraint that binds mid-chain localrules (this file's docstring) +# does not bite here: exp_persist's only neighbours are exp_psf, which is too +# heavy to ever fuse, and clean_exposure, which is local itself. +# +# ONE DECLARED OUTPUT, AND IT IS A MANIFEST, NOT A directory(). The copies are +# not declared: a directory output would attest that a directory exists, where +# what we want written down is WHICH files were copied and how big each was — +# the provenance a rho-statistics run months from now needs in order to know +# what it is reading. The manifest is byte-stable, so a no-op rerun does not +# move its mtime and does not make clean_exposure look out of date. +# +# THE KEEP LIST RIDES ON params. That is the entire reason this is not three +# lines of cp appended to exp_psf's shell: `params` is a rerun trigger, so +# adding a pattern reruns the copy and leaves the PSF chain alone. +rule exp_persist: + input: + rules.exp_psf.output.manifest + output: + manifest = f"{PROD_EXP_DIR}/manifests/exp_persist.json" + # No `log:`: the script's only failure modes are "nothing matched" and a + # name collision, both of which it reports on stderr and neither of which + # has a per-CCD verdict worth a completeness record. + params: + patterns = " ".join(f"--pattern '{p}'" for p in PERSIST_EXP), + exp_dir = lambda wc: exp_dir(wc.exp), + dest = lambda wc: f"{prod_exp_dir(wc.exp)}/psf", + script_hash = PERSIST_HASH + threads: 1 + retries: 2 + resources: + mem_mb = 2000, + runtime = 10 + shell: + "set -euo pipefail\n" + f"python {SCRIPTS}/persist_exp.py" + " --exp-dir '{params.exp_dir}' --exp {wildcards.exp}" + " --dest '{params.dest}' --manifest {output.manifest}" + " {params.patterns}" + + # --- reclamation (D5) ------------------------------------------------------- # The one exception to "no reclamation in this file": clean_exposure OWNS # exposure-level deletion, and it is a real job, not temp() bookkeeping, because @@ -346,7 +405,16 @@ rule clean_exposure: # spatial neighbours. In-scope consumers keep their edge: they may run in # this DAG, so the clean must be ordered after them. lambda wc: [tile_manifest(t, "tile_vignets") - for t in clean_consumers(wc.exp) if t in READY_SET] + for t in clean_consumers(wc.exp) if t in READY_SET], + # The keepers must be off /scratch before the store goes. Unlike the + # consumer edges above, this edge does not depend on scope: it is the + # same exposure's own rule, so it drags nothing into the DAG that this + # exposure's chain did not already put there. It is conditional only on + # there being a keep list at all — with `persist_exp:` empty, "keep + # nothing" is a coherent instruction and must not become a dependency on + # a rule that would fail for having nothing to copy. + lambda wc: ([prod_exp_manifest(wc.exp, "exp_persist")] + if PERSIST_EXP else []) output: tombstone = f"{EXP_DIR}/cleaned.json" params: From e02309a9c32a68718d355117ef2f39f1d7cf4394 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Tue, 1 Sep 2026 10:05:25 -0400 Subject: [PATCH 3/5] docs(orchestration): exp_persist in the rule list, and why the report omits it run_report disk-scans the scratch run_dir, and exp_persist's manifest is the one exposure manifest that lives on products_dir instead -- the placement that makes it survive clean_exposure. Listed in EXP_STAGES it would read as "not run" for every exposure in the campaign, so it is deliberately absent, with the reason on the line. Co-Authored-By: Claude Opus 5 --- workflow/README.md | 17 ++++++++++++++++- workflow/scripts/run_report.py | 7 +++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/workflow/README.md b/workflow/README.md index 08bfd2291..1350c7953 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -154,7 +154,7 @@ workflow/ bin/sp committed launcher (module load + /project venv + launch code snapshot + run/report/container/cancel) rules/ prepare.smk tile get_images/uncompress/find_exposures - exposure.smk per-exposure: get_images, star_cat, split, mask, psf (no temp()) + exposure.smk per-exposure: get_images, star_cat, split, mask, psf, persist (no temp()) tile.smk per-tile: exp forest, merge_headers, mask, detect, vignets, ngmix, merge, make_cat scripts/ sp_rule.py the thin per-unit wrapper (isolation furniture, config copy, log-sync, count floor) @@ -163,6 +163,7 @@ workflow/ completeness.py the ported count-floor table (shared by sp_rule + run_report) run_report.py standalone report (NOT a DAG node; run_report hooks call it) container.py image layers + the resolution order behind `sp container` (stdlib-only) + persist_exp.py ONE exposure's keepable PSF products -> products_dir (the exp_persist rule) clean_exposure.py ONE exposure's store + manifests + logs -> tombstone (the clean_exposure rule) profiles/nibi/config.yaml SLURM executor; apptainer SDM; per-user jobs cap; keep-going ``` @@ -230,6 +231,20 @@ profiles/nibi/config.yaml SLURM executor; apptainer SDM; per-user jobs cap; kee trigger reads that cut as a reason to rerun the very tiles it protects. Know the consequence — `--forcerun` on a tile whose `final_cat` exists will not rebuild its reclaimed exposures. Delete the `final_cat` first. +- **PSF products leave scratch before the purge does.** `exp_persist` copies + the files named by `persist_exp:` in `config.yaml` (default: the psfex_interp + `validation_psf-*.fits`, the rho/tau statistics input) from the exposure's + scratch store into `/exp///psf/`, and writes ONE + manifest beside them recording the patterns, the files and their sizes. The + threat it answers is the /scratch purge, not `clean_exposure` — the store goes + in 60 days whether or not the workflow reclaimed it — so it runs even with + `clean: false`, requested directly by `rule all`. `clean_exposure` takes its + manifest as an input, so reclamation can never overtake the copy. It is a + rule of its own rather than a `cp` on the end of `exp_psf` because the keep + list rides on `params`: adding a pattern reruns seconds of copying, not four + hours of PSF fitting per exposure. A pattern that matches nothing is a + recorded warning (setools rejects sparse CCDs); matching nothing at all is a + failure. A `localrule`, like `exp_star_cat` and for the same arithmetic. - **A dead tile can be told to stop pinning exposures.** An exposure is cleanable only once every consuming tile has its vignets, so one permanently-failed tile holds its ~80 exposures for the life of the diff --git a/workflow/scripts/run_report.py b/workflow/scripts/run_report.py index f559e3276..0a48605b8 100644 --- a/workflow/scripts/run_report.py +++ b/workflow/scripts/run_report.py @@ -61,6 +61,13 @@ "tile_ngmix", "tile_merge_cats", "tile_make_cat"] EXP_STAGES = ["exp_get_images", "exp_star_cat", "exp_split", "exp_mask", "exp_psf"] +# exp_persist is DELIBERATELY NOT in that list. This report disk-scans the +# scratch run_dir, and exp_persist's manifest is the one exposure manifest that +# lives on products_dir instead — that placement is what makes it survive +# clean_exposure. Listed here it would read as "not run" for every exposure in +# the campaign. Reporting on the persisted products means scanning the second +# root, which is a report this one does not yet do. + # The manifests clean_tile leaves on disk (workflow/scripts/clean_tile.py names # the mechanism that owns each). Their presence is therefore NOT evidence that a # tile's chain was rebuilt, which is the one thing absorb_tombstones has to know From e47ee2f77a4e83449b9102fec430214ae2830ff4 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Wed, 2 Sep 2026 20:16:01 -0400 Subject: [PATCH 4/5] feat(orchestration): exp_persist packs one tar per exposure, not loose copies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inodes, not bytes, bind on /project (~1 M-file group quota): smk-m2 measured ~200 loose files per exposure with all candidates on — 25k for 64 tiles, ~2 M at DR6 scale, for 7 GB. persist_exp.py now writes /exp// /psf/.tar (uncompressed, flat members, deterministic: ownership zeroed, sorted, tmp-cmp-mv so a no-op rerun keeps the mtime) and the manifest lists every member. Manifest path, rule wiring and params are unchanged. config.yaml's candidate table carries the smk-m2 per-exposure sizes; psfex_cat and star_stat are marked unmeasured (no live store held them). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015qtLUV3bVLPV5p6un7aTFR --- workflow/README.md | 12 ++--- workflow/Snakefile | 4 +- workflow/config.yaml | 57 ++++++++++++++--------- workflow/rules/exposure.smk | 25 ++++++----- workflow/scripts/persist_exp.py | 80 ++++++++++++++++++++++----------- 5 files changed, 112 insertions(+), 66 deletions(-) diff --git a/workflow/README.md b/workflow/README.md index 1350c7953..e533ab725 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -163,7 +163,7 @@ workflow/ completeness.py the ported count-floor table (shared by sp_rule + run_report) run_report.py standalone report (NOT a DAG node; run_report hooks call it) container.py image layers + the resolution order behind `sp container` (stdlib-only) - persist_exp.py ONE exposure's keepable PSF products -> products_dir (the exp_persist rule) + persist_exp.py ONE exposure's keepable PSF products -> one tar on products_dir (the exp_persist rule) clean_exposure.py ONE exposure's store + manifests + logs -> tombstone (the clean_exposure rule) profiles/nibi/config.yaml SLURM executor; apptainer SDM; per-user jobs cap; keep-going ``` @@ -231,17 +231,19 @@ profiles/nibi/config.yaml SLURM executor; apptainer SDM; per-user jobs cap; kee trigger reads that cut as a reason to rerun the very tiles it protects. Know the consequence — `--forcerun` on a tile whose `final_cat` exists will not rebuild its reclaimed exposures. Delete the `final_cat` first. -- **PSF products leave scratch before the purge does.** `exp_persist` copies +- **PSF products leave scratch before the purge does.** `exp_persist` packs the files named by `persist_exp:` in `config.yaml` (default: the psfex_interp `validation_psf-*.fits`, the rho/tau statistics input) from the exposure's - scratch store into `/exp///psf/`, and writes ONE - manifest beside them recording the patterns, the files and their sizes. The + scratch store into ONE uncompressed tar, + `/exp///psf/.tar` (inodes, not bytes, bind + on /project), and writes ONE manifest beside it recording the patterns, the + members and their sizes. The threat it answers is the /scratch purge, not `clean_exposure` — the store goes in 60 days whether or not the workflow reclaimed it — so it runs even with `clean: false`, requested directly by `rule all`. `clean_exposure` takes its manifest as an input, so reclamation can never overtake the copy. It is a rule of its own rather than a `cp` on the end of `exp_psf` because the keep - list rides on `params`: adding a pattern reruns seconds of copying, not four + list rides on `params`: adding a pattern reruns seconds of packing, not four hours of PSF fitting per exposure. A pattern that matches nothing is a recorded warning (setools rejects sparse CCDs); matching nothing at all is a failure. A `localrule`, like `exp_star_cat` and for the same arithmetic. diff --git a/workflow/Snakefile b/workflow/Snakefile index 7d2d222c6..81e395951 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -456,7 +456,7 @@ PERSIST_EXP = list(config.get("persist_exp") or []) def persist_targets(): - """Which exposures this invocation must copy PSF products off scratch for. + """Which exposures this invocation must pack PSF products off scratch for. `rule all` requests these DIRECTLY rather than reaching them only through clean_exposure. Persistence and reclamation are different concerns — the @@ -645,7 +645,7 @@ include: "rules/tile.smk" # star_catalogue is one job either way, and local keeps its CDS concurrency the # explicit number its thread pool sets (see exposure.smk). # -# exp_persist joins them for the same arithmetic — a few MB of `cp` per exposure, +# exp_persist joins them for the same arithmetic — a few MB of `tar` per exposure, # ~20k of them at DR6 scale, each far shorter than the scheduling latency that # would submit it (exposure.smk argues the placement in full). # diff --git a/workflow/config.yaml b/workflow/config.yaml index e55714021..fd140f5b4 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -83,12 +83,17 @@ star_cats: /project/def-mjhudson/cdaley/sp-products/star-cat-cache # would otherwise have to rebuild from tile headers. index_db: /project/def-mjhudson/cdaley/sp-products/smk-g6/index/run_index.sqlite -# Per-exposure PSF products to COPY onto the persistent root before the scratch +# Per-exposure PSF products to carry onto the persistent root before the scratch # store goes (`exp_persist`, exposure.smk). A list of plain file-name globs, # matched recursively under the PSF chain's four module output dirs # (/exp///output/run_sp_exp_SxSePsfPi/*/output/ — # sextractor_runner, setools_runner, psfex_runner, psfex_interp_runner). -# Matches land flat in /exp///psf/. +# Matches are packed, flat, into ONE uncompressed tar per exposure: +# /exp///psf/.tar, with a manifest listing the +# members beside it. One tar rather than loose copies because inodes, not bytes, +# bind on /project (~1 M-file group quota; loose copies would be ~200 files per +# exposure, ~2 M at DR6 scale). FITS members read straight from the tar: +# fits.open(io.BytesIO(tarfile.open(t).extractfile(m).read())). # # WHY COPY RATHER THAN EXEMPT THESE FROM CLEANUP. Reclamation is not the threat. # run_dir is /scratch and is PURGED on a 60-day window whether or not @@ -98,7 +103,7 @@ index_db: /project/def-mjhudson/cdaley/sp-products/smk-g6/index/run_index.sqlite # as an input, so a store is never reclaimed before its keepers are written.) # # EDITING THIS LIST IS CHEAP. It rides on exp_persist's `params`, so a change -# reruns the copy (seconds) and NOT exp_psf (four hours per exposure). That +# reruns the packing (seconds) and NOT exp_psf (four hours per exposure). That # separation is the whole reason exp_persist is a rule of its own. # # The default is the minimum: the psfex_interp VALIDATION catalogue, one per @@ -106,35 +111,43 @@ index_db: /project/def-mjhudson/cdaley/sp-products/smk-g6/index/run_index.sqlite # diagnostics cannot be recomputed after a purge without rebuilding the exposure # chain from VOS. # -# OPT-IN CANDIDATES, and what each buys (sizes to be measured): -# star_split_ratio_80-*.fits setools' 80% TRAINING star sample, the set PSFEx -# actually fitted. Refit or perturb the model. -# -# star_split_ratio_20-*.fits the 20% VALIDATION sample — the positions the -# validation_psf rows correspond to, with the -# measured star shapes beside them. +# OPT-IN CANDIDATES, and what each buys. Sizes are per exposure (40 CCDs), +# measured on smk-m2 (127 exposures, 64 tiles); a 64-tile campaign with all of +# the measured ones on came to 7.2 GB: +# validation_psf-*.fits (the default) 2.0 MB +# *.psf the PSFEx model itself. Keeping it means the PSF +# can be re-interpolated at ANY position later +# without rebuilding the exposure chain — the +# single most capability-adding entry here. +# 2.8 MB +# psfex_cat-*.cat PSFEx's own output catalogue (FITS_LDAC): the +# per-star FLAGS_PSF / CHI2_PSF, i.e. WHICH stars +# outlier rejection clipped. Not recoverable from +# anything else (the .psf header keeps only the +# LOADED/ACCEPTED counts). unmeasured # star_selection-*.fits the PRE-SPLIT selection (setools writes it under # mask/). The only file that can answer "which -# stars were rejected, and why" — the split -# samples have already lost the rejects. -# +# stars were rejected by the selection cuts, and +# why" — the split samples have already lost the +# rejects. 24.5 MB +# star_split_ratio_80-*.fits setools' 80% TRAINING star sample, the set PSFEx +# actually fitted. Rows duplicate star_selection. +# 19.9 MB +# star_split_ratio_20-*.fits the 20% VALIDATION sample — the positions the +# validation_psf rows correspond to. Rows +# duplicate star_selection. 7.1 MB # star_stat-*.txt setools' per-CCD STAT block (star counts, # stars/deg^2, FWHM mode and cuts, under stat/): # the selection's summary without its catalogue. -# -# *.psf the PSFEx model itself. Keeping it means the PSF -# can be re-interpolated at ANY position later -# without rebuilding the exposure chain — the -# single most capability-adding entry here. -# -# psfex_cat*.cat PSFEx's own output catalogue (FITS_LDAC). -# +# unmeasured +# A production keep list is `validation_psf` + `*.psf` + `psfex_cat` (~5 MB per +# exposure); the star_split files are only worth it if star_selection is off. # PSFEx residual/check images and its XML diagnostics are NOT candidates as the # chain stands: the committed default.psfex sets CHECKIMAGE_TYPE NONE and # WRITE_XML N, so nothing is emitted to match. They are a config change first, # a pattern second. # -# NOTE ON products_dir DEFAULTING TO run_dir (a fixture or smoke test): the copy +# NOTE ON products_dir DEFAULTING TO run_dir (a fixture or smoke test): the tar # then lands beside the store on the same filesystem and buys nothing, and the # manifest sits in the exposure's own manifests/ dir, which clean_exposure # deletes wholesale — so a one-root run re-persists after every reclamation. diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk index c2a537891..d15bf64fa 100644 --- a/workflow/rules/exposure.smk +++ b/workflow/rules/exposure.smk @@ -13,9 +13,10 @@ run log. There is no `prepare_exposures` aggregation target: these chains hang off the compute DAG (`all` <- final_cat <- tile chain <- exposure manifests). ``exp_persist`` is the one rule here that writes to the PERSISTENT root: it -copies the PSF products named by `persist_exp:` off /scratch before the purge -(or clean_exposure) can take them. It is a separate rule from exp_psf precisely -so that editing that list costs a `cp` and not a four-hour refit; the full +packs the PSF products named by `persist_exp:` into one tar per exposure off +/scratch before the purge (or clean_exposure) can take them. It is a separate +rule from exp_psf precisely so that editing that list costs a `tar` and not a +four-hour refit; the full argument is in workflow/scripts/persist_exp.py. NO temp() anywhere in this file, ever (D5). Exposures overlap tiles by @@ -322,8 +323,8 @@ rule exp_psf: # --- persistence (D5) ------------------------------------------------------- -# The counterpart of reclamation, and it must come first in the DAG: this copies -# the exposure's keepable PSF products onto the persistent root, and +# The counterpart of reclamation, and it must come first in the DAG: this packs +# the exposure's keepable PSF products into one tar on the persistent root, and # clean_exposure below takes its manifest as an input so the store is never # reclaimed before the keepers have left /scratch. The purge would take them # anyway — that, not clean_exposure, is what this rule exists for @@ -331,23 +332,23 @@ rule exp_psf: # `persist_exp:` block carries the keep list and its candidates). # # A LOCALRULE (declared in the Snakefile), by exactly the arithmetic that made -# exp_star_cat one: the body is `cp` of a few MB from one shared filesystem to -# another, seconds of work, and one sbatch per exposure would be ~20k +# exp_star_cat one: the body is a `tar` of a few MB from one shared filesystem +# to another, seconds of work, and one sbatch per exposure would be ~20k # submissions at DR6 scale for jobs shorter than the scheduling latency. The # grouping constraint that binds mid-chain localrules (this file's docstring) # does not bite here: exp_persist's only neighbours are exp_psf, which is too # heavy to ever fuse, and clean_exposure, which is local itself. # -# ONE DECLARED OUTPUT, AND IT IS A MANIFEST, NOT A directory(). The copies are -# not declared: a directory output would attest that a directory exists, where -# what we want written down is WHICH files were copied and how big each was — +# ONE DECLARED OUTPUT, AND IT IS A MANIFEST, NOT THE TAR OR A directory(). The +# tar is not declared: a directory output would attest that a directory exists, +# where what we want written down is WHICH files were packed and how big each was — # the provenance a rho-statistics run months from now needs in order to know # what it is reading. The manifest is byte-stable, so a no-op rerun does not # move its mtime and does not make clean_exposure look out of date. # # THE KEEP LIST RIDES ON params. That is the entire reason this is not three -# lines of cp appended to exp_psf's shell: `params` is a rerun trigger, so -# adding a pattern reruns the copy and leaves the PSF chain alone. +# lines of tar appended to exp_psf's shell: `params` is a rerun trigger, so +# adding a pattern reruns the packing and leaves the PSF chain alone. rule exp_persist: input: rules.exp_psf.output.manifest diff --git a/workflow/scripts/persist_exp.py b/workflow/scripts/persist_exp.py index 1a17c40cb..a23ab150d 100644 --- a/workflow/scripts/persist_exp.py +++ b/workflow/scripts/persist_exp.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Copy ONE exposure's keepable PSF products off scratch, and record what went. +"""Pack ONE exposure's keepable PSF products into a tar off scratch, and record what went. Run as the shell of the in-DAG ``exp_persist`` rule, never by hand. @@ -36,21 +36,39 @@ we think it is, and writing a green manifest over that would let ``clean_exposure`` delete an exposure whose products were never saved. -The destination is FLAT — one ``psf/`` dir per exposure, no module subtree — -because the module a file came from is already in its name and the consumer -(rho/tau statistics) globs the directory. A name collision between two modules -is therefore a hard error rather than a silent overwrite; nothing in the current -config can produce one, and if a future one can we want to hear about it. +The manifest lists every member (name, pattern, source path, bytes), so a reader +knows what the tar holds without opening it. + +ONE UNCOMPRESSED TAR PER EXPOSURE, ``/.tar``, NOT LOOSE COPIES. +Inodes, not bytes, are what bind on /project: the group quota is ~1 M files, +and loose per-CCD copies are ~200 per exposure with all candidates on — ~25k for +a 64-tile campaign, ~2 M at DR6 scale, against ~7 GB of bytes. A tar collapses +that to one inode per exposure and costs nothing to read: FITS members go +``tarfile.open(t).extractfile(m).read()`` -> ``fits.open(io.BytesIO(...))``, +which is why a tar rather than a multi-HDU FITS bundle (the keep list mixes +FITS, ``.psf`` and ``.txt``; a FITS container could not hold the last two). +Uncompressed because FITS barely compresses and a plain tar is seekable. + +Members are FLAT — file name only, no module subtree — because the module a +file came from is already in its name and the consumer globs member names. A +name collision between two modules is therefore a hard error rather than a +silent overwrite; nothing in the current config can produce one, and if a +future one can we want to hear about it. + +The tar is written DETERMINISTICALLY (ownership zeroed, members in sorted +order, source mtimes kept), tmp-then-``cmp``-then-``mv``: a rerun over an +unchanged store produces a byte-identical tar and leaves the existing one's +mtime alone. The manifest is the rule's ONLY declared output, and it lives on the persistent -root beside the copies (``/exp///manifests/``), NOT in +root beside the tar (``/exp///manifests/``, beside the tar's ``psf/``), NOT in the exposure's scratch ``manifests/`` dir which ``clean_exposure`` deletes wholesale. It is deliberately NOT a ``directory()`` output: what was copied, and how big each file was, is provenance we want written down, and a directory output attests only that some directory exists. It carries no timestamp and is written tmp-then-``cmp``-then-``mv`` (the pattern -``exp_star_cat`` uses), so a rerun that copies the same files leaves the mtime +``exp_star_cat`` uses), so a rerun that packs the same files leaves the mtime alone — mtime is a rerun trigger, and an unconditional rewrite would make every downstream ``clean_exposure`` look out of date once per invocation. """ @@ -58,8 +76,8 @@ import argparse import filecmp import json -import shutil import sys +import tarfile from pathlib import Path # The PSF chain's run dir (RUN_NAME in config_exp_psfex.ini). Hardcoded rather @@ -92,7 +110,8 @@ def main() -> None: help="the exposure's scratch store") p.add_argument("--exp", required=True) p.add_argument("--dest", required=True, type=Path, - help="/exp///psf") + help="/exp///psf; the tar is " + "/.tar") p.add_argument("--manifest", required=True, type=Path) p.add_argument("--pattern", action="append", default=[], help="repeatable; a plain file-name glob") @@ -107,29 +126,40 @@ def main() -> None: f"{args.pattern} under {args.exp_dir}/output/{RUN_NAME}") args.dest.mkdir(parents=True, exist_ok=True) + tar_path = args.dest / f"{args.exp}.tar" seen, files = {}, [] for pat, hits in found.items(): for src in hits: - dst = args.dest / src.name if src.name in seen: sys.exit(f"persist_exp: {args.exp}: two source files are both " - f"named {src.name} ({seen[src.name]} and {src}); the " - f"destination is flat, so this would silently overwrite") - seen[src.name] = src - # Skip a byte-identical copy: it is not just an I/O saving, it keeps - # the destination's mtimes still for anything downstream that reads - # them. - if not (dst.exists() and filecmp.cmp(src, dst, shallow=False)): - tmp = dst.with_name(dst.name + ".tmp") - shutil.copy2(src, tmp) - tmp.replace(dst) # atomic: no half-copied product + f"named {src.name} ({seen[src.name][0]} and {src}); tar " + f"members are flat, so this would silently overwrite") + seen[src.name] = (src, pat) files.append({"name": src.name, "pattern": pat, - "src": str(src), "bytes": dst.stat().st_size}) + "src": str(src), "bytes": src.stat().st_size}) + files.sort(key=lambda f: f["name"]) + + def anonymous(ti: tarfile.TarInfo) -> tarfile.TarInfo: + # Ownership is the one thing that would differ between two writes of + # the same files from different accounts/nodes; drop it. mtime stays: + # it is the product's, and it is stable while the store is. + ti.uid = ti.gid = 0 + ti.uname = ti.gname = "" + return ti + + tmp = tar_path.with_name(tar_path.name + ".tmp") + with tarfile.open(tmp, "w", format=tarfile.PAX_FORMAT) as tf: + for f in files: + tf.add(seen[f["name"]][0], arcname=f["name"], filter=anonymous) + if tar_path.exists() and filecmp.cmp(tmp, tar_path, shallow=False): + tmp.unlink() # unchanged: leave the mtime alone + else: + tmp.replace(tar_path) # atomic: no half-written archive body = { "stage": "exp_persist", "level": "exp", "unit": args.exp, "status": "complete", - "dest": str(args.dest), + "tar": str(tar_path), "patterns": list(args.pattern), # The warning the docstring argues for: named patterns that matched # nothing. Present as a key even when empty, so a reader never has to @@ -137,7 +167,7 @@ def main() -> None: "patterns_unmatched": empty, "n_files": len(files), "bytes": sum(f["bytes"] for f in files), - "files": sorted(files, key=lambda f: f["name"]), + "files": files, } args.manifest.parent.mkdir(parents=True, exist_ok=True) tmp = args.manifest.with_name(args.manifest.name + ".tmp") @@ -149,7 +179,7 @@ def main() -> None: warn = f" ({len(empty)} pattern(s) matched nothing: {empty})" if empty else "" print(f"[persist_exp] {args.exp}: {len(files)} file(s), " - f"{body['bytes'] / 1e6:.1f} MB -> {args.dest}{warn}") + f"{body['bytes'] / 1e6:.1f} MB -> {tar_path}{warn}") if __name__ == "__main__": From 0d7c313b0a017ec2c20e758c17a4a60e75e61cbf Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Wed, 2 Sep 2026 20:19:22 -0400 Subject: [PATCH 5/5] fix(orchestration): persist_exp never leaves a .tmp behind on failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An orphaned tar.tmp on /project is an inode nothing revisits — the leak the tar design exists to avoid. try/finally around both tmp writes. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015qtLUV3bVLPV5p6un7aTFR --- workflow/scripts/persist_exp.py | 33 +++++++++++++++++++++------------ 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/workflow/scripts/persist_exp.py b/workflow/scripts/persist_exp.py index a23ab150d..c13fca604 100644 --- a/workflow/scripts/persist_exp.py +++ b/workflow/scripts/persist_exp.py @@ -147,14 +147,20 @@ def anonymous(ti: tarfile.TarInfo) -> tarfile.TarInfo: ti.uname = ti.gname = "" return ti + # tmp-then-cmp-then-mv, and the tmp NEVER outlives a failure: an orphaned + # .tmp on /project is an inode nothing revisits — the leak this whole tar + # design exists to avoid, one per failed attempt at DR6 scale. tmp = tar_path.with_name(tar_path.name + ".tmp") - with tarfile.open(tmp, "w", format=tarfile.PAX_FORMAT) as tf: - for f in files: - tf.add(seen[f["name"]][0], arcname=f["name"], filter=anonymous) - if tar_path.exists() and filecmp.cmp(tmp, tar_path, shallow=False): - tmp.unlink() # unchanged: leave the mtime alone - else: - tmp.replace(tar_path) # atomic: no half-written archive + try: + with tarfile.open(tmp, "w", format=tarfile.PAX_FORMAT) as tf: + for f in files: + tf.add(seen[f["name"]][0], arcname=f["name"], filter=anonymous) + if tar_path.exists() and filecmp.cmp(tmp, tar_path, shallow=False): + tmp.unlink() # unchanged: leave the mtime alone + else: + tmp.replace(tar_path) # atomic: no half-written archive + finally: + tmp.unlink(missing_ok=True) body = { "stage": "exp_persist", "level": "exp", "unit": args.exp, @@ -171,11 +177,14 @@ def anonymous(ti: tarfile.TarInfo) -> tarfile.TarInfo: } args.manifest.parent.mkdir(parents=True, exist_ok=True) tmp = args.manifest.with_name(args.manifest.name + ".tmp") - tmp.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n") - if args.manifest.exists() and filecmp.cmp(tmp, args.manifest, shallow=False): - tmp.unlink() # unchanged: leave the mtime alone - else: - tmp.replace(args.manifest) + try: + tmp.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n") + if args.manifest.exists() and filecmp.cmp(tmp, args.manifest, shallow=False): + tmp.unlink() # unchanged: leave the mtime alone + else: + tmp.replace(args.manifest) + finally: + tmp.unlink(missing_ok=True) warn = f" ({len(empty)} pattern(s) matched nothing: {empty})" if empty else "" print(f"[persist_exp] {args.exp}: {len(files)} file(s), "