Extract the module graphs through a staging directory

.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

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:

tests/concurrency_test.py

@@ -0,0 +1,95 @@
+#!/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. 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
+
+        # 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())