feat(nextflow): self-heal torn-shard/IO failures on per-position steps + default to non-preemptible cpu partition

nextflow/modules/common.nf

@@ -40,6 +40,48 @@ def biahub_cmd() {
 }
 
 
+// Wrap a per-position command with clean-and-retry self-healing. A per-position
+// task killed mid-write (SLURM preemption / timeout / OOM, or a transient storage
+// I/O error) can leave a torn or partial zarr v3 shard in the output position.
+// Because flat-field / deskew / apply-inv-tf / virtual-stain write partial chunks
+// via read-modify-write, the NEXT attempt (a Nextflow retry, or a later `-resume`)
+// reads that torn shard back and zarr's `zarrs` codec pipeline aborts with a
+// non-signal exit (1) that the global errorStrategy would otherwise `terminate`
+// on — so retries and `-resume` keep dying on the same position. The same root
+// cause surfaces as several different messages depending on which codec hits the
+// corruption (see czbiohub-sf/iohub#415, czbiohub-sf/biahub#286):
+//   "the checksum is invalid" / "encoded shard is smaller than the expected size"
+//   / "blosc encoded value is invalid" / ...
+//
+// Rather than enumerate every message, this heals on ANY failure. A per-position
+// task always fully recomputes its position from the input, so on failure it is
+// always safe to remove that position's chunk data (the zarr v3 `c/` directories
+// under `${output_zarr}/${position}`, preserving every `zarr.json` scaffold from
+// the `--init` step) and run `cmd` once more from clean chunks: this recovers
+// every torn-shard flavour (present and future) and gives transient errors a
+// second chance. If the retry ALSO fails, its exit status propagates and the
+// global errorStrategy handles it exactly as before — so genuine bugs are not
+// masked, they just cost one extra attempt. No-op on success (nothing is
+// deleted), and the delete is scoped to a single position group, so concurrent
+// per-position tasks never touch each other.
+//
+// `cmd` must be a single shell command (no trailing backslashes): it is run, and
+// re-run verbatim on failure. `|| heal_status=$?` captures the exit code without
+// tripping the `-e` shell option so the retry logic can run.
+def checksum_heal(output_zarr, position, cmd) {
+    def pos_dir = "${output_zarr}/${position}"
+    return """
+    heal_status=0
+    ${cmd} || heal_status=\$?
+    if [ "\$heal_status" -ne 0 ]; then
+        echo "[self-heal] ${position}: attempt failed (exit \$heal_status); clearing chunk data under ${pos_dir} and retrying once"
+        find "${pos_dir}" -type d -name c -prune -exec rm -rf {} + 2>/dev/null || true
+        ${cmd}
+    fi
+    """.stripIndent()
+}
+
+
 // List the position keys of a plate zarr, one per line, for fan-out.
 process list_positions {
     label 'cpu_local'

nextflow/modules/deskew.nf

@@ -14,7 +14,7 @@
 // Nextflow task.  See also:
 // examples/submitit_debug_nextflow/2026-05-27-submitit-debug-nextflow-concerns.md
 
-include { parse_resources; biahub_cmd; slurm_logs; slurm_log_dir } from './common'
+include { parse_resources; biahub_cmd; checksum_heal; slurm_logs; slurm_log_dir } from './common'
 
 
 process init_deskew {
@@ -57,11 +57,9 @@ process run_deskew {
     val position
 
     script:
+    def cmd = "${biahub_cmd()} deskew --cluster debug -i \"${input_zarr}/${position}\" -o \"${output_zarr}\" -c \"${config}\""
     """
-    ${biahub_cmd()} deskew --cluster debug \
-        -i "${input_zarr}/${position}" \
-        -o "${output_zarr}" \
-        -c "${config}"
+    ${checksum_heal(output_zarr, position, cmd)}
     """
 }
 

nextflow/modules/flat_field.nf

@@ -13,7 +13,7 @@
 // the Nextflow task.  See also:
 // examples/submitit_debug_nextflow/2026-05-27-submitit-debug-nextflow-concerns.md
 
-include { parse_resources; biahub_cmd; slurm_logs; slurm_log_dir } from './common'
+include { parse_resources; biahub_cmd; checksum_heal; slurm_logs; slurm_log_dir } from './common'
 
 
 process init_flat_field {
@@ -56,11 +56,9 @@ process run_flat_field {
     val position
 
     script:
+    def cmd = "${biahub_cmd()} flat-field --cluster debug -i \"${input_zarr}/${position}\" -o \"${output_zarr}\" -c \"${config}\""
     """
-    ${biahub_cmd()} flat-field --cluster debug \
-        -i "${input_zarr}/${position}" \
-        -o "${output_zarr}" \
-        -c "${config}"
+    ${checksum_heal(output_zarr, position, cmd)}
     """
 }
 

nextflow/modules/reconstruct.nf

@@ -20,7 +20,7 @@
 // resource scheduling, so the CLI must NOT submit its own SLURM jobs.
 // See: examples/submitit_debug_nextflow/2026-05-27-submitit-debug-nextflow-concerns.md
 
-include { parse_resources; biahub_cmd; slurm_logs; slurm_log_dir } from './common'
+include { parse_resources; biahub_cmd; checksum_heal; slurm_logs; slurm_log_dir } from './common'
 
 
 process init_apply_inv_tf {
@@ -104,12 +104,9 @@ process run_apply_inv_tf {
     val position
 
     script:
+    def cmd = "${biahub_cmd()} apply-inv-tf --cluster debug -i \"${input_zarr}/${position}\" -t \"${tf_zarr}\" -o \"${output_zarr}\" -c \"${config}\""
     """
-    ${biahub_cmd()} apply-inv-tf --cluster debug \
-        -i "${input_zarr}/${position}" \
-        -t "${tf_zarr}" \
-        -o "${output_zarr}" \
-        -c "${config}"
+    ${checksum_heal(output_zarr, position, cmd)}
     """
 }
 

nextflow/modules/virtual_stain.nf

@@ -29,7 +29,7 @@
 // extra (cytoland → viscy-utils provides the `viscy` console script), so the
 // tasks here run in that extra's environment rather than the plain biahub env.
 
-include { parse_resources; slurm_logs; slurm_log_dir } from './common'
+include { parse_resources; checksum_heal; slurm_logs; slurm_log_dir } from './common'
 
 // Command prefix for tools that require biahub's `stain` extra. Both
 // `biahub virtual-stain` (it imports cytoland) and `viscy preprocess` need it.
@@ -118,11 +118,9 @@ process run_virtual_stain {
     val position
 
     script:
+    def cmd = "${stain_cmd('biahub')} virtual-stain --cluster debug -i \"${input_zarr}/${position}\" -o \"${output_zarr}\" -c \"${config}\""
     """
-    ${stain_cmd('biahub')} virtual-stain --cluster debug \
-        -i "${input_zarr}/${position}" \
-        -o "${output_zarr}" \
-        -c "${config}"
+    ${checksum_heal(output_zarr, position, cmd)}
     """
 }
 

nextflow/nextflow.config

@@ -86,16 +86,21 @@ profiles {
     // SLURM execution, with each label routed to a partition:
     //   cpu_local -> local executor (lightweight init / list-positions steps
     //                stay on the head node; only the memory ceiling differs).
-    //   cpu       -> `preempted` partition. This is the bulk per-position work
-    //                plus the one-shot compute-tf / preprocess steps. When SLURM
-    //                reclaims one of these jobs it is CANCELLED (Nextflow submits
-    //                every job with `#SBATCH --no-requeue`, so SLURM cannot
-    //                requeue it), Nextflow's executor sees the failure, and the
-    //                global `errorStrategy 'retry'` (maxRetries above) resubmits
-    //                it as a fresh, tracked task. Do NOT add SLURM `--requeue`
-    //                here: Nextflow does not follow a requeued job, so it reads
-    //                the killed run's exit code as a failure AND SLURM reruns the
-    //                original — a double execution racing on the same output.
+    //   cpu       -> `cpu` partition (non-preemptible). This is the bulk
+    //                per-position work plus the one-shot compute-tf / preprocess
+    //                steps. It was previously `preempted` (cheaper but reclaimable),
+    //                but SLURM reclaiming a job mid-write leaves a partially-written
+    //                zarr shard that fails every subsequent read-modify-write with a
+    //                checksum / "encoded shard is smaller" error and keeps aborting
+    //                on `-resume` (see czbiohub-sf/biahub#286, #287). Running on the
+    //                non-preemptible `cpu` partition removes that mid-write-kill
+    //                source. The global `errorStrategy 'retry'` (maxRetries above)
+    //                still covers genuine time-limit / OOM kills (exit 130..145).
+    //                To trade cost for reclaim risk, switch this back to
+    //                `preempted`; if you do, do NOT add SLURM `--requeue` (Nextflow
+    //                does not follow a requeued job, so it would read the killed
+    //                run's exit code as a failure AND SLURM would rerun the original
+    //                — a double execution racing on the same output).
     //   gpu       -> `gpu` partition (high PriorityTier, effectively not
     //                preempted), for virtual-stain prediction.
     slurm {
@@ -113,7 +118,7 @@ profiles {
                 memory   = '12 GB'
             }
             withLabel: 'cpu' {
-                queue = 'preempted'
+                queue = 'cpu'
             }
             withLabel: 'gpu' {
                 queue = 'gpu'