From f7d7cc6a75350cf9cefecc4ef0ec878f5aea6ab2 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Sat, 10 Oct 2026 09:58:44 -0400 Subject: [PATCH 1/6] Persist tile epoch-cut counts in published completeness manifests Capture each ngmix summary before temp cleanup, aggregate the full chunk set at make-cat, and publish the manifest beside final_cat. Refuse reclamation without validated durable counts. This changes completeness params across stages and the make-cat/clean shell fingerprints; deploy between campaigns, not as an unreviewed resume. Co-Authored-By: GPT-6.1 Sol via Claude Opus 5.5 --- workflow/Snakefile | 7 ++- workflow/rules/tile.smk | 9 ++- workflow/scripts/clean_tile.py | 53 ++++++++++------- workflow/scripts/completeness.py | 16 ++++++ workflow/scripts/epoch_cuts.py | 97 ++++++++++++++++++++++++++++++++ 5 files changed, 158 insertions(+), 24 deletions(-) create mode 100644 workflow/scripts/epoch_cuts.py diff --git a/workflow/Snakefile b/workflow/Snakefile index 1d9ba94af..5f06933ea 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -484,10 +484,13 @@ def path_hash(path): f"that needs the file will say so when it runs.") return "missing" -SCRIPT_HASH = script_hash("completeness.py") +# Epoch-count parsing/aggregation is part of the completeness contract too. +SCRIPT_HASH = ":".join((script_hash("completeness.py"), + script_hash("epoch_cuts.py"))) FOREST_HASH = script_hash("build_forest.py") CLEAN_HASH = script_hash("clean_exposure.py") -CLEAN_TILE_HASH = script_hash("clean_tile.py") +CLEAN_TILE_HASH = ":".join((script_hash("clean_tile.py"), + script_hash("epoch_cuts.py"))) PERSIST_HASH = script_hash("persist_exp.py") MERGE_STAR_HASH = script_hash("merge_star_cat.py") # Three files, one trigger: the rule's script, the column extraction it calls, diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index e370b36d5..9e23846ef 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -886,10 +886,12 @@ rule tile_make_cat: output: manifest = f"{TILE_DIR}/manifests/tile_make_cat.json", final_cat = f"{PROD_TILE_DIR}/final_cat-{{tile}}.fits", + stats = f"{PROD_TILE_DIR}/tile_make_cat.json", log: f"{TILE_DIR}/logs/tile_make_cat.json" params: pre = lambda wc: unit_pre("tile_make_cat", wc.tile, + env={"NGMIX_N_CHUNKS": NGMIX_CHUNKS}, pre_run=[tile_local(wc.tile), TILE_VIGNET_REQUIRED, TILE_CLEAN]), script_hash = SCRIPT_HASH @@ -908,6 +910,8 @@ rule tile_make_cat: post="if [ $rc -eq 0 ]; then\n" ' cp -f "$(ls -1 "$SP_RUN"/output/run_sp_tile_Mc/make_cat_runner' '/output/final_cat*.fits | head -1)" {output.final_cat}\n' + ' cp -f {output.manifest} {output.stats}.tmp\n' + ' mv -f {output.stats}.tmp {output.stats}\n' "fi\n") @@ -986,7 +990,8 @@ rule clean_tile: # the PROD_TILE_DIR pattern so there is exactly one definition of this # path (final_cat() in the Snakefile), the same way clean_exposure keys # off wildcards.exp alone. - lambda wc: final_cat(wc.tile) + cat = lambda wc: final_cat(wc.tile), + stats = rules.tile_make_cat.output.stats, output: tombstone = f"{TILE_DIR}/cleaned.json" params: @@ -1000,7 +1005,7 @@ rule clean_tile: shell: f"python {SCRIPTS}/clean_tile.py" " --tile-dir $(dirname {output.tombstone}) --tile {wildcards.tile}" - " --tombstone {output.tombstone}" + " --tombstone {output.tombstone} --final-cat {input.cat}" # --- the campaign's shear catalogue ----------------------------------------- diff --git a/workflow/scripts/clean_tile.py b/workflow/scripts/clean_tile.py index bdeefb794..15c2ccf28 100644 --- a/workflow/scripts/clean_tile.py +++ b/workflow/scripts/clean_tile.py @@ -1,9 +1,11 @@ #!/usr/bin/env python3 """Reclaim one finished tile's scratch store and leave a tombstone (PRD #848 D5). -Run as the shell of the in-DAG ``clean_tile`` rule, never by hand: the rule's -``input:`` is the tile's ``final_cat`` on the PERSISTENT root, so by the time -this executes the tile has published its final catalogue. +Run as the shell of the in-DAG ``clean_tile`` rule, or through +``backfill_epoch_cuts.py`` for an idle campaign. The rule's input is the tile's +``final_cat`` on the PERSISTENT root. Before any deletion, validate the +``tile_make_cat.json`` beside it, including all chunks' epoch-cut counts. +The persistent manifest is not scratch DAG currency and is never pruned. A tile's scratch store has no reader outside that tile — tiles read exposures, nothing reads another tile's store — so unlike the exposure case there is no consumer set to close over and no eligibility test to make. Writer, then @@ -113,7 +115,10 @@ next should know that the worst case is not a scratch store they could rebuild — it is a rmtree walking into half a terabyte of shared, backed-up survey data. -Logs are deleted, not absorbed, as in ``clean_exposure``. Here the +Module logs are deleted too: epoch-cut counts have already been captured in +chunk completeness manifests and summed in the persistent make-cat manifest. +No log is needed to recover these diagnostics after reclamation. +Verdict logs are deleted, not absorbed, as in ``clean_exposure``. Here the duplication is exact: on a finished tile every ``logs/.json`` is BYTE-IDENTICAL to the ``manifests/.json`` beside it (verified across all 16 stage records of 186.307), because a tile with a failed stage has no @@ -272,38 +277,46 @@ def prune(root: Path, keep: set, removed: list) -> None: removed.append(str(entry)) -def main() -> None: - p = argparse.ArgumentParser(description=__doc__) - p.add_argument("--tile-dir", required=True, type=Path) - p.add_argument("--tile", required=True) - p.add_argument("--tombstone", required=True, type=Path) - args = p.parse_args() +def reclaim(tile_dir: Path, tile: str, tombstone: Path, final_cat: Path) -> None: + """Validate durable diagnostics, write the tombstone, then prune scratch.""" + from epoch_cuts import require_durable_manifest - survivors = survivor_paths(args.tile_dir, args.tile) - require_survivors(survivors, args.tile) + require_durable_manifest(final_cat, tile) + survivors = survivor_paths(tile_dir, tile) + require_survivors(survivors, tile) - mdir = args.tile_dir / "manifests" - manifests, benchmarks = previous_record(args.tombstone) + mdir = tile_dir / "manifests" + manifests, benchmarks = previous_record(tombstone) manifests.update(absorb_manifests(mdir)) benchmarks.update(absorb_benchmarks(mdir)) # Tombstone first, complete — then delete (see the module docstring). - args.tombstone.parent.mkdir(parents=True, exist_ok=True) - tmp = args.tombstone.with_suffix(".json.tmp") + tombstone.parent.mkdir(parents=True, exist_ok=True) + tmp = tombstone.with_suffix(".json.tmp") tmp.write_text(json.dumps({ - "tile": args.tile, + "tile": tile, "cleaned_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "kept": sorted(str(p) for p in survivors.values()), "manifests": manifests, "benchmarks": benchmarks, }, indent=2, sort_keys=True) + "\n") - tmp.replace(args.tombstone) # atomic: no half-written tombstone, ever + tmp.replace(tombstone) # atomic: no half-written tombstone, ever removed: list = [] - prune(args.tile_dir, set(survivors.values()) | {args.tombstone}, removed) - print(f"[clean_tile] {args.tile}: removed {len(removed)} path(s); kept " + prune(tile_dir, set(survivors.values()) | {tombstone}, removed) + print(f"[clean_tile] {tile}: removed {len(removed)} path(s); kept " f"{len(survivors)} survivor(s) + the tombstone") +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--tile-dir", required=True, type=Path) + p.add_argument("--tile", required=True) + p.add_argument("--tombstone", required=True, type=Path) + p.add_argument("--final-cat", required=True, type=Path) + args = p.parse_args() + reclaim(args.tile_dir, args.tile, args.tombstone, args.final_cat) + + if __name__ == "__main__": main() diff --git a/workflow/scripts/completeness.py b/workflow/scripts/completeness.py index b39a951ed..bf4345b97 100644 --- a/workflow/scripts/completeness.py +++ b/workflow/scripts/completeness.py @@ -72,6 +72,8 @@ import sys from pathlib import Path +from epoch_cuts import gather_manifests, read_chunk_logs + # How the tile's galaxy sample is defined: SExtractor on the tile image alone, # or those detections joined to the UNIONS per-tile catalogue, whose NUMBER # they take. The run config's `tile_detection:` picks one. @@ -365,6 +367,20 @@ def build_manifest(stage, run_dir, unit, stage_subdir=None): "warn": warn, "status": status, "reasons": scrape_reasons(stage_dir, runner), }) + # Capture ngmix's diagnostics while its temp() directory still exists. + # Make-cat consumes only structured records, never the retained log copies. + if ok and stage in ("tile_ngmix", "tile_make_cat"): + try: + manifest["epoch_cuts"] = ( + read_chunk_logs(stage_dir) if stage == "tile_ngmix" else + gather_manifests(run_dir, int(os.environ["NGMIX_N_CHUNKS"])) + ) + except (OSError, ValueError, KeyError) as exc: + ok = False + manifest["failures"].append({ + "runner": "epoch_cuts", "found": 0, "expect": 1, + "warn": False, "status": "failed", "reasons": [str(exc)], + }) manifest["status"] = "failed" if not ok else ("warn" if short else "complete") return manifest, ok diff --git a/workflow/scripts/epoch_cuts.py b/workflow/scripts/epoch_cuts.py new file mode 100644 index 000000000..01bd8fe40 --- /dev/null +++ b/workflow/scripts/epoch_cuts.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Archive PR #922 epoch-cut diagnostics, independently of module-log retention. + +Chunk completeness checks read the single end-of-loop summary before temp() +removes the chunk directory. The make-cat check gathers those structured chunk +records, and its manifest is published beside final_cat. Backfills use the same +schema and parser. Missing or ambiguous records are errors, never zero counts. +These are diagnostics only: no cuts or catalogue values are changed. +""" + +import json +import re +from pathlib import Path + +FIELDS = ("considered", "masked_fraction", "central_veto", "failed", + "objects_emptied") + + +def validate_counts(counts): + """Return an exact, non-negative integer count record or raise ValueError.""" + if not isinstance(counts, dict) or set(counts) != set(FIELDS): + raise ValueError(f"epoch cuts must contain exactly {FIELDS}") + if any(type(v) is not int or v < 0 for v in counts.values()): + raise ValueError("epoch cuts must be non-negative integers") + if sum(counts[k] for k in ("masked_fraction", "central_veto", "failed")) > counts["considered"]: + raise ValueError("epoch cuts exceed epochs considered") + if counts["objects_emptied"] > counts["considered"]: + raise ValueError("objects emptied exceeds epochs considered") + return dict(counts) + + +def parse_epoch_cuts(text): + """Parse exactly one complete end-of-loop summary; reject retry ambiguity.""" + lines = [line.split("epoch cuts:", 1)[1].strip() + for line in text.splitlines() if "epoch cuts:" in line] + if len(lines) != 1: + raise ValueError(f"expected one epoch cuts summary, found {len(lines)}") + counts = {} + for token in lines[0].split(): + match = re.fullmatch(r"([a-z_]+)=(\d+)", token) + if not match or match[1] in counts: + raise ValueError(f"invalid epoch cuts token: {token!r}") + counts[match[1]] = int(match[2]) + return validate_counts(counts) + + +def read_chunk_logs(run_root): + """Read ngmix worker logs only, not copies or ShapePipe's aggregate log.""" + paths = sorted((Path(run_root) / "ngmix_runner" / "logs").glob("process-*.log")) + if len(paths) != 1: + raise ValueError(f"expected one ngmix worker log under {run_root}, found {len(paths)}") + return parse_epoch_cuts(paths[0].read_text()) + + +def aggregate_chunks(chunks, n_chunks): + """Validate the complete 1..N chunk set and sum each diagnostic once.""" + if type(n_chunks) is not int or n_chunks < 1: + raise ValueError("n_chunks must be a positive integer") + expected = {str(k) for k in range(1, n_chunks + 1)} + if set(chunks) != expected: + raise ValueError(f"expected chunks 1..{n_chunks}, found {sorted(chunks)}") + chunks = {str(k): validate_counts(chunks[str(k)]) + for k in range(1, n_chunks + 1)} + return {"schema_version": 1, "n_chunks": n_chunks, "chunks": chunks, + "totals": {key: sum(c[key] for c in chunks.values()) for key in FIELDS}} + + +def gather_manifests(tile_dir, n_chunks): + """Gather successful chunk completeness records, after the merge DAG edge.""" + chunks = {} + for k in range(1, n_chunks + 1): + path = Path(tile_dir) / "manifests" / f"tile_ngmix_{k}.json" + manifest = json.loads(path.read_text()) + if (manifest.get("stage") != "tile_ngmix" + or manifest.get("unit") != Path(tile_dir).name + or manifest.get("status") != "complete"): + raise ValueError(f"not a successful chunk manifest: {path}") + chunks[str(k)] = manifest.get("epoch_cuts") + return aggregate_chunks(chunks, n_chunks) + + +def require_durable_manifest(final_cat, tile, n_chunks=None): + """Validate catalogue existence and its persistent completeness/count record.""" + final_cat = Path(final_cat) + if not final_cat.is_file() or final_cat.stat().st_size == 0: + raise ValueError(f"no persisted final_cat: {final_cat}") + path = final_cat.parent / "tile_make_cat.json" + manifest = json.loads(path.read_text()) + if (manifest.get("stage") != "tile_make_cat" or manifest.get("unit") != tile + or manifest.get("status") != "complete"): + raise ValueError(f"not a successful make-cat manifest for {tile}: {path}") + record = manifest.get("epoch_cuts") or {} + expected = aggregate_chunks(record.get("chunks", {}), + n_chunks if n_chunks is not None else record.get("n_chunks")) + if record != expected: + raise ValueError(f"invalid epoch cuts aggregate: {path}") + return manifest From ef49304f73f11729f4b84c70d68a93af7516d234 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Sat, 10 Oct 2026 09:59:46 -0400 Subject: [PATCH 2/6] Document safe tile cleaning with durable epoch-cut diagnostics Keep clean_tiles enabled by default, remove the stale false-default comment, and document the ngmix summary prerequisite, persistent schema, backfill procedure and campaign-boundary deployment. Co-Authored-By: GPT-6.1 Sol via Claude Opus 5.5 --- workflow/README.md | 22 ++++++++++++++++++++++ workflow/config.yaml | 10 +++++++--- workflow/rules/tile.smk | 6 +++--- 3 files changed, 32 insertions(+), 6 deletions(-) diff --git a/workflow/README.md b/workflow/README.md index 4f0b135c3..1666a9e20 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -317,6 +317,28 @@ profiles/nibi/config.yaml SLURM executor; apptainer SDM; per-user jobs cap; kee since the missing tombstones schedule exactly the outstanding clean jobs. The tombstone is written *before* anything is deleted, so a crash can cost disk but never the record. +- **Tile epoch-cut diagnostics survive cleaning.** + `tile_ngmix` completeness checks capture PR #922's five counters before each chunk's temporary directory goes away. + `tile_make_cat` sums all configured chunks and publishes `tile_make_cat.json` beside the tile's persistent `final_cat`. + Its `epoch_cuts` field contains `schema_version`, `n_chunks`, per-chunk `chunks`, and tile `totals` for `considered`, `masked_fraction`, `central_veto`, `failed`, and `objects_emptied`. + Counts are sequential veto counts, not independent cut incidences; `objects_emptied` counts objects whose considered epochs were all removed before fitting. + Missing, malformed, or duplicate summaries fail completeness rather than becoming zeros. + `clean_tile` validates the durable count record before touching scratch, so `clean_tiles: true` needs no module-log retention override. + This requires an ngmix build with PR #922's end-of-loop `epoch cuts:` summary, including on empty chunks. + Older code without that summary fails the ngmix completeness check and leaves scratch intact. + + For an **idle existing campaign** with kept logs, use `workflow/scripts/backfill_epoch_cuts.py --run-dir --products-dir --n-chunks --dry-run` first. + Alternatively pass `--config ` to resolve products and chunk count from the run's frozen `workflow/config.yaml` plus its run config. + The script skips tiles without a nonempty persisted catalogue and refuses missing or ambiguous chunk evidence. + Without `--dry-run`, it publishes the same manifest schema and reclaims through `clean_tile.reclaim`, including its survivor checks and tombstone. + Don't use it while jobs or another cleaner are writing the campaign. + Dry-run reads metadata and small logs only, writes nothing, and reports logical regular-file bytes (not allocated blocks), without following symlinks. + + **Deploy at a campaign boundary.** + The completeness/helper hash is part of every ShapePipe rule's `params`; this change also alters make-cat's params, outputs and shell, and clean-tile's inputs and shell. + An ordinary resume against old completed tiles can therefore schedule science stages again, even after their exposure stores have been reclaimed. + Backfill old campaigns without changing their frozen code or resuming them against this workflow; removing the log-retention override is appropriate for new campaigns running this contract. + - **A finished tile declares no reclaimed exposures.** Deleting an exposure's manifests would otherwise rerun every other tile that reads it, and those reruns spread across the exposure-overlap component. So a tile whose diff --git a/workflow/config.yaml b/workflow/config.yaml index 68e4a517a..a2e73a7d7 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -174,8 +174,9 @@ machines: # selection studies and cost an order of magnitude more. # # THIS LIST IS EXPOSURE-SIDE ONLY. Tile-side retention is not configurable: the -# only tile product that persists today is final_cat, written by tile_make_cat -# straight to products_dir. A tile keep list is #844 follow-up. +# tile products that persist are final_cat and tile_make_cat.json, written by +# tile_make_cat straight to products_dir. The manifest contains epoch-cut +# counts summed over all ngmix chunks. A tile keep list is #844 follow-up. # # 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 @@ -191,7 +192,10 @@ clean: true # Per-tile store reclamation after the tile's final_cat. Needed above ~850 # tiles (scratch quota); deletes the per-tile data sp_tilecost/sp_costmodel -# read. Set both false for a small control run you want to re-measure. +# read. Epoch-cut counts survive in products_dir/tiles/// +# tile_make_cat.json; keeping module logs does NOT require disabling cleaning. +# Cleaning refuses a tile without this complete durable count record. +# Set both false for a small control run you want to re-measure timings on. clean_tiles: true # Failed tiles that should not keep their exposures from being cleaned. diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index 9e23846ef..219582484 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -739,9 +739,9 @@ rule tile_ngmix: # reclamation ON the tombstoned tile has lost tile_detect.json and the # structural "Input files updated by another job" propagation puts # tile_vignets back in the group (measured at clean_tile below), so the - # store is rebuilt and nothing trips. The SHIPPED DEFAULT - # clean_tiles: false is the dangerous configuration -- the opposite of - # how reclamation reads everywhere else in this file. + # store is rebuilt and nothing trips. A campaign override setting + # clean_tiles: false leaves the dangerous configuration; the shipped + # default is true. Epoch-cut diagnostics no longer need that override. # # So: land this hash, and every later edit to ngmix_range.py, at a # campaign boundary on a fresh root. The same rule and direct command as From fbdef6cba1bf2c2031c4228bee547052a0d626ee Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Sat, 10 Oct 2026 10:02:14 -0400 Subject: [PATCH 3/6] Backfill epoch-cut manifests and reclaim idle campaign tiles safely Use frozen run defaults, require persisted final_cat and complete successful chunk evidence, publish counts atomically, then reuse clean_tile.reclaim. Dry-run uses the same whitelist traversal and never follows links. smk-b01 dry-run: 25 tiles, 53,290,513,588 logical bytes (49.631 GiB), no skips or refusals; nothing written or deleted. Co-Authored-By: GPT-6.1 Sol via Claude Opus 5.5 --- workflow/scripts/backfill_epoch_cuts.py | 159 ++++++++++++++++++++++++ workflow/scripts/clean_tile.py | 19 ++- 2 files changed, 175 insertions(+), 3 deletions(-) create mode 100644 workflow/scripts/backfill_epoch_cuts.py diff --git a/workflow/scripts/backfill_epoch_cuts.py b/workflow/scripts/backfill_epoch_cuts.py new file mode 100644 index 000000000..3be22294d --- /dev/null +++ b/workflow/scripts/backfill_epoch_cuts.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Persist epoch-cut counts and reclaim finished tiles of an IDLE campaign. + +Run --dry-run first. No DAG or Snakemake resume is needed. Products must be +outside scratch's run root. Both historical .fits and .hdf5 catalogues are +supported; a missing, empty or ambiguous persisted catalogue is never cleaned. +The parser and schema are shared with the live completeness checks, and +clean_tile owns all keep rules, the dry-run traversal and actual reclamation. +Do not run while jobs or another cleaner are writing this campaign. +""" + +import argparse +import json +import re +import sys +from pathlib import Path + +import clean_tile +from completeness import write_if_changed +from epoch_cuts import (aggregate_chunks, read_chunk_logs, + require_durable_manifest, validate_counts) + + +def resolve_config(config_path): + """Resolve the run's frozen base defaults plus its overlay, not today's defaults.""" + import yaml + from run_config import apply_defaults, merge + + base = config_path.parent / "workflow" / "config.yaml" + config = apply_defaults(merge(yaml.safe_load(base.read_text()), + yaml.safe_load(config_path.read_text()))) + return (Path(config["outputs"]["run_dir"]), + Path(config["outputs"]["products_dir"]), + int(config["ngmix_chunks"])) + + +def catalogue_path(products_dir, tile): + """Return one nonempty persisted catalogue, or None if none exists.""" + root = products_dir / "tiles" / tile[:2] / tile + paths = [root / f"final_cat-{tile}.{ext}" for ext in ("fits", "hdf5")] + paths = [p for p in paths if p.is_file() and p.stat().st_size > 0] + if len(paths) > 1: + raise ValueError(f"ambiguous persisted final_cat for {tile}: {paths}") + return paths[0] if paths else None + + +def backfill_manifest(tile_dir, final_cat, n_chunks): + """Build the durable make-cat manifest, or validate and reuse an existing one.""" + durable = final_cat.parent / "tile_make_cat.json" + if durable.exists(): + return require_durable_manifest(final_cat, tile_dir.name, n_chunks) + path = tile_dir / "manifests" / "tile_make_cat.json" + manifest = json.loads(path.read_text()) + if (manifest.get("stage") != "tile_make_cat" + or manifest.get("unit") != tile_dir.name + or manifest.get("status") != "complete"): + raise ValueError(f"not a successful make-cat manifest: {path}") + chunks = {} + for k in range(1, n_chunks + 1): + chunk_path = tile_dir / "manifests" / f"tile_ngmix_{k}.json" + chunk = json.loads(chunk_path.read_text()) + if (chunk.get("stage") != "tile_ngmix" + or chunk.get("unit") != tile_dir.name + or chunk.get("status") != "complete"): + raise ValueError(f"not a successful chunk manifest: {chunk_path}") + if "epoch_cuts" in chunk: + chunks[str(k)] = validate_counts(chunk["epoch_cuts"]) + else: + name = f"run_sp_tile_ngmix_Ng{k}u" + root = tile_dir / "logs" / "modules" / name + if not root.exists(): + root = tile_dir / "output" / name + chunks[str(k)] = read_chunk_logs(root) + manifest["epoch_cuts"] = aggregate_chunks(chunks, n_chunks) + return manifest + + +def process_tile(tile_dir, products_dir, n_chunks, dry_run): + """Validate one tile, report reclaimable bytes and optionally persist/reclaim.""" + tile = tile_dir.name + final_cat = catalogue_path(products_dir, tile) + if final_cat is None: + print(f"[backfill] {tile}: SKIP no persisted final_cat") + return "skipped", 0, None + survivors = clean_tile.survivor_paths(tile_dir, tile) + clean_tile.require_survivors(survivors, tile) + manifest = backfill_manifest(tile_dir, final_cat, n_chunks) + tombstone = tile_dir / "cleaned.json" + freed = clean_tile.prune(tile_dir, set(survivors.values()) | {tombstone}, + [], dry_run=True) + record = manifest["epoch_cuts"] + print(f"[backfill] {tile}: {'WOULD FREE' if dry_run else 'FREE'} {freed} bytes; " + f"{n_chunks} chunks; totals={json.dumps(record['totals'], sort_keys=True)}") + if not dry_run: + # Publish atomically BEFORE the cleaner can touch scratch. Never run a + # Snakemake science stage merely to retrofit this small durable record. + write_if_changed(final_cat.parent / "tile_make_cat.json", + json.dumps(manifest, indent=2, sort_keys=True) + "\n") + clean_tile.reclaim(tile_dir, tile, tombstone, final_cat) + return "ready", freed, record["totals"] + + +def main(argv=None): + """Run the backfill CLI; refusals leave the affected tile completely intact.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run-dir", type=Path, required=True) + parser.add_argument("--products-dir", type=Path) + parser.add_argument("--n-chunks", type=int) + parser.add_argument("--config", type=Path, + help="run overlay beside frozen workflow/config.yaml") + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + if args.config: + run, products, chunks = resolve_config(args.config) + if run.resolve() != args.run_dir.resolve(): + parser.error("--run-dir disagrees with the run config") + if args.products_dir and args.products_dir.resolve() != products.resolve(): + parser.error("--products-dir disagrees with the run config") + if args.n_chunks is not None and args.n_chunks != chunks: + parser.error("--n-chunks disagrees with the run config") + args.products_dir, args.n_chunks = products, chunks + if args.products_dir is None or args.n_chunks is None or args.n_chunks < 1: + parser.error("provide --config, or both --products-dir and positive --n-chunks") + run = args.run_dir.resolve() + products = args.products_dir.resolve() + if products == run or run in products.parents: + parser.error("products must be outside the scratch run root") + if not (run / "tiles").is_dir(): + parser.error("run has no tiles directory") + totals = {} + counts = dict(ready=0, skipped=0, refused=0) + freed = 0 + for tile_dir in sorted((run / "tiles").glob("*/*")): + if not re.fullmatch(r"\d+\.\d+", tile_dir.name): + continue + try: + if (tile_dir.is_symlink() or tile_dir.parent.is_symlink() + or tile_dir.parent.name != tile_dir.name[:2]): + raise ValueError(f"unexpected tile/shard path: {tile_dir}") + status, size, record = process_tile(tile_dir, products, + args.n_chunks, args.dry_run) + counts[status] += 1 + freed += size + if record: + for key, value in record.items(): + totals[key] = totals.get(key, 0) + value + except (OSError, ValueError, SystemExit) as exc: + counts["refused"] += 1 + print(f"[backfill] {tile_dir.name}: REFUSED {exc}", file=sys.stderr) + print(f"[backfill] {'DRY RUN' if args.dry_run else 'DONE'}: {counts['ready']} ready, " + f"{counts['skipped']} skipped, {counts['refused']} refused; " + f"{freed} bytes ({freed / 2**30:.3f} GiB) " + f"{'would be freed' if args.dry_run else 'reclaimed'}") + print(f"[backfill] campaign totals={json.dumps(totals, sort_keys=True)}") + return 1 if counts["refused"] else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/workflow/scripts/clean_tile.py b/workflow/scripts/clean_tile.py index 15c2ccf28..a542b553f 100644 --- a/workflow/scripts/clean_tile.py +++ b/workflow/scripts/clean_tile.py @@ -250,7 +250,16 @@ def absorb_benchmarks(mdir: Path) -> dict: return out -def prune(root: Path, keep: set, removed: list) -> None: +def logical_bytes(path: Path) -> int: + """Count regular-file bytes without following any symlink, even a dangling one.""" + if path.is_symlink(): + return 0 + if path.is_dir(): + return sum(logical_bytes(p) for p in path.iterdir()) + return path.stat().st_size if path.is_file() else 0 + + +def prune(root: Path, keep: set, removed: list, *, dry_run=False) -> int: """Delete everything under ``root`` except ``keep`` and the dirs leading to it. A whitelist walk rather than an rmtree-with-exceptions, so adding a survivor @@ -262,19 +271,23 @@ def prune(root: Path, keep: set, removed: list) -> None: see the module docstring on the two symlink classes. """ ancestors = {p for k in keep for p in k.parents} + freed = 0 for entry in sorted(root.iterdir()): if entry in keep: continue if entry in ancestors and not entry.is_symlink(): - prune(entry, keep, removed) + freed += prune(entry, keep, removed, dry_run=dry_run) continue - if entry.is_symlink(): + if dry_run: + freed += logical_bytes(entry) + elif entry.is_symlink(): entry.unlink() elif entry.is_dir(): shutil.rmtree(entry) # unlinks nested symlinks, never follows else: entry.unlink() removed.append(str(entry)) + return freed def reclaim(tile_dir: Path, tile: str, tombstone: Path, final_cat: Path) -> None: From f197c754dc25bc33856e480a72faf3f4c4c1fdec Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Sat, 10 Oct 2026 10:09:31 -0400 Subject: [PATCH 4/6] Test epoch-cut parsing, aggregation, durable publication and safe backfill Cover missing/ambiguous summaries, full chunk coverage, zero counts, duplicate log copies, stable manifest mtimes, cleanup refusal, read-only byte accounting, publication-before-cleaning and idempotent retries. Include the exact PR #922 producer line. Container verification after restacking: 199 passed, 1 skipped across targeted, structural and ngmix producer suites; ASTRA validation passes. Co-Authored-By: GPT-6.1 Sol via Claude Opus 5.5 --- tests/unit/test_epoch_cuts_persist.py | 430 ++++++++++++++++++++++++ workflow/scripts/backfill_epoch_cuts.py | 110 ++++-- workflow/scripts/epoch_cuts.py | 88 +++-- 3 files changed, 566 insertions(+), 62 deletions(-) create mode 100644 tests/unit/test_epoch_cuts_persist.py diff --git a/tests/unit/test_epoch_cuts_persist.py b/tests/unit/test_epoch_cuts_persist.py new file mode 100644 index 000000000..3e3f6f011 --- /dev/null +++ b/tests/unit/test_epoch_cuts_persist.py @@ -0,0 +1,430 @@ +"""Epoch-cut diagnostics survive cleaning without retaining module logs.""" + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.unions +SCRIPTS = Path(__file__).resolve().parents[2] / "workflow" / "scripts" +TILE = "259.283" +COUNTS = dict( + considered=100, + masked_fraction=7, + central_veto=3, + failed=2, + objects_emptied=1, +) + + +def load(name): + """Import a workflow script with its sibling imports available.""" + sys.path.insert(0, str(SCRIPTS)) + try: + spec = importlib.util.spec_from_file_location( + name, SCRIPTS / f"{name}.py" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + finally: + sys.path.remove(str(SCRIPTS)) + + +epoch = load("epoch_cuts") +completeness = load("completeness") +clean = load("clean_tile") +backfill = load("backfill_epoch_cuts") + + +def summary(counts=COUNTS): + """Render the production end-of-loop summary.""" + return ( + "10/10/2026 07:47:51 epoch cuts: " + + " ".join(f"{k}={v}" for k, v in counts.items()) + + "\n" + ) + + +def write(path, text): + """Create one fixture file and its parent directories.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return path + + +def verdict(stage, **extra): + """Build a successful completeness verdict.""" + return dict( + stage=stage, + unit=TILE, + status="complete", + failures=[], + runners={}, + **extra, + ) + + +def test_parse_timestamp_and_unrelated_lines(): + """Only the summary supplies counts.""" + assert ( + epoch.parse_epoch_cuts( + "ngmix loop finished\n" + summary() + "all done\n" + ) + == COUNTS + ) + + +def test_parse_pr922_exact_line(): + """Accept the exact field names and spacing emitted by PR #922.""" + assert epoch.parse_epoch_cuts( + "epoch cuts: considered=8181 masked_fraction=235 " + "central_veto=271 failed=0 objects_emptied=16" + ) == dict( + considered=8181, + masked_fraction=235, + central_veto=271, + failed=0, + objects_emptied=16, + ) + + +def test_zero_is_data_not_missing(): + """An explicit zero summary is valid; absent evidence is not.""" + zeros = dict.fromkeys(epoch.FIELDS, 0) + assert epoch.parse_epoch_cuts(summary(zeros)) == zeros + with pytest.raises(ValueError, match="found 0"): + epoch.parse_epoch_cuts("no summary\n") + + +@pytest.mark.parametrize( + "text", + [ + summary() * 2, + "epoch cuts: considered=1\n", + summary().replace("failed=2", "failed=-2"), + summary().replace("failed=2", "failed=2.5"), + summary().replace("failed=2", "failed=2 failed=2"), + summary().replace("failed=2", "failed=2 extra=1"), + summary().replace("considered=100", "considered=1"), + summary().replace("objects_emptied=1", "objects_emptied=101"), + ], +) +def test_malformed_and_ambiguous_counts_refused(text): + """Never salvage partial records or combine retry summaries.""" + with pytest.raises(ValueError): + epoch.parse_epoch_cuts(text) + + +@pytest.mark.parametrize("bad", [True, -1, 1.5, "1"]) +def test_counts_require_actual_nonnegative_integers(bad): + """JSON booleans and strings are not counts.""" + with pytest.raises(ValueError): + epoch.validate_counts(dict(COUNTS, failed=bad)) + + +def test_aggregate_all_fields_once_in_numeric_chunk_order(): + """All five counters sum over exactly the configured chunks.""" + chunks = {str(k): dict(COUNTS) for k in reversed(range(1, 17))} + record = epoch.aggregate_chunks(chunks, 16) + assert record["schema_version"] == 1 + assert list(record["chunks"]) == [str(k) for k in range(1, 17)] + assert record["totals"] == {k: 16 * v for k, v in COUNTS.items()} + assert record["n_chunks"] == 16 + + +@pytest.mark.parametrize( + "chunks,n", + [ + ({"1": COUNTS}, 2), + ({"1": COUNTS, "3": COUNTS}, 2), + ({"1": COUNTS, "2": COUNTS}, 1), + ({}, 0), + ({}, None), + ({"1": dict(COUNTS, unknown=0)}, 1), + ], +) +def test_partial_extra_or_invalid_chunks_never_become_zeros(chunks, n): + """The chunk set and each count record must be complete.""" + with pytest.raises(ValueError): + epoch.aggregate_chunks(chunks, n) + + +def test_chunk_check_reads_worker_not_duplicate_shape_pipe_log(tmp_path): + """Read one worker, not its stage's duplicate aggregate log.""" + root = tmp_path / TILE + stage = root / "output" / "run_sp_tile_ngmix_Ng1u" + write(stage / "ngmix_runner/output/ngmix.fits", "fits fixture") + write(stage / "ngmix_runner/logs/process-259-283.log", summary()) + write(stage / "logs/log_sp.log", summary()) + record, ok = completeness.build_manifest( + "tile_ngmix", root, TILE, stage.name + ) + assert ok and record["epoch_cuts"] == COUNTS + write(stage / "ngmix_runner/logs/process-other.log", summary()) + record, ok = completeness.build_manifest( + "tile_ngmix", root, TILE, stage.name + ) + assert not ok and record["status"] == "failed" + assert "epoch_cuts" not in record + + +def test_make_cat_gathers_structured_records_and_preserves_mtime( + tmp_path, monkeypatch +): + """Make-cat needs no logs, and identical manifests keep their mtime.""" + root = tmp_path / TILE + stage = root / "output" / "run_sp_tile_Mc" + write(stage / "make_cat_runner/output/final_cat.fits", "fits fixture") + for k in (1, 2): + write( + root / f"manifests/tile_ngmix_{k}.json", + json.dumps(verdict("tile_ngmix", epoch_cuts=COUNTS)), + ) + monkeypatch.setenv("NGMIX_N_CHUNKS", "2") + record, ok = completeness.build_manifest("tile_make_cat", root, TILE) + assert ok + assert record["epoch_cuts"]["totals"] == { + k: 2 * v for k, v in COUNTS.items() + } + path = tmp_path / "persistent.json" + text = json.dumps(record, sort_keys=True) + completeness.write_if_changed(path, text) + before = path.stat().st_mtime_ns + completeness.write_if_changed(path, text) + assert path.stat().st_mtime_ns == before + write( + root / "manifests/tile_ngmix_2.json", + json.dumps( + dict(verdict("tile_ngmix", epoch_cuts=COUNTS), status="failed") + ), + ) + record, ok = completeness.build_manifest("tile_make_cat", root, TILE) + assert not ok and "epoch_cuts" not in record + + +@pytest.fixture +def campaign(tmp_path): + """Create an idle two-chunk campaign with retained logs and symlinks.""" + run = tmp_path / "run" + products = tmp_path / "products" + tile = run / "tiles" / TILE[:2] / TILE + final_cat = write( + products / "tiles" / TILE[:2] / TILE / f"final_cat-{TILE}.hdf5", + "persistent catalogue", + ) + for path in clean.survivor_paths(tile, TILE).values(): + write(path, "{}" if path.suffix == ".json" else "123456p\n") + write( + tile / "manifests/tile_make_cat.json", + json.dumps(verdict("tile_make_cat")), + ) + for k in (1, 2): + write( + tile / f"manifests/tile_ngmix_{k}.json", + json.dumps(verdict("tile_ngmix")), + ) + name = f"run_sp_tile_ngmix_Ng{k}u" + write( + tile + / f"logs/modules/{name}/ngmix_runner/logs/process-259-283.log", + summary(), + ) + # Original and ShapePipe log copies must not count twice. + write( + tile / f"output/{name}/ngmix_runner/logs/process-259-283.log", + summary(), + ) + write(tile / f"logs/modules/{name}/logs/log_sp.log", summary()) + write(tile / "output/bulk/data", "x" * 1024) + precious = write(tmp_path / "outside/precious", "x" * 5000) + (tile / "output/bulk/link").symlink_to( + precious.parent, target_is_directory=True + ) + (tile / "output/bulk/dangling").symlink_to(tmp_path / "absent") + return run, products, tile, final_cat, precious + + +def tree_bytes(root): + """Snapshot fixture bytes and mtimes without reading symlink targets.""" + return { + str(p): (p.read_bytes(), p.stat().st_mtime_ns) + for p in root.rglob("*") + if p.is_file() and not p.is_symlink() + } + + +def test_dry_run_is_read_only_and_uses_cleaner_keep_rules( + campaign, monkeypatch, capsys +): + """Preview exact whitelist bytes without modifying either filesystem.""" + run, products, tile, cat, precious = campaign + before = tree_bytes(run.parent) + monkeypatch.setattr( + backfill.clean_tile, + "reclaim", + lambda *a: pytest.fail("dry-run must never reclaim"), + ) + assert ( + backfill.main( + [ + "--run-dir", + str(run), + "--products-dir", + str(products), + "--n-chunks", + "2", + "--dry-run", + ] + ) + == 0 + ) + assert tree_bytes(run.parent) == before + assert not (cat.parent / "tile_make_cat.json").exists() + # Same traversal as actual cleaning; no symlink target or keep-file bytes. + keep = set(clean.survivor_paths(tile, TILE).values()) | { + tile / "cleaned.json" + } + expected = sum( + len(data) + for path, (data, _) in before.items() + if Path(path).is_relative_to(tile) and Path(path) not in keep + ) + output = capsys.readouterr().out + assert f"WOULD FREE {expected} bytes" in output + assert "1 ready, 0 skipped, 0 refused" in output + assert precious.stat().st_size == 5000 + + +def test_backfill_publishes_before_calling_shared_cleaner( + campaign, monkeypatch +): + """The real backfill calls only the shared cleaner, after publication.""" + run, products, tile, cat, _ = campaign + called = [] + + def reclaim(tile_dir, tile_id, tomb, final): + record = epoch.require_durable_manifest(final, tile_id, 2) + called.append((tile_dir, tomb, record)) + + monkeypatch.setattr(backfill.clean_tile, "reclaim", reclaim) + assert ( + backfill.main( + [ + "--run-dir", + str(run), + "--products-dir", + str(products), + "--n-chunks", + "2", + ] + ) + == 0 + ) + assert len(called) == 1 + assert called[0][2]["epoch_cuts"]["totals"] == { + k: 2 * v for k, v in COUNTS.items() + } + path = cat.parent / "tile_make_cat.json" + before = path.stat().st_mtime_ns + # Retry reads the durable structured record, no longer requiring logs. + monkeypatch.setattr( + backfill, "read_chunk_logs", lambda *a: pytest.fail("already durable") + ) + backfill.process_tile(tile, products, 2, dry_run=False) + assert path.stat().st_mtime_ns == before + + +def test_missing_catalogue_is_skipped_without_reading_logs( + campaign, monkeypatch, capsys +): + """Never touch unfinished tiles.""" + run, products, tile, cat, _ = campaign + # Point to an empty products root; leave the fixture catalogue intact. + monkeypatch.setattr( + backfill, + "read_chunk_logs", + lambda *a: pytest.fail("unfinished tile must not read logs"), + ) + status, size, record = backfill.process_tile( + tile, products / "absent", 2, True + ) + assert (status, size, record) == ("skipped", 0, None) + assert "SKIP no persisted final_cat" in capsys.readouterr().out + + +def test_missing_summary_refuses_without_publishing_or_cleaning( + campaign, monkeypatch +): + """A partial calibration record leaves all scratch data intact.""" + run, products, tile, cat, _ = campaign + write( + tile + / "logs/modules/run_sp_tile_ngmix_Ng2u/ngmix_runner/logs" + / "process-259-283.log", + "no summary", + ) + monkeypatch.setattr( + backfill.clean_tile, + "reclaim", + lambda *a: pytest.fail("missing evidence must not clean"), + ) + assert ( + backfill.main( + [ + "--run-dir", + str(run), + "--products-dir", + str(products), + "--n-chunks", + "2", + "--dry-run", + ] + ) + == 1 + ) + assert not (cat.parent / "tile_make_cat.json").exists() + assert (tile / "output/bulk/data").exists() + + +def test_cleaner_refuses_without_durable_counts(campaign): + """A persisted catalogue alone no longer authorizes cleaning.""" + _, _, tile, cat, _ = campaign + with pytest.raises(FileNotFoundError): + clean.reclaim(tile, TILE, tile / "cleaned.json", cat) + write( + cat.parent / "tile_make_cat.json", json.dumps(verdict("tile_make_cat")) + ) + with pytest.raises(ValueError): + clean.reclaim(tile, TILE, tile / "cleaned.json", cat) + assert not (tile / "cleaned.json").exists() + assert (tile / "output/bulk/data").exists() + + +def test_durable_record_detects_tampered_total(campaign): + """Validate stored totals against the complete per-chunk evidence.""" + _, _, tile, cat, _ = campaign + record = backfill.backfill_manifest(tile, cat, 2) + record["epoch_cuts"]["totals"]["failed"] += 1 + write(cat.parent / "tile_make_cat.json", json.dumps(record)) + with pytest.raises(ValueError, match="invalid epoch cuts aggregate"): + epoch.require_durable_manifest(cat, TILE) + + +def test_one_root_refused_before_any_writes(campaign): + """Products within scratch cannot be durable survivors.""" + run, _, _, _, _ = campaign + with pytest.raises(SystemExit): + backfill.main( + [ + "--run-dir", + str(run), + "--products-dir", + str(run), + "--n-chunks", + "2", + "--dry-run", + ] + ) diff --git a/workflow/scripts/backfill_epoch_cuts.py b/workflow/scripts/backfill_epoch_cuts.py index 3be22294d..d42951330 100644 --- a/workflow/scripts/backfill_epoch_cuts.py +++ b/workflow/scripts/backfill_epoch_cuts.py @@ -17,21 +17,31 @@ import clean_tile from completeness import write_if_changed -from epoch_cuts import (aggregate_chunks, read_chunk_logs, - require_durable_manifest, validate_counts) +from epoch_cuts import ( + aggregate_chunks, + read_chunk_logs, + require_durable_manifest, + validate_counts, +) def resolve_config(config_path): - """Resolve the run's frozen base defaults plus its overlay, not today's defaults.""" + """Resolve frozen base defaults plus the run's overlay.""" import yaml from run_config import apply_defaults, merge base = config_path.parent / "workflow" / "config.yaml" - config = apply_defaults(merge(yaml.safe_load(base.read_text()), - yaml.safe_load(config_path.read_text()))) - return (Path(config["outputs"]["run_dir"]), - Path(config["outputs"]["products_dir"]), - int(config["ngmix_chunks"])) + config = apply_defaults( + merge( + yaml.safe_load(base.read_text()), + yaml.safe_load(config_path.read_text()), + ) + ) + return ( + Path(config["outputs"]["run_dir"]), + Path(config["outputs"]["products_dir"]), + int(config["ngmix_chunks"]), + ) def catalogue_path(products_dir, tile): @@ -45,23 +55,27 @@ def catalogue_path(products_dir, tile): def backfill_manifest(tile_dir, final_cat, n_chunks): - """Build the durable make-cat manifest, or validate and reuse an existing one.""" + """Build the durable manifest, or validate and reuse an existing one.""" durable = final_cat.parent / "tile_make_cat.json" if durable.exists(): return require_durable_manifest(final_cat, tile_dir.name, n_chunks) path = tile_dir / "manifests" / "tile_make_cat.json" manifest = json.loads(path.read_text()) - if (manifest.get("stage") != "tile_make_cat" - or manifest.get("unit") != tile_dir.name - or manifest.get("status") != "complete"): + if ( + manifest.get("stage") != "tile_make_cat" + or manifest.get("unit") != tile_dir.name + or manifest.get("status") != "complete" + ): raise ValueError(f"not a successful make-cat manifest: {path}") chunks = {} for k in range(1, n_chunks + 1): chunk_path = tile_dir / "manifests" / f"tile_ngmix_{k}.json" chunk = json.loads(chunk_path.read_text()) - if (chunk.get("stage") != "tile_ngmix" - or chunk.get("unit") != tile_dir.name - or chunk.get("status") != "complete"): + if ( + chunk.get("stage") != "tile_ngmix" + or chunk.get("unit") != tile_dir.name + or chunk.get("status") != "complete" + ): raise ValueError(f"not a successful chunk manifest: {chunk_path}") if "epoch_cuts" in chunk: chunks[str(k)] = validate_counts(chunk["epoch_cuts"]) @@ -76,7 +90,7 @@ def backfill_manifest(tile_dir, final_cat, n_chunks): def process_tile(tile_dir, products_dir, n_chunks, dry_run): - """Validate one tile, report reclaimable bytes and optionally persist/reclaim.""" + """Validate one tile, report reclaimable bytes and optionally reclaim.""" tile = tile_dir.name final_cat = catalogue_path(products_dir, tile) if final_cat is None: @@ -86,41 +100,56 @@ def process_tile(tile_dir, products_dir, n_chunks, dry_run): clean_tile.require_survivors(survivors, tile) manifest = backfill_manifest(tile_dir, final_cat, n_chunks) tombstone = tile_dir / "cleaned.json" - freed = clean_tile.prune(tile_dir, set(survivors.values()) | {tombstone}, - [], dry_run=True) + freed = clean_tile.prune( + tile_dir, set(survivors.values()) | {tombstone}, [], dry_run=True + ) record = manifest["epoch_cuts"] - print(f"[backfill] {tile}: {'WOULD FREE' if dry_run else 'FREE'} {freed} bytes; " - f"{n_chunks} chunks; totals={json.dumps(record['totals'], sort_keys=True)}") + action = "WOULD FREE" if dry_run else "FREE" + totals_text = json.dumps(record["totals"], sort_keys=True) + print( + f"[backfill] {tile}: {action} {freed} bytes; " + f"{n_chunks} chunks; totals={totals_text}" + ) if not dry_run: # Publish atomically BEFORE the cleaner can touch scratch. Never run a # Snakemake science stage merely to retrofit this small durable record. - write_if_changed(final_cat.parent / "tile_make_cat.json", - json.dumps(manifest, indent=2, sort_keys=True) + "\n") + write_if_changed( + final_cat.parent / "tile_make_cat.json", + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + ) clean_tile.reclaim(tile_dir, tile, tombstone, final_cat) return "ready", freed, record["totals"] def main(argv=None): - """Run the backfill CLI; refusals leave the affected tile completely intact.""" + """Run the backfill CLI; refusals leave the affected tile intact.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-dir", type=Path, required=True) parser.add_argument("--products-dir", type=Path) parser.add_argument("--n-chunks", type=int) - parser.add_argument("--config", type=Path, - help="run overlay beside frozen workflow/config.yaml") + parser.add_argument( + "--config", + type=Path, + help="run overlay beside frozen workflow/config.yaml", + ) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args(argv) if args.config: run, products, chunks = resolve_config(args.config) if run.resolve() != args.run_dir.resolve(): parser.error("--run-dir disagrees with the run config") - if args.products_dir and args.products_dir.resolve() != products.resolve(): + if ( + args.products_dir + and args.products_dir.resolve() != products.resolve() + ): parser.error("--products-dir disagrees with the run config") if args.n_chunks is not None and args.n_chunks != chunks: parser.error("--n-chunks disagrees with the run config") args.products_dir, args.n_chunks = products, chunks if args.products_dir is None or args.n_chunks is None or args.n_chunks < 1: - parser.error("provide --config, or both --products-dir and positive --n-chunks") + parser.error( + "provide --config, or both --products-dir and positive --n-chunks" + ) run = args.run_dir.resolve() products = args.products_dir.resolve() if products == run or run in products.parents: @@ -134,11 +163,15 @@ def main(argv=None): if not re.fullmatch(r"\d+\.\d+", tile_dir.name): continue try: - if (tile_dir.is_symlink() or tile_dir.parent.is_symlink() - or tile_dir.parent.name != tile_dir.name[:2]): + if ( + tile_dir.is_symlink() + or tile_dir.parent.is_symlink() + or tile_dir.parent.name != tile_dir.name[:2] + ): raise ValueError(f"unexpected tile/shard path: {tile_dir}") - status, size, record = process_tile(tile_dir, products, - args.n_chunks, args.dry_run) + status, size, record = process_tile( + tile_dir, products, args.n_chunks, args.dry_run + ) counts[status] += 1 freed += size if record: @@ -146,11 +179,16 @@ def main(argv=None): totals[key] = totals.get(key, 0) + value except (OSError, ValueError, SystemExit) as exc: counts["refused"] += 1 - print(f"[backfill] {tile_dir.name}: REFUSED {exc}", file=sys.stderr) - print(f"[backfill] {'DRY RUN' if args.dry_run else 'DONE'}: {counts['ready']} ready, " - f"{counts['skipped']} skipped, {counts['refused']} refused; " - f"{freed} bytes ({freed / 2**30:.3f} GiB) " - f"{'would be freed' if args.dry_run else 'reclaimed'}") + print( + f"[backfill] {tile_dir.name}: REFUSED {exc}", file=sys.stderr + ) + action = "DRY RUN" if args.dry_run else "DONE" + print( + f"[backfill] {action}: {counts['ready']} ready, " + f"{counts['skipped']} skipped, {counts['refused']} refused; " + f"{freed} bytes ({freed / 2**30:.3f} GiB) " + f"{'would be freed' if args.dry_run else 'reclaimed'}" + ) print(f"[backfill] campaign totals={json.dumps(totals, sort_keys=True)}") return 1 if counts["refused"] else 0 diff --git a/workflow/scripts/epoch_cuts.py b/workflow/scripts/epoch_cuts.py index 01bd8fe40..31a563d02 100644 --- a/workflow/scripts/epoch_cuts.py +++ b/workflow/scripts/epoch_cuts.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Archive PR #922 epoch-cut diagnostics, independently of module-log retention. +"""Archive PR #922 epoch-cut diagnostics without retaining module logs. Chunk completeness checks read the single end-of-loop summary before temp() removes the chunk directory. The make-cat check gathers those structured chunk @@ -12,17 +12,25 @@ import re from pathlib import Path -FIELDS = ("considered", "masked_fraction", "central_veto", "failed", - "objects_emptied") +FIELDS = ( + "considered", + "masked_fraction", + "central_veto", + "failed", + "objects_emptied", +) def validate_counts(counts): - """Return an exact, non-negative integer count record or raise ValueError.""" + """Validate an exact, non-negative integer count record.""" if not isinstance(counts, dict) or set(counts) != set(FIELDS): raise ValueError(f"epoch cuts must contain exactly {FIELDS}") if any(type(v) is not int or v < 0 for v in counts.values()): raise ValueError("epoch cuts must be non-negative integers") - if sum(counts[k] for k in ("masked_fraction", "central_veto", "failed")) > counts["considered"]: + if ( + sum(counts[k] for k in ("masked_fraction", "central_veto", "failed")) + > counts["considered"] + ): raise ValueError("epoch cuts exceed epochs considered") if counts["objects_emptied"] > counts["considered"]: raise ValueError("objects emptied exceeds epochs considered") @@ -30,11 +38,16 @@ def validate_counts(counts): def parse_epoch_cuts(text): - """Parse exactly one complete end-of-loop summary; reject retry ambiguity.""" - lines = [line.split("epoch cuts:", 1)[1].strip() - for line in text.splitlines() if "epoch cuts:" in line] + """Parse one complete end-of-loop summary; reject retry ambiguity.""" + lines = [ + line.split("epoch cuts:", 1)[1].strip() + for line in text.splitlines() + if "epoch cuts:" in line + ] if len(lines) != 1: - raise ValueError(f"expected one epoch cuts summary, found {len(lines)}") + raise ValueError( + f"expected one epoch cuts summary, found {len(lines)}" + ) counts = {} for token in lines[0].split(): match = re.fullmatch(r"([a-z_]+)=(\d+)", token) @@ -46,9 +59,14 @@ def parse_epoch_cuts(text): def read_chunk_logs(run_root): """Read ngmix worker logs only, not copies or ShapePipe's aggregate log.""" - paths = sorted((Path(run_root) / "ngmix_runner" / "logs").glob("process-*.log")) + paths = sorted( + (Path(run_root) / "ngmix_runner" / "logs").glob("process-*.log") + ) if len(paths) != 1: - raise ValueError(f"expected one ngmix worker log under {run_root}, found {len(paths)}") + raise ValueError( + f"expected one ngmix worker log under {run_root}, " + f"found {len(paths)}" + ) return parse_epoch_cuts(paths[0].read_text()) @@ -58,40 +76,58 @@ def aggregate_chunks(chunks, n_chunks): raise ValueError("n_chunks must be a positive integer") expected = {str(k) for k in range(1, n_chunks + 1)} if set(chunks) != expected: - raise ValueError(f"expected chunks 1..{n_chunks}, found {sorted(chunks)}") - chunks = {str(k): validate_counts(chunks[str(k)]) - for k in range(1, n_chunks + 1)} - return {"schema_version": 1, "n_chunks": n_chunks, "chunks": chunks, - "totals": {key: sum(c[key] for c in chunks.values()) for key in FIELDS}} + raise ValueError( + f"expected chunks 1..{n_chunks}, found {sorted(chunks)}" + ) + chunks = { + str(k): validate_counts(chunks[str(k)]) for k in range(1, n_chunks + 1) + } + return { + "schema_version": 1, + "n_chunks": n_chunks, + "chunks": chunks, + "totals": { + key: sum(c[key] for c in chunks.values()) for key in FIELDS + }, + } def gather_manifests(tile_dir, n_chunks): - """Gather successful chunk completeness records, after the merge DAG edge.""" + """Gather successful chunk records after the merge DAG edge.""" chunks = {} for k in range(1, n_chunks + 1): path = Path(tile_dir) / "manifests" / f"tile_ngmix_{k}.json" manifest = json.loads(path.read_text()) - if (manifest.get("stage") != "tile_ngmix" - or manifest.get("unit") != Path(tile_dir).name - or manifest.get("status") != "complete"): + if ( + manifest.get("stage") != "tile_ngmix" + or manifest.get("unit") != Path(tile_dir).name + or manifest.get("status") != "complete" + ): raise ValueError(f"not a successful chunk manifest: {path}") chunks[str(k)] = manifest.get("epoch_cuts") return aggregate_chunks(chunks, n_chunks) def require_durable_manifest(final_cat, tile, n_chunks=None): - """Validate catalogue existence and its persistent completeness/count record.""" + """Validate the persisted catalogue and its completeness/count record.""" final_cat = Path(final_cat) if not final_cat.is_file() or final_cat.stat().st_size == 0: raise ValueError(f"no persisted final_cat: {final_cat}") path = final_cat.parent / "tile_make_cat.json" manifest = json.loads(path.read_text()) - if (manifest.get("stage") != "tile_make_cat" or manifest.get("unit") != tile - or manifest.get("status") != "complete"): - raise ValueError(f"not a successful make-cat manifest for {tile}: {path}") + if ( + manifest.get("stage") != "tile_make_cat" + or manifest.get("unit") != tile + or manifest.get("status") != "complete" + ): + raise ValueError( + f"not a successful make-cat manifest for {tile}: {path}" + ) record = manifest.get("epoch_cuts") or {} - expected = aggregate_chunks(record.get("chunks", {}), - n_chunks if n_chunks is not None else record.get("n_chunks")) + expected = aggregate_chunks( + record.get("chunks", {}), + n_chunks if n_chunks is not None else record.get("n_chunks"), + ) if record != expected: raise ValueError(f"invalid epoch cuts aggregate: {path}") return manifest From 44b6478ba804dee596b7b24e304834aec57a62be Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Mon, 12 Oct 2026 11:51:39 -0400 Subject: [PATCH 5/6] Regenerate params pins after merging feat/masked-pixels --- tests/workflow/params_pin.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/workflow/params_pin.json b/tests/workflow/params_pin.json index dec484c31..99c8950df 100644 --- a/tests/workflow/params_pin.json +++ b/tests/workflow/params_pin.json @@ -3,7 +3,7 @@ "rules": { "all": "572122d8d1901e12ff591b30adf405f8920be8183129befececbb819f3392ed8", "clean_exposure": "22cb76b13a5205d20a02a9bd3b8c8bea5ea2801555e24dd2b84b11145f7e79d9", - "clean_tile": "a5c07b0461526ed407df36a291deb866d4181524b4c8e0fbc3dd047fd9d28479", + "clean_tile": "c6a91af666e86bd9735effa005dee862e7d3682079eac23c56a877505bdbe76e", "exp_get_images": "a9e9eca0a99a348e43e0cd44d0c0aae0f890773d03e35007fb77f886b7d21f5f", "exp_maps": "6114b2bb508217eb1a87c89a295485c4cc17f329391a27832548f525a3ef1d2e", "exp_persist": "302e2837542bc1102430c27c81c600b7cda32e8bddcb5fd60d33950987609fff", @@ -18,7 +18,7 @@ "tile_find_exposures": "c883495a7ec196c61ef770507865f43834b08523793d337e8ddeaef0ad62f117", "tile_get_catalogue": "84ffd03c4bde248c06aa58a0f4b7f399390a0ff3ecbcf07e03d718df4fe0cd7f", "tile_get_images": "96e235dad9cdcde8ecf11f4516bc2303dbba1c8115ad343cdb26d3d1513da671", - "tile_make_cat": "831f2b104d20326c3836e29d252118b8d2dd9a277be26857f47162b8250e3ca2", + "tile_make_cat": "28e38d428333b7998511e91a2ae5f71beb18da4e8342633a045a92ff706033c2", "tile_merge_cats": "0783c93deeb7f90bdb9772bbc4117851dff1a671518c81f67edd4c7c79cc7e6a", "tile_merge_headers": "3e674f961fd5bb93ce1f3b0e2bb072a25a421f2828315a3db45a960c44a42111", "tile_ngmix": "a35a6326f345e2fac77444724e917f5a36c3c19377a6c3c66750c8ee32250e62", @@ -26,7 +26,7 @@ "tile_vignets": "1d434b4a6a4bcdeb8f1e5e9a5e4f228b67ade400245b41b033a2c883fb6a3267" }, "schema": 1, - "sha256": "8065298d1fb835b0346c75c9946bbe2f25f4b99131f025662341e388a71c7fbd", + "sha256": "6c1a7cf52778e4423ab1e3d4f1424dc7f039a8173266216206556a52491f2935", "unit_pre": { "exp_get_images": "dac6685a207dae3ea81d296ce53e9ba75d4636cf1f81e6399f2dda452680c992", "exp_psf": "87fc8ea1153a709ab7ba56542cfacc041f5f8b7f6cf2f38a9bc9d8004d1cecb4", From d731900e0f201130bba364826da7f02606b58672 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Mon, 12 Oct 2026 11:55:47 -0400 Subject: [PATCH 6/6] test_epoch_cuts_persist: the make_cat fixture is the HDF5 final catalogue --- tests/unit/test_epoch_cuts_persist.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_epoch_cuts_persist.py b/tests/unit/test_epoch_cuts_persist.py index 3e3f6f011..f31b3a2f7 100644 --- a/tests/unit/test_epoch_cuts_persist.py +++ b/tests/unit/test_epoch_cuts_persist.py @@ -177,7 +177,7 @@ def test_make_cat_gathers_structured_records_and_preserves_mtime( """Make-cat needs no logs, and identical manifests keep their mtime.""" root = tmp_path / TILE stage = root / "output" / "run_sp_tile_Mc" - write(stage / "make_cat_runner/output/final_cat.fits", "fits fixture") + write(stage / "make_cat_runner/output/final_cat.hdf5", "hdf5 fixture") for k in (1, 2): write( root / f"manifests/tile_ngmix_{k}.json",