diff --git a/workflow/README.md b/workflow/README.md index 08bfd2291..e533ab725 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 -> 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 ``` @@ -230,6 +231,22 @@ 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` 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 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 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. - **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/Snakefile b/workflow/Snakefile index b925adc06..81e395951 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 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 + /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 `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). +# # 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/config.yaml b/workflow/config.yaml index 8e4de8506..fd140f5b4 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -83,6 +83,78 @@ 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 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 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 +# 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 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 +# 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 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 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. +# 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 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. +# 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/rules/exposure.smk b/workflow/rules/exposure.smk index 9fe3e7b29..d15bf64fa 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,13 @@ 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 +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 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 +322,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 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 +# (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 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 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 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 + 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 +406,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: diff --git a/workflow/scripts/persist_exp.py b/workflow/scripts/persist_exp.py new file mode 100644 index 000000000..c13fca604 --- /dev/null +++ b/workflow/scripts/persist_exp.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""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. + +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 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 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 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. +""" + +import argparse +import filecmp +import json +import sys +import tarfile +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; 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") + 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) + tar_path = args.dest / f"{args.exp}.tar" + seen, files = {}, [] + for pat, hits in found.items(): + for src in hits: + if src.name in seen: + sys.exit(f"persist_exp: {args.exp}: two source files are both " + 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": 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-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") + 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, + "status": "complete", + "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 + # wonder whether an old manifest predates the field. + "patterns_unmatched": empty, + "n_files": len(files), + "bytes": sum(f["bytes"] for f in files), + "files": files, + } + args.manifest.parent.mkdir(parents=True, exist_ok=True) + tmp = args.manifest.with_name(args.manifest.name + ".tmp") + 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), " + f"{body['bytes'] / 1e6:.1f} MB -> {tar_path}{warn}") + + +if __name__ == "__main__": + main() 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