fix(download): per-model serialization + airtight cross-process lock

Address the download-lock review items:

- Per-model in-process lock. Replace the single global download_mutex_ with a
  per-model mutex keyed on the resolved cache path. Two downloads of the same
  model serialize; downloads of different models run concurrently in-process
  instead of queuing behind each other's (up to 3 h) cross-process waits.

- Close the POSIX flock()+unlink() orphan-inode race. After flock() succeeds,
  verify (fstat vs stat) that the inode we locked is still the file at the lock
  path; if a racing releaser unlinked it and a third process recreated it, drop
  the stale lock and report contention so the caller retries. This makes the
  self-cleaning unlink-on-release provably safe and guarantees two processes can
  never both believe they hold the lock - so a model can never be downloaded to
  the same directory twice at once, across any number of processes or apps.

- Fix the misleading Windows ACCESS_DENIED comment: it is the DELETE_ON_CLOSE
  delete-pending window (STATUS_DELETE_PENDING), not "narrower access rights".

- Document why the POSIX unlink-before-close is safe (fresh-fd non-blocking
  waiters; no work between unlink and close; the inode check above).

- Decouple the lock-wait cadence from the progress heartbeat: poll the
  cancellation/heartbeat callback once per poll_interval instead of every
  100 ms, so a user callback is not invoked ~10x/s for the whole wait.

- Tests: add ConcurrentDownloadsOfDifferentModelsRunConcurrently (proves
  different models do not serialize). Existing same-model serialize and
  CrossProcessFileLock acquire/release/wait/cancel/timeout tests still pass
  (66 download-suite tests green; POSIX branch syntax-checked under g++).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: bmehta001

sdk_v2/cpp/src/download/cross_process_file_lock.cc

@@ -22,6 +22,7 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <sys/file.h>
+#include <sys/stat.h>
 #include <unistd.h>
 #endif
 
