fix: lake: deduplicate artifact transfers (#14060)

This PR has Lake deduplicate artifacts by their hash when uploading or
downloading to the cache (e.g., in `lake cache put` or `lake cache
get`). This fixes possible errors when `curl` was asked to transfer to
the same file and/or URL multiple times.

🤖 Claude Code assisted with code review and writing tests.

Author: tydeu

src/lake/Lake/Config/Cache.lean

@@ -703,8 +703,39 @@ inductive TransferKind
 
 structure TransferInfo where
   url : String
+  hash : Hash
   path : FilePath
-  descr : ArtifactDescr
+  /-- Additional paths for a downloaded artifact. -/
+  extraPaths : Array FilePath := #[]
+
+@[inline] def TransferInfo.addPath (self : TransferInfo) (path : FilePath) (extra := true) : TransferInfo :=
+  if extra then
+    {self with extraPaths := self.extraPaths.push path}
+  else
+    {self with path, extraPaths := self.extraPaths.push self.path}
+
+structure TransferDict where
+  infos : Array TransferInfo
+  indices : Std.HashMap Hash Nat
+
+@[inline] def TransferDict.empty : TransferDict :=
+  ⟨#[], ∅⟩
+
+@[inline] def TransferDict.push
+  (self : TransferDict) (url : String) (hash : Hash) (path : FilePath)
+: TransferDict := ⟨self.infos.push {url, hash, path}, self.indices.insert hash self.infos.size⟩
+
+@[inline] def TransferDict.addIfNew
+  (self : TransferDict) (url : String) (hash : Hash) (path : FilePath)
+: TransferDict := if self.indices.contains hash then self else self.push url hash path
+
+@[inline] def TransferDict.add
+  (self : TransferDict) (url : String) (hash : Hash) (path : FilePath) (extra := true)
+: TransferDict :=
+  if let some j := self.indices.get? hash then
+    {self with infos := self.infos.modify j (·.addPath path extra)}
+  else
+    self.push url hash path
 
 structure TransferConfig where
   kind : TransferKind
@@ -716,6 +747,12 @@ structure TransferState where
   didError : Bool := false
   numSuccesses : Nat := 0
 
+def createExtraPaths (path : FilePath) (extraPaths : Array FilePath) : IO Unit := do
+  -- Note: No intra-cache hard links (breaks permissions/pruning), so we copy
+  let contents ← IO.FS.readBinFile path
+  for extraPath in extraPaths do
+    IO.FS.writeBinFile extraPath contents
+
 partial def monitorTransfer
   (cfg : TransferConfig) (h hOut : IO.FS.Handle) (s : TransferState)
 : LoggerIO TransferState := do
@@ -726,7 +763,7 @@ partial def monitorTransfer
     let s ← (·.2) <$> StateT.run (s := s) do
       match Json.parse line >>= fromJson? with
       | .ok (out : JsonObject) =>
-        let some info@{url, path, descr} := getInfo? out
+        let some info@{url, hash, path, extraPaths} := getInfo? out
           | logError s!"{cfg.scope}: unidentifiable transfer completed: {line.trimAscii}"
             modify ({· with didError := true})
             return
@@ -735,18 +772,20 @@ partial def monitorTransfer
         | .ok 201 =>
           match cfg.kind with
           | .get =>
-            logInfo s!"{cfg.scope}: downloaded artifact {descr.hash}\
+            logInfo s!"{cfg.scope}: downloaded artifact {hash}\
               \n  local path: {path}\
               \n  remote URL: {url}"
             let actualHash ← computeFileHash path
-            if actualHash != descr.hash then
+            if actualHash != hash then
               logError s!"{path}: downloaded artifact hash mismatch, got {actualHash}"
               IO.FS.removeFile path
               modify ({· with didError := true})
             else
+              unless extraPaths.isEmpty do
+                createExtraPaths path extraPaths
               modify fun s => {s with numSuccesses := s.numSuccesses + 1}
           | .put =>
-            logInfo s!"{cfg.scope}: uploaded artifact {descr.hash}\
+            logInfo s!"{cfg.scope}: uploaded artifact {hash}\
               \n  local path: {path}\
               \n  remote URL: {url}"
             modify fun s => {s with numSuccesses := s.numSuccesses + 1}
@@ -764,7 +803,7 @@ where
     | _ => none
   handleFailure info code? out line : LoggerIO Unit := do
     let action := match cfg.kind with | .get => "download" | .put => "upload"
-    let mut msg := s!"{cfg.scope}: failed to {action} artifact {info.descr.hash}"
+    let mut msg := s!"{cfg.scope}: failed to {action} artifact {info.hash}"
     if let .ok code := code? then
       msg := s!"{msg} (status code: {code})"
     if let .ok errMsg := out.getAs String "errormsg" then
@@ -851,14 +890,25 @@ public def downloadArtifacts
   if descrs.isEmpty then
     logWarning "no artifacts to download"
     return
-  let infos ← descrs.foldlM (init := #[]) fun s descr => do
+  let {infos, ..} ← descrs.foldlM (init := TransferDict.empty) fun s {descr} => do
+    let hash := descr.hash
+    let url := service.artifactUrl hash scope
     let path := cache.artifactDir / descr.relPath
     if force then
       removeFileIfExists path
+      return s.add url hash path
     else if (← path.pathExists) then
-      return s
-    let url := service.artifactUrl descr.hash scope
-    return s.push {url, path, descr}
+      return s.add url hash path (extra := false)
+    else
+      return s.add url hash path
+  let infos ← infos.filterM fun info => do
+    if info.extraPaths.isEmpty then
+      not <$> info.path.pathExists
+    else
+      match (← createExtraPaths info.path info.extraPaths |>.toBaseIO) with
+      | .ok _ => return false
+      | .error (.noFileOrDirectory ..) => return true
+      | .error e => error s!"failed to copy artifact: {e}"
   if infos.isEmpty then
     return
   let infos ← id do
@@ -871,7 +921,7 @@ public def downloadArtifacts
   transferArtifacts {scope, infos, kind := .get}
 where
   fetchUrls url infos := IO.FS.withTempFile fun h path => do
-    let body := Json.arr <| infos.map (toJson ·.descr.hash)
+    let body := Json.arr <| infos.map (toJson ·.hash)
     h.putStr body.compress
     h.flush
     let args := #[
@@ -940,9 +990,10 @@ public def uploadArtifacts
   if n = 0 then
     logWarning "no artifacts to upload"
     return
-  let infos ← n.foldM (init := #[]) fun i _ s => do
-    let url := service.artifactUrl descrs[i].hash scope
-    return s.push {url, path := paths[i], descr := descrs[i]}
+  let {infos, ..} ← n.foldM (init := TransferDict.empty) fun i _ s => do
+    let hash := descrs[i].hash
+    let url := service.artifactUrl hash scope
+    return s.addIfNew url hash paths[i]
   transferArtifacts {scope, infos, kind := .put, key := service.impl.key}
 
 /-! ### Output Transfer -/

tests/lake/tests/cache/test.sh

@@ -271,6 +271,31 @@ test_exp -d "$CACHE_DIR"
 test_run cache clean
 test_exp ! -d "$CACHE_DIR"
 
+# Verify cached artifacts are not re-downloaded and that
+# artifacts sharing a content hash across extensions are restored locally
+# (and thus that artifacts are deduplicated by hash)
+rm -rf "$CACHE_DIR"
+mkdir -p "$CACHE_DIR/artifacts"
+hsh=0123456789abcdef
+schema_ver=2026-03-17 # cache map schema version
+echo "arbitrary artifact contents" > "$CACHE_DIR/artifacts/$hsh.o"
+# An already-cached artifact is not re-downloaded
+cat <<EOF > .lake/dummy-outputs.jsonl
+"$schema_ver"
+["aaaaaaaaaaaaaaaa","$hsh.o"]
+EOF
+test_run cache get .lake/dummy-outputs.jsonl --scope=test
+# A shared content hash restores each extension by local copy
+test_exp ! -f "$CACHE_DIR/artifacts/$hsh.dup"
+cat <<EOF > .lake/dummy-outputs.jsonl
+"$schema_ver"
+["aaaaaaaaaaaaaaaa","$hsh.o"]
+["bbbbbbbbbbbbbbbb","$hsh.dup"]
+EOF
+test_run cache get .lake/dummy-outputs.jsonl --scope=test
+test_exp -f "$CACHE_DIR/artifacts/$hsh.dup"
+test_cmd cmp -s "$CACHE_DIR/artifacts/$hsh.o" "$CACHE_DIR/artifacts/$hsh.dup"
+
 # Verify all artifacts restore from the cache and
 # use the build directory with `restoreAllArtifacts`
 test_cmd rm -rf "$CACHE_DIR" .lake/build