From fec7fc3679cc86c16620b2f80b11334b420931cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 01:54:05 -0700 Subject: [PATCH 01/20] feat(serverless): add VolumeCache skeleton with availability gating --- runpod/serverless/utils/rp_volume_cache.py | 41 +++++++++++++++++++ .../test_utils/test_rp_volume_cache.py | 32 +++++++++++++++ 2 files changed, 73 insertions(+) create mode 100644 runpod/serverless/utils/rp_volume_cache.py create mode 100644 tests/test_serverless/test_utils/test_rp_volume_cache.py diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py new file mode 100644 index 00000000..f6b67a42 --- /dev/null +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -0,0 +1,41 @@ +"""Bidirectional warm-cache sync between local directories and a network volume.""" + +import os +import time +import uuid +import threading + +from runpod.serverless.modules.rp_logger import RunPodLogger + +log = RunPodLogger() + + +class VolumeCache: + """Hydrate configured dirs from a network volume on cold start; sync their + delta back as a worker-owned tar shard. Best-effort and stdlib-only.""" + + def __init__( + self, + dirs, + *, + namespace=None, + volume_path="/runpod-volume", + max_size_gb=None, + best_effort=True, + ): + self._dirs = [os.path.abspath(os.fspath(d)) for d in dirs] + self._namespace = namespace or os.environ.get("RUNPOD_ENDPOINT_ID") or "" + self._volume_path = os.fspath(volume_path) + self._max_size_gb = max_size_gb + self._best_effort = best_effort + self._worker_id = os.environ.get("RUNPOD_POD_ID") or uuid.uuid4().hex[:12] + self._baseline = time.time() + self._lock = threading.Lock() + + @property + def _shard_dir(self): + return os.path.join(self._volume_path, ".cache", self._namespace) + + @property + def available(self): + return bool(self._namespace) and os.path.isdir(self._volume_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 new file mode 100644 index 00000000..dd843907 --- /dev/null +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -0,0 +1,32 @@ +import os +import pytest +from runpod.serverless.utils.rp_volume_cache import VolumeCache + + +def test_unavailable_when_volume_dir_missing(tmp_path): + vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", + volume_path=str(tmp_path / "no-volume")) + assert vc.available is False + + +def test_available_when_volume_present_and_namespace_set(tmp_path): + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol)) + assert vc.available is True + + +def test_unavailable_without_namespace(tmp_path, monkeypatch): + monkeypatch.delenv("RUNPOD_ENDPOINT_ID", raising=False) + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "cache")], volume_path=str(vol)) + assert vc.available is False + + +def test_namespace_defaults_to_endpoint_id(tmp_path, monkeypatch): + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "endpoint-xyz") + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "cache")], volume_path=str(vol)) + assert vc._shard_dir == os.path.join(str(vol), ".cache", "endpoint-xyz") From 2853f386be901d7eb2944fc03ecfdc7e3828d0b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 01:58:53 -0700 Subject: [PATCH 02/20] feat(serverless): VolumeCache.sync writes collision-free delta shards --- runpod/serverless/utils/rp_volume_cache.py | 50 +++++++++++++++++ .../test_utils/test_rp_volume_cache.py | 56 +++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index f6b67a42..7163a93a 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -1,6 +1,7 @@ """Bidirectional warm-cache sync between local directories and a network volume.""" import os +import tarfile import time import uuid import threading @@ -14,6 +15,8 @@ class VolumeCache: """Hydrate configured dirs from a network volume on cold start; sync their delta back as a worker-owned tar shard. Best-effort and stdlib-only.""" + _EXCLUDE_SUBSTRINGS = (os.sep + "refs" + os.sep, os.sep + ".no_exist" + os.sep) + def __init__( self, dirs, @@ -39,3 +42,50 @@ def _shard_dir(self): @property def available(self): return bool(self._namespace) and os.path.isdir(self._volume_path) + + def _list_shards(self): + d = self._shard_dir + if not os.path.isdir(d): + return [] + shards = [os.path.join(d, f) for f in os.listdir(d) if f.endswith(".tar")] + return sorted(shards, key=os.path.getmtime) + + def _iter_delta_files(self): + for root in self._dirs: + if not os.path.isdir(root): + continue + for dirpath, _dirs, files in os.walk(root): + for name in files: + path = os.path.join(dirpath, name) + if name.endswith(".lock") or name.startswith(".rp_volume_cache"): + continue + if any(sub in path for sub in self._EXCLUDE_SUBSTRINGS): + continue + try: + if os.path.getmtime(path) > self._baseline: + yield path + except OSError: + continue + + def _guard(self, fn, default): + return fn() + + def sync(self): + if not self.available: + return False + return self._guard(self._do_sync, False) + + def _do_sync(self): + files = list(self._iter_delta_files()) + if not files: + log.debug("VolumeCache: no delta files to sync") + return False + os.makedirs(self._shard_dir, exist_ok=True) + final = os.path.join(self._shard_dir, f"{self._worker_id}-{time.time_ns():020d}.tar") + tmp = final + ".tmp" + with tarfile.open(tmp, "w") as tar: + for path in files: + tar.add(path, arcname=os.path.relpath(path, "/")) + os.replace(tmp, final) + log.info(f"VolumeCache: synced {len(files)} files to {final}") + return True 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 dd843907..a02d9d8e 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1,4 +1,6 @@ import os +import tarfile +import time import pytest from runpod.serverless.utils.rp_volume_cache import VolumeCache @@ -30,3 +32,57 @@ def test_namespace_defaults_to_endpoint_id(tmp_path, monkeypatch): vol.mkdir() vc = VolumeCache([str(tmp_path / "cache")], volume_path=str(vol)) assert vc._shard_dir == os.path.join(str(vol), ".cache", "endpoint-xyz") + + +def _mk_cache_with_volume(tmp_path): + cache = tmp_path / "cache" + cache.mkdir() + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + return vc, cache, vol + + +def test_sync_packs_files_created_after_baseline(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + (cache / "model.bin").write_text("weights") + assert vc.sync() is True + shards = vc._list_shards() + assert len(shards) == 1 + with tarfile.open(shards[0]) as tar: + names = tar.getnames() + assert os.path.relpath(str(cache / "model.bin"), "/") in names + + +def test_sync_excludes_files_older_than_baseline(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + old = cache / "old.bin" + old.write_text("x") + os.utime(old, (time.time() - 100, time.time() - 100)) + vc._baseline = time.time() + assert vc.sync() is False + + +def test_sync_skips_excluded_paths(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + (cache / "refs").mkdir() + (cache / "refs" / "main").write_text("ref") + assert vc.sync() is False + + +def test_sync_shard_names_are_unique_per_call(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + (cache / "a.bin").write_text("a") + vc.sync() + (cache / "b.bin").write_text("b") + vc.sync() + assert len(vc._list_shards()) == 2 + + +def test_sync_noop_when_unavailable(tmp_path): + vc = VolumeCache([str(tmp_path / "c")], namespace="ep1", + volume_path=str(tmp_path / "missing")) + assert vc.sync() is False From f0a76f832714c0606ee491cdab30c9c6da0dd009 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 02:07:06 -0700 Subject: [PATCH 03/20] fix(serverless): tolerate coarse network-fs mtime granularity in VolumeCache delta --- runpod/serverless/utils/rp_volume_cache.py | 5 ++++- .../test_utils/test_rp_volume_cache.py | 16 ++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 7163a93a..770c21a4 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -10,6 +10,8 @@ log = RunPodLogger() +_BASELINE_EPSILON_SECONDS = 2.0 # tolerate coarse (1s) network-filesystem mtime granularity + class VolumeCache: """Hydrate configured dirs from a network volume on cold start; sync their @@ -32,7 +34,7 @@ def __init__( self._max_size_gb = max_size_gb self._best_effort = best_effort self._worker_id = os.environ.get("RUNPOD_POD_ID") or uuid.uuid4().hex[:12] - self._baseline = time.time() + self._baseline = time.time() - _BASELINE_EPSILON_SECONDS self._lock = threading.Lock() @property @@ -88,4 +90,5 @@ def _do_sync(self): tar.add(path, arcname=os.path.relpath(path, "/")) os.replace(tmp, final) log.info(f"VolumeCache: synced {len(files)} files to {final}") + self._baseline = time.time() - _BASELINE_EPSILON_SECONDS return True 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 a02d9d8e..769b1685 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -86,3 +86,19 @@ def test_sync_noop_when_unavailable(tmp_path): vc = VolumeCache([str(tmp_path / "c")], namespace="ep1", volume_path=str(tmp_path / "missing")) assert vc.sync() is False + + +def test_sync_tolerates_coarse_mtime_granularity(tmp_path): + # A fresh instance sets baseline to now - epsilon. A file whose mtime is + # floored to the current integer second (as coarse NFS filesystems report) + # must still be picked up, not silently dropped. + cache = tmp_path / "cache" + cache.mkdir() + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + f = cache / "model.bin" + f.write_text("weights") + now = time.time() + os.utime(f, (float(int(now)), float(int(now)))) # floor mtime to integer second + assert vc.sync() is True From 4b0f6df9fe388f0a6be40bcc8ddee4770ffea551 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 02:10:20 -0700 Subject: [PATCH 04/20] feat(serverless): VolumeCache.hydrate extracts shards with idempotent marker --- runpod/serverless/utils/rp_volume_cache.py | 45 +++++++++++++++++++ .../test_utils/test_rp_volume_cache.py | 40 +++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 770c21a4..180ae778 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -2,6 +2,7 @@ import os import tarfile +import tempfile import time import uuid import threading @@ -92,3 +93,47 @@ def _do_sync(self): log.info(f"VolumeCache: synced {len(files)} files to {final}") self._baseline = time.time() - _BASELINE_EPSILON_SECONDS return True + + @property + def _marker_path(self): + base = os.path.join(tempfile.gettempdir(), "rp_volume_cache") + return os.path.join(base, f"{self._namespace}.hydrated") + + def _newest_shard_mtime(self): + shards = self._list_shards() + return os.path.getmtime(shards[-1]) if shards else 0.0 + + def _clear_marker_for_test(self): + if os.path.exists(self._marker_path): + os.remove(self._marker_path) + + def hydrate(self): + if not self.available: + return False + return self._guard(self._do_hydrate, False) + + def _do_hydrate(self): + shards = self._list_shards() + if not shards: + return False + newest = self._newest_shard_mtime() + if os.path.exists(self._marker_path) and os.path.getmtime(self._marker_path) >= newest: + log.debug("VolumeCache: cache already hydrated, skipping") + return False + extracted = False + for shard in shards: # oldest -> newest (last wins) + with tarfile.open(shard) as tar: + safe = [m for m in tar.getmembers() if self._is_safe_member(m)] + tar.extractall(path="/", members=safe, filter=tarfile.data_filter) + extracted = extracted or bool(safe) + os.makedirs(os.path.dirname(self._marker_path), exist_ok=True) + with open(self._marker_path, "w") as fh: + fh.write(str(newest)) + os.utime(self._marker_path, (newest, newest)) + self._baseline = time.time() - _BASELINE_EPSILON_SECONDS + if extracted: + log.info(f"VolumeCache: hydrated from {len(shards)} shard(s)") + return extracted + + def _is_safe_member(self, member): + return member.isfile() or member.isdir() 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 769b1685..39714ab3 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -102,3 +102,43 @@ def test_sync_tolerates_coarse_mtime_granularity(tmp_path): now = time.time() os.utime(f, (float(int(now)), float(int(now)))) # floor mtime to integer second assert vc.sync() is True + + +def test_hydrate_noop_when_no_shards(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + assert vc.hydrate() is False + + +def test_hydrate_restores_files_to_absolute_paths(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + (cache / "model.bin").write_text("weights") + vc.sync() + (cache / "model.bin").unlink() # simulate a fresh cold worker + assert vc.hydrate() is True + assert (cache / "model.bin").read_text() == "weights" + + +def test_hydrate_later_shard_overwrites_earlier(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + f = cache / "model.bin" + f.write_text("v1") + vc.sync() + time.sleep(0.01) + f.write_text("v2") + vc._baseline = time.time() - 5 + vc.sync() + f.unlink() + vc._clear_marker_for_test() + vc.hydrate() + assert f.read_text() == "v2" + + +def test_hydrate_is_idempotent_via_marker(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._baseline = time.time() - 5 + (cache / "model.bin").write_text("weights") + vc.sync() + assert vc.hydrate() is True # first hydrate extracts + assert vc.hydrate() is False # marker current -> no-op From b95e1f14c043bfd5bb5dbe8cdc925a4b6d8e558a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 02:39:30 -0700 Subject: [PATCH 05/20] fix(serverless): drop 3.12-only tarfile filter kwarg to honor py3.10 floor --- runpod/serverless/utils/rp_volume_cache.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 180ae778..5bfb330d 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -124,7 +124,7 @@ def _do_hydrate(self): for shard in shards: # oldest -> newest (last wins) with tarfile.open(shard) as tar: safe = [m for m in tar.getmembers() if self._is_safe_member(m)] - tar.extractall(path="/", members=safe, filter=tarfile.data_filter) + tar.extractall(path="/", members=safe) extracted = extracted or bool(safe) os.makedirs(os.path.dirname(self._marker_path), exist_ok=True) with open(self._marker_path, "w") as fh: From 6fc8250d489e7674f8fa851dd58b540f4e11704c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 02:42:52 -0700 Subject: [PATCH 06/20] fix(serverless): apply tar data filter when available, honoring py3.10 floor --- runpod/serverless/utils/rp_volume_cache.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 5bfb330d..cbc66fd4 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -121,10 +121,14 @@ def _do_hydrate(self): log.debug("VolumeCache: cache already hydrated, skipping") return False extracted = False + # Use the tar data filter for defense-in-depth where the runtime provides + # it (Python 3.12+, and 3.10.12+/3.11.4+ backports) without requiring it -- + # the >=3.10 floor may predate the API. _is_safe_member is the primary guard. + extract_kwargs = {"filter": "data"} if hasattr(tarfile, "data_filter") else {} for shard in shards: # oldest -> newest (last wins) with tarfile.open(shard) as tar: safe = [m for m in tar.getmembers() if self._is_safe_member(m)] - tar.extractall(path="/", members=safe) + tar.extractall(path="/", members=safe, **extract_kwargs) extracted = extracted or bool(safe) os.makedirs(os.path.dirname(self._marker_path), exist_ok=True) with open(self._marker_path, "w") as fh: From 5b76d8d480342e2c5cf597f53a14aa9551db506b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 02:45:58 -0700 Subject: [PATCH 07/20] feat(serverless): harden VolumeCache extract against traversal and symlinks --- runpod/serverless/utils/rp_volume_cache.py | 8 ++++++- .../test_utils/test_rp_volume_cache.py | 24 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index cbc66fd4..6d4a402a 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -140,4 +140,10 @@ def _do_hydrate(self): return extracted def _is_safe_member(self, member): - return member.isfile() or member.isdir() + if not (member.isfile() or member.isdir()): + return False # reject symlink/hardlink/device/fifo + target = os.path.realpath(os.path.join("/", member.name)) + return any( + target == d or target.startswith(d + os.sep) + for d in self._dirs + ) 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 39714ab3..e9b0b5b2 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -142,3 +142,27 @@ def test_hydrate_is_idempotent_via_marker(tmp_path): vc.sync() assert vc.hydrate() is True # first hydrate extracts assert vc.hydrate() is False # marker current -> no-op + + +def test_rejects_member_outside_configured_dirs(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + m = tarfile.TarInfo(name="etc/passwd") # resolves to /etc/passwd, outside cache + m.type = tarfile.REGTYPE + assert vc._is_safe_member(m) is False + + +def test_rejects_symlink_member(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + rel = os.path.relpath(str(cache / "link"), "/") + m = tarfile.TarInfo(name=rel) + m.type = tarfile.SYMTYPE + m.linkname = "/etc/passwd" + assert vc._is_safe_member(m) is False + + +def test_accepts_regular_member_inside_dirs(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + rel = os.path.relpath(str(cache / "model.bin"), "/") + m = tarfile.TarInfo(name=rel) + m.type = tarfile.REGTYPE + assert vc._is_safe_member(m) is True From b52bb4822e439cfdc4f03701bea0a562f1a30b3a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 14:36:25 -0700 Subject: [PATCH 08/20] fix(serverless): realpath-normalize cache dirs so symlinked mounts match --- runpod/serverless/utils/rp_volume_cache.py | 2 +- .../test_utils/test_rp_volume_cache.py | 21 +++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 6d4a402a..693f7714 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -29,7 +29,7 @@ def __init__( max_size_gb=None, best_effort=True, ): - self._dirs = [os.path.abspath(os.fspath(d)) for d in dirs] + self._dirs = [os.path.realpath(os.fspath(d)) for d in dirs] self._namespace = namespace or os.environ.get("RUNPOD_ENDPOINT_ID") or "" self._volume_path = os.fspath(volume_path) self._max_size_gb = max_size_gb 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 e9b0b5b2..ce5fa6a2 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -166,3 +166,24 @@ def test_accepts_regular_member_inside_dirs(tmp_path): m = tarfile.TarInfo(name=rel) m.type = tarfile.REGTYPE assert vc._is_safe_member(m) is True + + +def test_rejects_sibling_prefix_collision(tmp_path): + # An allowed dir ".../cache" must NOT match a sibling ".../cache-evil"; + # this locks the separator-anchored prefix check against substring regressions. + vc, cache, vol = _mk_cache_with_volume(tmp_path) + evil = tmp_path / "cache-evil" + evil.mkdir() + rel = os.path.relpath(str(evil / "x"), "/") + m = tarfile.TarInfo(name=rel) + m.type = tarfile.REGTYPE + assert vc._is_safe_member(m) is False + + +def test_rejects_hardlink_member(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + rel = os.path.relpath(str(cache / "hl"), "/") + m = tarfile.TarInfo(name=rel) + m.type = tarfile.LNKTYPE + m.linkname = "root/.cache/x" + assert vc._is_safe_member(m) is False From fd19f6598ea4500e8668e4cb66cb79746645f3f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 14:39:09 -0700 Subject: [PATCH 09/20] feat(serverless): add size-capped retention to VolumeCache --- runpod/serverless/utils/rp_volume_cache.py | 18 ++++++++++++++ .../test_utils/test_rp_volume_cache.py | 24 +++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 693f7714..be38cb9b 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -92,8 +92,26 @@ def _do_sync(self): os.replace(tmp, final) log.info(f"VolumeCache: synced {len(files)} files to {final}") self._baseline = time.time() - _BASELINE_EPSILON_SECONDS + self._enforce_retention() return True + def _enforce_retention(self): + if not self._max_size_gb: + return + cap = self._max_size_gb * (1024 ** 3) + shards = self._list_shards() # oldest first + total = sum(os.path.getsize(s) for s in shards) + for shard in shards: + if total <= cap: + break + size = os.path.getsize(shard) + try: + os.remove(shard) + total -= size + log.info(f"VolumeCache: pruned old shard {shard}") + except OSError as exc: + log.warn(f"VolumeCache: failed to prune {shard}: {exc}") + @property def _marker_path(self): base = os.path.join(tempfile.gettempdir(), "rp_volume_cache") 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 ce5fa6a2..fd107570 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -187,3 +187,27 @@ def test_rejects_hardlink_member(tmp_path): m.type = tarfile.LNKTYPE m.linkname = "root/.cache/x" assert vc._is_safe_member(m) is False + + +def test_retention_prunes_oldest_shards_past_cap(tmp_path): + cache = tmp_path / "cache"; cache.mkdir() + vol = tmp_path / "volume"; vol.mkdir() + cap_bytes = 1500 + vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol), + max_size_gb=cap_bytes / (1024 ** 3)) + for i in range(4): + vc._baseline = time.time() - 5 + (cache / f"f{i}.bin").write_text("x" * 800) + vc.sync() + time.sleep(0.01) + total = sum(os.path.getsize(s) for s in vc._list_shards()) + assert total <= cap_bytes + + +def test_no_retention_when_cap_is_none(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + for i in range(3): + vc._baseline = time.time() - 5 + (cache / f"f{i}.bin").write_text("data") + vc.sync() + assert len(vc._list_shards()) == 3 From 68b56d3355ca5c91f5f113f3942de8e831ba69ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 14:42:58 -0700 Subject: [PATCH 10/20] feat(serverless): add best-effort guard and warm() context manager --- runpod/serverless/utils/rp_volume_cache.py | 17 +++++++- .../test_utils/test_rp_volume_cache.py | 41 +++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index be38cb9b..81d80978 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -1,5 +1,6 @@ """Bidirectional warm-cache sync between local directories and a network volume.""" +import contextlib import os import tarfile import tempfile @@ -71,7 +72,13 @@ def _iter_delta_files(self): continue def _guard(self, fn, default): - return fn() + try: + return fn() + except Exception as exc: # best-effort: never break the worker + if not self._best_effort: + raise + log.warn(f"VolumeCache operation failed: {exc}") + return default def sync(self): if not self.available: @@ -165,3 +172,11 @@ def _is_safe_member(self, member): target == d or target.startswith(d + os.sep) for d in self._dirs ) + + @contextlib.contextmanager + def warm(self): + self.hydrate() + try: + yield self + finally: + self.sync() 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 fd107570..a6257d72 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -211,3 +211,44 @@ def test_no_retention_when_cap_is_none(tmp_path): (cache / f"f{i}.bin").write_text("data") vc.sync() assert len(vc._list_shards()) == 3 + + +def test_best_effort_swallows_and_returns_default(tmp_path, monkeypatch): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + def boom(): + raise RuntimeError("disk exploded") + monkeypatch.setattr(vc, "_do_sync", boom) + (cache / "x.bin").write_text("x") + vc._baseline = time.time() - 5 + assert vc.sync() is False # swallowed, no raise + + +def test_best_effort_false_reraises(tmp_path, monkeypatch): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + vc._best_effort = False + def boom(): + raise RuntimeError("disk exploded") + monkeypatch.setattr(vc, "_do_sync", boom) + with pytest.raises(RuntimeError): + vc.sync() + + +def test_warm_hydrates_on_enter_and_syncs_on_exit(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + calls = [] + vc.hydrate = lambda: calls.append("hydrate") + vc.sync = lambda: calls.append("sync") + with vc.warm(): + calls.append("body") + assert calls == ["hydrate", "body", "sync"] + + +def test_warm_syncs_even_on_exception(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + calls = [] + vc.hydrate = lambda: calls.append("hydrate") + vc.sync = lambda: calls.append("sync") + with pytest.raises(ValueError): + with vc.warm(): + raise ValueError("boom") + assert calls == ["hydrate", "sync"] From c3a9e988bf3574e793dca1c72b7a1f3de5766298 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 14:47:37 -0700 Subject: [PATCH 11/20] feat(serverless): export VolumeCache from runpod.serverless --- runpod/serverless/__init__.py | 4 +++- runpod/serverless/utils/__init__.py | 4 +++- tests/test_serverless/test_init.py | 17 ++++++++-------- .../test_utils/test_rp_volume_cache.py | 7 +++++++ tests/test_serverless/test_utils_init.py | 20 ++++++++++--------- 5 files changed, 33 insertions(+), 19 deletions(-) diff --git a/runpod/serverless/__init__.py b/runpod/serverless/__init__.py index 7905731f..05245207 100644 --- a/runpod/serverless/__init__.py +++ b/runpod/serverless/__init__.py @@ -17,12 +17,14 @@ from .modules.rp_logger import RunPodLogger from .modules.rp_progress import progress_update from .modules.rp_fitness import register_fitness_check +from .utils.rp_volume_cache import VolumeCache __all__ = [ "start", "progress_update", "register_fitness_check", - "runpod_version" + "runpod_version", + "VolumeCache", ] log = RunPodLogger() diff --git a/runpod/serverless/utils/__init__.py b/runpod/serverless/utils/__init__.py index 9036a4dd..f0c44cc8 100644 --- a/runpod/serverless/utils/__init__.py +++ b/runpod/serverless/utils/__init__.py @@ -2,9 +2,11 @@ from .rp_download import download_files_from_urls from .rp_upload import upload_file_to_bucket, upload_in_memory_object +from .rp_volume_cache import VolumeCache __all__ = [ "download_files_from_urls", "upload_file_to_bucket", - "upload_in_memory_object" + "upload_in_memory_object", + "VolumeCache", ] diff --git a/tests/test_serverless/test_init.py b/tests/test_serverless/test_init.py index de212041..a41c05a2 100644 --- a/tests/test_serverless/test_init.py +++ b/tests/test_serverless/test_init.py @@ -24,7 +24,8 @@ def test_expected_public_symbols(self): 'start', 'progress_update', 'register_fitness_check', - 'runpod_version' + 'runpod_version', + 'VolumeCache' } actual_symbols = set(runpod.serverless.__all__) assert expected_symbols == actual_symbols, f"Expected {expected_symbols}, got {actual_symbols}" @@ -79,24 +80,24 @@ def test_private_symbols_not_exported(self): def test_all_covers_public_api_only(self): """Test that __all__ contains only the intended public API.""" # Get all non-private attributes from the module - module_attrs = {name for name in dir(runpod.serverless) + module_attrs = {name for name in dir(runpod.serverless) if not name.startswith('_')} - + # Filter out imported modules and types that shouldn't be public expected_private_attrs = { - 'argparse', 'json', 'os', 'signal', 'sys', 'time', + 'argparse', 'json', 'os', 'signal', 'sys', 'time', 'worker', 'rp_fastapi', 'log', 'parser', 'Any', 'Dict', # Type hints 'modules', 'utils', # Sub-modules 'RunPodLogger' # Internal logger class } - + public_attrs = module_attrs - expected_private_attrs all_symbols = set(runpod.serverless.__all__) - + # All symbols in __all__ should be actual public API assert all_symbols.issubset(public_attrs), f"__all__ contains non-public symbols: {all_symbols - public_attrs}" - + # Expected public API should be exactly what's in __all__ - expected_public_api = {'start', 'progress_update', 'register_fitness_check', 'runpod_version'} + expected_public_api = {'start', 'progress_update', 'register_fitness_check', 'runpod_version', 'VolumeCache'} assert all_symbols == expected_public_api, f"Expected {expected_public_api}, got {all_symbols}" 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 a6257d72..f12036da 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -252,3 +252,10 @@ def test_warm_syncs_even_on_exception(tmp_path): with vc.warm(): raise ValueError("boom") assert calls == ["hydrate", "sync"] + + +def test_volumecache_exported_from_serverless(): + import runpod.serverless as sls + import runpod.serverless.utils as utils + assert sls.VolumeCache is utils.VolumeCache + assert "VolumeCache" in sls.__all__ diff --git a/tests/test_serverless/test_utils_init.py b/tests/test_serverless/test_utils_init.py index 296f4b9a..bff2bb33 100644 --- a/tests/test_serverless/test_utils_init.py +++ b/tests/test_serverless/test_utils_init.py @@ -23,7 +23,8 @@ def test_expected_public_symbols(self): expected_symbols = { 'download_files_from_urls', 'upload_file_to_bucket', - 'upload_in_memory_object' + 'upload_in_memory_object', + 'VolumeCache' } actual_symbols = set(runpod.serverless.utils.__all__) assert expected_symbols == actual_symbols, f"Expected {expected_symbols}, got {actual_symbols}" @@ -69,23 +70,24 @@ def test_no_duplicate_symbols_in_all(self): def test_all_covers_public_api_only(self): """Test that __all__ contains only the intended public API.""" # Get all non-private attributes from the module - module_attrs = {name for name in dir(runpod.serverless.utils) + module_attrs = {name for name in dir(runpod.serverless.utils) if not name.startswith('_')} - + # Filter out any imported modules that shouldn't be public expected_private_attrs = set() # No private imports in this module - + public_attrs = module_attrs - expected_private_attrs all_symbols = set(runpod.serverless.utils.__all__) - + # All symbols in __all__ should be actual public API assert all_symbols.issubset(public_attrs), f"__all__ contains non-public symbols: {all_symbols - public_attrs}" - + # Expected public API should be exactly what's in __all__ expected_public_api = { - 'download_files_from_urls', - 'upload_file_to_bucket', - 'upload_in_memory_object' + 'download_files_from_urls', + 'upload_file_to_bucket', + 'upload_in_memory_object', + 'VolumeCache' } assert all_symbols == expected_public_api, f"Expected {expected_public_api}, got {all_symbols}" From a8f466a671825168a460452e9ce09554bd0e7c07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 14:53:40 -0700 Subject: [PATCH 12/20] feat(serverless): auto-hydrate model cache from network volume in worker loop --- runpod/serverless/modules/rp_job.py | 2 + runpod/serverless/utils/rp_volume_cache.py | 48 ++++++++++++++++ runpod/serverless/worker.py | 7 +++ .../test_utils/test_rp_volume_cache.py | 55 +++++++++++++++++++ 4 files changed, 112 insertions(+) diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py index a45cebc6..7f824892 100644 --- a/runpod/serverless/modules/rp_job.py +++ b/runpod/serverless/modules/rp_job.py @@ -15,6 +15,7 @@ from ...version import __version__ as runpod_version from ..utils import rp_debugger +from ..utils.rp_volume_cache import sync_after_job from .rp_handler import is_generator from .rp_http import send_result, stream_result from .rp_tips import check_return_size @@ -267,6 +268,7 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: error_msg = job_output.pop("error", None) refresh_worker = job_output.pop("refresh_worker", None) run_result["output"] = job_output + sync_after_job() if error_msg: run_result["error"] = error_msg diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 81d80978..010c73eb 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -180,3 +180,51 @@ def warm(self): yield self finally: self.sync() + + +_ACTIVE_CACHE = None +_SYNCED = False +_sync_lock = threading.Lock() + +_DISABLED_VALUES = ("0", "false", "no") + + +def _discover_model_dirs(): + dirs = [os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface")] + for var in ("HF_HUB_CACHE", "TORCH_HOME"): + if os.environ.get(var): + dirs.append(os.environ[var]) + extra = os.environ.get("RUNPOD_CACHE_DIRS") + if extra: + dirs.extend(p for p in extra.split(os.pathsep) if p) + return list(dict.fromkeys(dirs)) # de-dupe, preserve order + + +def build_default_cache(): + if os.environ.get("RUNPOD_VOLUME_CACHE", "1").lower() in _DISABLED_VALUES: + return None + max_gb = float(os.environ.get("RUNPOD_VOLUME_CACHE_MAX_GB", "50")) + vc = VolumeCache(_discover_model_dirs(), max_size_gb=max_gb) + return vc if vc.available else None + + +def set_active_cache(vc): + global _ACTIVE_CACHE + _ACTIVE_CACHE = vc + + +def sync_after_job(): + global _SYNCED + if _ACTIVE_CACHE is None: + return + with _sync_lock: + if _SYNCED: + return + _SYNCED = True + threading.Thread(target=_ACTIVE_CACHE.sync, daemon=True).start() + + +def reset_builtin_state_for_test(): + global _ACTIVE_CACHE, _SYNCED + _ACTIVE_CACHE = None + _SYNCED = False diff --git a/runpod/serverless/worker.py b/runpod/serverless/worker.py index 90053ec7..d50df6ec 100644 --- a/runpod/serverless/worker.py +++ b/runpod/serverless/worker.py @@ -9,6 +9,7 @@ from runpod.serverless.modules import rp_logger, rp_local, rp_ping, rp_scale from runpod.serverless.modules.rp_fitness import run_fitness_checks +from runpod.serverless.utils.rp_volume_cache import build_default_cache, set_active_cache log = rp_logger.RunPodLogger() heartbeat = rp_ping.Heartbeat() @@ -48,6 +49,12 @@ def run_worker(config: Dict[str, Any]) -> None: # Start pinging Runpod to show that the worker is alive. heartbeat.start_ping(mirror) + # Warm the local cache from a mounted network volume before any job runs. + volume_cache = build_default_cache() + if volume_cache is not None: + volume_cache.hydrate() + set_active_cache(volume_cache) + # Create a JobScaler responsible for adjusting the concurrency job_scaler = rp_scale.JobScaler(config) job_scaler.start() 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 f12036da..79a8987f 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -259,3 +259,58 @@ def test_volumecache_exported_from_serverless(): import runpod.serverless.utils as utils assert sls.VolumeCache is utils.VolumeCache assert "VolumeCache" in sls.__all__ + + +import runpod.serverless.utils.rp_volume_cache as vcmod + + +def test_build_default_cache_disabled_by_env(tmp_path, monkeypatch): + monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "0") + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") + assert vcmod.build_default_cache() is None + + +def test_build_default_cache_none_when_no_volume(tmp_path, monkeypatch): + monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") + monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: False)) + assert vcmod.build_default_cache() is None + + +def test_discover_model_dirs_includes_hf_home_and_extras(monkeypatch): + monkeypatch.setenv("HF_HOME", "/models/hf") + monkeypatch.setenv("RUNPOD_CACHE_DIRS", "/a" + os.pathsep + "/b") + dirs = vcmod._discover_model_dirs() + assert "/models/hf" in dirs and "/a" in dirs and "/b" in dirs + + +def test_sync_after_job_runs_once(monkeypatch): + vcmod.reset_builtin_state_for_test() + counter = {"n": 0} + fake = type("F", (), {"sync": lambda self: counter.__setitem__("n", counter["n"] + 1)})() + joined = [] + class FakeThread: + def __init__(self, target, daemon=None): self.target = target + def start(self): self.target(); joined.append(True) + monkeypatch.setattr(vcmod.threading, "Thread", FakeThread) + vcmod.set_active_cache(fake) + vcmod.sync_after_job() + vcmod.sync_after_job() + assert counter["n"] == 1 + + +def test_run_worker_hydrates_registered_cache(monkeypatch): + from runpod.serverless import worker + + async def _noop_fitness_checks(): + return None + + fake = type("F", (), {"hydrate": lambda self: fake_calls.append("h")})() + fake_calls = [] + monkeypatch.setattr(worker, "build_default_cache", lambda: fake, raising=False) + monkeypatch.setattr(worker.rp_scale, "JobScaler", + lambda config: type("J", (), {"start": lambda self: None})()) + monkeypatch.setattr(worker.heartbeat, "start_ping", lambda mirror: None) + monkeypatch.setattr(worker, "run_fitness_checks", _noop_fitness_checks) + worker.run_worker({"handler": lambda job: job}) + assert "h" in fake_calls From 1e8fdf10f93f3d0517ef363dd836a7c9a255923b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 15:00:18 -0700 Subject: [PATCH 13/20] fix(serverless): guard build_default_cache and isolate worker-wiring test state Wrap build_default_cache() parse+construct in try/except to prevent malformed RUNPOD_VOLUME_CACHE_MAX_GB from crashing worker startup; degrades to cache-disabled instead. Add sync() no-op to fake cache in test_run_worker_hydrates_registered_cache and wrap test body with reset_builtin_state_for_test() to prevent module-state leakage to subsequent tests (sync_after_job could fire and hit AttributeError). --- runpod/serverless/utils/rp_volume_cache.py | 8 +++- .../test_utils/test_rp_volume_cache.py | 39 +++++++++++++------ 2 files changed, 33 insertions(+), 14 deletions(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 010c73eb..ca486e79 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -203,8 +203,12 @@ def _discover_model_dirs(): def build_default_cache(): if os.environ.get("RUNPOD_VOLUME_CACHE", "1").lower() in _DISABLED_VALUES: return None - max_gb = float(os.environ.get("RUNPOD_VOLUME_CACHE_MAX_GB", "50")) - vc = VolumeCache(_discover_model_dirs(), max_size_gb=max_gb) + try: + max_gb = float(os.environ.get("RUNPOD_VOLUME_CACHE_MAX_GB", "50")) + vc = VolumeCache(_discover_model_dirs(), max_size_gb=max_gb) + except Exception as exc: # never let cache setup crash worker startup + log.warn(f"VolumeCache: failed to build default cache, disabling: {exc}") + return None return vc if vc.available else None 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 79a8987f..fa125f76 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -302,15 +302,30 @@ def start(self): self.target(); joined.append(True) def test_run_worker_hydrates_registered_cache(monkeypatch): from runpod.serverless import worker - async def _noop_fitness_checks(): - return None - - fake = type("F", (), {"hydrate": lambda self: fake_calls.append("h")})() - fake_calls = [] - monkeypatch.setattr(worker, "build_default_cache", lambda: fake, raising=False) - monkeypatch.setattr(worker.rp_scale, "JobScaler", - lambda config: type("J", (), {"start": lambda self: None})()) - monkeypatch.setattr(worker.heartbeat, "start_ping", lambda mirror: None) - monkeypatch.setattr(worker, "run_fitness_checks", _noop_fitness_checks) - worker.run_worker({"handler": lambda job: job}) - assert "h" in fake_calls + vcmod.reset_builtin_state_for_test() + try: + async def _noop_fitness_checks(): + return None + + fake = type("F", (), { + "hydrate": lambda self: fake_calls.append("h"), + "sync": lambda self: None, + })() + fake_calls = [] + monkeypatch.setattr(worker, "build_default_cache", lambda: fake, raising=False) + monkeypatch.setattr(worker.rp_scale, "JobScaler", + lambda config: type("J", (), {"start": lambda self: None})()) + monkeypatch.setattr(worker.heartbeat, "start_ping", lambda mirror: None) + monkeypatch.setattr(worker, "run_fitness_checks", _noop_fitness_checks) + worker.run_worker({"handler": lambda job: job}) + assert "h" in fake_calls + finally: + vcmod.reset_builtin_state_for_test() + + +def test_build_default_cache_survives_bad_max_gb(monkeypatch): + vcmod.reset_builtin_state_for_test() + monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") + monkeypatch.setenv("RUNPOD_VOLUME_CACHE_MAX_GB", "not-a-number") + assert vcmod.build_default_cache() is None # degrades, does not raise From f8fe045b1925c862fa1478bae0f82365ebacc15e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 15:09:54 -0700 Subject: [PATCH 14/20] fix(serverless): guard cache sync dispatch, sweep temp shards, cover all outputs --- runpod/serverless/modules/rp_job.py | 3 +- runpod/serverless/utils/rp_volume_cache.py | 34 +++++++++++++++---- .../test_utils/test_rp_volume_cache.py | 31 +++++++++++++++++ 3 files changed, 60 insertions(+), 8 deletions(-) diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py index 7f824892..9525a5e1 100644 --- a/runpod/serverless/modules/rp_job.py +++ b/runpod/serverless/modules/rp_job.py @@ -268,7 +268,6 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: error_msg = job_output.pop("error", None) refresh_worker = job_output.pop("refresh_worker", None) run_result["output"] = job_output - sync_after_job() if error_msg: run_result["error"] = error_msg @@ -284,6 +283,8 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: if run_result.get("output") == {}: run_result.pop("output") + sync_after_job() # fire-and-forget cache warm on any successful output + check_return_size(run_result) # Checks the size of the return body. except Exception as err: diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index ca486e79..0b7b55bf 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -1,6 +1,7 @@ """Bidirectional warm-cache sync between local directories and a network volume.""" import contextlib +import glob import os import tarfile import tempfile @@ -37,7 +38,6 @@ def __init__( self._best_effort = best_effort self._worker_id = os.environ.get("RUNPOD_POD_ID") or uuid.uuid4().hex[:12] self._baseline = time.time() - _BASELINE_EPSILON_SECONDS - self._lock = threading.Lock() @property def _shard_dir(self): @@ -91,6 +91,11 @@ def _do_sync(self): log.debug("VolumeCache: no delta files to sync") return False os.makedirs(self._shard_dir, exist_ok=True) + for stale in glob.glob(os.path.join(self._shard_dir, f"{self._worker_id}-*.tar.tmp")): + try: + os.remove(stale) + except OSError: + pass final = os.path.join(self._shard_dir, f"{self._worker_id}-{time.time_ns():020d}.tar") tmp = final + ".tmp" with tarfile.open(tmp, "w") as tar: @@ -201,6 +206,18 @@ def _discover_model_dirs(): def build_default_cache(): + """Build the built-in VolumeCache from RUNPOD_* env vars, or None if disabled/unavailable. + + Known operational limitations (accepted, not bugs): + 1. Cold-scale write amplification: when N workers cold-start simultaneously, + each misses the still-empty cache and each writes its own full-size shard + on first sync. Bounded by retention (RUNPOD_VOLUME_CACHE_MAX_GB). + 2. Lost first warm on aggressive recycle: sync() runs in a daemon thread + dispatched after the job response; a large-model sync may not finish + before the worker is recycled, killing the thread mid-write. The partial + shard is cleaned up on the next sync from this worker, and the warm is + simply retried then. + """ if os.environ.get("RUNPOD_VOLUME_CACHE", "1").lower() in _DISABLED_VALUES: return None try: @@ -219,13 +236,16 @@ def set_active_cache(vc): def sync_after_job(): global _SYNCED - if _ACTIVE_CACHE is None: - return - with _sync_lock: - if _SYNCED: + try: + if _ACTIVE_CACHE is None: return - _SYNCED = True - threading.Thread(target=_ACTIVE_CACHE.sync, daemon=True).start() + with _sync_lock: + if _SYNCED: + return + _SYNCED = True + threading.Thread(target=_ACTIVE_CACHE.sync, daemon=True).start() + except Exception as exc: # cache glue must never affect job outcome + log.warn(f"VolumeCache: sync_after_job failed to dispatch: {exc}") def reset_builtin_state_for_test(): 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 fa125f76..e328bebd 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -329,3 +329,34 @@ def test_build_default_cache_survives_bad_max_gb(monkeypatch): monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") monkeypatch.setenv("RUNPOD_VOLUME_CACHE_MAX_GB", "not-a-number") assert vcmod.build_default_cache() is None # degrades, does not raise + + +def test_two_workers_produce_independently_hydratable_shards(tmp_path): + # Distinct worker_ids must write non-colliding shards that both hydrate (spec acceptance). + cache = tmp_path / "cache"; cache.mkdir() + vol = tmp_path / "volume"; vol.mkdir() + a = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + a._worker_id = "workerA" + a._baseline = time.time() - 5 + (cache / "a.bin").write_text("aaa") + assert a.sync() is True + b = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + b._worker_id = "workerB" + b._baseline = time.time() - 5 + (cache / "b.bin").write_text("bbb") + assert b.sync() is True + (cache / "a.bin").unlink(); (cache / "b.bin").unlink() + reader = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + reader._clear_marker_for_test() + assert reader.hydrate() is True + assert (cache / "a.bin").read_text() == "aaa" + assert (cache / "b.bin").read_text() == "bbb" + + +def test_build_default_cache_returns_instance_when_available(monkeypatch): + vcmod.reset_builtin_state_for_test() + monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") + monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: True)) + vc = vcmod.build_default_cache() + assert isinstance(vc, vcmod.VolumeCache) From ede2d5cd3427d17869357de9430bc276d6fbe4e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 17:38:27 -0700 Subject: [PATCH 15/20] fix(serverless): address review - streaming sync, retention resilience, namespace validation, lint --- runpod/serverless/modules/rp_job.py | 2 + runpod/serverless/utils/rp_volume_cache.py | 21 +++++- .../test_serverless/test_modules/test_job.py | 66 +++++++++++++++++++ .../test_utils/test_rp_volume_cache.py | 60 ++++++++++++++--- 4 files changed, 139 insertions(+), 10 deletions(-) diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py index 9525a5e1..94de75b0 100644 --- a/runpod/serverless/modules/rp_job.py +++ b/runpod/serverless/modules/rp_job.py @@ -332,6 +332,8 @@ async def run_job_generator( log.debug(f"Generator output: {output_partial}", job["id"]) yield {"output": output_partial} + sync_after_job() # fire-and-forget cache warm on successful generator completion + except Exception as err: log.error(err, job["id"]) yield {"error": f"handler: {str(err)} \ntraceback: {traceback.format_exc()}"} diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 0b7b55bf..408d41c3 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -33,6 +33,16 @@ def __init__( ): self._dirs = [os.path.realpath(os.fspath(d)) for d in dirs] self._namespace = namespace or os.environ.get("RUNPOD_ENDPOINT_ID") or "" + if self._namespace and ( + os.path.isabs(self._namespace) + or os.sep in self._namespace + or "/" in self._namespace + or "\\" in self._namespace + or self._namespace in (".", "..") + ): + raise ValueError( + f"namespace must be a single safe path component, got {self._namespace!r}" + ) self._volume_path = os.fspath(volume_path) self._max_size_gb = max_size_gb self._best_effort = best_effort @@ -95,6 +105,7 @@ def _do_sync(self): try: os.remove(stale) except OSError: + # best-effort cleanup: ignore temp files that vanish or can't be removed pass final = os.path.join(self._shard_dir, f"{self._worker_id}-{time.time_ns():020d}.tar") tmp = final + ".tmp" @@ -110,13 +121,19 @@ def _do_sync(self): def _enforce_retention(self): if not self._max_size_gb: return + def _size(p): + try: + return os.path.getsize(p) + except OSError: + return 0 + cap = self._max_size_gb * (1024 ** 3) shards = self._list_shards() # oldest first - total = sum(os.path.getsize(s) for s in shards) + total = sum(_size(s) for s in shards) for shard in shards: if total <= cap: break - size = os.path.getsize(shard) + size = _size(shard) try: os.remove(shard) total -= size diff --git a/tests/test_serverless/test_modules/test_job.py b/tests/test_serverless/test_modules/test_job.py index 96590c1e..410d01fa 100644 --- a/tests/test_serverless/test_modules/test_job.py +++ b/tests/test_serverless/test_modules/test_job.py @@ -439,3 +439,69 @@ async def test_run_job_generator_exception(self): assert mock_log.error.call_count == 1 assert mock_log.info.call_count == 1 mock_log.info.assert_called_with("Finished running generator.", "123") + + async def test_run_job_generator_success_syncs_cache(self): + """ + Tests that run_job_generator triggers sync_after_job on the success path. + """ + handler = self.handler_gen_success + job = {"id": "123"} + counter = {"n": 0} + + def fake_sync_after_job(): + counter["n"] += 1 + + with patch( + "runpod.serverless.modules.rp_job.log", new_callable=Mock + ), patch( + "runpod.serverless.modules.rp_job.sync_after_job", + side_effect=fake_sync_after_job, + ) as mock_sync: + result = [i async for i in rp_job.run_job_generator(handler, job)] + + assert result == [ + {"output": "partial_output_1"}, + {"output": "partial_output_2"}, + ] + mock_sync.assert_called_once() + assert counter["n"] == 1 + + async def test_run_job_generator_success_syncs_cache_async(self): + """ + Tests that run_job_generator triggers sync_after_job on the success path + for an async generator handler. + """ + handler = self.handler_async_gen_success + job = {"id": "123"} + + with patch( + "runpod.serverless.modules.rp_job.log", new_callable=Mock + ), patch( + "runpod.serverless.modules.rp_job.sync_after_job" + ) as mock_sync: + result = [i async for i in rp_job.run_job_generator(handler, job)] + + assert result == [ + {"output": "partial_output_1"}, + {"output": "partial_output_2"}, + ] + mock_sync.assert_called_once() + + async def test_run_job_generator_exception_does_not_sync_cache(self): + """ + Tests that run_job_generator does NOT trigger sync_after_job on the + exception path. + """ + handler = self.handler_fail + job = {"id": "123"} + + with patch( + "runpod.serverless.modules.rp_job.log", new_callable=Mock + ), patch( + "runpod.serverless.modules.rp_job.sync_after_job" + ) as mock_sync: + result = [i async for i in rp_job.run_job_generator(handler, job)] + + assert len(result) == 1 + assert "error" in result[0] + mock_sync.assert_not_called() 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 e328bebd..eab9c877 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -2,7 +2,9 @@ import tarfile import time import pytest -from runpod.serverless.utils.rp_volume_cache import VolumeCache +import runpod.serverless.utils.rp_volume_cache as vcmod + +VolumeCache = vcmod.VolumeCache def test_unavailable_when_volume_dir_missing(tmp_path): @@ -26,6 +28,21 @@ def test_unavailable_without_namespace(tmp_path, monkeypatch): assert vc.available is False +@pytest.mark.parametrize("bad_namespace", ["../evil", "a/b", "/etc", ".."]) +def test_namespace_rejects_unsafe_values(tmp_path, bad_namespace): + vol = tmp_path / "volume" + vol.mkdir() + with pytest.raises(ValueError): + VolumeCache([str(tmp_path / "cache")], namespace=bad_namespace, volume_path=str(vol)) + + +def test_namespace_accepts_normal_value(tmp_path): + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol)) + assert vc._namespace == "ep1" + + def test_namespace_defaults_to_endpoint_id(tmp_path, monkeypatch): monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "endpoint-xyz") vol = tmp_path / "volume" @@ -204,6 +221,32 @@ def test_retention_prunes_oldest_shards_past_cap(tmp_path): assert total <= cap_bytes +def test_retention_tolerates_shard_removed_concurrently(tmp_path, monkeypatch): + # A concurrent worker prunes/removes a shard between _list_shards() and the + # getsize() lookups; the size lookup must degrade to 0 instead of raising. + cache = tmp_path / "cache"; cache.mkdir() + vol = tmp_path / "volume"; vol.mkdir() + cap_bytes = 1000 + vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol), + max_size_gb=cap_bytes / (1024 ** 3)) + for i in range(3): + vc._baseline = time.time() - 5 + (cache / f"f{i}.bin").write_text("x" * 800) + vc.sync() + time.sleep(0.01) + + real_getsize = os.path.getsize + + def flaky_getsize(path): + if str(path).endswith(".tar") and "f0" not in str(path): + raise FileNotFoundError(path) + return real_getsize(path) + + monkeypatch.setattr(os.path, "getsize", flaky_getsize) + # Should not raise despite getsize() failures on some shards. + vc._enforce_retention() + + def test_no_retention_when_cap_is_none(tmp_path): vc, cache, vol = _mk_cache_with_volume(tmp_path) for i in range(3): @@ -248,22 +291,23 @@ def test_warm_syncs_even_on_exception(tmp_path): calls = [] vc.hydrate = lambda: calls.append("hydrate") vc.sync = lambda: calls.append("sync") - with pytest.raises(ValueError): + raised = False + try: with vc.warm(): raise ValueError("boom") + except ValueError: + raised = True + assert raised assert calls == ["hydrate", "sync"] def test_volumecache_exported_from_serverless(): - import runpod.serverless as sls - import runpod.serverless.utils as utils - assert sls.VolumeCache is utils.VolumeCache + from runpod import serverless as sls + from runpod.serverless import utils as sls_utils + assert sls.VolumeCache is sls_utils.VolumeCache assert "VolumeCache" in sls.__all__ -import runpod.serverless.utils.rp_volume_cache as vcmod - - def test_build_default_cache_disabled_by_env(tmp_path, monkeypatch): monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "0") monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") From eeb26e6160e2ffdf179591904a07c2299460ed7d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Mon, 6 Jul 2026 18:44:06 -0700 Subject: [PATCH 16/20] feat(serverless): make network-volume warm cache opt-in; add usage docs --- README.md | 20 +++ docs/serverless/volume_cache.md | 139 ++++++++++++++++++ runpod/serverless/utils/rp_volume_cache.py | 38 ++++- .../test_utils/test_rp_volume_cache.py | 15 +- 4 files changed, 204 insertions(+), 8 deletions(-) create mode 100644 docs/serverless/volume_cache.md diff --git a/README.md b/README.md index b34014b1..b84c089f 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,26 @@ runpod.serverless.start({"handler": handler}) See [Worker Fitness Checks](https://github.com/runpod/runpod-python/blob/main/docs/serverless/worker_fitness_checks.md) documentation for more examples and best practices. +### Network-Volume Warm Cache + +When a network volume is attached, `VolumeCache` warms local directories (such as a model cache) across cold starts — restoring them on startup and syncing new files back — so a repeated multi-GB model download becomes a one-time cost per endpoint. It is opt-in and best-effort. + +```bash +# Enable the built-in warm cache (requires a mounted network volume) +RUNPOD_VOLUME_CACHE=1 +``` + +Or use it explicitly around a model load: + +```python +from runpod.serverless import VolumeCache + +with VolumeCache(dirs=["/root/.cache/huggingface"]).warm(): + model = load_model() +``` + +See [Network-Volume Warm Cache](https://github.com/runpod/runpod-python/blob/main/docs/serverless/volume_cache.md) documentation for configuration and details. + ## 📚 | API Language Library (GraphQL Wrapper) When interacting with the Runpod API you can use this library to make requests to the API. diff --git a/docs/serverless/volume_cache.md b/docs/serverless/volume_cache.md new file mode 100644 index 00000000..20fb4598 --- /dev/null +++ b/docs/serverless/volume_cache.md @@ -0,0 +1,139 @@ +# Network-Volume Warm Cache (VolumeCache) + +`VolumeCache` warms local directories across serverless workers using a mounted +network volume. On cold start it restores previously-synced directories (for +example a model cache) from the volume; after use it syncs newly written files +back so the next cold worker starts warm. This turns a repeated multi-GB model +download on every cold start into a one-time cost per endpoint. + +It is stdlib-only and best-effort: any failure degrades to a cold worker and +never raises into your handler or the worker loop. + +## Requirements + +- A **network volume** attached to the endpoint (mounted at `/runpod-volume`). +- Set on serverless automatically: `RUNPOD_ENDPOINT_ID` (used to scope the cache + per endpoint). + +If no volume is mounted, every operation is a safe no-op. + +## Option 1: Built-in (opt-in, zero code) + +The worker can hydrate a model cache at startup and sync it after the first job, +with no changes to your handler. It is **off by default** — enable it with an +environment variable: + +```bash +# Enable the built-in warm cache (opt-in) +RUNPOD_VOLUME_CACHE=1 +``` + +When enabled and a volume is mounted, the worker: + +1. Hydrates the model cache from the volume before the first job runs. +2. Syncs new files to the volume once, after the first successful job. + +By default it caches the directories pointed to by `HF_HOME`, `HF_HUB_CACHE`, +and `TORCH_HOME` (whichever are set; `HF_HOME` defaults to +`~/.cache/huggingface`). Add more directories with `RUNPOD_CACHE_DIRS`. + +### Configuration + +| Variable | Default | Purpose | +| --- | --- | --- | +| `RUNPOD_VOLUME_CACHE` | unset (off) | Set to `1`/`true`/`yes`/`on` to enable the built-in. | +| `RUNPOD_CACHE_DIRS` | — | Extra directories to cache, `os.pathsep`-separated (`:` on Linux). | +| `RUNPOD_VOLUME_CACHE_MAX_GB` | `50` | Prune oldest shards once the endpoint's cache exceeds this size. | +| `HF_HOME` / `HF_HUB_CACHE` / `TORCH_HOME` | — | Auto-discovered model-cache locations. | + +### Example + +For most model-serving workers, enabling the built-in and letting the model +download into `HF_HOME` on the first request is all that is needed: + +```python +# my_worker.py +import runpod + +def handler(job): + # First cold worker downloads the model into HF_HOME; the built-in syncs it + # to the volume. Subsequent cold workers hydrate it and skip the download. + ... + +runpod.serverless.start({"handler": handler}) +``` + +```bash +# Template / container env +RUNPOD_VOLUME_CACHE=1 +``` + +> Note: hydration only helps model loads that happen **after** the worker starts +> (lazy or in-handler loading). If your model loads at module import time, use +> Option 2 to place `warm()` around the load. + +## Option 2: Explicit API + +Import `VolumeCache` and control caching yourself. This works for any +directories, and the `warm()` context manager guarantees hydration happens +before your model load regardless of when it runs. + +```python +from runpod.serverless import VolumeCache + +vc = VolumeCache(dirs=["/root/.cache/huggingface"]) + +with vc.warm(): # hydrate on enter, sync the delta on exit + model = load_model() # downloads land in the cached directory +``` + +You can also call the phases directly when they happen at different points in +your worker's lifecycle: + +```python +vc = VolumeCache(dirs=["/data/models"], namespace="my-model-cache", max_size_gb=100) + +vc.hydrate() # restore cached files (e.g. at startup) +model = load_model() # populate the cache +vc.sync() # persist new files back to the volume +``` + +### Constructor + +| Argument | Default | Purpose | +| --- | --- | --- | +| `dirs` | required | Local directories to cache. | +| `namespace` | `RUNPOD_ENDPOINT_ID` | Isolation key for the on-volume shards. Must be a single safe path component. | +| `volume_path` | `/runpod-volume` | Network-volume mount point. | +| `max_size_gb` | `None` | Prune oldest shards past this cap; `None` = no cap. | +| `best_effort` | `True` | Swallow and log errors instead of raising. Set `False` while debugging. | + +## How it works + +- **Per-endpoint, sharded storage.** Each worker writes its delta as its own tar + shard under `//.cache//`, published atomically. Per-worker + shards are collision-free under concurrent workers — no shared file is + rewritten. Hydration extracts all shards for the namespace (newest wins on + overlap). +- **Delta only.** `sync()` packs just the files written since the last hydrate, + not the whole cache. +- **Retention.** With `max_size_gb` set, the oldest shards are pruned once the + namespace exceeds the cap. +- **Safety.** Extraction rejects members that would escape the configured + directories (traversal, symlinks, absolute paths). + +## Limitations + +- **Cold-scale write amplification.** If N workers cold-start at the same time, + each misses the still-empty cache, downloads the model, and writes a full-size + shard. Total volume writes on the first scale-up are ~N×; bounded thereafter by + retention. +- **Lost first warm on aggressive recycle.** `sync()` runs in a background thread + after the job response. A large-model sync may not finish before the worker is + recycled; the partial shard is cleaned up and the warm is retried on the next + worker. + +## Status + +The built-in is opt-in for now (`RUNPOD_VOLUME_CACHE=1`). Whether it should +default on when a volume is present is still under review. diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 408d41c3..1523526f 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -17,8 +17,33 @@ class VolumeCache: - """Hydrate configured dirs from a network volume on cold start; sync their - delta back as a worker-owned tar shard. Best-effort and stdlib-only.""" + """Warm-cache local directories across serverless workers via a network volume. + + On cold start, ``hydrate()`` extracts previously-synced directories from a + mounted network volume into place; after use, ``sync()`` packs the newly + written files into a per-worker tar shard on the volume so the next cold + worker starts warm. Stdlib-only and best-effort: any failure degrades to a + cold worker and never raises into the caller. + + Requires a network volume mounted at ``volume_path`` (default + ``/runpod-volume``) and a non-empty ``namespace`` (default + ``RUNPOD_ENDPOINT_ID``); otherwise ``available`` is False and all operations + are no-ops. + + Args: + dirs: Local directories to cache (e.g. a model cache like ``HF_HOME``). + namespace: Isolation key for the on-volume shards. Must be a single safe + path component. Defaults to ``RUNPOD_ENDPOINT_ID``. + volume_path: Network-volume mount point. Defaults to ``/runpod-volume``. + max_size_gb: Prune oldest shards past this cap. ``None`` = no cap. + best_effort: When True (default), swallow and log errors instead of + raising. + + Example: + >>> vc = VolumeCache(dirs=["/root/.cache/huggingface"]) + >>> with vc.warm(): # hydrate on enter, sync delta on exit + ... model = load_model() # downloads land in the cached dir + """ _EXCLUDE_SUBSTRINGS = (os.sep + "refs" + os.sep, os.sep + ".no_exist" + os.sep) @@ -208,7 +233,7 @@ def warm(self): _SYNCED = False _sync_lock = threading.Lock() -_DISABLED_VALUES = ("0", "false", "no") +_ENABLED_VALUES = ("1", "true", "yes", "on") def _discover_model_dirs(): @@ -223,7 +248,10 @@ def _discover_model_dirs(): def build_default_cache(): - """Build the built-in VolumeCache from RUNPOD_* env vars, or None if disabled/unavailable. + """Build the built-in VolumeCache when opt-in is enabled, else None. + + The built-in is OFF by default. It activates only when RUNPOD_VOLUME_CACHE is + set to a truthy value ("1"/"true"/"yes"/"on") AND a network volume is mounted. Known operational limitations (accepted, not bugs): 1. Cold-scale write amplification: when N workers cold-start simultaneously, @@ -235,7 +263,7 @@ def build_default_cache(): shard is cleaned up on the next sync from this worker, and the warm is simply retried then. """ - if os.environ.get("RUNPOD_VOLUME_CACHE", "1").lower() in _DISABLED_VALUES: + if os.environ.get("RUNPOD_VOLUME_CACHE", "").lower() not in _ENABLED_VALUES: return None try: max_gb = float(os.environ.get("RUNPOD_VOLUME_CACHE_MAX_GB", "50")) 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 eab9c877..0a216e60 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -308,6 +308,15 @@ def test_volumecache_exported_from_serverless(): assert "VolumeCache" in sls.__all__ +def test_build_default_cache_off_by_default(tmp_path, monkeypatch): + # Opt-in: with RUNPOD_VOLUME_CACHE unset, the built-in is disabled even + # when a volume would be available. + monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") + monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: True)) + assert vcmod.build_default_cache() is None + + def test_build_default_cache_disabled_by_env(tmp_path, monkeypatch): monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "0") monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") @@ -315,7 +324,7 @@ def test_build_default_cache_disabled_by_env(tmp_path, monkeypatch): def test_build_default_cache_none_when_no_volume(tmp_path, monkeypatch): - monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: False)) assert vcmod.build_default_cache() is None @@ -369,7 +378,7 @@ async def _noop_fitness_checks(): def test_build_default_cache_survives_bad_max_gb(monkeypatch): vcmod.reset_builtin_state_for_test() - monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") monkeypatch.setenv("RUNPOD_VOLUME_CACHE_MAX_GB", "not-a-number") assert vcmod.build_default_cache() is None # degrades, does not raise @@ -399,7 +408,7 @@ def test_two_workers_produce_independently_hydratable_shards(tmp_path): def test_build_default_cache_returns_instance_when_available(monkeypatch): vcmod.reset_builtin_state_for_test() - monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) + monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: True)) vc = vcmod.build_default_cache() From b184d21e059b4af897b26f67fe4ae6a5d6d3d542 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Tue, 7 Jul 2026 15:35:02 -0700 Subject: [PATCH 17/20] refactor(serverless): VolumeCache mirror+reconcile closure; drop env-var built-in --- README.md | 11 +- docs/serverless/volume_cache.md | 148 ++--- runpod/serverless/modules/rp_job.py | 5 - runpod/serverless/utils/rp_volume_cache.py | 388 ++++++------ runpod/serverless/worker.py | 7 - .../test_serverless/test_modules/test_job.py | 65 -- .../test_utils/test_rp_volume_cache.py | 562 ++++++++---------- 7 files changed, 468 insertions(+), 718 deletions(-) diff --git a/README.md b/README.md index b84c089f..6ad9c666 100644 --- a/README.md +++ b/README.md @@ -151,19 +151,12 @@ See [Worker Fitness Checks](https://github.com/runpod/runpod-python/blob/main/do ### Network-Volume Warm Cache -When a network volume is attached, `VolumeCache` warms local directories (such as a model cache) across cold starts — restoring them on startup and syncing new files back — so a repeated multi-GB model download becomes a one-time cost per endpoint. It is opt-in and best-effort. - -```bash -# Enable the built-in warm cache (requires a mounted network volume) -RUNPOD_VOLUME_CACHE=1 -``` - -Or use it explicitly around a model load: +When a network volume is attached, `VolumeCache` warms local directories (such as a model cache) across cold starts — hydrating them on startup and syncing new files back on exit — so a repeated multi-GB model download becomes a one-time cost per endpoint. It is stdlib-only and best-effort. ```python from runpod.serverless import VolumeCache -with VolumeCache(dirs=["/root/.cache/huggingface"]).warm(): +with VolumeCache(dirs=["/root/.cache/huggingface"]): model = load_model() ``` diff --git a/docs/serverless/volume_cache.md b/docs/serverless/volume_cache.md index 20fb4598..be780458 100644 --- a/docs/serverless/volume_cache.md +++ b/docs/serverless/volume_cache.md @@ -1,139 +1,89 @@ # Network-Volume Warm Cache (VolumeCache) `VolumeCache` warms local directories across serverless workers using a mounted -network volume. On cold start it restores previously-synced directories (for -example a model cache) from the volume; after use it syncs newly written files -back so the next cold worker starts warm. This turns a repeated multi-GB model +network volume. It keeps a browsable mirror of your cache directories on the +volume and reconciles it against the container on each use: on cold start it +restores previously-cached files into place, and after use it copies newly +written files back to the volume. This turns a repeated multi-GB model download on every cold start into a one-time cost per endpoint. It is stdlib-only and best-effort: any failure degrades to a cold worker and -never raises into your handler or the worker loop. +never raises into your handler or worker loop. ## Requirements - A **network volume** attached to the endpoint (mounted at `/runpod-volume`). -- Set on serverless automatically: `RUNPOD_ENDPOINT_ID` (used to scope the cache - per endpoint). +- Set on serverless automatically: `RUNPOD_ENDPOINT_ID` (used to scope the + mirror per endpoint/namespace). -If no volume is mounted, every operation is a safe no-op. +If no volume is mounted, or the namespace is empty, every operation is a safe +no-op. -## Option 1: Built-in (opt-in, zero code) +## Usage -The worker can hydrate a model cache at startup and sync it after the first job, -with no changes to your handler. It is **off by default** — enable it with an -environment variable: - -```bash -# Enable the built-in warm cache (opt-in) -RUNPOD_VOLUME_CACHE=1 -``` - -When enabled and a volume is mounted, the worker: - -1. Hydrates the model cache from the volume before the first job runs. -2. Syncs new files to the volume once, after the first successful job. - -By default it caches the directories pointed to by `HF_HOME`, `HF_HUB_CACHE`, -and `TORCH_HOME` (whichever are set; `HF_HOME` defaults to -`~/.cache/huggingface`). Add more directories with `RUNPOD_CACHE_DIRS`. - -### Configuration - -| Variable | Default | Purpose | -| --- | --- | --- | -| `RUNPOD_VOLUME_CACHE` | unset (off) | Set to `1`/`true`/`yes`/`on` to enable the built-in. | -| `RUNPOD_CACHE_DIRS` | — | Extra directories to cache, `os.pathsep`-separated (`:` on Linux). | -| `RUNPOD_VOLUME_CACHE_MAX_GB` | `50` | Prune oldest shards once the endpoint's cache exceeds this size. | -| `HF_HOME` / `HF_HUB_CACHE` / `TORCH_HOME` | — | Auto-discovered model-cache locations. | - -### Example - -For most model-serving workers, enabling the built-in and letting the model -download into `HF_HOME` on the first request is all that is needed: - -```python -# my_worker.py -import runpod - -def handler(job): - # First cold worker downloads the model into HF_HOME; the built-in syncs it - # to the volume. Subsequent cold workers hydrate it and skip the download. - ... - -runpod.serverless.start({"handler": handler}) -``` - -```bash -# Template / container env -RUNPOD_VOLUME_CACHE=1 -``` - -> Note: hydration only helps model loads that happen **after** the worker starts -> (lazy or in-handler loading). If your model loads at module import time, use -> Option 2 to place `warm()` around the load. - -## Option 2: Explicit API - -Import `VolumeCache` and control caching yourself. This works for any -directories, and the `warm()` context manager guarantees hydration happens -before your model load regardless of when it runs. +`VolumeCache` is a context-manager closure — using it around a model load +hydrates the cache before the block runs and syncs any changes back after: ```python from runpod.serverless import VolumeCache -vc = VolumeCache(dirs=["/root/.cache/huggingface"]) - -with vc.warm(): # hydrate on enter, sync the delta on exit +with VolumeCache(dirs=["/root/.cache/huggingface"]): model = load_model() # downloads land in the cached directory ``` +- **On enter**, `hydrate()` copies files that are missing or newer on the + volume mirror into the container. +- **On exit**, `sync()` copies files that are missing or newer in the + container onto the volume mirror. By default this runs on a background + daemon thread and returns immediately, so the `with` block doesn't block on + the sync; a process-exit hook joins any outstanding syncs so short-lived + processes (including local test runs) still complete the sync before + exiting. + You can also call the phases directly when they happen at different points in your worker's lifecycle: ```python -vc = VolumeCache(dirs=["/data/models"], namespace="my-model-cache", max_size_gb=100) +vc = VolumeCache(dirs=["/data/models"], namespace="my-model-cache") -vc.hydrate() # restore cached files (e.g. at startup) -model = load_model() # populate the cache -vc.sync() # persist new files back to the volume +vc.hydrate() # restore cached files (e.g. at startup) +model = load_model() # populate the cache +vc.sync(background=False) # persist new files back to the volume, inline ``` -### Constructor +## Constructor | Argument | Default | Purpose | | --- | --- | --- | | `dirs` | required | Local directories to cache. | -| `namespace` | `RUNPOD_ENDPOINT_ID` | Isolation key for the on-volume shards. Must be a single safe path component. | +| `namespace` | `RUNPOD_ENDPOINT_ID` | Isolation key for the on-volume mirror. Must be a single safe path component. | | `volume_path` | `/runpod-volume` | Network-volume mount point. | -| `max_size_gb` | `None` | Prune oldest shards past this cap; `None` = no cap. | | `best_effort` | `True` | Swallow and log errors instead of raising. Set `False` while debugging. | ## How it works -- **Per-endpoint, sharded storage.** Each worker writes its delta as its own tar - shard under `//.cache//`, published atomically. Per-worker - shards are collision-free under concurrent workers — no shared file is - rewritten. Hydration extracts all shards for the namespace (newest wins on - overlap). -- **Delta only.** `sync()` packs just the files written since the last hydrate, - not the whole cache. -- **Retention.** With `max_size_gb` set, the oldest shards are pruned once the - namespace exceeds the cap. -- **Safety.** Extraction rejects members that would escape the configured - directories (traversal, symlinks, absolute paths). +- **Directory mirror.** Cached files live at `{volume_path}/.cache/{namespace}`, + laid out at the same absolute path they occupy in the container (so the + mirror is directly browsable — no archive format to unpack). +- **Per-file, atomic reconcile.** Each direction (`hydrate`/`sync`) walks the + source tree and copies any file whose size differs or whose mtime is newer + than the destination's (beyond a small tolerance for coarse network-filesystem + mtimes). Copies are written to a temp file and atomically renamed into place + (`os.replace`), so a crash mid-copy never leaves a half-written file visible. +- **Idempotent.** Copies preserve mtime (`shutil.copy2`), so re-running + `hydrate`/`sync` after nothing has changed copies zero files. +- **Last-writer-wins.** Under concurrent workers, the mirror simply reflects + whichever worker synced most recently — there's no locking or merge. +- **Safety.** Symlinked sources are never followed/copied. Hydration destinations + are checked to resolve inside one of the configured `dirs` before any write, + so a mirror entry can't be used to write outside the cached directories. ## Limitations - **Cold-scale write amplification.** If N workers cold-start at the same time, - each misses the still-empty cache, downloads the model, and writes a full-size - shard. Total volume writes on the first scale-up are ~N×; bounded thereafter by - retention. -- **Lost first warm on aggressive recycle.** `sync()` runs in a background thread - after the job response. A large-model sync may not finish before the worker is - recycled; the partial shard is cleaned up and the warm is retried on the next - worker. - -## Status - -The built-in is opt-in for now (`RUNPOD_VOLUME_CACHE=1`). Whether it should -default on when a volume is present is still under review. + each may miss the still-empty mirror, download the model, and sync a full + copy back. There's no coordination between concurrent syncs. +- **Background sync on short-lived processes.** `sync()` schedules the copy on + a daemon thread; if the process exits without going through normal + interpreter shutdown (e.g. `os._exit`, `SIGKILL`), the atexit hook never + runs and the sync may not complete. diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py index 94de75b0..a45cebc6 100644 --- a/runpod/serverless/modules/rp_job.py +++ b/runpod/serverless/modules/rp_job.py @@ -15,7 +15,6 @@ from ...version import __version__ as runpod_version from ..utils import rp_debugger -from ..utils.rp_volume_cache import sync_after_job from .rp_handler import is_generator from .rp_http import send_result, stream_result from .rp_tips import check_return_size @@ -283,8 +282,6 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: if run_result.get("output") == {}: run_result.pop("output") - sync_after_job() # fire-and-forget cache warm on any successful output - check_return_size(run_result) # Checks the size of the return body. except Exception as err: @@ -332,8 +329,6 @@ async def run_job_generator( log.debug(f"Generator output: {output_partial}", job["id"]) yield {"output": output_partial} - sync_after_job() # fire-and-forget cache warm on successful generator completion - except Exception as err: log.error(err, job["id"]) yield {"error": f"handler: {str(err)} \ntraceback: {traceback.format_exc()}"} diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 1523526f..a1a2a829 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -1,61 +1,61 @@ -"""Bidirectional warm-cache sync between local directories and a network volume.""" +"""Directory-mirror warm cache between local directories and a network volume. -import contextlib -import glob +``VolumeCache`` keeps a browsable mirror of one or more local directories on a +mounted network volume, and reconciles the two directions on demand: + +- ``hydrate()``: copy files that are missing or newer on the volume mirror + into the container (used on cold start, before the cache is populated). +- ``sync()``: copy files that are missing or newer in the container onto the + volume mirror (used after the cache has been populated/updated). + +Both directions are per-file, atomic (write-to-temp + ``os.replace``), and +idempotent: unchanged files are skipped on repeat calls because ``shutil.copy2`` +preserves mtimes. Stdlib-only and best-effort: any failure degrades to a cold +worker and never raises into the caller (unless ``best_effort=False``). +""" + +import atexit import os -import tarfile -import tempfile -import time -import uuid +import shutil import threading from runpod.serverless.modules.rp_logger import RunPodLogger log = RunPodLogger() -_BASELINE_EPSILON_SECONDS = 2.0 # tolerate coarse (1s) network-filesystem mtime granularity +_MTIME_TOLERANCE = 2.0 # seconds; tolerate coarse (NFS) mtime granularity class VolumeCache: """Warm-cache local directories across serverless workers via a network volume. - On cold start, ``hydrate()`` extracts previously-synced directories from a - mounted network volume into place; after use, ``sync()`` packs the newly - written files into a per-worker tar shard on the volume so the next cold - worker starts warm. Stdlib-only and best-effort: any failure degrades to a - cold worker and never raises into the caller. + Maintains a mirror of ``dirs`` at ``{volume_path}/.cache/{namespace}``. + ``hydrate()`` reconciles volume -> container; ``sync()`` reconciles + container -> volume. Used as a context manager, the object is itself the + "warm cache" closure: hydrate on enter, sync (in the background by + default) on exit. Requires a network volume mounted at ``volume_path`` (default ``/runpod-volume``) and a non-empty ``namespace`` (default - ``RUNPOD_ENDPOINT_ID``); otherwise ``available`` is False and all operations - are no-ops. + ``RUNPOD_ENDPOINT_ID``); otherwise ``available`` is False and all + operations are no-ops. Args: dirs: Local directories to cache (e.g. a model cache like ``HF_HOME``). - namespace: Isolation key for the on-volume shards. Must be a single safe - path component. Defaults to ``RUNPOD_ENDPOINT_ID``. + namespace: Isolation key for the on-volume mirror. Must be a single + safe path component. Defaults to ``RUNPOD_ENDPOINT_ID``. volume_path: Network-volume mount point. Defaults to ``/runpod-volume``. - max_size_gb: Prune oldest shards past this cap. ``None`` = no cap. best_effort: When True (default), swallow and log errors instead of raising. Example: - >>> vc = VolumeCache(dirs=["/root/.cache/huggingface"]) - >>> with vc.warm(): # hydrate on enter, sync delta on exit + >>> with VolumeCache(dirs=["/root/.cache/huggingface"]): ... model = load_model() # downloads land in the cached dir """ _EXCLUDE_SUBSTRINGS = (os.sep + "refs" + os.sep, os.sep + ".no_exist" + os.sep) - def __init__( - self, - dirs, - *, - namespace=None, - volume_path="/runpod-volume", - max_size_gb=None, - best_effort=True, - ): + def __init__(self, dirs, *, namespace=None, volume_path="/runpod-volume", best_effort=True): self._dirs = [os.path.realpath(os.fspath(d)) for d in dirs] self._namespace = namespace or os.environ.get("RUNPOD_ENDPOINT_ID") or "" if self._namespace and ( @@ -69,231 +69,175 @@ def __init__( f"namespace must be a single safe path component, got {self._namespace!r}" ) self._volume_path = os.fspath(volume_path) - self._max_size_gb = max_size_gb self._best_effort = best_effort - self._worker_id = os.environ.get("RUNPOD_POD_ID") or uuid.uuid4().hex[:12] - self._baseline = time.time() - _BASELINE_EPSILON_SECONDS @property - def _shard_dir(self): + def _mirror_root(self): return os.path.join(self._volume_path, ".cache", self._namespace) @property def available(self): + """True iff volume_path is a mounted dir AND namespace is non-empty.""" return bool(self._namespace) and os.path.isdir(self._volume_path) - def _list_shards(self): - d = self._shard_dir - if not os.path.isdir(d): - return [] - shards = [os.path.join(d, f) for f in os.listdir(d) if f.endswith(".tar")] - return sorted(shards, key=os.path.getmtime) - - def _iter_delta_files(self): - for root in self._dirs: - if not os.path.isdir(root): - continue - for dirpath, _dirs, files in os.walk(root): - for name in files: - path = os.path.join(dirpath, name) - if name.endswith(".lock") or name.startswith(".rp_volume_cache"): - continue - if any(sub in path for sub in self._EXCLUDE_SUBSTRINGS): - continue - try: - if os.path.getmtime(path) > self._baseline: - yield path - except OSError: + # ----------------------------------------------------------------- # + # reconcile helpers + # ----------------------------------------------------------------- # + + def _iter_files(self, root): + for dirpath, _dirs, files in os.walk(root): + for name in files: + path = os.path.join(dirpath, name) + if name.endswith(".lock") or name.startswith(".rpvc"): + continue + if any(sub in path for sub in self._EXCLUDE_SUBSTRINGS): + continue + try: + if os.path.islink(path): continue + except OSError: + continue + yield path - def _guard(self, fn, default): + @staticmethod + def _needs_copy(src_path, dst_path): try: - return fn() - except Exception as exc: # best-effort: never break the worker - if not self._best_effort: - raise - log.warn(f"VolumeCache operation failed: {exc}") - return default - - def sync(self): - if not self.available: + s = os.stat(src_path) + except OSError: return False - return self._guard(self._do_sync, False) + try: + d = os.stat(dst_path) + except OSError: + return True + return s.st_size != d.st_size or s.st_mtime > d.st_mtime + _MTIME_TOLERANCE - def _do_sync(self): - files = list(self._iter_delta_files()) - if not files: - log.debug("VolumeCache: no delta files to sync") - return False - os.makedirs(self._shard_dir, exist_ok=True) - for stale in glob.glob(os.path.join(self._shard_dir, f"{self._worker_id}-*.tar.tmp")): + 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): + tmp = dst + ".rpvc.tmp" + try: + os.makedirs(os.path.dirname(dst), exist_ok=True) + shutil.copy2(src, tmp) # preserves mtime so future diffs converge + os.replace(tmp, dst) # atomic; last-writer-wins under concurrency + return True + except OSError as exc: + log.debug(f"VolumeCache: skip {src} -> {dst}: {exc}") try: - os.remove(stale) + if os.path.exists(tmp): + os.remove(tmp) except OSError: - # best-effort cleanup: ignore temp files that vanish or can't be removed pass - final = os.path.join(self._shard_dir, f"{self._worker_id}-{time.time_ns():020d}.tar") - tmp = final + ".tmp" - with tarfile.open(tmp, "w") as tar: - for path in files: - tar.add(path, arcname=os.path.relpath(path, "/")) - os.replace(tmp, final) - log.info(f"VolumeCache: synced {len(files)} files to {final}") - self._baseline = time.time() - _BASELINE_EPSILON_SECONDS - self._enforce_retention() - return True - - def _enforce_retention(self): - if not self._max_size_gb: - return - def _size(p): - try: - return os.path.getsize(p) - except OSError: - return 0 - - cap = self._max_size_gb * (1024 ** 3) - shards = self._list_shards() # oldest first - total = sum(_size(s) for s in shards) - for shard in shards: - if total <= cap: - break - size = _size(shard) - try: - os.remove(shard) - total -= size - log.info(f"VolumeCache: pruned old shard {shard}") - except OSError as exc: - log.warn(f"VolumeCache: failed to prune {shard}: {exc}") - - @property - def _marker_path(self): - base = os.path.join(tempfile.gettempdir(), "rp_volume_cache") - return os.path.join(base, f"{self._namespace}.hydrated") + return False - def _newest_shard_mtime(self): - shards = self._list_shards() - return os.path.getmtime(shards[-1]) if shards else 0.0 + def _guard(self, fn, default): + try: + return fn() + except Exception as exc: # best-effort: never break the worker + if not self._best_effort: + raise + log.warn(f"VolumeCache operation failed: {exc}") + return default - def _clear_marker_for_test(self): - if os.path.exists(self._marker_path): - os.remove(self._marker_path) + # ----------------------------------------------------------------- # + # hydrate / sync + # ----------------------------------------------------------------- # def hydrate(self): + """Reconcile volume mirror -> container. + + Copy files missing or newer in the container. Returns the number of + files copied. No-op (0) if unavailable. + """ if not self.available: - return False - return self._guard(self._do_hydrate, False) + return 0 + return self._guard(self._do_hydrate, 0) def _do_hydrate(self): - shards = self._list_shards() - if not shards: - return False - newest = self._newest_shard_mtime() - if os.path.exists(self._marker_path) and os.path.getmtime(self._marker_path) >= newest: - log.debug("VolumeCache: cache already hydrated, skipping") - return False - extracted = False - # Use the tar data filter for defense-in-depth where the runtime provides - # it (Python 3.12+, and 3.10.12+/3.11.4+ backports) without requiring it -- - # the >=3.10 floor may predate the API. _is_safe_member is the primary guard. - extract_kwargs = {"filter": "data"} if hasattr(tarfile, "data_filter") else {} - for shard in shards: # oldest -> newest (last wins) - with tarfile.open(shard) as tar: - safe = [m for m in tar.getmembers() if self._is_safe_member(m)] - tar.extractall(path="/", members=safe, **extract_kwargs) - extracted = extracted or bool(safe) - os.makedirs(os.path.dirname(self._marker_path), exist_ok=True) - with open(self._marker_path, "w") as fh: - fh.write(str(newest)) - os.utime(self._marker_path, (newest, newest)) - self._baseline = time.time() - _BASELINE_EPSILON_SECONDS - if extracted: - log.info(f"VolumeCache: hydrated from {len(shards)} shard(s)") - return extracted - - def _is_safe_member(self, member): - if not (member.isfile() or member.isdir()): - return False # reject symlink/hardlink/device/fifo - target = os.path.realpath(os.path.join("/", member.name)) - return any( - target == d or target.startswith(d + os.sep) - for d in self._dirs - ) - - @contextlib.contextmanager - def warm(self): - self.hydrate() - try: - yield self - finally: - self.sync() - - -_ACTIVE_CACHE = None -_SYNCED = False -_sync_lock = threading.Lock() + root = self._mirror_root + if not os.path.isdir(root): + return 0 + copied = 0 + for m in self._iter_files(root): + dst = os.path.join("/", os.path.relpath(m, root)) + if not self._is_safe_dest(dst): + continue # skip anything that would escape configured dirs + if self._needs_copy(m, dst) and self._copy_file(m, dst): + copied += 1 + if copied: + log.info(f"VolumeCache: hydrated {copied} file(s) from {root}") + return copied + + def sync(self, *, background=True): + """Reconcile container -> volume mirror. + + Copy files missing or newer on the volume. When ``background=True`` + (default), run on a daemon thread and return immediately; a + process-exit hook joins outstanding syncs so short-lived processes + still complete. When False, run inline and return when done. + """ + if not self.available: + return + if background: + t = threading.Thread(target=lambda: self._guard(self._do_sync, 0), daemon=True) + _register_pending(t) + t.start() + else: + self._guard(self._do_sync, 0) -_ENABLED_VALUES = ("1", "true", "yes", "on") + def _do_sync(self): + os.makedirs(self._mirror_root, exist_ok=True) + copied = 0 + for root in self._dirs: + if not os.path.isdir(root): + continue + for f in self._iter_files(root): + dst = os.path.join(self._mirror_root, os.path.relpath(f, "/")) + if self._needs_copy(f, dst) and self._copy_file(f, dst): + copied += 1 + if copied: + log.info(f"VolumeCache: synced {copied} file(s) to {self._mirror_root}") + return copied + + # ----------------------------------------------------------------- # + # context manager + # ----------------------------------------------------------------- # + + def __enter__(self): + self.hydrate() + return self + def __exit__(self, *exc): + self.sync(background=True) + return None -def _discover_model_dirs(): - dirs = [os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface")] - for var in ("HF_HUB_CACHE", "TORCH_HOME"): - if os.environ.get(var): - dirs.append(os.environ[var]) - extra = os.environ.get("RUNPOD_CACHE_DIRS") - if extra: - dirs.extend(p for p in extra.split(os.pathsep) if p) - return list(dict.fromkeys(dirs)) # de-dupe, preserve order +# ----------------------------------------------------------------------- # +# background sync completion +# ----------------------------------------------------------------------- # -def build_default_cache(): - """Build the built-in VolumeCache when opt-in is enabled, else None. +_pending_syncs = [] +_atexit_registered = False - The built-in is OFF by default. It activates only when RUNPOD_VOLUME_CACHE is - set to a truthy value ("1"/"true"/"yes"/"on") AND a network volume is mounted. - Known operational limitations (accepted, not bugs): - 1. Cold-scale write amplification: when N workers cold-start simultaneously, - each misses the still-empty cache and each writes its own full-size shard - on first sync. Bounded by retention (RUNPOD_VOLUME_CACHE_MAX_GB). - 2. Lost first warm on aggressive recycle: sync() runs in a daemon thread - dispatched after the job response; a large-model sync may not finish - before the worker is recycled, killing the thread mid-write. The partial - shard is cleaned up on the next sync from this worker, and the warm is - simply retried then. - """ - if os.environ.get("RUNPOD_VOLUME_CACHE", "").lower() not in _ENABLED_VALUES: - return None - try: - max_gb = float(os.environ.get("RUNPOD_VOLUME_CACHE_MAX_GB", "50")) - vc = VolumeCache(_discover_model_dirs(), max_size_gb=max_gb) - except Exception as exc: # never let cache setup crash worker startup - log.warn(f"VolumeCache: failed to build default cache, disabling: {exc}") - return None - return vc if vc.available else None +def _register_pending(thread): + global _atexit_registered + _pending_syncs[:] = [t for t in _pending_syncs if t.is_alive()] + _pending_syncs.append(thread) + if not _atexit_registered: + atexit.register(_join_pending_syncs) + _atexit_registered = True -def set_active_cache(vc): - global _ACTIVE_CACHE - _ACTIVE_CACHE = vc +def _join_pending_syncs(): + for t in list(_pending_syncs): + try: + t.join() + except Exception: + pass + _pending_syncs.clear() -def sync_after_job(): - global _SYNCED - try: - if _ACTIVE_CACHE is None: - return - with _sync_lock: - if _SYNCED: - return - _SYNCED = True - threading.Thread(target=_ACTIVE_CACHE.sync, daemon=True).start() - except Exception as exc: # cache glue must never affect job outcome - log.warn(f"VolumeCache: sync_after_job failed to dispatch: {exc}") - - -def reset_builtin_state_for_test(): - global _ACTIVE_CACHE, _SYNCED - _ACTIVE_CACHE = None - _SYNCED = False +def _reset_pending_for_test(): + _pending_syncs.clear() diff --git a/runpod/serverless/worker.py b/runpod/serverless/worker.py index d50df6ec..90053ec7 100644 --- a/runpod/serverless/worker.py +++ b/runpod/serverless/worker.py @@ -9,7 +9,6 @@ from runpod.serverless.modules import rp_logger, rp_local, rp_ping, rp_scale from runpod.serverless.modules.rp_fitness import run_fitness_checks -from runpod.serverless.utils.rp_volume_cache import build_default_cache, set_active_cache log = rp_logger.RunPodLogger() heartbeat = rp_ping.Heartbeat() @@ -49,12 +48,6 @@ def run_worker(config: Dict[str, Any]) -> None: # Start pinging Runpod to show that the worker is alive. heartbeat.start_ping(mirror) - # Warm the local cache from a mounted network volume before any job runs. - volume_cache = build_default_cache() - if volume_cache is not None: - volume_cache.hydrate() - set_active_cache(volume_cache) - # Create a JobScaler responsible for adjusting the concurrency job_scaler = rp_scale.JobScaler(config) job_scaler.start() diff --git a/tests/test_serverless/test_modules/test_job.py b/tests/test_serverless/test_modules/test_job.py index 410d01fa..1ec5ce35 100644 --- a/tests/test_serverless/test_modules/test_job.py +++ b/tests/test_serverless/test_modules/test_job.py @@ -440,68 +440,3 @@ async def test_run_job_generator_exception(self): assert mock_log.info.call_count == 1 mock_log.info.assert_called_with("Finished running generator.", "123") - async def test_run_job_generator_success_syncs_cache(self): - """ - Tests that run_job_generator triggers sync_after_job on the success path. - """ - handler = self.handler_gen_success - job = {"id": "123"} - counter = {"n": 0} - - def fake_sync_after_job(): - counter["n"] += 1 - - with patch( - "runpod.serverless.modules.rp_job.log", new_callable=Mock - ), patch( - "runpod.serverless.modules.rp_job.sync_after_job", - side_effect=fake_sync_after_job, - ) as mock_sync: - result = [i async for i in rp_job.run_job_generator(handler, job)] - - assert result == [ - {"output": "partial_output_1"}, - {"output": "partial_output_2"}, - ] - mock_sync.assert_called_once() - assert counter["n"] == 1 - - async def test_run_job_generator_success_syncs_cache_async(self): - """ - Tests that run_job_generator triggers sync_after_job on the success path - for an async generator handler. - """ - handler = self.handler_async_gen_success - job = {"id": "123"} - - with patch( - "runpod.serverless.modules.rp_job.log", new_callable=Mock - ), patch( - "runpod.serverless.modules.rp_job.sync_after_job" - ) as mock_sync: - result = [i async for i in rp_job.run_job_generator(handler, job)] - - assert result == [ - {"output": "partial_output_1"}, - {"output": "partial_output_2"}, - ] - mock_sync.assert_called_once() - - async def test_run_job_generator_exception_does_not_sync_cache(self): - """ - Tests that run_job_generator does NOT trigger sync_after_job on the - exception path. - """ - handler = self.handler_fail - job = {"id": "123"} - - with patch( - "runpod.serverless.modules.rp_job.log", new_callable=Mock - ), patch( - "runpod.serverless.modules.rp_job.sync_after_job" - ) as mock_sync: - result = [i async for i in rp_job.run_job_generator(handler, job)] - - assert len(result) == 1 - assert "error" in result[0] - mock_sync.assert_not_called() 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 0a216e60..a2789a3e 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1,23 +1,41 @@ import os -import tarfile +import shutil import time + import pytest + import runpod.serverless.utils.rp_volume_cache as vcmod VolumeCache = vcmod.VolumeCache -def test_unavailable_when_volume_dir_missing(tmp_path): - vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", - volume_path=str(tmp_path / "no-volume")) - assert vc.available is False +@pytest.fixture(autouse=True) +def _reset_pending(): + vcmod._reset_pending_for_test() + yield + vcmod._join_pending_syncs() + vcmod._reset_pending_for_test() -def test_available_when_volume_present_and_namespace_set(tmp_path): +def _mk_cache_with_volume(tmp_path, namespace="ep1"): + cache = tmp_path / "cache" + cache.mkdir() vol = tmp_path / "volume" vol.mkdir() - vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol)) - assert vc.available is True + vc = VolumeCache([str(cache)], namespace=namespace, volume_path=str(vol)) + return vc, cache, vol + + +# --------------------------------------------------------------------------- # +# available +# --------------------------------------------------------------------------- # + + +def test_unavailable_when_volume_dir_missing(tmp_path): + vc = VolumeCache( + [str(tmp_path / "cache")], namespace="ep1", volume_path=str(tmp_path / "no-volume") + ) + assert vc.available is False def test_unavailable_without_namespace(tmp_path, monkeypatch): @@ -28,7 +46,25 @@ def test_unavailable_without_namespace(tmp_path, monkeypatch): assert vc.available is False -@pytest.mark.parametrize("bad_namespace", ["../evil", "a/b", "/etc", ".."]) +def test_available_when_volume_present_and_namespace_set(tmp_path): + vc, _cache, _vol = _mk_cache_with_volume(tmp_path) + assert vc.available is True + + +def test_namespace_defaults_to_endpoint_id(tmp_path, monkeypatch): + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "endpoint-xyz") + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "cache")], volume_path=str(vol)) + assert vc._mirror_root == os.path.join(str(vol), ".cache", "endpoint-xyz") + + +# --------------------------------------------------------------------------- # +# namespace validation +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize("bad_namespace", ["../evil", "a/b", "/etc", "..", "a\\b", "."]) def test_namespace_rejects_unsafe_values(tmp_path, bad_namespace): vol = tmp_path / "volume" vol.mkdir() @@ -37,379 +73,283 @@ def test_namespace_rejects_unsafe_values(tmp_path, bad_namespace): def test_namespace_accepts_normal_value(tmp_path): - vol = tmp_path / "volume" - vol.mkdir() - vc = VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol)) + vc, _cache, _vol = _mk_cache_with_volume(tmp_path, namespace="ep1") assert vc._namespace == "ep1" -def test_namespace_defaults_to_endpoint_id(tmp_path, monkeypatch): - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "endpoint-xyz") - vol = tmp_path / "volume" - vol.mkdir() - vc = VolumeCache([str(tmp_path / "cache")], volume_path=str(vol)) - assert vc._shard_dir == os.path.join(str(vol), ".cache", "endpoint-xyz") +# --------------------------------------------------------------------------- # +# sync -> hydrate round trip +# --------------------------------------------------------------------------- # -def _mk_cache_with_volume(tmp_path): - cache = tmp_path / "cache" - cache.mkdir() - vol = tmp_path / "volume" - vol.mkdir() - vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) - return vc, cache, vol +def test_sync_then_hydrate_round_trip(tmp_path): + vc, cache, vol = _mk_cache_with_volume(tmp_path) + (cache / "model.bin").write_text("weights") + + vc.sync(background=False) + (cache / "model.bin").unlink() + + fresh = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + copied = fresh.hydrate() + + assert copied == 1 + assert (cache / "model.bin").read_text() == "weights" + + +def test_sync_is_idempotent_on_unchanged_file(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + (cache / "model.bin").write_text("weights") + assert vc.sync(background=False) is None + first_copied = vc._do_sync() + assert first_copied == 0 # already synced above; nothing new to copy -def test_sync_packs_files_created_after_baseline(tmp_path): + +def test_sync_recopies_modified_file(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + f = cache / "model.bin" + f.write_text("v1") + vc.sync(background=False) + + time.sleep(0.01) + f.write_text("v2-longer") + os.utime(f, (time.time() + 10, time.time() + 10)) + + copied = vc._do_sync() + assert copied == 1 + + mirror_file = os.path.join(vc._mirror_root, os.path.relpath(str(f), "/")) + assert open(mirror_file).read() == "v2-longer" + + +def test_hydrate_skips_unchanged_after_first_hydrate(tmp_path): vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 (cache / "model.bin").write_text("weights") - assert vc.sync() is True - shards = vc._list_shards() - assert len(shards) == 1 - with tarfile.open(shards[0]) as tar: - names = tar.getnames() - assert os.path.relpath(str(cache / "model.bin"), "/") in names + vc.sync(background=False) + (cache / "model.bin").unlink() + + fresh = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + assert fresh.hydrate() == 1 + assert fresh.hydrate() == 0 # already up to date -def test_sync_excludes_files_older_than_baseline(tmp_path): +def test_hydrate_does_not_overwrite_newer_container_file(tmp_path): + # Same size, later mtime: _needs_copy must key off mtime here since size + # alone can't distinguish the versions. vc, cache, vol = _mk_cache_with_volume(tmp_path) - old = cache / "old.bin" - old.write_text("x") - os.utime(old, (time.time() - 100, time.time() - 100)) - vc._baseline = time.time() - assert vc.sync() is False + f = cache / "model.bin" + f.write_text("weightsA") + vc.sync(background=False) + + # Container file is newer than the mirror copy (same size). + f.write_text("weightsB") + os.utime(f, (time.time() + 100, time.time() + 100)) + + fresh = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) + fresh.hydrate() + assert f.read_text() == "weightsB" + + +# --------------------------------------------------------------------------- # +# exclusions +# --------------------------------------------------------------------------- # def test_sync_skips_excluded_paths(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 + vc, cache, _vol = _mk_cache_with_volume(tmp_path) (cache / "refs").mkdir() (cache / "refs" / "main").write_text("ref") - assert vc.sync() is False + (cache / "locked.lock").write_text("lock") + copied = vc._do_sync() + assert copied == 0 -def test_sync_shard_names_are_unique_per_call(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 - (cache / "a.bin").write_text("a") - vc.sync() - (cache / "b.bin").write_text("b") - vc.sync() - assert len(vc._list_shards()) == 2 +def test_sync_skips_symlink_source(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + real = cache / "real.bin" + real.write_text("real-data") + link = cache / "link.bin" + try: + link.symlink_to(real) + except (OSError, NotImplementedError): + pytest.skip("symlinks not supported on this platform") -def test_sync_noop_when_unavailable(tmp_path): - vc = VolumeCache([str(tmp_path / "c")], namespace="ep1", - volume_path=str(tmp_path / "missing")) - assert vc.sync() is False + copied = vc._do_sync() + assert copied == 1 # only real.bin, not the symlink + mirror_link = os.path.join(vc._mirror_root, os.path.relpath(str(link), "/")) + assert not os.path.exists(mirror_link) -def test_sync_tolerates_coarse_mtime_granularity(tmp_path): - # A fresh instance sets baseline to now - epsilon. A file whose mtime is - # floored to the current integer second (as coarse NFS filesystems report) - # must still be picked up, not silently dropped. - cache = tmp_path / "cache" - cache.mkdir() - vol = tmp_path / "volume" - vol.mkdir() - vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) - f = cache / "model.bin" - f.write_text("weights") - now = time.time() - os.utime(f, (float(int(now)), float(int(now)))) # floor mtime to integer second - assert vc.sync() is True + +# --------------------------------------------------------------------------- # +# hydrate destination safety +# --------------------------------------------------------------------------- # -def test_hydrate_noop_when_no_shards(tmp_path): +def test_hydrate_skips_unsafe_destination(tmp_path): vc, cache, vol = _mk_cache_with_volume(tmp_path) - assert vc.hydrate() is False + # Craft a mirror file whose mapped destination escapes the configured dirs. + escape_rel = os.path.relpath(str(tmp_path / "outside" / "evil.txt"), "/") + mirror_file = os.path.join(vc._mirror_root, escape_rel) + os.makedirs(os.path.dirname(mirror_file), exist_ok=True) + with open(mirror_file, "w") as fh: + fh.write("malicious") -def test_hydrate_restores_files_to_absolute_paths(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 - (cache / "model.bin").write_text("weights") - vc.sync() - (cache / "model.bin").unlink() # simulate a fresh cold worker - assert vc.hydrate() is True - assert (cache / "model.bin").read_text() == "weights" + copied = vc.hydrate() + assert copied == 0 + assert not (tmp_path / "outside" / "evil.txt").exists() -def test_hydrate_later_shard_overwrites_earlier(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 - f = cache / "model.bin" - f.write_text("v1") - vc.sync() - time.sleep(0.01) - f.write_text("v2") - vc._baseline = time.time() - 5 - vc.sync() - f.unlink() - vc._clear_marker_for_test() - vc.hydrate() - assert f.read_text() == "v2" +# --------------------------------------------------------------------------- # +# context manager +# --------------------------------------------------------------------------- # -def test_hydrate_is_idempotent_via_marker(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - vc._baseline = time.time() - 5 +def test_context_manager_hydrates_on_enter_and_syncs_on_exit(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) (cache / "model.bin").write_text("weights") - vc.sync() - assert vc.hydrate() is True # first hydrate extracts - assert vc.hydrate() is False # marker current -> no-op + with vc as ctx: + assert ctx is vc -def test_rejects_member_outside_configured_dirs(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - m = tarfile.TarInfo(name="etc/passwd") # resolves to /etc/passwd, outside cache - m.type = tarfile.REGTYPE - assert vc._is_safe_member(m) is False + vcmod._join_pending_syncs() + mirror_file = os.path.join(vc._mirror_root, os.path.relpath(str(cache / "model.bin"), "/")) + assert os.path.exists(mirror_file) -def test_rejects_symlink_member(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - rel = os.path.relpath(str(cache / "link"), "/") - m = tarfile.TarInfo(name=rel) - m.type = tarfile.SYMTYPE - m.linkname = "/etc/passwd" - assert vc._is_safe_member(m) is False +def test_context_manager_does_not_suppress_exceptions(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + (cache / "model.bin").write_text("weights") -def test_accepts_regular_member_inside_dirs(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - rel = os.path.relpath(str(cache / "model.bin"), "/") - m = tarfile.TarInfo(name=rel) - m.type = tarfile.REGTYPE - assert vc._is_safe_member(m) is True + with pytest.raises(ValueError): + with vc: + raise ValueError("boom") + vcmod._join_pending_syncs() + mirror_file = os.path.join(vc._mirror_root, os.path.relpath(str(cache / "model.bin"), "/")) + assert os.path.exists(mirror_file) -def test_rejects_sibling_prefix_collision(tmp_path): - # An allowed dir ".../cache" must NOT match a sibling ".../cache-evil"; - # this locks the separator-anchored prefix check against substring regressions. - vc, cache, vol = _mk_cache_with_volume(tmp_path) - evil = tmp_path / "cache-evil" - evil.mkdir() - rel = os.path.relpath(str(evil / "x"), "/") - m = tarfile.TarInfo(name=rel) - m.type = tarfile.REGTYPE - assert vc._is_safe_member(m) is False +# --------------------------------------------------------------------------- # +# best-effort +# --------------------------------------------------------------------------- # -def test_rejects_hardlink_member(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - rel = os.path.relpath(str(cache / "hl"), "/") - m = tarfile.TarInfo(name=rel) - m.type = tarfile.LNKTYPE - m.linkname = "root/.cache/x" - assert vc._is_safe_member(m) is False - - -def test_retention_prunes_oldest_shards_past_cap(tmp_path): - cache = tmp_path / "cache"; cache.mkdir() - vol = tmp_path / "volume"; vol.mkdir() - cap_bytes = 1500 - vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol), - max_size_gb=cap_bytes / (1024 ** 3)) - for i in range(4): - vc._baseline = time.time() - 5 - (cache / f"f{i}.bin").write_text("x" * 800) - vc.sync() - time.sleep(0.01) - total = sum(os.path.getsize(s) for s in vc._list_shards()) - assert total <= cap_bytes - - -def test_retention_tolerates_shard_removed_concurrently(tmp_path, monkeypatch): - # A concurrent worker prunes/removes a shard between _list_shards() and the - # getsize() lookups; the size lookup must degrade to 0 instead of raising. - cache = tmp_path / "cache"; cache.mkdir() - vol = tmp_path / "volume"; vol.mkdir() - cap_bytes = 1000 - vc = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol), - max_size_gb=cap_bytes / (1024 ** 3)) - for i in range(3): - vc._baseline = time.time() - 5 - (cache / f"f{i}.bin").write_text("x" * 800) - vc.sync() - time.sleep(0.01) - - real_getsize = os.path.getsize - - def flaky_getsize(path): - if str(path).endswith(".tar") and "f0" not in str(path): - raise FileNotFoundError(path) - return real_getsize(path) - - monkeypatch.setattr(os.path, "getsize", flaky_getsize) - # Should not raise despite getsize() failures on some shards. - vc._enforce_retention() - - -def test_no_retention_when_cap_is_none(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - for i in range(3): - vc._baseline = time.time() - 5 - (cache / f"f{i}.bin").write_text("data") - vc.sync() - assert len(vc._list_shards()) == 3 +def test_best_effort_swallows_sync_failure_inline(tmp_path, monkeypatch): + vc, _cache, _vol = _mk_cache_with_volume(tmp_path) -def test_best_effort_swallows_and_returns_default(tmp_path, monkeypatch): - vc, cache, vol = _mk_cache_with_volume(tmp_path) def boom(): raise RuntimeError("disk exploded") + monkeypatch.setattr(vc, "_do_sync", boom) - (cache / "x.bin").write_text("x") - vc._baseline = time.time() - 5 - assert vc.sync() is False # swallowed, no raise + vc.sync(background=False) # must not raise -def test_best_effort_false_reraises(tmp_path, monkeypatch): - vc, cache, vol = _mk_cache_with_volume(tmp_path) +def test_best_effort_false_reraises_inline(tmp_path, monkeypatch): + vc, _cache, _vol = _mk_cache_with_volume(tmp_path) vc._best_effort = False + def boom(): raise RuntimeError("disk exploded") + monkeypatch.setattr(vc, "_do_sync", boom) with pytest.raises(RuntimeError): - vc.sync() + vc.sync(background=False) -def test_warm_hydrates_on_enter_and_syncs_on_exit(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - calls = [] - vc.hydrate = lambda: calls.append("hydrate") - vc.sync = lambda: calls.append("sync") - with vc.warm(): - calls.append("body") - assert calls == ["hydrate", "body", "sync"] +def test_best_effort_swallows_sync_failure_background(tmp_path, monkeypatch): + vc, _cache, _vol = _mk_cache_with_volume(tmp_path) + def boom(): + raise RuntimeError("disk exploded") -def test_warm_syncs_even_on_exception(tmp_path): - vc, cache, vol = _mk_cache_with_volume(tmp_path) - calls = [] - vc.hydrate = lambda: calls.append("hydrate") - vc.sync = lambda: calls.append("sync") - raised = False - try: - with vc.warm(): - raise ValueError("boom") - except ValueError: - raised = True - assert raised - assert calls == ["hydrate", "sync"] + monkeypatch.setattr(vc, "_do_sync", boom) + vc.sync(background=True) + vcmod._join_pending_syncs() # must not raise / must not crash the thread -def test_volumecache_exported_from_serverless(): - from runpod import serverless as sls - from runpod.serverless import utils as sls_utils - assert sls.VolumeCache is sls_utils.VolumeCache - assert "VolumeCache" in sls.__all__ +# --------------------------------------------------------------------------- # +# unavailable no-ops +# --------------------------------------------------------------------------- # + + +# --------------------------------------------------------------------------- # +# low-level helper edge cases (coverage of defensive branches) +# --------------------------------------------------------------------------- # -def test_build_default_cache_off_by_default(tmp_path, monkeypatch): - # Opt-in: with RUNPOD_VOLUME_CACHE unset, the built-in is disabled even - # when a volume would be available. - monkeypatch.delenv("RUNPOD_VOLUME_CACHE", raising=False) - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") - monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: True)) - assert vcmod.build_default_cache() is None +def test_iter_files_swallows_islink_oserror(tmp_path, monkeypatch): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + (cache / "f.bin").write_text("data") + def flaky_islink(path): + raise OSError("boom") -def test_build_default_cache_disabled_by_env(tmp_path, monkeypatch): - monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "0") - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") - assert vcmod.build_default_cache() is None + monkeypatch.setattr(os.path, "islink", flaky_islink) + assert list(vc._iter_files(str(cache))) == [] -def test_build_default_cache_none_when_no_volume(tmp_path, monkeypatch): - monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") - monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: False)) - assert vcmod.build_default_cache() is None +def test_needs_copy_false_when_src_missing(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + assert vc._needs_copy(str(cache / "missing.bin"), str(cache / "dst.bin")) is False -def test_discover_model_dirs_includes_hf_home_and_extras(monkeypatch): - monkeypatch.setenv("HF_HOME", "/models/hf") - monkeypatch.setenv("RUNPOD_CACHE_DIRS", "/a" + os.pathsep + "/b") - dirs = vcmod._discover_model_dirs() - assert "/models/hf" in dirs and "/a" in dirs and "/b" in dirs +def test_do_sync_skips_missing_dirs(tmp_path): + vol = tmp_path / "volume" + vol.mkdir() + vc = VolumeCache([str(tmp_path / "does-not-exist")], namespace="ep1", volume_path=str(vol)) + assert vc._do_sync() == 0 -def test_sync_after_job_runs_once(monkeypatch): - vcmod.reset_builtin_state_for_test() - counter = {"n": 0} - fake = type("F", (), {"sync": lambda self: counter.__setitem__("n", counter["n"] + 1)})() - joined = [] - class FakeThread: - def __init__(self, target, daemon=None): self.target = target - def start(self): self.target(); joined.append(True) - monkeypatch.setattr(vcmod.threading, "Thread", FakeThread) - vcmod.set_active_cache(fake) - vcmod.sync_after_job() - vcmod.sync_after_job() - assert counter["n"] == 1 +def test_copy_file_returns_false_and_cleans_tmp_on_failure(tmp_path, monkeypatch): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + src = cache / "f.bin" + src.write_text("data") + dst = cache / "dst.bin" + def flaky_copy2(_src, _dst): + raise OSError("disk full") -def test_run_worker_hydrates_registered_cache(monkeypatch): - from runpod.serverless import worker + monkeypatch.setattr(shutil, "copy2", flaky_copy2) + assert vc._copy_file(str(src), str(dst)) is False + assert not os.path.exists(str(dst) + ".rpvc.tmp") - vcmod.reset_builtin_state_for_test() - try: - async def _noop_fitness_checks(): - return None - - fake = type("F", (), { - "hydrate": lambda self: fake_calls.append("h"), - "sync": lambda self: None, - })() - fake_calls = [] - monkeypatch.setattr(worker, "build_default_cache", lambda: fake, raising=False) - monkeypatch.setattr(worker.rp_scale, "JobScaler", - lambda config: type("J", (), {"start": lambda self: None})()) - monkeypatch.setattr(worker.heartbeat, "start_ping", lambda mirror: None) - monkeypatch.setattr(worker, "run_fitness_checks", _noop_fitness_checks) - worker.run_worker({"handler": lambda job: job}) - assert "h" in fake_calls - finally: - vcmod.reset_builtin_state_for_test() - - -def test_build_default_cache_survives_bad_max_gb(monkeypatch): - vcmod.reset_builtin_state_for_test() - monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") - monkeypatch.setenv("RUNPOD_VOLUME_CACHE_MAX_GB", "not-a-number") - assert vcmod.build_default_cache() is None # degrades, does not raise - - -def test_two_workers_produce_independently_hydratable_shards(tmp_path): - # Distinct worker_ids must write non-colliding shards that both hydrate (spec acceptance). - cache = tmp_path / "cache"; cache.mkdir() - vol = tmp_path / "volume"; vol.mkdir() - a = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) - a._worker_id = "workerA" - a._baseline = time.time() - 5 - (cache / "a.bin").write_text("aaa") - assert a.sync() is True - b = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) - b._worker_id = "workerB" - b._baseline = time.time() - 5 - (cache / "b.bin").write_text("bbb") - assert b.sync() is True - (cache / "a.bin").unlink(); (cache / "b.bin").unlink() - reader = VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol)) - reader._clear_marker_for_test() - assert reader.hydrate() is True - assert (cache / "a.bin").read_text() == "aaa" - assert (cache / "b.bin").read_text() == "bbb" - - -def test_build_default_cache_returns_instance_when_available(monkeypatch): - vcmod.reset_builtin_state_for_test() - monkeypatch.setenv("RUNPOD_VOLUME_CACHE", "1") - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep1") - monkeypatch.setattr(vcmod.VolumeCache, "available", property(lambda self: True)) - vc = vcmod.build_default_cache() - assert isinstance(vc, vcmod.VolumeCache) + +def test_join_pending_syncs_swallows_join_exception(): + class FlakyThread: + def join(self): + raise RuntimeError("join blew up") + + vcmod._pending_syncs.append(FlakyThread()) + vcmod._join_pending_syncs() # must not raise + assert vcmod._pending_syncs == [] + + +def test_hydrate_noop_when_unavailable(tmp_path): + vc = VolumeCache( + [str(tmp_path / "cache")], namespace="ep1", volume_path=str(tmp_path / "missing") + ) + assert vc.hydrate() == 0 + + +def test_sync_noop_when_unavailable(tmp_path): + vc = VolumeCache( + [str(tmp_path / "cache")], namespace="ep1", volume_path=str(tmp_path / "missing") + ) + assert vc.sync(background=False) is None + + +# --------------------------------------------------------------------------- # +# exports +# --------------------------------------------------------------------------- # + + +def test_volumecache_exported_from_serverless(): + from runpod import serverless as sls + from runpod.serverless import utils as sls_utils + + assert sls.VolumeCache is sls_utils.VolumeCache + assert "VolumeCache" in sls.__all__ From 87438a61d82df0dd731ad229ce371b1cdae39bf5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Tue, 7 Jul 2026 15:43:40 -0700 Subject: [PATCH 18/20] fix(serverless): bound atexit sync join, unique temp names, exclude temp files --- runpod/serverless/utils/rp_volume_cache.py | 12 ++++-- .../test_utils/test_rp_volume_cache.py | 38 +++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index a1a2a829..27fe73df 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -24,6 +24,7 @@ log = RunPodLogger() _MTIME_TOLERANCE = 2.0 # seconds; tolerate coarse (NFS) mtime granularity +_JOIN_TIMEOUT_SECONDS = 30.0 # bound the atexit wait so stalled volume I/O can't hang shutdown class VolumeCache: @@ -88,7 +89,7 @@ def _iter_files(self, root): for dirpath, _dirs, files in os.walk(root): for name in files: path = os.path.join(dirpath, name) - if name.endswith(".lock") or name.startswith(".rpvc"): + if name.endswith(".lock") or name.endswith(".rpvc.tmp"): continue if any(sub in path for sub in self._EXCLUDE_SUBSTRINGS): continue @@ -116,7 +117,7 @@ def _is_safe_dest(self, dst_abs): return any(target == d or target.startswith(d + os.sep) for d in self._dirs) def _copy_file(self, src, dst): - tmp = dst + ".rpvc.tmp" + 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 @@ -233,7 +234,12 @@ def _register_pending(thread): def _join_pending_syncs(): for t in list(_pending_syncs): try: - t.join() + t.join(timeout=_JOIN_TIMEOUT_SECONDS) + if t.is_alive(): + log.warn( + "VolumeCache: background sync did not finish within " + f"{_JOIN_TIMEOUT_SECONDS:.0f}s at exit; cache may be incomplete" + ) except Exception: pass _pending_syncs.clear() 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 a2789a3e..352eec14 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1,5 +1,6 @@ import os import shutil +import threading import time import pytest @@ -353,3 +354,40 @@ def test_volumecache_exported_from_serverless(): assert sls.VolumeCache is sls_utils.VolumeCache assert "VolumeCache" in sls.__all__ + + +# --------------------------------------------------------------------------- # +# v2 review fixes +# --------------------------------------------------------------------------- # + + +def test_iter_files_skips_inflight_temp_files(tmp_path): + vc, cache, _vol = _mk_cache_with_volume(tmp_path) + (cache / "model.bin").write_text("real") + (cache / "model.bin.12345.67890.rpvc.tmp").write_text("partial") + copied = vc._do_sync() + assert copied == 1 # only model.bin, not the .rpvc.tmp file + tmp_mirror = os.path.join( + vc._mirror_root, os.path.relpath(str(cache / "model.bin.12345.67890.rpvc.tmp"), "/") + ) + assert not os.path.exists(tmp_mirror) + + +def test_join_pending_syncs_bounded_by_timeout(monkeypatch): + vcmod._reset_pending_for_test() + monkeypatch.setattr(vcmod, "_JOIN_TIMEOUT_SECONDS", 0.05) + started = threading.Event() + release = threading.Event() + + def slow(): + started.set() + release.wait(5) + + t = threading.Thread(target=slow, daemon=True) + vcmod._register_pending(t) + t.start() + started.wait(1) + # Must return promptly despite the thread still running (bounded join). + vcmod._join_pending_syncs() + release.set() + vcmod._reset_pending_for_test() From 9469d3a88a882408b33b374ae40b3dd1addc17e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Tue, 7 Jul 2026 18:30:48 -0700 Subject: [PATCH 19/20] fix(serverless): clear CodeQL alerts (empty-except, unreachable test, open() in assert, unused global) --- runpod/serverless/utils/rp_volume_cache.py | 10 +++++----- .../test_serverless/test_utils/test_rp_volume_cache.py | 10 ++++++++-- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index 27fe73df..6c6d1342 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -129,6 +129,7 @@ def _copy_file(self, src, dst): if os.path.exists(tmp): os.remove(tmp) except OSError: + # best-effort cleanup; a leftover temp file is overwritten next sync pass return False @@ -219,16 +220,11 @@ def __exit__(self, *exc): # ----------------------------------------------------------------------- # _pending_syncs = [] -_atexit_registered = False def _register_pending(thread): - global _atexit_registered _pending_syncs[:] = [t for t in _pending_syncs if t.is_alive()] _pending_syncs.append(thread) - if not _atexit_registered: - atexit.register(_join_pending_syncs) - _atexit_registered = True def _join_pending_syncs(): @@ -241,9 +237,13 @@ def _join_pending_syncs(): f"{_JOIN_TIMEOUT_SECONDS:.0f}s at exit; cache may be incomplete" ) except Exception: + # best-effort shutdown: a join failure must not block interpreter exit pass _pending_syncs.clear() def _reset_pending_for_test(): _pending_syncs.clear() + + +atexit.register(_join_pending_syncs) 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 352eec14..4c85bd74 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -120,7 +120,9 @@ def test_sync_recopies_modified_file(tmp_path): assert copied == 1 mirror_file = os.path.join(vc._mirror_root, os.path.relpath(str(f), "/")) - assert open(mirror_file).read() == "v2-longer" + with open(mirror_file) as fh: + content = fh.read() + assert content == "v2-longer" def test_hydrate_skips_unchanged_after_first_hydrate(tmp_path): @@ -225,9 +227,13 @@ def test_context_manager_does_not_suppress_exceptions(tmp_path): vc, cache, _vol = _mk_cache_with_volume(tmp_path) (cache / "model.bin").write_text("weights") - with pytest.raises(ValueError): + raised = False + try: with vc: raise ValueError("boom") + except ValueError: + raised = True + assert raised vcmod._join_pending_syncs() mirror_file = os.path.join(vc._mirror_root, os.path.relpath(str(cache / "model.bin"), "/")) From 351f35d924b2cc6bb11a40ea7e9c9631a94e05de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Tue, 7 Jul 2026 21:53:39 -0700 Subject: [PATCH 20/20] docs(serverless): add VolumeCache warm-cache handler example --- examples/serverless/volume_cache_handler.py | 48 +++++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 examples/serverless/volume_cache_handler.py diff --git a/examples/serverless/volume_cache_handler.py b/examples/serverless/volume_cache_handler.py new file mode 100644 index 00000000..a0b4e08c --- /dev/null +++ b/examples/serverless/volume_cache_handler.py @@ -0,0 +1,48 @@ +"""Warm-cache model weights across cold starts with VolumeCache. + +`VolumeCache` keeps a browsable mirror of local cache directories on a mounted +network volume and reconciles the two on each use: it restores cached files on +cold start (`hydrate`) and syncs new downloads back afterward (`sync`). Wrapping +a model load in `with VolumeCache(...)` turns a repeated multi-GB download into a +one-time cost per endpoint. + +Why not just point HF_HOME at the volume? Because then every read hits the +network mount. Here the Hugging Face cache stays on fast local disk and +VolumeCache mirrors it to the volume, so inference reads are local while cold +starts stay warm. + +Requirements to run: +- Attach a network volume to the endpoint (mounted at /runpod-volume). +- RUNPOD_ENDPOINT_ID is set automatically on Runpod serverless (it scopes the + cache per endpoint). Without a mounted volume or endpoint id, VolumeCache is a + safe no-op and the handler still works. +- pip install "transformers" "torch" (the model library used below). + +Local test: + python volume_cache_handler.py --rp_serve_api +""" + +import os + +import runpod +from runpod.serverless import VolumeCache + +# Hugging Face caches models here by default. Keep it on local disk and let +# VolumeCache mirror it to the network volume. +HF_CACHE = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface")) + +# Hydrate the cache from the volume before loading, then sync new files back on +# exit. On a warm endpoint the model is already local and nothing is downloaded. +with VolumeCache(dirs=[HF_CACHE]): + from transformers import pipeline + + classifier = pipeline("sentiment-analysis") + + +def handler(job): + """Classify the sentiment of an input string.""" + text = job["input"].get("text", "") + return {"input": text, "predictions": classifier(text)} + + +runpod.serverless.start({"handler": handler})