fix(serverless): VolumeCache copies files on volumes that reject utime/chmod

runpod/serverless/utils/rp_volume_cache.py

@@ -219,19 +219,49 @@ def _needs_copy(src_path, dst_path):
             return True
         return s.st_size != d.st_size or s.st_mtime > d.st_mtime + _MTIME_TOLERANCE
 
+    @staticmethod
+    def _sync_needs_copy(src_path, dst_path, prior):
+        """Container -> mirror: does dst need a fresh copy of src?
+
+        ``prior`` is src's entry in the previous manifest. It records the
+        source metadata the mirror copy was made from, which stays reliable
+        when the volume can't carry mtimes (the mirror then holds copy-time
+        mtimes). Without a prior entry, only an exact size+mtime match counts
+        as current.
+        """
+        try:
+            s = os.stat(src_path)
+        except OSError:
+            return False
+        try:
+            d = os.stat(dst_path)
+        except OSError:
+            return True
+        if s.st_size != d.st_size:
+            return True
+        if prior is None:
+            return abs(s.st_mtime - d.st_mtime) > _MTIME_TOLERANCE
+        # Only recopy when the source is NEWER than the metadata the mirror was
+        # made from. A source older than the manifest entry is a stale worker or
+        # a sibling that downloaded identical bytes at a different time; recopying
+        # there churns every big file across workers and can overwrite newer
+        # mirror bytes with older ones.
+        return prior["size"] != s.st_size or s.st_mtime > prior["mtime"] + _MTIME_TOLERANCE
+
     def _is_safe_dest(self, dst_abs):
         target = os.path.realpath(dst_abs)
         return any(target == d or target.startswith(d + os.sep) for d in self._dirs)
 
-    def _copy_file(self, src, dst):
+    def _copy_file(self, src, dst, mtime=None):
         tmp = f"{dst}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp"
         try:
             os.makedirs(os.path.dirname(dst), exist_ok=True)
-            shutil.copy2(src, tmp)  # preserves mtime so future diffs converge
+            shutil.copyfile(src, tmp)
+            _copy_metadata(src, tmp, mtime)
             os.replace(tmp, dst)  # atomic; last-writer-wins under concurrency
             return True
         except OSError as exc:
-            log.debug(f"VolumeCache: skip {src} -> {dst}: {exc}")
+            log.warn(f"VolumeCache: failed to copy {src} -> {dst}: {exc}")
             try:
                 if os.path.exists(tmp):
                     os.remove(tmp)
@@ -240,29 +270,35 @@ def _copy_file(self, src, dst):
                 pass
             return False
 
-    def _copy_parallel(self, pairs):
+    def _copy_parallel(self, pairs, needs_copy=None, mtime_of=None):
         """Copy every (src, dst) that needs it, in parallel.
 
+        ``needs_copy(src, dst)`` decides which pairs to copy (default: compare
+        the two files). ``mtime_of(src)`` gives the mtime to stamp on dst
+        (default: src's own).
+
         Returns ``(copied, current)`` sets of source paths:
         - ``copied``  -- sources transferred successfully this call.
         - ``current`` -- sources skipped because the destination was already
           up to date.
         Their union is the set of sources the destination now holds a current
         copy of; a source in neither failed to copy. Callers that record a
         "mirror complete" manifest MUST include only sources in that union, or
-        a failed copy would be marked present and later served stale. copy2 +
+        a failed copy would be marked present and later served stale. copyfile +
         atomic os.replace means a source in ``copied`` matches its destination
         at that instant, so no re-stat is needed to confirm it.
         """
-        todo = [(s, d) for s, d in pairs if self._needs_copy(s, d)]
+        needs_copy = needs_copy or self._needs_copy
+        mtime_of = mtime_of or (lambda _src: None)
+        todo = [(s, d) for s, d in pairs if needs_copy(s, d)]
         current = {s for s, _ in pairs} - {s for s, _ in todo}
         if not todo:
             return set(), current
         with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
             # pool.map preserves input order, so zip pairs each result with its
             # source. This holds one path per copied file (unavoidable: per-file
             # identity is required to keep failed copies out of the manifest).
-            results = pool.map(lambda sd: self._copy_file(sd[0], sd[1]), todo)
+            results = pool.map(lambda sd: self._copy_file(sd[0], sd[1], mtime=mtime_of(sd[0])), todo)
             copied = {s for (s, _), ok in zip(todo, results) if ok}
         return copied, current
 
