Files
local-ci-cd-system/src/ci_orchestrator/cargo_cache.py
T
Simone fce7cfd86d feat(cache): content-addressed store for cargo dependency artifacts
One immutable file per key, published by rename(2) inside the store. Two jobs
publishing the same key both produce a valid entry and the last wins; a reader
holding the previous one keeps reading it. No advisory locking, which is the
property a VMware shared folder cannot offer with several jobs in flight --
the objection raised in the furyrs thread.

The key joins its parts with NUL so ("ab","c") and ("a","bc") cannot collide,
and carries a format version so the whole store can be invalidated at once
when the prune rules change and older entries stop being trustworthy.

The feature set is deliberately absent from the key: cargo already encodes it
in each artifact's own hash suffix, so several feature graphs coexist in one
entry rather than fragmenting the store into one entry per variant.
2026-07-22 20:50:22 +02:00

122 lines
5.0 KiB
Python

"""Content-addressed store for compiled Rust dependency artifacts.
A cache entry is one immutable file per key. Publishing is a ``rename(2)``
within the store directory, so a reader either sees the previous entry or the
new one and never a half-written file — no advisory locking, which is exactly
what a VMware shared folder cannot give us with several jobs in flight.
Nothing here knows about SSH or guests: it takes a finished tarball and files
it, or hands back the path of an existing one. That keeps the interesting
logic testable without a VM.
"""
from __future__ import annotations
import hashlib
import os
import re
import shutil
from datetime import datetime
from pathlib import Path
# Bump to invalidate every entry at once — e.g. when the prune script changes
# what it strips, making older entries untrustworthy.
CACHE_FORMAT = 1
_SAFE_KEY = re.compile(r"^v\d+-[0-9a-f]{32,64}$")
_INCOMING = ".incoming"
def cache_key(*, lock_digest: str, rustc_version: str, host_triple: str) -> str:
"""Derive the entry key from the inputs that make artifacts incompatible.
``lock_digest`` is the sha256 of ``Cargo.lock``: it changes exactly when
the dependency set changes, which is the property that makes this cache
sound. The feature set is deliberately *not* part of the key — cargo
encodes it in each artifact's own hash suffix, so several feature graphs
coexist inside one entry instead of fragmenting it.
The parts are joined with NUL, which cannot occur in any of them, so
``("ab", "c")`` and ``("a", "bc")`` cannot collide.
"""
digest = hashlib.sha256(
"\0".join((lock_digest, rustc_version, host_triple)).encode("utf-8")
).hexdigest()
return f"v{CACHE_FORMAT}-{digest}"
class CargoCacheStore:
"""A directory of immutable cache entries, one file per key."""
def __init__(self, root: Path, *, suffix: str = ".tar.zst") -> None:
self.root = Path(root)
self.suffix = suffix
# ── paths ────────────────────────────────────────────────────────────
def _checked(self, key: str) -> str:
if not _SAFE_KEY.match(key):
raise ValueError(f"unsafe cache key: {key!r}")
return key
def path_for(self, key: str) -> Path:
"""Absolute path of the entry for ``key``. Raises on a malformed key."""
return self.root / f"{self._checked(key)}{self.suffix}"
# ── read ─────────────────────────────────────────────────────────────
def lookup(self, key: str) -> Path | None:
"""Return the entry path for ``key``, or ``None`` on a miss."""
candidate = self.path_for(key)
return candidate if candidate.is_file() else None
# ── write ────────────────────────────────────────────────────────────
def publish(self, key: str, source: Path) -> Path:
"""Move ``source`` into the store under ``key``. Returns the entry path.
The move goes through a staging directory on the same filesystem so the
final step is a rename: concurrent publishers of the same key both
produce a valid entry and the last one wins, with no window in which a
reader can observe a partial file.
"""
source = Path(source)
if not source.is_file():
raise FileNotFoundError(f"nothing to publish: {source}")
target = self.path_for(key)
staging = self.root / _INCOMING
staging.mkdir(parents=True, exist_ok=True)
tmp = staging / f"{key}.{os.getpid()}{self.suffix}"
try:
os.replace(source, tmp)
except OSError:
# Different filesystem — copy into staging instead, which still
# leaves the publish itself an atomic rename.
shutil.copy2(source, tmp)
source.unlink(missing_ok=True)
os.replace(tmp, target)
return target
# ── maintenance ──────────────────────────────────────────────────────
def prune_older_than(
self, cutoff: datetime, *, what_if: bool = False
) -> list[Path]:
"""Delete entries last modified before ``cutoff``. Returns what went."""
if not self.root.is_dir():
return []
removed: list[Path] = []
for entry in sorted(self.root.glob(f"*{self.suffix}")):
if not entry.is_file():
continue
mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=cutoff.tzinfo)
if mtime >= cutoff:
continue
removed.append(entry)
if not what_if:
entry.unlink(missing_ok=True)
return removed
__all__ = ["CACHE_FORMAT", "CargoCacheStore", "cache_key"]