From 3401adae17d21d3e67fc6263697e3bf91d698bae Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 13 Aug 2026 08:20:01 -0700 Subject: [PATCH 1/3] Add a failing test for concurrent module graph extraction Nextflow dispatches one BLIMMP task per genome, so several processes extract the module graphs into the same directory at once. On a 9-genome run, 4 of 8 tasks died with [FATAL] Failed to extract KEGG_Graphs_Generated_March26.zip before running any analysis. This commit adds the test without the fix, so CI records the failure first. The existing smoke test runs a single process and stays green throughout, which is why the bug shipped. --- .github/workflows/test.yml | 6 +++ tests/concurrency_test.py | 94 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 tests/concurrency_test.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a092b61..aa6d1cb 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -36,3 +36,9 @@ jobs: docker run --rm --read-only --tmpfs /tmp \ -e BLIMMP_CACHE_DIR=/tmp/blimmp-cache \ blimmp:ci python /app/tests/smoke_test.py + + # Nextflow runs one BLIMMP task per genome, so several processes extract + # the module graphs at once. That raced and killed half the tasks on a + # 9-genome run. + - name: Concurrent extraction + run: docker run --rm blimmp:ci python /app/tests/concurrency_test.py diff --git a/tests/concurrency_test.py b/tests/concurrency_test.py new file mode 100644 index 0000000..6e409e0 --- /dev/null +++ b/tests/concurrency_test.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Several BLIMMP processes must be able to extract the module graphs at once. + +Nextflow dispatches one BLIMMP task per genome, so a run with N genomes puts N +processes on a node at the same time. When the graphs are not already +extracted, every one of them extracts into the same directory. + +Extracting straight into that shared directory made them destroy each other's +work: one process removed __MACOSX or flattened the nested folder while another +was still reading from it, and the resulting OSError looked to the caller like +an unwritable destination. On a 9-genome run, 4 tasks died with a FATAL +"failed to extract" before any analysis ran. A single-process test cannot see +this, which is how it shipped. +""" + +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +WORKERS = 8 +EXPECTED_GRAPHS = 340 + +CHILD = """ +import sys +from BLIMMP_Scripts.module_detection import ensure_module_graphs +print(ensure_module_graphs(sys.argv[1])) +""" + + +def main() -> int: + import BLIMMP_Scripts + + packaged = Path(BLIMMP_Scripts.__file__).parent / "Graph_Dependencies" + zip_name = "KEGG_Graphs_Generated_March26.zip" + if not (packaged / zip_name).is_file(): + print(f"FAIL: {zip_name} is not in the installed package at {packaged}") + return 1 + + scratch = Path(tempfile.mkdtemp(prefix="blimmp-concurrency-")) + try: + # A Graph_Dependencies directory holding only the archive, so every + # worker has to extract rather than finding graphs already in place. + graph_dir = scratch / "Graph_Dependencies" + graph_dir.mkdir() + shutil.copy2(packaged / zip_name, graph_dir) + + procs = [ + subprocess.Popen( + [sys.executable, "-c", CHILD, str(graph_dir)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for _ in range(WORKERS) + ] + results = [(p.wait(), *p.communicate()) for p in procs] + + failed = [(i, err or out) for i, (rc, out, err) in enumerate(results) if rc != 0] + if failed: + print(f"FAIL: {len(failed)}/{WORKERS} workers exited non-zero") + for i, msg in failed[:3]: + last = msg.strip().splitlines()[-1] if msg.strip() else "(no output)" + print(f" worker {i}: {last}") + return 1 + + extracted = graph_dir / "KEGG_Graphs_Generated_March26" + graphs = list(extracted.glob("module_*_nodes.json")) + if len(graphs) != EXPECTED_GRAPHS: + print(f"FAIL: expected {EXPECTED_GRAPHS} module graphs, found {len(graphs)}") + return 1 + + # Every worker must agree on where the graphs ended up. + returned = {out.strip() for _, out, _ in results} + if returned != {str(extracted)}: + print(f"FAIL: workers disagreed on the graph directory: {sorted(returned)}") + return 1 + + # A crashed or abandoned extraction leaves its staging directory behind. + leftovers = [p.name for p in graph_dir.iterdir() if p.name.startswith(".")] + if leftovers: + print(f"FAIL: staging directories left behind: {leftovers}") + return 1 + + print(f"PASS: {WORKERS} concurrent workers, {len(graphs)} graphs, no leftovers") + return 0 + finally: + shutil.rmtree(scratch, ignore_errors=True) + + +if __name__ == "__main__": + sys.exit(main()) From 1efc959303de8953d612d6b0261822477ffbc4ab Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 13 Aug 2026 08:26:36 -0700 Subject: [PATCH 2/3] Extract the module graphs through a staging directory Nextflow dispatches one BLIMMP task per genome, so several processes extract the module graphs into the same directory at once. Extracting straight into that directory let them destroy each other's work: one process would rmtree __MACOSX or flatten the nested folder while another was still reading from it. The OSError that followed looked to the caller like an unwritable destination, so it moved on to the next candidate and then reported that all of them had failed. Extraction now happens in a private staging directory that is renamed into position. A rename onto a missing or empty directory is atomic on POSIX, so the first process to finish wins and the others adopt its copy instead of failing. The test added in the previous commit went from 6 of 8 workers dead to all 8 agreeing on one directory with 340 graphs and no staging directories left behind. 0.1.4 makes this unreachable in the published image, since the graphs ship pre-extracted and the extraction path is never entered. It still runs for anyone installing from source or from a wheel. Implemented with assistance from Claude (Opus 5) --- BLIMMP_Scripts/module_detection.py | 48 ++++++++++++++++++++++-------- 1 file changed, 36 insertions(+), 12 deletions(-) diff --git a/BLIMMP_Scripts/module_detection.py b/BLIMMP_Scripts/module_detection.py index e28cafa..f423075 100644 --- a/BLIMMP_Scripts/module_detection.py +++ b/BLIMMP_Scripts/module_detection.py @@ -2255,21 +2255,45 @@ def _extract_module_graphs(zip_path, destination) -> None: The archive was built on macOS, so it carries __MACOSX metadata and wraps everything in a redundant top-level folder. Both are flattened away. The source zip is deliberately left in place so extraction stays repeatable. + + Extraction happens in a private staging directory that is then renamed into + position, because several BLIMMP processes on one machine share a cache + directory and will race here. Extracting straight into `destination` let + them tear up each other's files: one process would rmtree __MACOSX or + rmdir the nested folder while another was still reading from it, and the + OSError that followed looked to the caller like an unwritable destination. + A rename onto a missing or empty directory is atomic on POSIX, so the first + process to finish wins and the rest adopt its copy. """ print(f"Extracting {os.path.basename(str(zip_path))} to {destination} ...") destination = Path(destination) - with zipfile.ZipFile(str(zip_path), "r") as z: - z.extractall(str(destination)) - - macosx_path = destination / "__MACOSX" - if macosx_path.is_dir(): - shutil.rmtree(str(macosx_path)) - - nested = destination / destination.name - if nested.is_dir(): - for item in os.listdir(str(nested)): - shutil.move(str(nested / item), str(destination / item)) - nested.rmdir() + parent = destination.parent + parent.mkdir(parents=True, exist_ok=True) + + staging = Path(tempfile.mkdtemp(prefix=f".{destination.name}.", dir=str(parent))) + try: + with zipfile.ZipFile(str(zip_path), "r") as z: + z.extractall(str(staging)) + + macosx_path = staging / "__MACOSX" + if macosx_path.is_dir(): + shutil.rmtree(str(macosx_path)) + + nested = staging / destination.name + if nested.is_dir(): + for item in os.listdir(str(nested)): + shutil.move(str(nested / item), str(staging / item)) + nested.rmdir() + + try: + os.replace(str(staging), str(destination)) + except OSError: + # Another process finished first and its directory is non-empty, so + # the rename is refused. Its graphs are as good as ours. + if not _graphs_present(destination): + raise + finally: + shutil.rmtree(str(staging), ignore_errors=True) def validate_paths(paths: "Paths") -> None: From bc68edab2f76ed78b999d99c04b8a05dad9e7960 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 13 Aug 2026 08:28:47 -0700 Subject: [PATCH 3/3] Read the graph path off the last stdout line in the concurrency test ensure_module_graphs prints a progress line when it does the extracting, so the worker that wins the race emits two lines and the others emit one. Comparing whole stdout blobs made the workers look like they disagreed when they had all returned the same directory. --- tests/concurrency_test.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/concurrency_test.py b/tests/concurrency_test.py index 6e409e0..450e7fc 100644 --- a/tests/concurrency_test.py +++ b/tests/concurrency_test.py @@ -72,8 +72,9 @@ def main() -> int: print(f"FAIL: expected {EXPECTED_GRAPHS} module graphs, found {len(graphs)}") return 1 - # Every worker must agree on where the graphs ended up. - returned = {out.strip() for _, out, _ in results} + # Every worker must agree on where the graphs ended up. The extracting + # worker also prints a progress line, so read the path off the last one. + returned = {out.strip().splitlines()[-1] for _, out, _ in results if out.strip()} if returned != {str(extracted)}: print(f"FAIL: workers disagreed on the graph directory: {sorted(returned)}") return 1