fix(workflow): keep vignette dependencies unconditional on resume

Fix forward the unsafe conditional edge in ca224f7. A completion manifest does not mean a chunk will not run: after the vignets manifest is swept, a newer tile_detect input schedules all ngmix chunks, but the conditional removed their dependency on the writer. They could start before the store exists or race its rebuild.

Remove ngmix_vignets and restore every chunk unconditional vignets input. Keep the unconditional make_cat input. The accepted trade-off is that sweeping a tile now rebuilds its store and reruns all ngmix chunks and their merge, even if earlier outputs survived. Correct ordering is worth the extra ngmix time on these resumed tiles; no completed-output inference may remove a reader edge.

Replace the reuse-oriented tests and documentation. Test the original swept-store scenario and the review counterexample with newer tile_detect, in data and image-sims modes, with completed, reclaimed, partial and missing chunk products. Assert every scheduled chunk and make_cat depends directly on tile_vignets. The new merged-state regressions fail on ca224f7. Shell bodies and params.pre are unchanged.

Author: cailmdaley

tests/workflow/test_resume.py

@@ -1,4 +1,4 @@
-"""Resume a tile after its node-local vignette manifest has been swept."""
+"""Every resumed reader must wait for the rebuilt node-local vignette store."""
 
 import os
 import shutil
@@ -9,11 +9,6 @@
 from snakemake.io import is_flagged
 
 
-def test_resume_edge_cut_requires_input_trigger_off(profile):
-    """Changing a satisfied chunk's input set must not invalidate it."""
-    assert "input" not in profile["rerun-triggers"]
-
-
 def seed_outputs(campaign, resolve_dag):
     """Materialise satisfied outputs without executing the science pipeline."""
     with resolve_dag(campaign) as dag:
@@ -32,74 +27,57 @@ def seed_outputs(campaign, resolve_dag):
 
 
 @pytest.mark.parametrize("state", [
-    "merged", "chunks_done", "partial", "missing_chunkdir",
+    "merged", "merged_reclaimed", "chunks_done", "partial", "missing_chunkdir",
+])
+@pytest.mark.parametrize("invalidate_detect", [False, True], ids=[
+    "swept_only", "swept_and_upstream_invalidated",
 ])
-def test_resume_after_vignets_swept(campaign, resolve_dag, state):
-    """Only unfinished chunks read the rebuilt store; make_cat always does."""
+def test_resume_after_vignets_swept(
+    campaign, resolve_dag, state, invalidate_detect,
+):
+    """Sweep alone and sweep plus upstream invalidation both preserve ordering.
+
+    The newer tile_detect input reproduces the review's counterexample:
+    completed chunks can still be scheduled and must retain their writer edge.
+    """
     seed_outputs(campaign, resolve_dag)
     tile = next(iter(campaign.ready))
     root = campaign.tile_manifest(tile, "tile_vignets").parents[1]
     campaign.tile_manifest(tile, "tile_vignets").unlink()
     campaign.tile_manifest(tile, "tile_make_cat").unlink()
     campaign.final_cat(tile).unlink()
-    if state == "merged":
+    if state == "merged_reclaimed":
         # merge_cats reclaims its temp inputs in production.
         for path in (root / "output").glob("run_sp_tile_ngmix_Ng*u"):
             shutil.rmtree(path)
-    else:
+    elif state != "merged":
         campaign.tile_manifest(tile, "tile_merge_cats").unlink()
     if state == "partial":
         campaign.tile_manifest(tile, "tile_ngmix_8").unlink()
     if state == "missing_chunkdir":
         shutil.rmtree(root / "output/run_sp_tile_ngmix_Ng8u")
+    if invalidate_detect:
+        campaign.tile_manifest(tile, "tile_detect").touch()
     with resolve_dag(campaign) as dag:
         pending = list(dag.workflow.dag.needrun_jobs())
         counts = Counter(job.rule.name for job in pending)
