gw#597+gw#598: multi-writer CAS safety + digest-verified reuse for volume-backed roots

src/gen_worker/executor.py

@@ -911,14 +911,21 @@ def component_sizes(self, ref: str) -> Dict[str, int]:
 
     def rescan_disk(self) -> None:
         """Boot-time truth: re-register still-present downloads from the
-        persisted ref index so Hello.models and GC see what disk holds."""
+        persisted ref index so Hello.models and GC see what disk holds.
+
+        Also sweeps abandoned writer-unique CAS temp artifacts (th#850): on
+        pod-local disk those died with the pod, but a CAS root pointed at a
+        persistent volume keeps them until swept."""
         for ref, ent in self._index.entries().items():
             p = Path(str(ent.get("path") or ""))
             if p.exists():
                 if self.residency.tier(ref) is None:
                     self.residency.track_disk(ref, p)
             else:
                 self._index.remove(ref)
+        removed = disk_gc.sweep_stale_writer_temp(self._cache_dir)
+        if removed:
+            logger.info("disk-gc: swept %d abandoned writer temp artifact(s)", removed)
 
     def lru_disk_refs(self, *, exclude: Tuple[str, ...] = ()) -> List[str]:
         """Idle DISK refs in persisted last-use order, oldest first."""

src/gen_worker/models/cache_paths.py

@@ -1,27 +1,26 @@
 from __future__ import annotations
 
-import os
 from pathlib import Path
 
 from ..config import get_settings
 
 
 TENSORHUB_CACHE_DIR = "/tmp/tensorhub-cache"
 
-# RunPod network volumes are attached at pod creation. Tensorhub reserves this
-# mount solely for an endpoint-owned immutable CAS tier; ordinary pod volumes
-# continue using their existing path. This is a provider contract, not an
-# operator-tunable cache knob.
-ENDPOINT_SHARED_CACHE_MOUNT = Path("/tensorhub-endpoint-cache")
-
 
 def tensorhub_cache_dir() -> Path:
-    """TensorHub cache root directory.
-
-    Honors the ``TENSORHUB_CACHE_DIR`` environment variable when set, so the
-    cozy local runner can point the CAS at a persistent ``~/.cache/tensorhub``
-    (weights survive reboots) instead of ``/tmp``. Falls back to the ``/tmp``
-    default when unset, preserving existing worker/orchestrator behavior.
+    """TensorHub cache root directory — the worker's CAS root.
+
+    Honors the ``TENSORHUB_CACHE_DIR`` environment variable when set. This is
+    the ONE knob for where the CAS lives: the cozy-local runner points it at
+    a persistent ``~/.cache/tensorhub`` (weights survive reboots); tensorhub
+    points it at a mounted RunPod endpoint volume (`/tensorhub-endpoint-cache`)
+    when the pod has one attached, so the ordinary R2 prefetch that
+    materializes blobs here on the first boot warms the volume for later
+    same-endpoint pods too (th#850) — no separate shared/read-through tier.
+    Falls back to the ``/tmp`` default when unset. The CAS implementation
+    itself is deliberately backend-agnostic: nothing branches on what's
+    mounted here.
     """
     env = get_settings().tensorhub_cache_dir.strip()
     if env:
@@ -32,14 +31,3 @@ def tensorhub_cache_dir() -> Path:
 def tensorhub_cas_dir() -> Path:
     """Worker CAS root: <TENSORHUB_CACHE_DIR>/cas."""
     return tensorhub_cache_dir() / "cas"
-
-
-def endpoint_shared_cas_dir() -> Path | None:
-    """Return the endpoint-owned shared CAS only when it is really mounted.
-
-    A plain directory in an image or on the container disk must never be
-    mistaken for the endpoint-isolated provider volume.
-    """
-    if os.path.ismount(ENDPOINT_SHARED_CACHE_MOUNT):
-        return ENDPOINT_SHARED_CACHE_MOUNT / "tensorhub-cas-v1"
-    return None

src/gen_worker/models/cozy_cas.py

@@ -3,8 +3,11 @@
 import asyncio
 import errno
 import logging
+import os
 import random
+import threading
 import time
+import uuid
 from pathlib import Path
 from typing import Callable, Dict, Optional
 
@@ -56,12 +59,20 @@ async def _download_one_file(
     started = time.monotonic()
     transient_failures = 0
     verify_failures = 0
+    # Writer-unique for the lifetime of this call (stable across its own
+    # retries, so HTTP Range resume-on-retry still works) but distinct from
+    # every OTHER writer of this digest — including a different PROCESS.
+    # dst's CAS root may now be a network volume shared by concurrent
+    # same-endpoint pods; a fixed ".part" name would let two pods racing on
+    # the same missing blob interleave writes into the same file (th#850).
+    writer_id = f"{os.getpid()}-{threading.get_ident()}-{uuid.uuid4().hex}"
     while True:
         try:
             await loop.run_in_executor(
                 None,
                 lambda: _download_one_file_sync(
-                    url, dst, expected_size, expected_blake3, on_bytes=on_bytes
+                    url, dst, expected_size, expected_blake3,
+                    on_bytes=on_bytes, writer_id=writer_id,
                 ),
             )
             return
@@ -92,14 +103,19 @@ def _download_one_file_sync(
     expected_size: int,
     expected_blake3: str,
     on_bytes: Optional[Callable[[int], None]] = None,
+    writer_id: Optional[str] = None,
 ) -> None:
     """Download a single file with HTTP Range resume, size + blake3 validation.
 
     Caller runs this in a worker thread; the transfer itself uses the same
     requests stack as the Tensorhub control plane fallback path.
-    """
-    import logging
 
+    ``writer_id`` must be unique per concurrent writer (distinct calls, and
+    critically distinct PROCESSES when ``dst``'s CAS root is a network volume
+    shared by several pods) so two writers of the same missing blob never
+    stage into the same temp path. Falls back to a fresh one-off id when not
+    supplied, which loses cross-retry resume but is still collision-safe.
+    """
     log = logging.getLogger("gen_worker.download")
 
     def _human_size(n: int) -> str:
@@ -125,7 +141,9 @@ def _human_size(n: int) -> str:
         except Exception:
             pass  # re-download
 
-    tmp = dst.with_suffix(dst.suffix + ".part")
+    if not writer_id:
+        writer_id = f"{os.getpid()}-{threading.get_ident()}-{uuid.uuid4().hex}"
+    tmp = dst.parent / f".{dst.name}.part-{writer_id}"
 
     # Resume from partial download if available.
     offset = 0