diff --git a/build_system/builder/cache/inventory.py b/build_system/builder/cache/inventory.py index 3c0e93962..da0504854 100644 --- a/build_system/builder/cache/inventory.py +++ b/build_system/builder/cache/inventory.py @@ -9,7 +9,7 @@ from .cargounits import unaccounted_size, unit_entries from .contract import PruneStrategy from .inventorymodels import RetentionInventory -from .leases import active_path +from .leases import active_path, lease_key from .measure import measure from .models import CacheEntry, CacheInventory, CachePolicy, StageInventory from .objectunits import object_entries @@ -25,6 +25,23 @@ def _lease_active(stage_path: Path, template: str | None, key: str) -> bool: return active_path(lease) +def _managed(stage_policy, name: str) -> bool: + return any(fnmatch.fnmatchcase(name, pattern) for pattern in stage_policy.managed_globs) + + +def _lease_files(template: str | None, children: list[Path], stage_policy) -> dict[str, str]: + """Map each lease file's name to the generation key it holds.""" + if template is None: + return {} + found = {} + for child in children: + key = lease_key(template, child.name) + if key is not None and not _managed(stage_policy, child.name) and not child.is_symlink() \ + and child.is_file(): + found[child.name] = key + return found + + def _entry_size(path: Path, allocated_seen: set[tuple[int, int]]) -> tuple[int, int]: measured = measure(path, allocated_seen) return measured.logical_bytes, measured.allocated_bytes @@ -54,25 +71,43 @@ def _stage_inventory( unmanaged_allocated = 0 busy = any(active_path(stage_root / lock) for lock in stage_policy.mutation_locks) if stage_path.is_dir(): - for child in sorted(stage_path.iterdir(), key=lambda item: item.name): + children = sorted(stage_path.iterdir(), key=lambda item: item.name) + names = {child.name for child in children} + managed_names = {name for name in names if _managed(stage_policy, name)} + leases = _lease_files(stage_policy.lease_template, children, stage_policy) + for child in children: + key = leases.get(child.name) + if key is not None and key in managed_names: + continue # removed and accounted with its generation + if key in names: + key = None # the lease of an unmanaged sibling stays unmanaged logical, allocated = _entry_size(child, allocated_seen) stat = child.lstat() - managed = any( - fnmatch.fnmatchcase(child.name, pattern) for pattern in stage_policy.managed_globs - ) + lease_only = key is not None + managed = lease_only or child.name in managed_names + members: tuple[Path, ...] = () + if managed and not lease_only and stage_policy.lease_template is not None: + lease = stage_policy.lease_template.format(key=child.name) + if lease in leases: + members = (Path(lease),) + lease_logical, lease_allocated = _entry_size(stage_path / lease, allocated_seen) + logical += lease_logical + allocated += lease_allocated entries.append( CacheEntry( - key=child.name, + key=key or child.name, relative_path=Path(child.name), + member_paths=members, logical_bytes=logical, allocated_bytes=allocated, created_ns=stat.st_ctime_ns, last_used_ns=stat.st_atime_ns, managed=managed, + lease_only=lease_only, protected=managed and ( - busy or child.name in referenced - or _lease_active(stage_path, stage_policy.lease_template, child.name) + busy or (key or child.name) in referenced + or _lease_active(stage_path, stage_policy.lease_template, key or child.name) ), ) ) diff --git a/build_system/builder/cache/leases.py b/build_system/builder/cache/leases.py index 49e56d60b..a65ee5c57 100644 --- a/build_system/builder/cache/leases.py +++ b/build_system/builder/cache/leases.py @@ -26,22 +26,72 @@ def retain_path(path: Path) -> BinaryIO: - """Hold one shared lease until explicit release or process exit.""" + """Hold one shared lease until explicit release or process exit. + + The lock waits out an exclusive holder: a pruner holds one only while it + checks or removes a dead generation. A lease file the pruner unlinked + meanwhile is a lock nobody can see, so it is retaken on the live path. + """ lease = path.absolute() existing = _HELD.get(lease) if existing is not None and not existing.closed: return existing lease.parent.mkdir(parents=True, exist_ok=True) - descriptor = os.fdopen(os.open(lease, os.O_APPEND | os.O_CREAT | os.O_RDWR, 0o600), "a+b") - try: - fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB) - except BaseException: + while True: + descriptor = os.fdopen( + os.open(lease, os.O_APPEND | os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600), "a+b" + ) + try: + fcntl.flock(descriptor, fcntl.LOCK_SH) + if _same_file(descriptor, lease): + break + except BaseException: + descriptor.close() + raise descriptor.close() - raise _HELD[lease] = descriptor return descriptor +def _same_file(descriptor: BinaryIO, path: Path) -> bool: + try: + current = path.lstat() + except FileNotFoundError: + return False + held = os.fstat(descriptor.fileno()) + return (held.st_dev, held.st_ino) == (current.st_dev, current.st_ino) + + +def lease_key(template: str, name: str) -> str | None: + """The generation key a lease file name encodes under `template`, if any.""" + prefix, suffix = template.split("{key}") + if len(name) <= len(prefix) + len(suffix): + return None + if not (name.startswith(prefix) and name.endswith(suffix)): + return None + return name[len(prefix):len(name) - len(suffix)] + + +@contextmanager +def exclusive_path(path: Path) -> Iterator[bool]: + """Hold a lease exclusively while its generation is removed. + + Yields False when an owner holds it. Creating the file when it is missing + is deliberate: an owner leases before it creates its generation, so a + concurrent owner then waits on this inode and retakes the live path. + """ + descriptor = os.fdopen( + os.open(path, os.O_APPEND | os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600), "a+b" + ) + with descriptor: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + yield False + return + yield True + + def retain_generation(paths: CachePaths, stage_id: str, key: str) -> BinaryIO: """Hold the configured lease for one managed stage generation.""" stage = paths.policy.stages[stage_id] diff --git a/build_system/builder/cache/models.py b/build_system/builder/cache/models.py index 306d71c9e..3b035e033 100644 --- a/build_system/builder/cache/models.py +++ b/build_system/builder/cache/models.py @@ -160,6 +160,8 @@ class CacheEntry(BaseModel): last_used_ns: Annotated[StrictInt, Field(ge=0)] managed: StrictBool = True protected: StrictBool = False + #: A lease file whose generation is gone: collected, never counted. + lease_only: StrictBool = False class StageInventory(BaseModel): @@ -223,6 +225,8 @@ class ApplyResult(BaseModel): removed: tuple[Path, ...] missing: tuple[Path, ...] + #: Planned paths kept because an owner leased their generation meanwhile. + busy: tuple[Path, ...] = () journal: Path diff --git a/build_system/builder/cache/operations.py b/build_system/builder/cache/operations.py index 403e8f356..ada94beff 100644 --- a/build_system/builder/cache/operations.py +++ b/build_system/builder/cache/operations.py @@ -8,12 +8,14 @@ import stat import time from collections.abc import Iterator +from contextlib import ExitStack from pathlib import Path from pydantic import ValidationError -from .leases import mutation_locks -from .models import AdmissionEvent, ApplyResult, PrunePlan +from .leases import exclusive_path, mutation_locks, release_path +from .measure import measure +from .models import AdmissionEvent, ApplyResult, PruneAction, PrunePlan from .paths import CachePaths JOURNAL_PATH = Path("state/events/cache.jsonl") @@ -72,16 +74,28 @@ def apply_prune(paths: CachePaths, plan: PrunePlan, *, reason: str) -> ApplyResu if not reason.strip(): raise ValueError("cache mutation reason must be non-empty") targets = tuple(paths.contained_entry(action.stage_id, action.path) for action in plan.actions) + generations: dict[tuple[str, str], list[Path]] = {} + for action, target in zip(plan.actions, targets, strict=True): + generations.setdefault((action.stage_id, action.key), []).append(target) removed: list[Path] = [] missing: list[Path] = [] + busy: list[Path] = [] with mutation_locks(paths, (action.stage_id for action in plan.actions)) as locks: - for selected in targets: - for target in _unlocked_targets(selected, locks): - if target.exists() or target.is_symlink(): - _remove(target) - removed.append(target) - else: - missing.append(target) + for (stage_id, key), selected in generations.items(): + template = paths.policy.stages[stage_id].lease_template + root = paths.stage(stage_id) + lease = None if template is None or not root.is_dir() else root / template.format(key=key) + with ExitStack() as stack: + if lease is not None and not stack.enter_context(exclusive_path(lease)): + busy.extend(selected) + continue + for target in (*selected, *(() if lease is None else (lease,))): + for unlocked in _unlocked_targets(target, locks): + if unlocked.exists() or unlocked.is_symlink(): + _remove(unlocked) + removed.append(unlocked) + elif unlocked != lease: + missing.append(unlocked) journal = paths.root / JOURNAL_PATH journal.parent.mkdir(parents=True, exist_ok=True) event = { @@ -91,11 +105,37 @@ def apply_prune(paths: CachePaths, plan: PrunePlan, *, reason: str) -> ApplyResu "reason": reason, "removed": [str(path) for path in removed], "missing": [str(path) for path in missing], + "busy": [str(path) for path in busy], "reclaim_bytes": plan.reclaim_bytes, } with journal.open("a", encoding="utf-8") as stream: stream.write(json.dumps(event, sort_keys=True) + "\n") - return ApplyResult(removed=tuple(removed), missing=tuple(missing), journal=journal) + return ApplyResult( + removed=tuple(removed), missing=tuple(missing), busy=tuple(busy), journal=journal + ) + + +def reclaim_generation(paths: CachePaths, stage_id: str, key: str, *, reason: str) -> ApplyResult: + """End one generation this process leased: release the lease, then remove + the generation and its lease through the same guarded, journaled path a + prune takes. A concurrent prune that wins the lease first does the same + removal; whichever loses sees it busy or already gone.""" + stage = paths.policy.stages[stage_id] + if stage.lease_template is None: + raise ValueError(f"cache stage {stage_id!r} has no generation lease") + root = paths.stage(stage_id) + generation = root / key + release_path(root / stage.lease_template.format(key=key)) + logical = measure(generation, set()).logical_bytes if generation.exists() else 0 + plan = PrunePlan( + generated_ns=time.time_ns(), + reclaim_bytes=logical, + actions=(PruneAction( + stage_id=stage_id, key=key, path=generation, logical_bytes=logical, reason=reason, + ),), + violations=(), + ) + return apply_prune(paths, plan, reason=reason) def record_admission_event(root: Path, state_path: Path, event: AdmissionEvent) -> Path: diff --git a/build_system/builder/cache/planner.py b/build_system/builder/cache/planner.py index fa6bbcb1c..484aaea28 100644 --- a/build_system/builder/cache/planner.py +++ b/build_system/builder/cache/planner.py @@ -20,6 +20,21 @@ def _age_clock(strategy: PruneStrategy, entry: CacheEntry) -> int: return entry.last_used_ns if strategy is PruneStrategy.LRU else entry.created_ns +def _actions(stage, entry: CacheEntry, reason: str) -> tuple[PruneAction, ...]: + """A generation with members is removed whole and in order; its bytes are + carried once, on the first path.""" + return tuple( + PruneAction( + stage_id=stage.stage_id, + key=entry.key, + path=stage.path / relative, + logical_bytes=entry.logical_bytes if index == 0 else 0, + reason=reason, + ) + for index, relative in enumerate((entry.relative_path, *entry.member_paths)) + ) + + def plan_prune(inventory: CacheInventory | RetentionInventory, policy: CachePolicy) -> PrunePlan: """Select expired, surplus, and pressure candidates without touching pinned state.""" actions: list[PruneAction] = [] @@ -27,24 +42,13 @@ def plan_prune(inventory: CacheInventory | RetentionInventory, policy: CachePoli selected: dict[str, set[str]] = {stage.stage_id: set() for stage in inventory.stages} def choose(stage, entry, reason: str) -> None: - # A generation with members is removed whole and in order; its bytes - # are carried once, on the first path. - for index, relative in enumerate((entry.relative_path, *entry.member_paths)): - actions.append( - PruneAction( - stage_id=stage.stage_id, - key=entry.key, - path=stage.path / relative, - logical_bytes=entry.logical_bytes if index == 0 else 0, - reason=reason, - ) - ) + actions.extend(_actions(stage, entry, reason)) selected[stage.stage_id].add(entry.key) for stage in inventory.stages: stage_policy = policy.stages[stage.stage_id] remaining = stage.logical_bytes - remaining_count = sum(entry.managed for entry in stage.entries) + remaining_count = sum(entry.managed and not entry.lease_only for entry in stage.entries) if stage_policy.prune_strategy is PruneStrategy.NONE: if remaining > stage_policy.max_size_bytes: violations.append( @@ -66,7 +70,20 @@ def choose(stage, entry, reason: str) -> None: ) maximum_age = stage_policy.maximum_age_hours * NANOSECONDS_PER_HOUR over_max = remaining > stage_policy.max_size_bytes + # A leased ephemeral generation lives exactly as long as its owner: + # once no process holds the lease, it is garbage whatever its age. + ownerless = ( + stage_policy.prune_strategy is PruneStrategy.EPHEMERAL + and stage_policy.lease_template is not None + ) for entry in ordered: + if entry.protected: + continue + if entry.lease_only or ownerless: + choose(stage, entry, "orphaned lease" if entry.lease_only else "no live owner") + remaining -= entry.logical_bytes + remaining_count -= not entry.lease_only + continue expired = ( inventory.generated_ns >= _age_clock(stage_policy.prune_strategy, entry) + maximum_age @@ -76,7 +93,7 @@ def choose(stage, entry, reason: str) -> None: stage_policy.maximum_count is not None and remaining_count > stage_policy.maximum_count ) - if entry.protected or not (expired or recover_to_warm or over_count): + if not (expired or recover_to_warm or over_count): continue reason = ( "expired" @@ -117,16 +134,11 @@ def plan_clean(inventory: CacheInventory, stage_id: str) -> PrunePlan: known = ", ".join(stage.stage_id for stage in inventory.stages) raise KeyError(f"unknown cache stage {stage_id!r}; expected one of: {known}, all") actions = tuple( - PruneAction( - stage_id=stage.stage_id, - key=entry.key, - path=stage.path / entry.relative_path, - logical_bytes=entry.logical_bytes, - reason="explicit clean", - ) + action for stage in selected for entry in stage.entries if entry.managed and not entry.protected + for action in _actions(stage, entry, "explicit clean") ) return PrunePlan( generated_ns=inventory.generated_ns, diff --git a/build_system/builder/gatelaunch.py b/build_system/builder/gatelaunch.py index 1dc8750b9..3e2636f2e 100644 --- a/build_system/builder/gatelaunch.py +++ b/build_system/builder/gatelaunch.py @@ -23,6 +23,7 @@ from __future__ import annotations +import atexit import hashlib import importlib import os @@ -206,7 +207,6 @@ def contained_environment(root: Path | None = None) -> dict[str, str]: generation = Path(python[PYCACHE]) pytest = _policy_stage(source, authority, "python-pytest") / generation.name test_tmp = _test_tmp(source, authority) - test_tmp.mkdir(parents=True, exist_ok=True) gate = _gate_policy(source) toolchain = gate["toolchain"] uv = _policy_stage(source, authority, "python-uv") @@ -251,7 +251,38 @@ def hold_environment(root: Path | None = None) -> None: test_tmp = _test_tmp(source, authority) _hold_generation(generation) _hold_generation(pytest) + # Leased before it exists, so a prune never sees this run without an owner. _hold_generation(test_tmp) + test_tmp.mkdir(parents=True, exist_ok=True) + if test_tmp not in _RUNS: + _RUNS.add(test_tmp) + atexit.register(_end_run, source, authority, test_tmp, os.getpid()) + + +#: Run directories this process owns and removes when it exits. +_RUNS: set[Path] = set() + + +def _end_run(source: Path, authority: Path, run: Path, owner: int) -> None: + """Remove this process's scratch run, success or failure. + + A run left behind was only ever reclaimed by a cold clean: one ~7 GB + directory per release precheck filled the disk. A process killed before + this runs leaves a run no lease holds, which `prune test-temp` reclaims. + """ + if os.getpid() != owner: # a forked child inherited the handler + return + from .cache.config import load_policy + from .cache.operations import reclaim_generation + from .cache.paths import CachePaths + + try: + paths = CachePaths(repository_root=authority, policy=load_policy(source)) + if paths.stage("test-temp") / run.name != run: + raise ValueError(f"{run} is not this authority's test-temp run") + reclaim_generation(paths, "test-temp", run.name, reason="test run ended") + except (OSError, ValueError) as error: + print(f"capsem: left {run} for `cache prune test-temp`: {error}", file=sys.stderr) def _hold_generation(generation: Path) -> BinaryIO: diff --git a/build_system/pyproject.toml b/build_system/pyproject.toml index 6dc3f4611..aa1d438d6 100644 --- a/build_system/pyproject.toml +++ b/build_system/pyproject.toml @@ -58,6 +58,9 @@ default = true testpaths = ["tests"] addopts = "--import-mode=importlib" filterwarnings = ["error"] +# A passing test's tmp_path is removed at its teardown, so a run's scratch +# peaks at what the failing and in-flight tests hold, not the whole suite. +tmp_path_retention_policy = "failed" pythonpath = ["..", "../tests", "tests/gate", "tests/release", "tests/release_site"] markers = [ "mcp: MCP black-box integration tests (require capsem-service + VM assets)", diff --git a/build_system/tests/cache/test_inventory.py b/build_system/tests/cache/test_inventory.py index 75a162ac6..a86f45d0a 100644 --- a/build_system/tests/cache/test_inventory.py +++ b/build_system/tests/cache/test_inventory.py @@ -135,7 +135,10 @@ def test_inventory_separates_metadata_and_protects_active_leases(tmp_path: Path) entries = {entry.key: entry for entry in report.stages[0].entries} assert entries["generation"].managed and entries["generation"].protected - assert not entries[".generation.lock"].managed + assert entries["generation"].member_paths == (Path(".generation.lock"),), ( + "a generation's lease leaves with it, or every finished run strands a lock file" + ) + assert ".generation.lock" not in entries def test_retention_inventory_skips_policy_protected_stages(tmp_path: Path) -> None: diff --git a/build_system/tests/cache/test_planner.py b/build_system/tests/cache/test_planner.py index b5c4f45ca..9b24b6fe9 100644 --- a/build_system/tests/cache/test_planner.py +++ b/build_system/tests/cache/test_planner.py @@ -173,3 +173,64 @@ def test_explicit_clean_preserves_metadata_and_active_leases() -> None: plan = plan_clean(inventory(metadata, leased, entry("generation", 5, 2)), "all") assert [action.key for action in plan.actions] == ["generation"] + + +def _leased_scratch() -> CachePolicy: + """The test-temp shape: ephemeral runs, each alive only while its lease is held.""" + stage = policy().stages["objects"].model_copy(update={ + "prune_strategy": PruneStrategy.EPHEMERAL, + "warm_size_bytes": 1000, + "max_size_bytes": 2000, + "maximum_age_hours": 24, + "maximum_count": 1, + "managed_globs": ("run-*",), + "lease_template": ".{key}.lock", + }) + return policy().model_copy(update={"stages": {"objects": stage}}) + + +def test_ephemeral_prune_reclaims_a_run_nobody_leases_whatever_its_age_or_size() -> None: + """A finished run's 7 GiB directory is garbage the moment its owner exits. + + Before, a leased ephemeral stage aged like any generation: a dead run sat + for `maximum_age_hours` below a 200 GiB maximum, and one run per release + precheck filled a 484 GB disk before prune saw anything to reclaim. + """ + dead = entry("run-1", 7, 99).model_copy( + update={"member_paths": (Path(".run-1.lock"),)} + ) + live = entry("run-2", 7, 99, protected=True) + + plan = plan_prune(inventory(dead, live), _leased_scratch()) + + assert [(action.path.name, action.reason) for action in plan.actions] == [ + ("run-1", "no live owner"), + (".run-1.lock", "no live owner"), + ] + assert plan.reclaim_bytes == 7 + + +def test_orphaned_leases_are_collected_but_never_counted_as_generations() -> None: + orphan = entry("run-3", 0, 99).model_copy( + update={"relative_path": Path(".run-3.lock"), "lease_only": True} + ) + held = entry("run-4", 0, 99, protected=True).model_copy( + update={"relative_path": Path(".run-4.lock"), "lease_only": True} + ) + live = entry("run-2", 7, 99, protected=True) + + plan = plan_prune(inventory(orphan, held, live), _leased_scratch()) + + assert [(action.path.name, action.reason) for action in plan.actions] == [ + (".run-3.lock", "orphaned lease"), + ] + assert plan.violations == (), "a lease file is not a generation for the count cap" + + +def test_explicit_clean_removes_a_generation_with_its_members() -> None: + generation = entry("run-1", 5, 1).model_copy(update={"member_paths": (Path(".run-1.lock"),)}) + + plan = plan_clean(inventory(generation), "all") + + assert [action.path.name for action in plan.actions] == ["run-1", ".run-1.lock"] + assert plan.reclaim_bytes == 5 diff --git a/build_system/tests/cache/test_policy.py b/build_system/tests/cache/test_policy.py index bfe673b4d..f95200540 100644 --- a/build_system/tests/cache/test_policy.py +++ b/build_system/tests/cache/test_policy.py @@ -105,7 +105,7 @@ def test_checked_in_policy_accounts_for_every_mechanism() -> None: assert policy.stages["test-temp"].external assert policy.stages["test-temp"].path == Path("/var/tmp/capsem-tests") assert policy.stages["test-temp"].warm_size_bytes == 8 * 1024**3 - assert policy.stages["test-temp"].max_size_bytes == 200 * 1024**3 + assert policy.stages["test-temp"].max_size_bytes == 32 * 1024**3 assert policy.stages["test-temp"].maximum_count is None assert policy.stages["cargo"].warm_size_bytes == 150 * 1024**3 assert policy.stages["cargo"].max_size_bytes == 180 * 1024**3 diff --git a/build_system/tests/cache/test_scratch_runs.py b/build_system/tests/cache/test_scratch_runs.py new file mode 100644 index 000000000..5f679ecdd --- /dev/null +++ b/build_system/tests/cache/test_scratch_runs.py @@ -0,0 +1,194 @@ +"""A leased scratch run lives exactly as long as the process that owns it. + +On 2026-09-28 `/var/tmp/capsem-tests/` held 42 GB: one ~7 GB +`run-` per release precheck, each left behind when its gate exited, and +~1,900 `.run-.lock` files back to early September. Nothing removed a run +when it ended, and `prune test-temp` offered 21 MB because a dead run aged like +any retained generation below a 200 GiB maximum. Only a cold clean got the +disk back. +""" + +from __future__ import annotations + +import fcntl +import os +import subprocess +import sys +from pathlib import Path + +import pytest +from capsem_builder.cache import leases +from capsem_builder.cache.config import load_policy +from capsem_builder.cache.inventory import scan_inventory +from capsem_builder.cache.models import CachePolicy, CacheScope, PruneStrategy, StagePolicy +from capsem_builder.cache.operations import apply_prune, reclaim_generation +from capsem_builder.cache.paths import CachePaths +from capsem_builder.cache.planner import plan_prune + +ROOT = Path(__file__).resolve().parents[3] +AUTHORITY = load_policy(ROOT).authority_environment + + +def scratch(tmp_path: Path) -> CachePaths: + policy = CachePolicy( + version=1, + root=Path("cache"), + authority_environment="CAPSEM_TEST_CACHE_AUTHORITY", + stages={ + "test-temp": StagePolicy( + path=tmp_path / "var/tmp/capsem-tests", + external=True, + description="test scratch", + scope=CacheScope.DISK, + warm_size_bytes=1 << 30, + max_size_bytes=1 << 31, + prune_strategy=PruneStrategy.EPHEMERAL, + maximum_age_hours=24, + managed_globs=("run-*",), + lease_template=".{key}.lock", + ) + }, + ) + return CachePaths(repository_root=tmp_path / "repository", policy=policy) + + +def run(paths: CachePaths, key: str) -> Path: + directory = paths.stage("test-temp") / key + (directory / "pytest").mkdir(parents=True) + (directory / "pytest/payload").write_bytes(b"x" * 64) + return directory + + +def prune(paths: CachePaths): + plan = plan_prune(scan_inventory(paths, paths.policy), paths.policy) + return plan, apply_prune(paths, plan, reason="test") + + +def test_prune_reclaims_a_dead_run_and_its_lease_but_keeps_a_live_one(tmp_path: Path) -> None: + paths = scratch(tmp_path) + stage = paths.stage("test-temp") + dead = run(paths, "run-1") + (stage / ".run-1.lock").touch() + unleased = run(paths, "run-2") + live = run(paths, "run-3") + held = stage / ".run-3.lock" + held.touch() + orphan = stage / ".run-4.lock" + orphan.touch() + + with held.open("rb") as descriptor: + fcntl.flock(descriptor, fcntl.LOCK_SH) + plan, _ = prune(paths) + + assert not plan.violations + assert sorted(path.name for path in stage.iterdir()) == [".run-3.lock", "run-3"] + assert not dead.exists() and not unleased.exists() and not orphan.exists() + assert (live / "pytest/payload").exists() + + +def test_apply_skips_a_run_whose_owner_appeared_after_the_plan(tmp_path: Path) -> None: + """A dead pid can be reused: a new owner that leases between plan and + apply must keep its directory.""" + paths = scratch(tmp_path) + directory = run(paths, "run-5") + plan = plan_prune(scan_inventory(paths, paths.policy), paths.policy) + assert [action.key for action in plan.actions] == ["run-5"] + + lease = paths.stage("test-temp") / ".run-5.lock" + leases.retain_path(lease) + try: + result = apply_prune(paths, plan, reason="test") + finally: + leases.release_path(lease) + + assert directory.exists() + assert result.removed == () + assert result.busy == (directory,) + + +def test_the_owner_reclaims_its_own_run_and_lease_at_the_end(tmp_path: Path) -> None: + paths = scratch(tmp_path) + directory = run(paths, "run-6") + lease_path = paths.stage("test-temp") / ".run-6.lock" + leases.retain_path(lease_path) + + result = reclaim_generation(paths, "test-temp", "run-6", reason="run ended") + + assert not directory.exists() + assert not lease_path.exists() + assert set(result.removed) == {directory, lease_path} + + +def test_a_lease_taken_across_a_concurrent_unlink_is_retaken_on_the_live_path( + tmp_path: Path, monkeypatch +) -> None: + """A pruner unlinks a lease it holds exclusively. A process that opened the + old inode first would otherwise hold a lock nobody can see.""" + lease = tmp_path / ".run-7.lock" + lease.touch() + real_flock = fcntl.flock + unlinked = [] + + def racing_flock(descriptor, operation): + if not unlinked: + unlinked.append(True) + lease.unlink() + return real_flock(descriptor, operation) + + monkeypatch.setattr(leases.fcntl, "flock", racing_flock) + descriptor = leases.retain_path(lease) + monkeypatch.undo() + try: + assert lease.exists() + assert os.fstat(descriptor.fileno()).st_ino == lease.stat().st_ino + assert leases.active_path(lease) + finally: + leases.release_path(lease) + + +def _launch_source(tmp_path: Path) -> Path: + source = tmp_path / "checkout" + (source / "config").mkdir(parents=True) + policy = (ROOT / "config/cache.toml").read_text(encoding="utf-8") + scratch_root = tmp_path / "scratch/capsem-tests" + policy = policy.replace('path = "/var/tmp/capsem-tests"', f'path = "{scratch_root}"') + (source / "config/cache.toml").write_text(policy, encoding="utf-8") + (source / "config/gate.toml").write_bytes((ROOT / "config/gate.toml").read_bytes()) + return source + + +@pytest.mark.parametrize("exit_code", [0, 3]) +def test_a_launched_process_leaves_no_run_directory_or_lease_behind( + tmp_path: Path, exit_code: int +) -> None: + source = _launch_source(tmp_path) + probe = f""" +import os, sys +from pathlib import Path +from capsem_builder import gatelaunch +source = Path({str(source)!r}) +environment = gatelaunch.contained_environment(source) +os.environ.update(environment) +gatelaunch.hold_environment(source) +run = Path(environment["TMPDIR"]) +assert run.is_dir(), "the run directory exists once its lease is held" +(run / "pytest").mkdir() +(run / "pytest" / "big").write_bytes(b"x" * 4096) +print(run) +sys.exit({exit_code}) +""" + result = subprocess.run( + [sys.executable, "-c", probe], + env={k: v for k, v in os.environ.items() if k != AUTHORITY}, + capture_output=True, + text=True, + timeout=60, + check=False, + ) + + assert result.returncode == exit_code, result.stderr + directory = Path(result.stdout.strip().splitlines()[-1]) + assert directory.is_relative_to(tmp_path / "scratch/capsem-tests") + assert not directory.exists(), "the run outlived its process" + assert not (directory.parent / f".{directory.name}.lock").exists() + assert result.stderr == "" diff --git a/build_system/tests/gate/test_gate_pycache_authority.py b/build_system/tests/gate/test_gate_pycache_authority.py index f2097616f..2e9a98da7 100644 --- a/build_system/tests/gate/test_gate_pycache_authority.py +++ b/build_system/tests/gate/test_gate_pycache_authority.py @@ -181,4 +181,5 @@ def test_live_gate_generation_holds_a_prune_lease(tmp_path: Path) -> None: entries = {entry.key: entry for entry in inventory.stages[0].entries} assert entries[generation.name].protected - assert not entries[f".{generation.name}.lock"].managed + assert entries[generation.name].member_paths == (Path(f".{generation.name}.lock"),) + assert f".{generation.name}.lock" not in entries diff --git a/build_system/tests/test_test_ownership.py b/build_system/tests/test_test_ownership.py index 509d28516..5ce8d7ba7 100644 --- a/build_system/tests/test_test_ownership.py +++ b/build_system/tests/test_test_ownership.py @@ -51,6 +51,7 @@ "build_system/tests/cache/test_runtime_adapters.py", "build_system/tests/cache/test_runtime_control.py", "build_system/tests/cache/test_registry.py", + "build_system/tests/cache/test_scratch_runs.py", "build_system/tests/cache/test_stats.py", "build_system/tests/cache/test_telemetry.py", "build_system/tests/cache/test_tool_adapters.py", diff --git a/config/cache.toml b/config/cache.toml index 8b04db1db..463c7fe6e 100644 --- a/config/cache.toml +++ b/config/cache.toml @@ -259,7 +259,12 @@ scope = "disk" path = "/var/tmp/capsem-tests" external = true warm_size_bytes = 8589934592 # 8 GiB -max_size_bytes = 214748364800 # 200 GiB +# Only live runs count here: a run leaves with its owner and prune reclaims any +# run nobody leases. 32 GiB is ~4x the largest run seen (a 7.2 GB release +# precheck); more live scratch than that is a leak to surface, not to absorb. +# It was 200 GiB, which on a 484 GB disk beside a 180 GiB Cargo target bounded +# nothing. +max_size_bytes = 34359738368 # 32 GiB prune_strategy = "ephemeral" maximum_age_hours = 24 managed_globs = ["run-*"] diff --git a/skills/dev-cache/SKILL.md b/skills/dev-cache/SKILL.md index 352166740..1958cf53b 100644 --- a/skills/dev-cache/SKILL.md +++ b/skills/dev-cache/SKILL.md @@ -122,6 +122,19 @@ reader and publisher holds `leases.shared_use(paths, "objects")`, so a prune never removes bytes in use. Before this the stage was `none`, grew past its cap, and failed every prune on the machine. +A leased `ephemeral` stage (`test-temp`) is scratch, not retention: a +generation lives exactly as long as a process holds its lease. The owner +leases `run-` before creating it and, at exit, releases the lease and +removes the run and its lease through `operations.reclaim_generation`, the same +guarded, journaled path a prune takes. A run no process leases (its owner was +killed) has no live owner, and `prune` reclaims it whatever its age or size; +a lease file whose generation is gone is collected too, and never counts +toward `maximum_count`. Removal takes each lease exclusively, so a reused pid +that leased after the plan keeps its directory (`ApplyResult.busy`). Before +this, a dead run aged like a retained generation below a 200 GiB maximum: +one ~7 GB run per release precheck and 1,900 lock files filled the disk while +`prune` offered 21 MB. + Retained cache lifetime ends only through these typed operations. Do not add consumer-boundary releases, post-test eviction hooks, or other subsystem lifecycle paths that bypass the owner's warm/max/age/count policy.