@@ -383,6 +419,7 @@ def _do_hydrate(self):
         if manifest.get("small"):
             restored += self._extract_small()
         big_pairs = []
+        entries = {}
         for entry in manifest.get("big", []):
             rel = os.path.relpath(entry["path"], "/")
             dst = os.path.join("/", rel)
@@ -392,7 +429,15 @@ def _do_hydrate(self):
             # big/ so a malicious entry can't read outside the mirror.
             if self._is_safe_dest(dst) and self._within(self._big_root, src):
                 big_pairs.append((src, dst))
-        copied, _current = self._copy_parallel(big_pairs)
+                entries[src] = entry
+        # Compare the container against the manifest's source metadata, not
+        # the mirror file's own mtime, and stamp that source mtime on restore.
+        copied, _current = self._copy_parallel(
+            big_pairs,
+            needs_copy=lambda src, _dst: os.path.isfile(src)
+            and not self._meta_satisfied_by_local(entries[src]),
+            mtime_of=lambda src: entries[src]["mtime"],
+        )
         restored += len(copied)
         if restored:
             log.info(f"VolumeCache: hydrated {restored} file(s) from {self._mirror_root}")
@@ -450,7 +495,11 @@ def _do_sync(self):
             _silent_remove(self._small_archive_path)
 
         big_pairs = [(f, os.path.join(self._big_root, os.path.relpath(f, "/"))) for f in big_files]
-        copied, current = self._copy_parallel(big_pairs)
+        prior_big = {e["path"]: e for e in (manifest.get("big", []) if manifest else [])}
+        copied, current = self._copy_parallel(
+            big_pairs,
+            needs_copy=lambda src, dst: self._sync_needs_copy(src, dst, prior_big.get(src)),
+        )
         # Record only sources the volume actually holds a current copy of.
         # A big file whose copy failed is in neither set, so it is omitted and
         # the manifest never claims it present -- hydrate then skips it rather
@@ -485,6 +534,30 @@ def __exit__(self, *exc):
 # ----------------------------------------------------------------------- #
 
 
+def _copy_metadata(src, dst, mtime=None):
+    """Stamp dst with an mtime and src's permission bits, each best-effort.
+
+    The mtime is ``mtime`` when given (hydrate passes the manifest's source
+    mtime), else src's own. Some volume filesystems (FUSE object stores such
+    as Global Store) reject utime and chmod with EPERM. The copy is still
+    valid then: reconciliation compares against the manifest's recorded
+    source metadata, not the mirror file's copy-time mtime. Only the
+    permission bits are lost on such a volume.
+    """
+    try:
+        if mtime is None:
+            st = os.stat(src)
+            os.utime(dst, ns=(st.st_atime_ns, st.st_mtime_ns))
+        else:
+            os.utime(dst, (mtime, mtime))
+    except OSError as exc:
+        log.debug(f"VolumeCache: kept copy-time mtime on {dst}: {exc}")
+    try:
+        shutil.copymode(src, dst)
+    except OSError as exc:
+        log.debug(f"VolumeCache: kept default mode on {dst}: {exc}")
+
+
 def _silent_remove(path):
     try:
         os.remove(path)

tests/test_serverless/test_utils/test_rp_volume_cache.py

@@ -1,3 +1,4 @@
+import errno
 import json
 import os
 import shutil
