fix: hold the data dir lock while packing it

scripts/generate_pack.py

@@ -62,6 +62,7 @@ from common import (
     yaml_load,
 )
 import packresolve
+from refresh_data_dirs import cache_lock
 import region as region_mod
 import slot as slot_mod
 import slots
@@ -345,22 +346,25 @@ def _data_directory_members(
                 dd_prefix = base_dest
             else:
                 dd_prefix = dd_dest
-            for root, _dirs, filenames in os.walk(local_path):
-                for fname in filenames:
-                    src = os.path.join(root, fname)
-                    rel = os.path.relpath(src, local_path)
-                    full = f"{dd_prefix}/{rel}"
-                    if full in seen_destinations or (
-                        full.lower() in seen_lower and case_insensitive
-                    ):
-                        continue
-                    if _has_path_conflict(full, seen_destinations, seen_parents):
-                        continue
-                    seen_destinations.add(full)
-                    _register_path(full, seen_destinations, seen_parents)
-                    if case_insensitive:
-                        seen_lower.add(full.lower())
-                    yield src, full
+            # Held while the caller writes what is yielded: a refresh swapping
+            # the tree waits until the walk is done.
+            with cache_lock(local_path, shared=True):
+                for root, _dirs, filenames in os.walk(local_path):
+                    for fname in filenames:
+                        src = os.path.join(root, fname)
+                        rel = os.path.relpath(src, local_path)
+                        full = f"{dd_prefix}/{rel}"
+                        if full in seen_destinations or (
+                            full.lower() in seen_lower and case_insensitive
+                        ):
+                            continue
+                        if _has_path_conflict(full, seen_destinations, seen_parents):
+                            continue
+                        seen_destinations.add(full)
+                        _register_path(full, seen_destinations, seen_parents)
+                        if case_insensitive:
+                            seen_lower.add(full.lower())
+                        yield src, full
 
 
 def _pack_data_directories(

scripts/refresh_data_dirs.py

@@ -52,8 +52,8 @@ def load_registry(registry_path: str = DEFAULT_REGISTRY) -> dict[str, dict]:
 
 
 @contextlib.contextmanager
-def _file_lock(lock_path: Path):
-    """Hold an exclusive lock on lock_path, waiting for it if taken.
+def _file_lock(lock_path: Path, shared: bool = False):
+    """Hold a lock on lock_path, waiting for it if taken.
 
     Several sessions refresh the same data directories: without it, two
     swaps of one tree interleave and the second lands inside the first.
@@ -65,14 +65,24 @@ def _file_lock(lock_path: Path):
         yield
         return
     lock_path.parent.mkdir(parents=True, exist_ok=True)
-    with open(lock_path, "w") as handle:
-        fcntl.flock(handle, fcntl.LOCK_EX)
+    with open(lock_path, "a") as handle:
+        fcntl.flock(handle, fcntl.LOCK_SH if shared else fcntl.LOCK_EX)
         try:
             yield
         finally:
             fcntl.flock(handle, fcntl.LOCK_UN)
 
 
+def cache_lock(cache_dir: str | Path, shared: bool = False):
+    """The lock a refresh holds while it swaps cache_dir.
+
+    A reader takes it shared: a pack walking data/sdlpal while another run
+    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)
+
+
 def _load_versions(versions_path: str = VERSIONS_FILE) -> dict[str, dict]:
     path = Path(versions_path)
     if not path.exists():
@@ -381,8 +391,7 @@ def refresh_entry(
     """
     if dry_run:
         return _refresh_entry(key, entry, force, dry_run, versions_path)
-    cache_dir = Path(entry["local_cache"])
-    with _file_lock(cache_dir.with_name(f".{cache_dir.name}.lock")):
+    with cache_lock(entry["local_cache"]):
         return _refresh_entry(key, entry, force, dry_run, versions_path)
 
 

tests/test_refresh_data_dirs.py

@@ -115,5 +115,28 @@ class RefreshConcurrency(unittest.TestCase):
         self.assertEqual(sorted(rdd._load_versions(self.versions)), sorted(keys))
 
 
+class PackWalkHoldsTheCache(unittest.TestCase):
+    def test_a_refresh_waits_for_the_walk(self):
+        import fcntl
+
+        from generate_pack import _data_directory_members
+
+        with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as tmp:
+            cache = Path(tmp) / "data" / "sdlpal"
+            cache.mkdir(parents=True)
+            (cache / "a.mkf").write_bytes(b"a")
+            (cache / "b.mkf").write_bytes(b"b")
+            systems = {"s": {"data_directories": [{"ref": "sdlpal", "destination": "sdlpal"}]}}
+            registry = {"sdlpal": {"local_cache": str(cache)}}
+            walk = _data_directory_members(systems, registry, "p", "", False, set(), set(), set())
+            next(walk)
+            with open(cache.with_name(".sdlpal.lock"), "a") as handle:
+                with self.assertRaises(BlockingIOError):
+                    fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            list(walk)
+            with open(cache.with_name(".sdlpal.lock"), "a") as handle:
+                fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                fcntl.flock(handle, fcntl.LOCK_UN)
+
 if __name__ == "__main__":
     unittest.main()