fix: claim the pack output before refreshing data

generate_pack refreshed the data directories before probing its output
lock. The refresh waits for every run holding a tree, and a pipeline holds
them all: a second build sat silent for hours instead of being refused.
The output lock is now taken first, and a refresh that has to wait for a
tree names it.

scripts/generate_pack.py

@@ -3223,46 +3223,48 @@ def _main(held: contextlib.ExitStack) -> None:
         return
 
     data_registry = load_data_dir_registry(args.platforms_dir)
-    if data_registry and not args.offline and not args.no_refresh_data:
-        from refresh_data_dirs import load_registry, refresh_all
+    # The output directory is claimed before the data directory refresh: a
+    # refresh waits for every run that holds a cache, so a second build
+    # during a pipeline sat silent for hours instead of being refused.
+    with _pack_output_lock(args.output_dir):
+        if data_registry and not args.offline and not args.no_refresh_data:
+            from refresh_data_dirs import load_registry, refresh_all
 
-        registry = load_registry(os.path.join(args.platforms_dir, "_data_dirs.yml"))
-        results = refresh_all(registry, force=args.refresh_data)
-        updated = sum(1 for v in results.values() if v)
-        if updated:
-            print(f"Refreshed {updated} data director{'ies' if updated > 1 else 'y'}")
-        failed = sorted(k for k, v in results.items() if v is None)
-        if failed:
-            print(f"WARNING: data directory refresh failed: {', '.join(failed)}")
-    if data_registry:
-        hold_data_dirs(held, data_registry)
+            registry = load_registry(os.path.join(args.platforms_dir, "_data_dirs.yml"))
+            results = refresh_all(registry, force=args.refresh_data)
+            updated = sum(1 for v in results.values() if v)
+            if updated:
+                print(f"Refreshed {updated} data director{'ies' if updated > 1 else 'y'}")
+            failed = sorted(k for k, v in results.items() if v is None)
+            if failed:
+                print(f"WARNING: data directory refresh failed: {', '.join(failed)}")
+        if data_registry:
+            hold_data_dirs(held, data_registry)
 
-    emu_profiles = load_emulator_profiles(args.emulators_dir)
+        emu_profiles = load_emulator_profiles(args.emulators_dir)
 
-    target_cores_cache: dict[str, set[str] | None] = {}
-    if args.target:
-        try:
-            target_cores_cache, platforms = build_target_cores_cache(
-                platforms,
-                args.target,
-                args.platforms_dir,
-                is_all=args.all,
-            )
-        except (FileNotFoundError, ValueError) as e:
-            print(f"ERROR: {e}", file=sys.stderr)
-            sys.exit(1)
+        target_cores_cache: dict[str, set[str] | None] = {}
+        if args.target:
+            try:
+                target_cores_cache, platforms = build_target_cores_cache(
+                    platforms,
+                    args.target,
+                    args.platforms_dir,
+                    is_all=args.all,
+                )
+            except (FileNotFoundError, ValueError) as e:
+                print(f"ERROR: {e}", file=sys.stderr)
+                sys.exit(1)
 
-    groups = group_identical_platforms(
-        platforms, args.platforms_dir, target_cores_cache if args.target else None
-    )
+        groups = group_identical_platforms(
+            platforms, args.platforms_dir, target_cores_cache if args.target else None
+        )
 
-    if args.manifest:
-        with _pack_output_lock(args.output_dir):
+        if args.manifest:
             _run_manifest_mode(
                 args, groups, db, zip_contents, emu_profiles, target_cores_cache
             )
-    else:
-        with _pack_output_lock(args.output_dir):
+        else:
             _run_platform_packs(
                 args,
                 groups,
@@ -3273,9 +3275,9 @@ def _main(held: contextlib.ExitStack) -> None:
                 target_cores_cache,
                 system_filter,
             )
-        if args.verify_packs:
-            with _pack_output_lock(args.output_dir, exclusive=False):
-                _run_verify_packs(args)
+    if not args.manifest and args.verify_packs:
+        with _pack_output_lock(args.output_dir, exclusive=False):
+            _run_verify_packs(args)
 
 
 # Manifest generation (JSON inventory for install.py)

scripts/refresh_data_dirs.py

@@ -23,6 +23,7 @@ import tempfile
 import urllib.error
 import urllib.request
 import zipfile
+from collections.abc import Iterator
 from pathlib import Path
 from artifacts import file_lock as _file_lock
 from common import platform_arg, require_platform, yaml_load
