fix: pin the large file inode just verified

_pin_verified returned an existing <sha1>/<name> link without checking
its inode. A link whose cache name was gone had one link left, and the
prune of the same download removed it after it was returned, so
fetch_large_file handed out a path that no longer existed. The link is
now replaced when it holds another inode, and a revision directory a
prune removed before the link is made again.

scripts/largefiles.py

@@ -250,23 +250,33 @@ def _pin_verified(
     passed off as the one verified. None when the name moved in between.
     """
     revision = os.path.join(dest_dir, VERIFIED_STORE, snapshot[0]["sha1"])
-    os.makedirs(revision, exist_ok=True)
     final = os.path.join(revision, name)
-    if os.path.isfile(final):
-        return final
-    scratch = os.path.join(revision, f".{os.getpid()}.{os.urandom(4).hex()}.tmp")
-    try:
-        os.link(path, scratch)
-    except FileNotFoundError:
-        return None
-    try:
-        if os.stat(scratch).st_ino != snapshot[2]:
-            return None
-        os.replace(scratch, final)
-        return final
-    finally:
+    # A prune removes an empty revision directory: one that went between
+    # makedirs and link is made again, once.
+    for _attempt in range(2):
+        os.makedirs(revision, exist_ok=True)
+        # The link must hold the inode just verified. An older link left by a
+        # copy that lost its cache name has one link left, and the prune of
+        # this very download removed it after it was returned.
         with contextlib.suppress(FileNotFoundError):
-            os.unlink(scratch)
+            if os.stat(final).st_ino == snapshot[2]:
+                return final
+        scratch = os.path.join(revision, f".{os.getpid()}.{os.urandom(4).hex()}.tmp")
+        try:
+            os.link(path, scratch)
+        except FileNotFoundError:
+            if not os.path.exists(path):
+                return None
+            continue
+        try:
+            if os.stat(scratch).st_ino != snapshot[2]:
+                return None
+            os.replace(scratch, final)
+            return final
+        finally:
+            with contextlib.suppress(FileNotFoundError):
+                os.unlink(scratch)
+    return None
 
 
 def _prune_verified(dest_dir: str) -> None:

tests/test_large_file_cache.py

@@ -226,6 +226,36 @@ class LargeFileCacheTest(unittest.TestCase):
         self.assertEqual(cached.read_bytes(), rebuilt)
         self.assertEqual(len(gets), 1)
 
+    def test_an_orphan_link_is_never_returned(self):
+        """A revision link whose cache name was removed has one link left.
+        Downloaded again with the same bytes, the old link was returned and
+        the prune of that same download removed it."""
+        self._serve(PAYLOAD_A, [])
+        sha1 = hashlib.sha1(PAYLOAD_A).hexdigest()
+        first = common.fetch_large_file("asset.bin", dest_dir=self.dir, expected_sha1=sha1)
+        self.assertTrue(first and os.path.isfile(first))
+        os.unlink(Path(self.dir) / "asset.bin")
+        with mock.patch.object(largefiles, "ORPHAN_GRACE", -1):
+            second = common.fetch_large_file(
+                "asset.bin", dest_dir=self.dir, expected_sha1=sha1
+            )
+        self.assertTrue(second and os.path.isfile(second))
+        self.assertTrue(os.path.samefile(second, Path(self.dir) / "asset.bin"))
+
+    def test_a_cache_hit_relinks_an_orphan(self):
+        """The cache name was replaced by a copy with the same bytes: the hit
+        handed out the old link, which a prune elsewhere would remove."""
+        self._serve(PAYLOAD_A, [])
+        sha1 = hashlib.sha1(PAYLOAD_A).hexdigest()
+        common.fetch_large_file("asset.bin", dest_dir=self.dir, expected_sha1=sha1)
+        cached = Path(self.dir) / "asset.bin"
+        os.unlink(cached)
+        cached.write_bytes(PAYLOAD_A)
+        hit = common.fetch_large_file(
+            "asset.bin", dest_dir=self.dir, expected_sha1=sha1, offline=True
+        )
+        self.assertTrue(os.path.samefile(hit, cached))
+
     def test_a_revision_the_release_still_serves_is_kept(self):
         cached = Path(self.dir) / "asset.bin"
         cached.write_bytes(PAYLOAD_A)