diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 35bcc170..cd654eb6 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/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,9 +270,13 @@ 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 @@ -250,11 +284,13 @@ def _copy_parallel(self, pairs): 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 @@ -262,7 +298,7 @@ def _copy_parallel(self, pairs): # 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) diff --git a/tests/test_serverless/test_utils/test_rp_volume_cache.py b/tests/test_serverless/test_utils/test_rp_volume_cache.py index 891deef2..179eb1a9 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/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()