From 6facb2e80888098b7d41f9cf890c78a24bcfe675 Mon Sep 17 00:00:00 2001 From: John Reeves Date: Tue, 29 Sep 2026 23:11:38 +1000 Subject: [PATCH] feat(cache): add 'dt cache relink' to symlink cache onto a local remote (0.28.0) Replace every regular file in the primary DVC cache with a symlink to its content-addressed twin in a locally-accessible remote, reclaiming the duplicated space. On a shared-filesystem HPC setup the local remote already holds a good copy of every pushed object, so the identical bytes in each clone's cache are pure duplication; the cache stays fully usable because 'dvc checkout' still resolves each object at its normal path. The match is content-addressed (a cache object and its remote copy share the md5 that names them), so the relink is existence-only and never re-hashes. Because that is only trustworthy on a known-good remote, 'dt remote verify' runs first by default (incremental via the remote's ledger) and the relink aborts if it finds bad/incomplete blobs; --force overrides, --skip-verify skips. Objects absent from the remote are left untouched; already-symlinked objects are skipped so re-runs are idempotent. Each replacement is atomic (temp symlink renamed over the object). Handles v3/v2/mixed layouts and .dir manifests, and only renames within existing prefix dirs so shared-cache directory permissions are never disturbed. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/cache.md | 44 +++++++ docs/commands.md | 1 + dt/__init__.py | 2 +- dt/cache_relink.py | 224 ++++++++++++++++++++++++++++++++ dt/cli.py | 61 +++++++++ pyproject.toml | 2 +- tests/unit/test_cache_relink.py | 202 ++++++++++++++++++++++++++++ 7 files changed, 534 insertions(+), 2 deletions(-) create mode 100644 dt/cache_relink.py create mode 100644 tests/unit/test_cache_relink.py diff --git a/docs/cache.md b/docs/cache.md index 6b77160..5a6980a 100644 --- a/docs/cache.md +++ b/docs/cache.md @@ -320,6 +320,50 @@ deliberately left in place. Worth running on a shared cache, where one user's interrupted pulls consume quota everyone shares. +## dt cache relink + +Replace every regular file in the primary cache with a symlink to its twin in +a locally-accessible remote, reclaiming the duplicated space. + +On a shared-filesystem HPC setup the local remote already holds a good copy of +every pushed object, so the identical bytes sitting in each clone's cache are +pure duplication. `dt cache relink` walks the cache and points each object at +the remote instead. The cache stays fully usable — `dvc checkout` still resolves +each object at its normal path, it is just a symlink now. + +The match is content-addressed: a cache object and its remote copy share the +md5 that names them, so the relink is **existence-only** and never re-hashes. +Because that is only trustworthy on a known-good remote, `dt cache relink` runs +[`dt remote verify`](remote.md#dt-remote-verify) first by default (incremental +via the remote's ledger, so repeat runs are cheap) and **aborts if it finds any +bad or incomplete blobs**. Fix those with +[`dt remote quarantine`](remote.md#dt-remote-quarantine) and re-push, or pass +`--force` to relink anyway. + +Objects present in the cache but **absent from the remote** (e.g. not yet +pushed) are left untouched — symlinking them would dangle. Objects that are +already symlinks are skipped, so re-running is safe and idempotent. + +```bash +dt cache relink # verify default remote, then relink +dt cache relink myremote # relink onto a named remote +dt cache relink --dry # preview + count reclaimable space +dt cache relink --skip-verify # skip verify (already verified) +dt cache relink --force # relink despite bad blobs +dt cache relink -v # list every object touched or missing +``` + +| Option | Description | +|--------|-------------| +| `--skip-verify` | Skip the integrity pass. Only safe if you have already run `dt remote verify` on this remote. | +| `--force` | Relink even if verification found bad/incomplete blobs. | +| `-j, --jobs N` | Hashing threads for the verify pass. | +| `--dry` | Report what would be relinked without touching the cache. | +| `-v, --verbose` | List every object relinked or missing from the remote. | + +Each replacement is atomic (a temp symlink renamed over the object), so an +interrupted run never leaves a half-written cache object behind. + ## Related Commands - [`dt init`](init.md) - Initialize projects with cache setup diff --git a/docs/commands.md b/docs/commands.md index 528c393..a25861f 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -265,6 +265,7 @@ Manage external shared caches. | `dt cache clean` | Remove abandoned .tmp files from interrupted transfers | | `dt cache perms` | Check/repair group-writable permissions on a shared cache | | `dt cache validate` | Verify cached blobs against the MD5 implied by their path | +| `dt cache relink` | Replace cache files with symlinks to a verified local remote | [Full documentation →](cache.md) diff --git a/dt/__init__.py b/dt/__init__.py index 8934495..41b74cf 100644 --- a/dt/__init__.py +++ b/dt/__init__.py @@ -1,3 +1,3 @@ """DVC Tools - Convenient tools for working with DVC in HPC environments.""" -__version__ = "0.27.1" +__version__ = "0.28.0" diff --git a/dt/cache_relink.py b/dt/cache_relink.py new file mode 100644 index 0000000..89a49e4 --- /dev/null +++ b/dt/cache_relink.py @@ -0,0 +1,224 @@ +"""Relink the local DVC cache onto a local remote. + +Replaces every regular file in the primary DVC cache with a symlink to the +corresponding content-addressed object in a locally-accessible remote. On a +shared-filesystem HPC setup the remote already holds a good copy of every +pushed object, so keeping a second physical copy in each clone's cache is pure +duplication. Relinking reclaims that space while leaving the cache fully +functional: ``dvc checkout`` still finds each object at its usual cache path, +it is just a symlink now. + +The mapping is trivial because both stores are content-addressed by the same +md5: a cache blob and its remote twin share the md5 that names them, so the +relink is *existence-only* -- we never re-hash. That is only safe if the remote +is known good, which is why :func:`relink_cache` runs ``dt remote verify`` +first by default (incremental via the remote's ledger, so repeat runs are +cheap) and refuses to relink onto a remote with bad blobs unless forced. + +Objects present in the cache but absent from the remote are left untouched: +they may not have been pushed yet, and symlinking them would dangle. +""" + +import os +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterator, List, Optional, Tuple + +from . import remote as remote_mod +from . import remote_verify as remote_verify_mod +from . import utils +from .errors import RemoteError + + +@dataclass +class RelinkStats: + """Outcome of a relink sweep.""" + + relinked: int = 0 + already_linked: int = 0 + missing: int = 0 + failed: int = 0 + bytes_reclaimed: int = 0 + missing_md5s: List[str] = field(default_factory=list) + failures: List[Tuple[str, str]] = field(default_factory=list) + + +def _iter_cache_blobs(cache_root: Path, layout: str) -> Iterator[Tuple[str, Path]]: + """Yield ``(md5, blob_path)`` for every object file in the cache. + + ``md5`` keeps a trailing ``.dir`` for directory manifests. Prefix + directories (the two-char ``ab/`` level) are skipped; only the object + files inside them are yielded. + """ + from .archive import operations as ops + + for _key, hex_prefix, prefix_dir in remote_verify_mod._prefix_dirs( + cache_root, layout): + try: + names = sorted(os.listdir(prefix_dir)) + except OSError: + continue + for name in names: + blob = prefix_dir / name + # Object files only; ignore stray subdirs / ledger dirs. + if not blob.is_file() and not os.path.islink(blob): + continue + yield hex_prefix + name, blob + + +def _remote_blob_path(remote_root: Path, remote_layout: str, + md5: str) -> Optional[Path]: + """Path of the object named ``md5`` in the remote, or None if absent. + + Checks the layout(s) the remote actually uses. For a mixed remote an + object may live under either ``files/md5/`` (v3) or the flat ``ab/`` + (v2) tree, so both are probed. + """ + from .archive import operations as ops + + prefix, rest = md5[:2], md5[2:] + candidates: List[Path] = [] + if remote_layout in (ops.LAYOUT_DVC_V3, ops.LAYOUT_DVC_MIXED): + candidates.append(remote_root / 'files' / 'md5' / prefix / rest) + if remote_layout in (ops.LAYOUT_DVC_V2, ops.LAYOUT_DVC_MIXED): + candidates.append(remote_root / prefix / rest) + for cand in candidates: + if cand.exists(): + return cand + return None + + +def _relink_one(blob: Path, target: Path) -> int: + """Replace regular file ``blob`` with a symlink to ``target``. + + Returns the number of bytes freed (the size of the file that was + replaced). Written atomically via a temp symlink + rename so an + interrupted run never leaves a half-written cache object. Raises OSError + on failure. + """ + reclaimed = blob.stat().st_size # follows to the regular file itself + tmp = blob.parent / (blob.name + '.dt-relink.tmp') + try: + os.unlink(tmp) + except OSError: + pass + os.symlink(os.path.abspath(target), tmp) + os.replace(tmp, blob) # atomic overwrite of the old regular file + return reclaimed + + +def relink_cache( + remote_name: Optional[str] = None, + *, + skip_verify: bool = False, + force: bool = False, + jobs: Optional[int] = None, + dry_run: bool = False, + verbose: bool = False, +) -> RelinkStats: + """Relink every cache object onto its twin in a local remote. + + Args: + remote_name: DVC remote to relink onto (default: the default remote). + skip_verify: Skip the ``dt remote verify`` integrity pass. + force: Relink even if verification found bad blobs. + jobs: Hashing threads for the verify pass. + dry_run: Report what would change without touching the cache. + verbose: List every missing/failed object instead of just counting. + + Returns: + A :class:`RelinkStats` describing the sweep. + + Raises: + RemoteError: if no local remote exists or verification fails and the + caller did not pass ``force``. + """ + from .archive import operations as ops + + name, url, remote_root = remote_verify_mod.resolve_local_remote(remote_name) + print(f"Local remote: {name} -> {remote_root}") + + if not skip_verify: + print(f"Verifying remote '{name}' (incremental)...") + totals, bad, incomplete, _layout = remote_verify_mod.verify_remote( + remote_root, jobs=jobs, progress=verbose) + n_bad = len(bad) + len(incomplete) + if n_bad: + msg = (f"Remote '{name}' has {n_bad} bad/incomplete blob(s); " + f"relinking onto it would point the cache at corrupt data. " + f"Fix with 'dt remote quarantine' then re-push, or pass " + f"--force to relink anyway.") + if not force: + raise RemoteError(msg) + print(f"WARNING: {msg}", file=sys.stderr) + else: + print(f" verify OK: {totals.get('objects', 0)} object(s) checked") + + cache_root = utils.get_cache_root() + if cache_root is None or not Path(cache_root).exists(): + raise RemoteError("No local DVC cache found (not in a DVC repo?).") + cache_root = Path(cache_root) + + try: + cache_layout = ops.detect_source_layout(cache_root) + except Exception as e: + raise RemoteError(f"Could not read cache layout at {cache_root}: {e}") + try: + remote_layout = ops.detect_source_layout(remote_root) + except Exception: + # An empty remote (no blobs yet) has no detectable layout. Fall back + # to probing both trees; every cache object then reports as missing. + remote_layout = ops.LAYOUT_DVC_MIXED + + stats = RelinkStats() + action = "Would relink" if dry_run else "Relinking" + print(f"{action} cache at {cache_root} (layout: {cache_layout})") + + for md5, blob in _iter_cache_blobs(cache_root, cache_layout): + if os.path.islink(blob): + stats.already_linked += 1 + continue + target = _remote_blob_path(remote_root, remote_layout, md5) + if target is None: + stats.missing += 1 + if verbose: + stats.missing_md5s.append(md5) + print(f" missing in remote: {md5}") + continue + if dry_run: + stats.relinked += 1 + try: + stats.bytes_reclaimed += blob.stat().st_size + except OSError: + pass + if verbose: + print(f" would relink: {md5}") + continue + try: + stats.bytes_reclaimed += _relink_one(blob, target) + stats.relinked += 1 + if verbose: + print(f" relinked: {md5}") + except OSError as e: + stats.failed += 1 + stats.failures.append((md5, str(e))) + print(f" FAILED {md5}: {e}", file=sys.stderr) + + return stats + + +def format_stats(stats: RelinkStats, dry_run: bool = False) -> str: + """One-block human summary of a relink sweep.""" + verb = "would be relinked" if dry_run else "relinked" + reclaim = "would reclaim" if dry_run else "reclaimed" + lines = [ + "", + f" {stats.relinked} object(s) {verb}", + f" {stats.already_linked} already symlinked (skipped)", + f" {stats.missing} not in remote (left as-is)", + ] + if stats.failed: + lines.append(f" {stats.failed} failed") + lines.append(f" {reclaim} {utils.format_size(stats.bytes_reclaimed)}") + return "\n".join(lines) diff --git a/dt/cli.py b/dt/cli.py index 570f409..4e3a718 100644 --- a/dt/cli.py +++ b/dt/cli.py @@ -1116,6 +1116,67 @@ def cache_perms(cache_path, do_fix, sticky, allow_other, jobs, json_output, ) +@cache.command('relink') +@click.argument('remote_name', required=False) +@click.option('--skip-verify', is_flag=True, + help='Skip the integrity pass. Only safe if you have already ' + 'run "dt remote verify" on this remote.') +@click.option('--force', is_flag=True, + help='Relink even if verification found bad/incomplete blobs.') +@click.option('--jobs', '-j', type=int, default=None, + help='Hashing threads for the verify pass.') +@click.option('--dry', 'dry_run', is_flag=True, + help='Report what would be relinked without touching the cache.') +@click.option('-v', '--verbose', is_flag=True, + help='List every object relinked or missing from the remote.') +def cache_relink(remote_name, skip_verify, force, jobs, dry_run, verbose): + """Replace cache objects with symlinks to a local remote. + + On a shared-filesystem HPC setup the local remote already holds a good + copy of every pushed object, so the identical bytes sitting in each clone's + cache are pure duplication. This walks the primary cache and replaces every + regular file with a symlink to its twin in the remote, reclaiming that space + while keeping the cache fully usable (`dvc checkout` still resolves each + object at its normal path). + + The match is content-addressed: a cache object and its remote copy share + the md5 that names them, so the relink is existence-only and never re-hashes. + That is only trustworthy on a known-good remote, so `dt remote verify` runs + first by default (incremental via the remote's ledger, so repeat runs are + cheap) and the relink aborts if it finds bad blobs. + + Objects not present in the remote (e.g. not yet pushed) are left untouched; + pass -v to list them. Objects already symlinked are skipped, so re-running + is safe. + + \b + Examples: + dt cache relink # verify default remote, then relink + dt cache relink myremote # relink onto a named remote + dt cache relink --dry # preview + count reclaimable space + dt cache relink --skip-verify # skip verify (already verified) + dt cache relink --force # relink despite bad blobs + dt cache relink -v # list every object touched/missing + """ + from . import cache_relink as relink_mod + + try: + stats = relink_mod.relink_cache( + remote_name, + skip_verify=skip_verify, + force=force, + jobs=jobs, + dry_run=dry_run, + verbose=verbose, + ) + except relink_mod.RemoteError as e: + raise click.ClickException(str(e)) + + click.echo(relink_mod.format_stats(stats, dry_run=dry_run)) + if stats.failed: + raise SystemExit(1) + + @cli.group() def remote(): """Manage remote storage.""" diff --git a/pyproject.toml b/pyproject.toml index e1e9460..fa4b687 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "dvc-tools" -version = "0.27.1" +version = "0.28.0" description = "Convenient tools for working with DVC in HPC environments" readme = "README.md" requires-python = ">=3.8" diff --git a/tests/unit/test_cache_relink.py b/tests/unit/test_cache_relink.py new file mode 100644 index 0000000..0e6c622 --- /dev/null +++ b/tests/unit/test_cache_relink.py @@ -0,0 +1,202 @@ +"""Unit tests for dt.cache_relink module.""" + +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from dt import cache_relink as cr +from dt.errors import RemoteError +from dt.utils import md5_file + + +def _place_v3_blob(root: Path, content: bytes, name: str = None) -> str: + """Write a v3-layout blob under ``root`` and return its md5 (or ``name``).""" + root.mkdir(parents=True, exist_ok=True) + scratch = root / "_scratch" + scratch.write_bytes(content) + h = md5_file(scratch) + scratch.unlink() + key = name if name is not None else h + d = root / "files" / "md5" / key[:2] + d.mkdir(parents=True, exist_ok=True) + (d / key[2:]).write_bytes(content) + return key + + +# ============================================================================= +# _remote_blob_path +# ============================================================================= + +class TestRemoteBlobPath: + def test_v3_hit(self, tmp_path): + remote = tmp_path / "remote" + md5 = _place_v3_blob(remote, b"hello") + got = cr._remote_blob_path(remote, "dvc-v3", md5) + assert got == remote / "files" / "md5" / md5[:2] / md5[2:] + + def test_v3_miss(self, tmp_path): + remote = tmp_path / "remote" + (remote / "files" / "md5").mkdir(parents=True) + assert cr._remote_blob_path(remote, "dvc-v3", "ab" + "0" * 30) is None + + def test_v2_hit(self, tmp_path): + remote = tmp_path / "remote" + md5 = "ab" + "1" * 30 + d = remote / md5[:2] + d.mkdir(parents=True) + (d / md5[2:]).write_bytes(b"x") + got = cr._remote_blob_path(remote, "dvc-v2", md5) + assert got == d / md5[2:] + + def test_mixed_checks_both(self, tmp_path): + remote = tmp_path / "remote" + # object only present in the v2 (flat) tree + md5 = "cd" + "2" * 30 + d = remote / md5[:2] + d.mkdir(parents=True) + (d / md5[2:]).write_bytes(b"y") + got = cr._remote_blob_path(remote, "dvc-mixed", md5) + assert got == d / md5[2:] + + +# ============================================================================= +# _relink_one +# ============================================================================= + +class TestRelinkOne: + def test_replaces_file_with_symlink(self, tmp_path): + target = tmp_path / "remote_obj" + target.write_bytes(b"payload-1234") + blob = tmp_path / "cache_obj" + blob.write_bytes(b"payload-1234") + + freed = cr._relink_one(blob, target) + + assert freed == len(b"payload-1234") + assert os.path.islink(blob) + assert Path(os.readlink(blob)) == target.resolve() + assert blob.read_bytes() == b"payload-1234" + # no temp file left behind + assert not (tmp_path / "cache_obj.dt-relink.tmp").exists() + + def test_atomic_over_readonly_blob(self, tmp_path): + target = tmp_path / "remote_obj" + target.write_bytes(b"z") + blob = tmp_path / "cache_obj" + blob.write_bytes(b"z") + os.chmod(blob, 0o444) # DVC keeps cache objects read-only + cr._relink_one(blob, target) + assert os.path.islink(blob) + + +# ============================================================================= +# relink_cache end-to-end (skip_verify) +# ============================================================================= + +class TestRelinkCache: + def _setup(self, tmp_path): + cache = tmp_path / "cache" + remote = tmp_path / "remote" + return cache, remote + + def _run(self, cache, remote, **kw): + with patch.object(cr.remote_verify_mod, "resolve_local_remote", + return_value=("local", str(remote), remote)), \ + patch.object(cr.utils, "get_cache_root", return_value=cache): + return cr.relink_cache(skip_verify=True, **kw) + + def test_relinks_present_objects(self, tmp_path): + cache, remote = self._setup(tmp_path) + md5 = _place_v3_blob(cache, b"shared-bytes") + _place_v3_blob(remote, b"shared-bytes") + + stats = self._run(cache, remote) + + assert stats.relinked == 1 + assert stats.missing == 0 + assert stats.bytes_reclaimed == len(b"shared-bytes") + blob = cache / "files" / "md5" / md5[:2] / md5[2:] + assert os.path.islink(blob) + + def test_missing_left_untouched(self, tmp_path): + cache, remote = self._setup(tmp_path) + md5 = _place_v3_blob(cache, b"only-in-cache") + (remote / "files" / "md5").mkdir(parents=True) + + stats = self._run(cache, remote, verbose=True) + + assert stats.relinked == 0 + assert stats.missing == 1 + assert md5 in stats.missing_md5s + blob = cache / "files" / "md5" / md5[:2] / md5[2:] + assert blob.is_file() and not os.path.islink(blob) + + def test_idempotent(self, tmp_path): + cache, remote = self._setup(tmp_path) + _place_v3_blob(cache, b"dup") + _place_v3_blob(remote, b"dup") + + first = self._run(cache, remote) + second = self._run(cache, remote) + + assert first.relinked == 1 + assert second.relinked == 0 + assert second.already_linked == 1 + + def test_dry_run_makes_no_changes(self, tmp_path): + cache, remote = self._setup(tmp_path) + md5 = _place_v3_blob(cache, b"preview") + _place_v3_blob(remote, b"preview") + + stats = self._run(cache, remote, dry_run=True) + + assert stats.relinked == 1 + assert stats.bytes_reclaimed == len(b"preview") + blob = cache / "files" / "md5" / md5[:2] / md5[2:] + assert blob.is_file() and not os.path.islink(blob) + + def test_dir_manifest_relinked(self, tmp_path): + cache, remote = self._setup(tmp_path) + dir_md5 = "ab" + "0" * 30 + ".dir" + _place_v3_blob(cache, b'[{"md5":"x"}]', name=dir_md5) + _place_v3_blob(remote, b'[{"md5":"x"}]', name=dir_md5) + + stats = self._run(cache, remote) + + assert stats.relinked == 1 + blob = cache / "files" / "md5" / dir_md5[:2] / dir_md5[2:] + assert os.path.islink(blob) + + +class TestVerifyGate: + def test_aborts_on_bad_blobs(self, tmp_path): + cache = tmp_path / "cache" + remote = tmp_path / "remote" + (remote / "files" / "md5").mkdir(parents=True) + with patch.object(cr.remote_verify_mod, "resolve_local_remote", + return_value=("local", str(remote), remote)), \ + patch.object(cr.remote_verify_mod, "verify_remote", + return_value=({"objects": 1}, + [{"path": "files/md5/ab/c"}], [], + "dvc-v3")): + with pytest.raises(RemoteError, match="bad/incomplete"): + cr.relink_cache() + + def test_force_proceeds_despite_bad_blobs(self, tmp_path): + cache = tmp_path / "cache" + remote = tmp_path / "remote" + md5 = _place_v3_blob(cache, b"data") + _place_v3_blob(remote, b"data") + with patch.object(cr.remote_verify_mod, "resolve_local_remote", + return_value=("local", str(remote), remote)), \ + patch.object(cr.remote_verify_mod, "verify_remote", + return_value=({"objects": 1}, + [{"path": "files/md5/ab/c"}], [], + "dvc-v3")), \ + patch.object(cr.utils, "get_cache_root", return_value=cache): + stats = cr.relink_cache(force=True) + assert stats.relinked == 1 + blob = cache / "files" / "md5" / md5[:2] / md5[2:] + assert os.path.islink(blob)