-        print(f"Resume dry-run jobs ({state}):", dict(sorted(counts.items())))
+        print(
+            f"Resume dry-run jobs ({state}, invalidate_detect={invalidate_detect}):",
+            dict(sorted(counts.items())),
+        )
         assert counts["tile_vignets"] == 1
+        assert counts["tile_ngmix"] == dag.namespace["NGMIX_CHUNKS"]
+        assert counts["tile_merge_cats"] == 1
         assert counts["tile_make_cat"] == 1
-        assert counts["tile_ngmix"] == (state in {"partial", "missing_chunkdir"})
-        assert counts["tile_merge_cats"] == (state != "merged")
-        vignets = next(j for j in pending if j.rule.name == "tile_vignets")
-        for job in pending:
-            if job.rule.name in {"tile_make_cat", "tile_ngmix"}:
-                assert vignets in dag.workflow.dag.dependencies[job]
-
-
-def test_finished_tile_after_store_only_rebuild(campaign, resolve_dag):
-    """A resumed tile must stay finished on the next invocation."""
-    seed_outputs(campaign, resolve_dag)
-    tile = next(iter(campaign.ready))
-    # The store-only rebuild is newer than the preserved chunk products;
-    # make_cat then succeeds. alloc_resume skips tiles with a final catalogue.
-    campaign.tile_manifest(tile, "tile_vignets").touch()
-    with resolve_dag(campaign) as dag:
-        outputs = [
-            Path(output)
-            for job in dag.jobs_for("tile_make_cat")
-            if job.wildcards.tile == tile
-            for output in job.output
+        writer = next(j for j in pending if j.rule.name == "tile_vignets")
+        readers = [
+            job for job in pending
+            if job.rule.name in {"tile_make_cat", "tile_ngmix"}
         ]
-    # Include any additional durable outputs, such as published tile stats.
-    for path in outputs:
-        path.touch()
-    with resolve_dag(campaign) as dag:
-        counts = Counter(j.rule.name for j in dag.workflow.dag.needrun_jobs())
-        assert counts["tile_ngmix"] == 0
-        assert counts["tile_merge_cats"] == 0
-        assert counts["tile_vignets"] == 0
-        assert counts["tile_make_cat"] == 0
-
-
-def test_real_vignets_rebuild_still_invalidates_chunks(campaign, resolve_dag):
-    """The storage-only exception must not hide upstream science changes."""
-    seed_outputs(campaign, resolve_dag)
-    tile = next(iter(campaign.ready))
-    campaign.tile_manifest(tile, "tile_detect").touch()
-    campaign.tile_manifest(tile, "tile_make_cat").unlink()
-    campaign.final_cat(tile).unlink()
-    with resolve_dag(campaign) as dag:
-        pending = list(dag.workflow.dag.needrun_jobs())
-        counts = Counter(j.rule.name for j in pending)
-        assert counts["tile_vignets"] == 1
-        assert counts["tile_ngmix"] == dag.namespace["NGMIX_CHUNKS"]
+        assert len(readers) == dag.namespace["NGMIX_CHUNKS"] + 1
+        for job in readers:
+            assert job.wildcards.tile == tile
+            assert writer in dag.workflow.dag.dependencies[job], (
+                f"pending reader {job.rule.name} {job.wildcards} has no writer "
+                f"dependency; inputs={list(job.input)}"
+            )

workflow/README.md

@@ -275,10 +275,9 @@ a multi-node allocation would split a tile's vignets from its ngmix chunks.
 A job cut short by its time limit is resumed by submitting the same script again.
 The store died with the job, so before it runs `sp` deletes `tile_vignets.json` for each listed tile that has no `final_cat` yet.
 `tile_make_cat` declares that manifest as an input, so `tile_vignets` rebuilds the store even if all chunks and their merge are complete.
-Completed chunks keep their products; only unfinished chunks wait for the rebuilt store and run again.
-A chunk whose temporary directory is missing must also rerun if its merge has not succeeded.
-This storage-only recovery trusts the chunk completion manifests: to explicitly recompute a completed chunk, remove its manifest too.
-To rebuild a finished tile, remove its final catalogue first, as for the exposure-edge cut in `tile.smk`.
+Every ngmix chunk also declares the vignets manifest unconditionally, so all chunks rerun after the store is rebuilt, followed by their merge and `make_cat`.
+This spends ngmix time on swept tiles, but guarantees that every scheduled reader waits for the writer, even if another input or rerun trigger invalidates a previously completed chunk.
+A completion manifest alone cannot prove that a chunk will not run in this invocation.
 `sp` also records the job holding the campaign's state dir, and releases that job's lock once the job has left the queue; any other lock stops the run.
 With `-n` both only report.
 

workflow/bin/sp

@@ -195,9 +195,10 @@ STATE_DIR="${SP_STATE_DIR:-${RUN_DIR}-state}"
 #     tile.smk's "node-local vignette store is missing" guard, so that
 #     manifest is deleted first. tile_make_cat declares it as an input, so
 #     tile_vignets rebuilds the store even if merge_cats is already complete.
-#     Completed chunks skip this storage-only rebuild (ngmix_vignets in
-#     tile.smk); unfinished chunks wait for it. Tiles with a final_cat are
-#     never touched.
+#     Every ngmix chunk retains its dependency on the writer and reruns after
+#     it, even if its previous output survived. This costs ngmix time on resume
+#     but ensures no reader can race the rebuild. Tiles with a final_cat are
+#     never swept.
 #   * A lock left by a killed in-allocation run is released. sp records the
 #     job that holds the state dir; when that job is no longer in the queue,
 #     its lock is stale. Any other lock (a run sp did not start in an

workflow/rules/tile.smk

@@ -614,30 +614,6 @@ rule tile_vignets:
         sp_shell("tile_vignets", f"config_tile_PiViVi_{PSF_MODEL}.ini",
                  check_args=' --run-dir "$SP_LOCAL" --unit {wildcards.tile}')
 
-def ngmix_vignets(wc):
-    """Keep completed chunks when only the node-local store was swept.
-
-    A missing temp chunk directory is harmless after merge_cats succeeds, but
-    before that merge it must be rebuilt and therefore needs the store edge.
-    Unfinished tiles with existing vignets manifests retain normal rerun
-    propagation. Finished tiles also cut the edge, so a restored manifest's
-    newer timestamp cannot wake old chunks on the next invocation. As with
-    the finished-tile exposure cut above, the profile disables the input
-    trigger. To force a chunk to recompute after a sweep, remove its manifest
-    too; to rebuild a finished tile, remove its final catalogue first.
-    """
-    root = Path(tile_dir(wc.tile))
-    vignets = root / "manifests/tile_vignets.json"
-    chunk = root / f"manifests/tile_ngmix_{wc.chunk}.json"
-    chunkdir = root / f"output/run_sp_tile_ngmix_Ng{wc.chunk}u"
-    merged = root / "manifests/tile_merge_cats.json"
-    if (not vignets.exists() or tile_finished(wc.tile)) and chunk.exists() and (
-        chunkdir.exists() or merged.exists()
-    ):
-        return []
-    return [str(vignets)]
-
-
 # ngmix shape measurement — N chunks per tile (D4). Each chunk LOOKS UP its own
 # CLOSED catalogue-row range in the file tile_vignets materialised at the top of this
 # group job (TILE_NGMIX_RANGES); the ranges are knowable only at EXECUTION time,
@@ -650,9 +626,9 @@ def ngmix_vignets(wc):
 rule tile_ngmix:
     group: TILE_GROUP
     input:
-        # A completed chunk doesn't read a rebuilt store on resume. Missing
-        # chunks keep the manifest edge and wait for this group's writer.
-        vignets  = ngmix_vignets,
+        # Unconditional: even a completed chunk can be invalidated by another
+        # input or rerun trigger. Every scheduled reader must wait for the store.
+        vignets  = rules.tile_vignets.output.manifest,
         sx       = rules.tile_detect.output.manifest,
     output:
         manifest = f"{TILE_DIR}/manifests/tile_ngmix_{{chunk}}.json",