@@ -71,8 +72,15 @@ struct CrossProcessFileLock::State {
   std::filesystem::path path;
   ~State() {
     if (fd >= 0) {
-      // Unlink before close so the file disappears at the same instant the
-      // lock releases; a concurrent acquirer simply recreates it.
+      // Unlink before close so the file disappears the instant the lock
+      // releases; a concurrent acquirer simply recreates it. This is the
+      // classic flock()+unlink() pattern, and it is safe here because every
+      // acquirer verifies, while holding the flock, that the inode it locked is
+      // still the one at `path` (see the fstat/stat check in
+      // TryAcquireForDirectory). An acquirer that raced in on the old inode
+      // between our unlink and a third party's recreate will see the inode
+      // mismatch and retry, so two processes never hold "the lock" at once.
+      // There is also no protected work between this unlink and close.
       ::unlink(path.c_str());
       ::close(fd);
     }
@@ -118,8 +126,14 @@ std::unique_ptr<CrossProcessFileLock> CrossProcessFileLock::TryAcquireForDirecto
   if (handle == INVALID_HANDLE_VALUE) {
     DWORD err = GetLastError();
     if (err == ERROR_SHARING_VIOLATION || err == ERROR_LOCK_VIOLATION || err == ERROR_ACCESS_DENIED) {
-      // ACCESS_DENIED can surface on FILE_SHARE_NONE collisions when the
-      // existing handle has narrower access rights — treat as contention.
+      // SHARING/LOCK_VIOLATION: another handle already holds the share-none
+      // lock. ACCESS_DENIED: the holder is mid-release — FILE_FLAG_DELETE_ON_CLOSE
+      // puts the file into STATUS_DELETE_PENDING during the close window, and a
+      // concurrent open of a delete-pending file is reported as ACCESS_DENIED.
+      // All three mean "another process has it"; treat as contention so the
+      // caller retries. (A genuine permission error also lands here and would
+      // poll until timeout, but the directory was just created successfully so
+      // that is improbable.)
       return nullptr;
     }
     FL_THROW(FOUNDRY_LOCAL_ERROR_INTERNAL,
@@ -149,6 +163,23 @@ std::unique_ptr<CrossProcessFileLock> CrossProcessFileLock::TryAcquireForDirecto
              "flock failed for '" + lock_path.string() + "' (errno=" + std::to_string(err) + ")");
   }
 
+  // Robust-flock inode check. We now hold an exclusive flock on whatever inode
+  // `fd` refers to, but a releaser unlink()s the lock file in its destructor —
+  // so between our open() and flock() the path may have been unlinked and a
+  // third process may have recreated it. If so, we are holding a lock on an
+  // orphaned inode that guards nothing while the live file at `lock_path` is a
+  // different inode. Confirm the inode we locked is still the one at the path;
+  // if not, drop it and report contention so the caller retries against the
+  // live file. This closes the flock()+unlink() orphan-inode race, which is
+  // what lets two processes never both believe they hold the lock.
+  struct stat fd_stat {};
+  struct stat path_stat {};
+  if (::fstat(fd, &fd_stat) != 0 || ::stat(lock_path.c_str(), &path_stat) != 0 ||
+      fd_stat.st_dev != path_stat.st_dev || fd_stat.st_ino != path_stat.st_ino) {
+    ::close(fd);  // releases the flock on the stale / orphaned inode
+    return nullptr;
+  }
+
   (void)::ftruncate(fd, 0);
   auto info = FormatProcessInfo();
   (void)::write(fd, info.data(), info.size());
@@ -170,9 +201,13 @@ std::unique_ptr<CrossProcessFileLock> WaitForLockForDirectory(
     std::chrono::milliseconds poll_interval,
     std::chrono::milliseconds timeout) {
   auto deadline = std::chrono::steady_clock::now() + timeout;
-  // Poll cancellation in slices of at most 100 ms so a long poll interval
-  // (1.25 s default) doesn't keep a cancelling caller waiting.
-  constexpr std::chrono::milliseconds kCancelSlice{100};
+  // `is_cancelled` is the caller's progress callback, which also serves as the
+  // liveness heartbeat — it emits 0% on every invocation. We therefore poll it
+  // on a single cadence (once per `poll_interval`) rather than on a separate
+  // fast cancellation tick: a faster tick would spam the user callback (~10x/s)
+  // for the entire wait, and cancelling a multi-minute cross-process wait a
+  // second sooner is imperceptible. There is no separate cancellation channel
+  // to decouple the heartbeat from.
   while (true) {
     if (is_cancelled && is_cancelled()) {
       FL_THROW(FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, "lock acquisition cancelled");
@@ -185,13 +220,7 @@ std::unique_ptr<CrossProcessFileLock> WaitForLockForDirectory(
       FL_THROW(FOUNDRY_LOCAL_ERROR_INTERNAL,
                "timed out waiting for cross-process download lock on '" + directory.string() + "'");
     }
-    auto slice_end = std::chrono::steady_clock::now() + poll_interval;
-    while (std::chrono::steady_clock::now() < slice_end) {
-      if (is_cancelled && is_cancelled()) {
-        FL_THROW(FOUNDRY_LOCAL_ERROR_OPERATION_CANCELLED, "lock acquisition cancelled");
-      }
-      std::this_thread::sleep_for(std::min(kCancelSlice, poll_interval));
-    }
+    std::this_thread::sleep_for(poll_interval);
   }
 }
 

sdk_v2/cpp/src/download/download_manager.cc

@@ -13,6 +13,8 @@
 #include <cctype>
 #include <filesystem>
 #include <fstream>
+#include <memory>
+#include <mutex>
 #include <string>
 #include <string_view>
 
@@ -212,15 +214,25 @@ std::string DownloadManager::ComputeModelPath(const ModelInfo& info) const {
   return full_path.string();
 }
 
+std::shared_ptr<std::mutex> DownloadManager::GetModelLock(const std::string& model_path) const {
+  std::lock_guard<std::mutex> guard(model_locks_mutex_);
+  auto& slot = model_locks_[model_path];
+  if (!slot) {
+    slot = std::make_shared<std::mutex>();
+  }
+  return slot;
+}
+
 std::string DownloadManager::DownloadModel(const ModelInfo& info,
                                            std::function<int(float)> progress_cb) {
-  // Serialize all downloads. Concurrent downloads of the same model would race into
-  // creating the same directory and double-writing inference_model.json; concurrent
-  // downloads of different models would compete for the same per-blob chunk parallelism.
-  // A single global lock keeps the model simple and predictable.
-  std::lock_guard<std::mutex> download_guard(download_mutex_);
-
+  // Resolve the cache path first, then serialize per model. Two downloads of the
+  // same model share one mutex and run one-at-a-time; downloads of different
+  // models take different mutexes and proceed concurrently. The cross-process
+  // file lock taken below extends the same-model guarantee across every process
+  // and app that shares this cache directory.
   auto model_path = ComputeModelPath(info);
+  auto model_lock = GetModelLock(model_path);
+  std::lock_guard<std::mutex> download_guard(*model_lock);
 
   // Fast path: serve the cache without taking the cross-process lock.
   // A valid cache hit requires: directory exists, no in-progress signal file, and

sdk_v2/cpp/src/download/download_manager.h

@@ -9,6 +9,7 @@
 
 #include <atomic>
 #include <functional>
+#include <map>
 #include <memory>
 #include <mutex>
 #include <string>
@@ -65,16 +66,28 @@ class DownloadManager {
   /// Uses {cache_dir}/{publisher}/{model_id_with_version_fix}
   std::string ComputeModelPath(const ModelInfo& info) const;
 
+  /// Get (creating on first use) the per-model serialization mutex for the
+  /// resolved cache path `model_path`. Downloads of the same model share one
+  /// mutex and run one-at-a-time; downloads of different models get distinct
+  /// mutexes and proceed concurrently in-process.
+  std::shared_ptr<std::mutex> GetModelLock(const std::string& model_path) const;
+
   std::string cache_directory_;
   int max_concurrency_;
   ILogger& logger_;
   std::unique_ptr<ModelRegistryClient> registry_client_;
   std::unique_ptr<IBlobDownloader> blob_downloader_;
 
-  /// Serializes all DownloadModel calls. Only one model downloads at a time — simpler
-  /// than per-model locking and avoids contending with the per-blob chunk parallelism
-  /// (`max_concurrency_`) inside a single download.
-  mutable std::mutex download_mutex_;
+  /// Guards `model_locks_`. Held only briefly to look up or insert a per-model
+  /// mutex — never across an actual download.
+  mutable std::mutex model_locks_mutex_;
+
+  /// Per-model serialization mutexes, keyed by resolved cache path. Bounded by
+  /// the number of distinct models this process downloads. The `shared_ptr`
+  /// keeps a mutex alive for an in-flight download even though its map entry
+  /// persists. Cross-process serialization is handled separately by
+  /// CrossProcessFileLock.
+  mutable std::map<std::string, std::shared_ptr<std::mutex>> model_locks_;
 };
 
 }  // namespace fl

sdk_v2/cpp/test/internal_api/download_test.cc

@@ -27,6 +27,7 @@
 
 #include <algorithm>
 #include <atomic>
+#include <condition_variable>
 #include <filesystem>
 #include <fstream>
 #include <functional>
@@ -981,6 +982,99 @@ TEST(DownloadManagerTest, ConcurrentDownloadsOfSameModelSerialize) {
   }
 }
 
+// With per-model locking, two *different* models must download concurrently —
+// they must not serialize through a shared in-process mutex. A rendezvous inside
+// the blob downloader proves both downloads occupy the critical section at the
+// same time: each arrival waits for its peer, so if the two ever serialized the
+// first arrival would time out waiting for a peer that can't enter yet.
+TEST(DownloadManagerTest, ConcurrentDownloadsOfDifferentModelsRunConcurrently) {
+  TempDir tmpdir;
+  DownloadManager manager(tmpdir.string(), "eastus", 64, fl::test::NullLog());
+
+  auto registry = std::make_unique<ModelRegistryClient>("eastus", fl::test::NullLog());
+  registry->SetHttpGet([](const std::string&) -> std::string {
+    return R"({"blobSasUri": "https://storage.blob.core.windows.net/c?sig=test"})";
+  });
+  manager.SetModelRegistryClient(std::move(registry));
+
+  class RendezvousDownloader : public IBlobDownloader {
+   public:
+    std::mutex m;
+    std::condition_variable cv;
+    int arrived = 0;
+    bool released = false;
+    std::atomic<int> timeouts{0};
+
+    std::vector<BlobItemInfo> ListBlobs(const std::string&) override {
+      return {{"variant-cpu/weights.bin", 16}};
+    }
+
+    void DownloadBlob(const std::string&, const std::string& blob_name,
+                      const std::string& local_path, int,
+                      BlobBytesWrittenFn bytes_written_cb,
+                      std::atomic<bool>*) override {
+      {
+        std::unique_lock<std::mutex> lk(m);
+        if (++arrived >= 2) {  // both concurrent downloads reached the rendezvous
+          released = true;
+          cv.notify_all();
+        } else if (!cv.wait_for(lk, std::chrono::seconds(5), [&] { return released; })) {
+          ++timeouts;  // peer never arrived within the window → downloads serialized
+        }
+      }
+
+      auto parent = fs::path(local_path).parent_path();
+      if (!parent.empty()) {
+        fs::create_directories(parent);
+      }
+      std::ofstream f(local_path);
+      f << "data for " << blob_name;
+      if (bytes_written_cb) {
+        bytes_written_cb(16);
+      }
+    }
+  };
+
+  auto rendezvous = std::make_unique<RendezvousDownloader>();
+  auto* rendezvous_raw = rendezvous.get();
+  manager.SetBlobDownloader(std::move(rendezvous));
+
+  auto make_info = [](const char* id, const char* publisher) {
+    ModelInfo info;
+    info.model_id = id;
+    info.name = id;
+    info.uri = std::string("azureml://registries/test/models/") + id + "/versions/1";
+    info.string_properties[FOUNDRY_LOCAL_MODEL_PROP_PUBLISHER_STR] = publisher;
+    return info;
+  };
+  auto info_a = make_info("model-a:1", "PubA");
+  auto info_b = make_info("model-b:1", "PubB");
+
+  std::atomic<int> exceptions{0};
+  std::thread t1([&] {
+    try {
+      manager.DownloadModel(info_a);
+    } catch (...) {
+      ++exceptions;
+    }
+  });
+  std::thread t2([&] {
+    try {
+      manager.DownloadModel(info_b);
+    } catch (...) {
+      ++exceptions;
+    }
+  });
+  t1.join();
+  t2.join();
+
+  EXPECT_EQ(exceptions.load(), 0);
+  EXPECT_TRUE(rendezvous_raw->released)
+      << "Both different-model downloads should have met at the rendezvous.";
+  EXPECT_EQ(rendezvous_raw->timeouts.load(), 0)
+      << "Downloads of different models must run concurrently, not serialize.";
+}
+
 // HasInferenceModelJson must return false instead of throwing when the path
 // it's asked about is not a directory (e.g. a regular file). Previously the
 // underlying directory_iterator would throw filesystem_error.