@@ -61,7 +62,34 @@ def cache_lock(
     swapped the tree shipped part of it, or nothing.
     """
     cache_dir = Path(cache_dir)
-    return _file_lock(cache_dir.with_name(f".{cache_dir.name}.lock"), shared=shared)
+    return _announced_lock(cache_dir.with_name(f".{cache_dir.name}.lock"), shared, cache_dir)
+
+
+@contextlib.contextmanager
+def _announced_lock(lock_path: Path, shared: bool, cache_dir: Path) -> Iterator[None]:
+    """file_lock, saying so when it has to wait.
+
+    A pipeline holds every cache for its whole run: a refresh started beside
+    it waited hours with nothing on screen.
+    """
+    try:
+        import fcntl
+    except ImportError:
+        with _file_lock(lock_path, shared=shared):
+            yield
+        return
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    mode = fcntl.LOCK_SH if shared else fcntl.LOCK_EX
+    with open(lock_path, "a") as handle:
+        try:
+            fcntl.flock(handle, mode | fcntl.LOCK_NB)
+        except BlockingIOError:
+            log.warning("waiting for %s: another run holds it", cache_dir)
+            fcntl.flock(handle, mode)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
 
 
 def hold_data_dirs(stack: contextlib.ExitStack, registry: dict) -> None:

tests/test_artifact_lock.py

@@ -110,6 +110,36 @@ class PackLockCliTest(unittest.TestCase):
         self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
         self.assertIn("is in use by another run", proc.stdout)
 
+    def test_a_held_output_is_refused_before_the_data_refresh(self):
+        """The refresh waits for every run holding a data tree, and a
+        pipeline holds them all for its whole run: probed after the refresh,
+        a second build sat silent for hours instead of being refused."""
+        from refresh_data_dirs import cache_lock  # noqa: PLC0415
+
+        platforms = Path(self.dir) / "platforms"
+        platforms.mkdir()
+        for name in ("_registry.yml", "_shared.yml", "misterfpga.yml"):
+            shutil.copyfile(REPO_ROOT / "platforms" / name, platforms / name)
+        cache = Path(self.dir) / "cache" / "probe"
+        cache.mkdir(parents=True)
+        (platforms / "_data_dirs.yml").write_text(
+            "data_directories:\n  probe:\n    source_type: zip\n"
+            f"    source_url: https://example.invalid/probe.zip\n"
+            f"    local_cache: {cache}\n"
+        )
+        output = Path(self.dir) / "out"
+        with artifact_lock(str(output)), cache_lock(cache):
+            try:
+                proc = subprocess.run(
+                    [sys.executable, "scripts/generate_pack.py", "--platform", "misterfpga",
+                     "--platforms-dir", str(platforms), "--output-dir", str(output)],
+                    capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=120,
+                )
+            except subprocess.TimeoutExpired:
+                self.fail("generate_pack waited on the data tree instead of refusing")
+        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
+        self.assertIn("is in use by another run", proc.stdout)
+
     def test_every_writing_mode_refuses_a_locked_output(self):
         """Two of seven modes took the lock, and five wrote straight into it.
 

tests/test_refresh_data_dirs.py

@@ -166,6 +166,33 @@ class EmulatorPackHoldsTheCache(unittest.TestCase):
             self.assertEqual(held, [True])
 
 
+class AWaitIsAnnounced(unittest.TestCase):
+    def test_a_held_tree_is_named_before_the_wait(self):
+        """A pipeline holds every tree for its run: the refresh beside it
+        waited with nothing on screen."""
+        with tempfile.TemporaryDirectory() as tmp:
+            cache = Path(tmp) / "probe"
+            taken = threading.Event()
+
+            def second_run() -> None:
+                with rdd.cache_lock(cache):
+                    taken.set()
+
+            with self.assertLogs(rdd.log, "WARNING") as logs:
+                with rdd.cache_lock(cache):
+                    worker = threading.Thread(target=second_run)
+                    worker.start()
+                    for _ in range(100):
+                        if logs.output:
+                            break
+                        threading.Event().wait(0.05)
+                    self.assertFalse(taken.is_set())
+                worker.join(10)
+            self.assertTrue(taken.is_set())
+            self.assertIn("waiting for", logs.output[0])
+            self.assertIn("probe", logs.output[0])
+
+
 class DryRunReport(unittest.TestCase):
     """What a dry run says: why each entry would be fetched, and how many."""