fix: check a tree before waiting to replace it

scripts/refresh_data_dirs.py

@@ -442,13 +442,45 @@ def refresh_entry(
     The decision and the swap run under one lock per directory: a run that
     waited for another session re-reads the version it recorded and finds
     nothing left to do.
+
+    The exclusive lock waits for every reader, and a pipeline reads the
+    trees for its whole run: a second build sat on "waiting" for a tree the
+    remote check would have found current. The check is made first without
+    it, and only a tree to replace waits.
     """
     if dry_run:
         return _refresh_entry(key, entry, force, dry_run, versions_path)
+    if not force:
+        needs, _remote, cached_tag = _decide(key, entry, False, versions_path)
+        if needs is None:
+            return None
+        if not needs:
+            log.info("[%s] up to date (tag: %s)", key, _short_tag(cached_tag))
+            return False
     with cache_lock(entry["local_cache"]):
         return _refresh_entry(key, entry, force, dry_run, versions_path)
 
 
+def _decide(
+    key: str, entry: dict, force: bool, versions_path: str
+) -> tuple[bool | None, str | None, str | None]:
+    """(whether the tree needs a refresh, the remote tag read, the recorded
+    tag). None when the remote could not be read."""
+    version = entry.get("version", "master")
+    cached = _load_versions(versions_path).get(key, {})
+    cached_tag = cached.get("sha") or cached.get("etag")
+    if force or not Path(entry["local_cache"]).exists():
+        return True, None, cached_tag
+    if entry.get("source_type", "tarball") == "zip":
+        remote_tag = _get_remote_etag(entry["source_url"].format(version=version))
+    else:
+        remote_tag = get_remote_sha(entry["source_url"], version)
+    if remote_tag is None:
+        log.warning("[%s] could not check remote, skipping", key)
+        return None, None, cached_tag
+    return remote_tag != cached_tag, remote_tag, cached_tag
+
+
 def _refresh_entry(
     key: str, entry: dict, force: bool, dry_run: bool, versions_path: str
 ) -> bool | None:
@@ -458,23 +490,9 @@ def _refresh_entry(
     local_cache = entry["local_cache"]
     exclude = entry.get("exclude", [])
 
-    versions = _load_versions(versions_path)
-    cached = versions.get(key, {})
-    cached_tag = cached.get("sha") or cached.get("etag")
-
-    needs_refresh = force or not Path(local_cache).exists()
-
-    remote_tag: str | None = None
-    if not needs_refresh:
-        if source_type == "zip":
-            remote_tag = _get_remote_etag(source_url)
-        else:
-            remote_tag = get_remote_sha(entry["source_url"], version)
-        if remote_tag is None:
-            log.warning("[%s] could not check remote, skipping", key)
-            return None
-        needs_refresh = remote_tag != cached_tag
-
+    needs_refresh, remote_tag, cached_tag = _decide(key, entry, force, versions_path)
+    if needs_refresh is None:
+        return None
     if not needs_refresh:
         log.info("[%s] up to date (tag: %s)", key, _short_tag(cached_tag))
         return False

tests/test_refresh_data_dirs.py

@@ -93,6 +93,31 @@ class RefreshConcurrency(unittest.TestCase):
             )
         self.assertEqual(seen, [True])
 
+    def test_a_current_tree_is_not_waited_for(self):
+        """A pipeline reads every tree for its whole run. A build beside it
+        waited on the exclusive lock before asking the remote, which would
+        have answered that nothing changed."""
+        import fcntl  # noqa: PLC0415
+
+        self.cache.mkdir(parents=True)
+        rdd._record_version("pak", {"sha": "e1"}, self.versions)
+        entry = {"source_type": "zip", "source_url": "https://x/pak.zip",
+                 "local_cache": str(self.cache)}
+        result: list = []
+        with open(self.cache.with_name(".pak.lock"), "a") as reader, \
+                mock.patch.object(rdd, "_get_remote_etag", return_value="e1"):
+            fcntl.flock(reader, fcntl.LOCK_SH)
+            worker = threading.Thread(target=lambda: result.append(
+                rdd.refresh_entry("pak", entry, versions_path=self.versions)
+            ), daemon=True)
+            worker.start()
+            worker.join(timeout=20)
+            finished = not worker.is_alive()
+            fcntl.flock(reader, fcntl.LOCK_UN)
+        worker.join(timeout=20)
+        self.assertTrue(finished, "the refresh waited for a reader")
+        self.assertEqual(result, [False])
+
     def test_version_file_is_never_left_truncated(self):
         Path(self.versions).parent.mkdir(parents=True)
         Path(self.versions).write_text(json.dumps({"keep": {"sha": "1"}}))