@@ -404,14 +405,203 @@ def test_copy_file_returns_false_and_cleans_tmp_on_failure(tmp_path, monkeypatch
     src.write_text("data")
     dst = cache / "dst.bin"
 
-    def flaky_copy2(_src, _dst):
+    def flaky_copyfile(_src, _dst, **_kwargs):
         raise OSError("disk full")
 
-    monkeypatch.setattr(shutil, "copy2", flaky_copy2)
+    monkeypatch.setattr(shutil, "copyfile", flaky_copyfile)
     assert vc._copy_file(str(src), str(dst)) is False
     assert not os.path.exists(str(dst) + ".rpvc.tmp")
 
 
+def test_copy_failure_is_logged_as_warning(tmp_path, monkeypatch):
+    vc, cache, _vol = _mk_cache_with_volume(tmp_path)
+    src = cache / "f.bin"
+    src.write_text("data")
+    warnings = []
+
+    def flaky_copyfile(_src, _dst, **_kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(shutil, "copyfile", flaky_copyfile)
+    monkeypatch.setattr(vcmod.log, "warn", lambda msg, *_a, **_kw: warnings.append(msg))
+    vc._copy_file(str(src), str(cache / "dst.bin"))
+
+    assert any("f.bin" in w and "disk full" in w for w in warnings)
+
+
+def test_copy_file_preserves_mtime_and_mode(tmp_path):
+    vc, cache, _vol = _mk_cache_with_volume(tmp_path)
+    src = cache / "run.sh"
+    src.write_text("#!/bin/sh\n")
+    os.chmod(src, 0o700)
+    os.utime(src, (1_000_000_000, 1_000_000_000))
+    dst = cache / "copy.sh"
+
+    assert vc._copy_file(str(src), str(dst)) is True
+    assert os.stat(dst).st_mtime == 1_000_000_000
+    assert os.stat(dst).st_mode & 0o777 == 0o700
+
+
+def _deny_metadata_ops(monkeypatch):
+    """Simulate a FUSE object store (such as Global Store) that rejects chmod/utime."""
+
+    def deny(*_args, **_kwargs):
+        raise PermissionError(errno.EPERM, "Operation not permitted")
+
+    monkeypatch.setattr(os, "chmod", deny)
+    monkeypatch.setattr(os, "utime", deny)
+
+
+def _deny_metadata_ops_under(monkeypatch, root):
+    """Like _deny_metadata_ops, but only for paths on the volume; local disk behaves normally."""
+    root = str(root)
+    real_chmod, real_utime = os.chmod, os.utime
+
+    def guarded(real):
+        def call(path, *args, **kwargs):
+            if str(path).startswith(root):
+                raise PermissionError(errno.EPERM, "Operation not permitted")
+            return real(path, *args, **kwargs)
+
+        return call
+
+    monkeypatch.setattr(os, "chmod", guarded(real_chmod))
+    monkeypatch.setattr(os, "utime", guarded(real_utime))
+
+
+def _big_payload(fill):
+    return bytes([fill]) * (vcmod._SMALL_FILE_THRESHOLD + 1)
+
+
+def _set_mtime(path, mtime):
+    os.utime(path, (mtime, mtime))
+
+
+def test_sync_recopies_same_size_update_when_volume_rejects_utime(tmp_path, monkeypatch):
+    # The mirror copy carries its copy time, not the source's mtime. A same-size
+    # update with an mtime between the original and that copy time must still
+    # be detected, via the manifest's recorded source metadata.
+    vc, cache, vol = _mk_cache_with_volume(tmp_path)
+    f = cache / "model.safetensors"
+    now = time.time()
+    f.write_bytes(_big_payload(1))
+    _set_mtime(f, now - 1000)
+    _deny_metadata_ops_under(monkeypatch, vol)
+    vc.sync(background=False)
+
+    f.write_bytes(_big_payload(2))
+    _set_mtime(f, now - 500)
+    vc.sync(background=False)
+    f.unlink()
+    VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)).hydrate()
+
+    assert f.read_bytes() == _big_payload(2)
+
+
+def test_hydrate_keeps_newer_local_file_when_volume_rejects_utime(tmp_path, monkeypatch):
+    vc, cache, vol = _mk_cache_with_volume(tmp_path)
+    f = cache / "model.safetensors"
+    now = time.time()
+    f.write_bytes(_big_payload(1))
+    _set_mtime(f, now - 1000)
+    _deny_metadata_ops_under(monkeypatch, vol)
+    vc.sync(background=False)
+
+    # Newer than the synced source, older than the mirror's copy-time mtime.
+    f.write_bytes(_big_payload(2))
+    _set_mtime(f, now - 500)
+    VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)).hydrate()
+
+    assert f.read_bytes() == _big_payload(2)
+
+
+def test_sync_needs_copy_ignores_source_older_than_manifest(tmp_path):
+    # Container -> mirror sync must only recopy when the source is NEWER than the
+    # metadata the mirror was last made from. A source whose mtime is older than
+    # the manifest entry (a stale worker, or a sibling that downloaded identical
+    # bytes at a different time) must not trigger a rewrite: recopying there
+    # churns every big file across workers and can clobber newer mirror bytes.
+    src = tmp_path / "src.bin"
+    dst = tmp_path / "dst.bin"
+    payload = b"x" * 1024
+    src.write_bytes(payload)
+    dst.write_bytes(payload)  # same size, so the size early-return does not fire
+    mid = time.time() - 1000
+    prior = {"size": len(payload), "mtime": mid}
+
+    _set_mtime(src, mid - 100)  # source older than the manifest record
+    assert VolumeCache._sync_needs_copy(str(src), str(dst), prior) is False
+
+    _set_mtime(src, mid + 100)  # source newer: a genuine update still copies
+    assert VolumeCache._sync_needs_copy(str(src), str(dst), prior) is True
+
+    # A size change relative to the manifest always recopies, regardless of mtime.
+    _set_mtime(src, mid - 100)
+    assert VolumeCache._sync_needs_copy(str(src), str(dst), {"size": len(payload) + 1, "mtime": mid}) is True
+
+
+def test_hydrate_restores_source_mtime_so_next_sync_is_noop(tmp_path, monkeypatch):
+    vc, cache, vol = _mk_cache_with_volume(tmp_path)
+    f = cache / "model.safetensors"
+    f.write_bytes(_big_payload(1))
+    _set_mtime(f, time.time() - 1000)
+    source_mtime = os.stat(f).st_mtime
+    _deny_metadata_ops_under(monkeypatch, vol)
+    vc.sync(background=False)
+    f.unlink()
+
+    fresh = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol))
+    assert fresh.hydrate() == 1
+    assert abs(os.stat(f).st_mtime - source_mtime) < 1e-3
+
+    copies = []
+    real_copy = fresh._copy_file
+    monkeypatch.setattr(fresh, "_copy_file", lambda src, dst, **kw: copies.append(src) or real_copy(src, dst, **kw))
+    fresh.sync(background=False)
+    assert copies == []
+
+
+def test_sync_recopies_when_mirror_file_is_missing(tmp_path):
+    vc, cache, vol = _mk_cache_with_volume(tmp_path)
+    f = cache / "model.safetensors"
+    f.write_bytes(_big_payload(1))
+    vc.sync(background=False)
+    mirrored = vol / ".cache" / "ep1" / "big" / os.path.relpath(str(f), "/")
+    mirrored.unlink()
+
+    vc.sync(background=False)
+
+    assert mirrored.read_bytes() == _big_payload(1)
+
+
+def test_big_file_round_trips_when_volume_rejects_metadata_ops(tmp_path, monkeypatch):
+    vc, cache, vol = _mk_cache_with_volume(tmp_path)
+    payload = os.urandom(vcmod._SMALL_FILE_THRESHOLD + 1)
+    (cache / "model.safetensors").write_bytes(payload)
+    _deny_metadata_ops(monkeypatch)
+
+    vc.sync(background=False)
+    (cache / "model.safetensors").unlink()
+    restored = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)).hydrate()
+
+    assert restored == 1
+    assert (cache / "model.safetensors").read_bytes() == payload
+
+
+def test_sync_converges_when_volume_rejects_metadata_ops(tmp_path, monkeypatch):
+    vc, cache, _vol = _mk_cache_with_volume(tmp_path)
+    (cache / "model.safetensors").write_bytes(os.urandom(vcmod._SMALL_FILE_THRESHOLD + 1))
+    _deny_metadata_ops(monkeypatch)
+    vc.sync(background=False)
+
+    copies = []
+    real_copy = vc._copy_file
+    monkeypatch.setattr(vc, "_copy_file", lambda src, dst, **kw: copies.append(src) or real_copy(src, dst, **kw))
+    vc.sync(background=False)
+
+    assert copies == []
+
+
 def test_join_pending_syncs_swallows_join_exception():
     class FlakyThread:
         def join(self):
@@ -684,10 +874,10 @@ def test_sync_omits_big_file_from_manifest_when_copy_fails(tmp_path, monkeypatch
     bad.write_bytes(b"new-v2" + big + b"-and-then-some-extra-bytes")
     real_copy_file = vc._copy_file
 
-    def flaky_copy(src, dst):
+    def flaky_copy(src, dst, **kwargs):
         if src == str(bad):
             return False  # simulate a failed overwrite (leaves stale v1 on volume)
-        return real_copy_file(src, dst)
+        return real_copy_file(src, dst, **kwargs)
 
     monkeypatch.setattr(vc, "_copy_file", flaky_copy)
     vc._do_sync()