1746 lines
62 KiB
Python
Executable File
1746 lines
62 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Measure and safely enforce Blacksite's bounded Cargo build storage policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from collections import defaultdict
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Mapping, Sequence
|
|
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError as error: # pragma: no cover - Python < 3.11
|
|
raise SystemExit("build_storage.py requires Python 3.11 or newer") from error
|
|
|
|
from cargo_lane import (
|
|
SENTINEL_NAME,
|
|
LaneError,
|
|
cache_root,
|
|
canonical_path,
|
|
cargo_home,
|
|
load_workflow,
|
|
read_sentinel,
|
|
sentinel_path,
|
|
toolchain_info,
|
|
utc_now,
|
|
validate_lane_name,
|
|
validate_sentinel,
|
|
workspace_hash,
|
|
workspace_root,
|
|
)
|
|
|
|
|
|
GIB = 1024**3
|
|
LEDGER_SCHEMA_VERSION = 1
|
|
CANDIDATE_PRESERVATION_SCHEMA_VERSION = 1
|
|
CANDIDATE_PRESERVATION_PATH = Path(".codex/session/candidate-prune-safe.json")
|
|
DEFAULT_POLICY: dict[str, Any] = {
|
|
"enabled": True,
|
|
"soft_total_gib": 55,
|
|
"hard_total_gib": 80,
|
|
"min_free_gib": 40,
|
|
"min_free_percent": 15,
|
|
"persistent_lane_max_gib": 40,
|
|
"disposable_lane_max_gib": 35,
|
|
"candidate_retention_hours": 24,
|
|
"hot_reload_retention_hours": 24,
|
|
"full_debug_retention_hours": 12,
|
|
"failed_lane_retention_hours": 24,
|
|
"auto_prune_disposable": True,
|
|
"delete_only_marked_lanes": True,
|
|
}
|
|
|
|
|
|
class StorageError(RuntimeError):
|
|
"""Storage state cannot be measured or changed safely."""
|
|
|
|
|
|
def _policy(workflow: Mapping[str, Any]) -> dict[str, Any]:
|
|
result = dict(DEFAULT_POLICY)
|
|
configured = workflow.get("build_storage", {})
|
|
if isinstance(configured, Mapping):
|
|
result.update(configured)
|
|
return result
|
|
|
|
|
|
def human_bytes(value: int) -> str:
|
|
amount = float(max(0, value))
|
|
for suffix in ("B", "KiB", "MiB", "GiB", "TiB"):
|
|
if amount < 1024 or suffix == "TiB":
|
|
return f"{amount:.1f} {suffix}"
|
|
amount /= 1024
|
|
return f"{amount:.1f} TiB"
|
|
|
|
|
|
def _allocated_bytes(stat_result: os.stat_result) -> int:
|
|
blocks = getattr(stat_result, "st_blocks", 0)
|
|
return blocks * 512 if blocks else stat_result.st_size
|
|
|
|
|
|
@dataclass
|
|
class PathUsage:
|
|
path: str
|
|
allocated_bytes: int = 0
|
|
apparent_bytes: int = 0
|
|
files: int = 0
|
|
directories: int = 0
|
|
symlinks: int = 0
|
|
top_level: dict[str, int] | None = None
|
|
error: str | None = None
|
|
|
|
|
|
def measure_tree(path: Path) -> PathUsage:
|
|
path = canonical_path(path)
|
|
usage = PathUsage(path=str(path), top_level={})
|
|
if not path.exists():
|
|
return usage
|
|
if path.is_symlink():
|
|
usage.symlinks = 1
|
|
usage.error = "root is a symlink"
|
|
return usage
|
|
|
|
seen: set[tuple[int, int]] = set()
|
|
try:
|
|
root_stat = path.lstat()
|
|
root_device = root_stat.st_dev
|
|
usage.allocated_bytes += _allocated_bytes(root_stat)
|
|
usage.apparent_bytes += root_stat.st_size
|
|
usage.directories += 1
|
|
seen.add((root_stat.st_dev, root_stat.st_ino))
|
|
except OSError as error:
|
|
usage.error = str(error)
|
|
return usage
|
|
|
|
for directory, names, filenames in os.walk(path, topdown=True, followlinks=False):
|
|
current = Path(directory)
|
|
relative = current.relative_to(path)
|
|
current_bucket = relative.parts[0] if relative.parts else None
|
|
|
|
kept_names: list[str] = []
|
|
for name in names:
|
|
child = current / name
|
|
try:
|
|
stat_result = child.lstat()
|
|
except OSError:
|
|
continue
|
|
if child.is_symlink():
|
|
usage.symlinks += 1
|
|
usage.allocated_bytes += _allocated_bytes(stat_result)
|
|
usage.apparent_bytes += stat_result.st_size
|
|
continue
|
|
if stat_result.st_dev != root_device:
|
|
continue
|
|
kept_names.append(name)
|
|
inode = (stat_result.st_dev, stat_result.st_ino)
|
|
if inode in seen:
|
|
continue
|
|
seen.add(inode)
|
|
allocated = _allocated_bytes(stat_result)
|
|
usage.allocated_bytes += allocated
|
|
usage.apparent_bytes += stat_result.st_size
|
|
usage.directories += 1
|
|
bucket = current_bucket or name
|
|
usage.top_level[bucket] = usage.top_level.get(bucket, 0) + allocated
|
|
names[:] = kept_names
|
|
|
|
for name in filenames:
|
|
child = current / name
|
|
try:
|
|
stat_result = child.lstat()
|
|
except OSError:
|
|
continue
|
|
if child.is_symlink():
|
|
usage.symlinks += 1
|
|
inode = (stat_result.st_dev, stat_result.st_ino)
|
|
if inode in seen:
|
|
continue
|
|
seen.add(inode)
|
|
allocated = _allocated_bytes(stat_result)
|
|
usage.allocated_bytes += allocated
|
|
usage.apparent_bytes += stat_result.st_size
|
|
usage.files += 1
|
|
bucket = current_bucket or name
|
|
usage.top_level[bucket] = usage.top_level.get(bucket, 0) + allocated
|
|
usage.top_level = dict(
|
|
sorted(usage.top_level.items(), key=lambda item: item[1], reverse=True)
|
|
)
|
|
return usage
|
|
|
|
|
|
def _nearest_existing(path: Path) -> Path:
|
|
current = canonical_path(path)
|
|
while not current.exists() and current != current.parent:
|
|
current = current.parent
|
|
return current
|
|
|
|
|
|
def _mount_point(path: Path) -> Path:
|
|
current = _nearest_existing(path)
|
|
while current != current.parent and not os.path.ismount(current):
|
|
current = current.parent
|
|
return current
|
|
|
|
|
|
def _filesystem_status(paths: Iterable[Path], policy: Mapping[str, Any]) -> list[dict[str, Any]]:
|
|
by_device: dict[int, dict[str, Any]] = {}
|
|
for requested in paths:
|
|
probe = _nearest_existing(requested)
|
|
try:
|
|
device = probe.stat().st_dev
|
|
usage = shutil.disk_usage(probe)
|
|
except OSError:
|
|
continue
|
|
if device in by_device:
|
|
by_device[device]["paths"].append(str(canonical_path(requested)))
|
|
continue
|
|
percentage_floor = int(usage.total * float(policy["min_free_percent"]) / 100)
|
|
fixed_floor = int(float(policy["min_free_gib"]) * GIB)
|
|
floor = max(fixed_floor, percentage_floor)
|
|
by_device[device] = {
|
|
"device": device,
|
|
"mount": str(_mount_point(probe)),
|
|
"paths": [str(canonical_path(requested))],
|
|
"total_bytes": usage.total,
|
|
"used_bytes": usage.used,
|
|
"free_bytes": usage.free,
|
|
"minimum_free_bytes": floor,
|
|
}
|
|
return list(by_device.values())
|
|
|
|
|
|
def _path_within(path: Path, parent: Path) -> bool:
|
|
try:
|
|
canonical_path(path).relative_to(canonical_path(parent))
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _read_proc_value(path: Path, *, binary: bool = False) -> str:
|
|
try:
|
|
if binary:
|
|
return path.read_bytes().replace(b"\0", b" ").decode(errors="replace").strip()
|
|
return path.read_text(encoding="utf-8", errors="replace").strip()
|
|
except (OSError, PermissionError):
|
|
return ""
|
|
|
|
|
|
BUILD_TOOL_NAMES = {
|
|
"cargo",
|
|
"rustc",
|
|
"rustdoc",
|
|
"clang",
|
|
"clang++",
|
|
"cc",
|
|
"c++",
|
|
"gcc",
|
|
"g++",
|
|
"ld",
|
|
"ld.lld",
|
|
"lld",
|
|
"mold",
|
|
}
|
|
|
|
|
|
def active_processes(root: Path, watched_paths: Sequence[Path]) -> list[dict[str, Any]]:
|
|
proc = Path("/proc")
|
|
if not proc.is_dir():
|
|
return []
|
|
root = canonical_path(root)
|
|
watched = [canonical_path(path) for path in watched_paths]
|
|
results: list[dict[str, Any]] = []
|
|
own_pid = os.getpid()
|
|
|
|
for entry in proc.iterdir():
|
|
if not entry.name.isdigit() or int(entry.name) == own_pid:
|
|
continue
|
|
pid = int(entry.name)
|
|
comm = _read_proc_value(entry / "comm")
|
|
command = _read_proc_value(entry / "cmdline", binary=True)
|
|
executable = ""
|
|
cwd = ""
|
|
try:
|
|
executable = os.readlink(entry / "exe")
|
|
except OSError:
|
|
pass
|
|
try:
|
|
cwd = os.readlink(entry / "cwd")
|
|
except OSError:
|
|
pass
|
|
|
|
command_name = Path(command.split(" ", 1)[0]).name if command else comm
|
|
is_build_tool = comm in BUILD_TOOL_NAMES or command_name in BUILD_TOOL_NAMES
|
|
is_packager = any(
|
|
token in command
|
|
for token in ("package-project", "process-assets", "cargo package")
|
|
)
|
|
is_editor = comm in {"editor", "project_launcher", "game"} or command_name in {
|
|
"editor",
|
|
"project_launcher",
|
|
"game",
|
|
}
|
|
|
|
environment = _read_proc_value(entry / "environ", binary=True)
|
|
references = " ".join((command, executable, cwd, environment))
|
|
used = [str(path) for path in watched if str(path) in references]
|
|
workspace_related = bool(cwd and _path_within(Path(cwd), root))
|
|
|
|
if is_editor and not used:
|
|
maps = _read_proc_value(entry / "maps")
|
|
used = [str(path) for path in watched if str(path) in maps]
|
|
|
|
if not used and not ((is_build_tool or is_packager) and workspace_related):
|
|
continue
|
|
results.append(
|
|
{
|
|
"pid": pid,
|
|
"comm": comm,
|
|
"command": command,
|
|
"cwd": cwd,
|
|
"exe": executable,
|
|
"build_tool": is_build_tool,
|
|
"packager": is_packager,
|
|
"editor": is_editor,
|
|
"workspace_related": workspace_related,
|
|
"uses": used,
|
|
}
|
|
)
|
|
return sorted(results, key=lambda item: item["pid"])
|
|
|
|
|
|
def _sccache_status(root: Path) -> dict[str, Any]:
|
|
wrapper = os.environ.get("RUSTC_WRAPPER", "")
|
|
config_path = root / ".cargo" / "config.toml"
|
|
if not wrapper and config_path.is_file():
|
|
try:
|
|
with config_path.open("rb") as handle:
|
|
config = tomllib.load(handle)
|
|
build = config.get("build", {})
|
|
if isinstance(build, Mapping):
|
|
wrapper = str(build.get("rustc-wrapper", ""))
|
|
except (OSError, tomllib.TOMLDecodeError):
|
|
pass
|
|
enabled = Path(wrapper).name == "sccache" if wrapper else False
|
|
result: dict[str, Any] = {
|
|
"enabled": enabled,
|
|
"wrapper": wrapper or None,
|
|
"executable": shutil.which("sccache"),
|
|
"allocated_bytes": 0,
|
|
}
|
|
if not enabled:
|
|
return result
|
|
|
|
configured_dir = os.environ.get("SCCACHE_DIR")
|
|
if configured_dir:
|
|
usage = measure_tree(Path(configured_dir))
|
|
result["cache_dir"] = usage.path
|
|
result["allocated_bytes"] = usage.allocated_bytes
|
|
executable = result["executable"]
|
|
if executable:
|
|
completed = subprocess.run(
|
|
[executable, "--show-stats", "--stats-format=json"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
if completed.returncode == 0:
|
|
try:
|
|
result["stats"] = json.loads(completed.stdout)
|
|
except json.JSONDecodeError:
|
|
result["stats_text"] = completed.stdout.strip()
|
|
return result
|
|
|
|
|
|
def _marker_record(path: Path, root: Path, role: str) -> dict[str, Any]:
|
|
document = read_sentinel(path)
|
|
lane = path.name
|
|
errors: list[str] = []
|
|
if document is None:
|
|
errors.append(f"missing or malformed {SENTINEL_NAME}")
|
|
else:
|
|
lane = str(document.get("lane", lane))
|
|
errors.extend(
|
|
validate_sentinel(path, document, root=root, lane=lane, role=role)
|
|
)
|
|
usage = measure_tree(path)
|
|
if usage.error:
|
|
errors.append(usage.error)
|
|
return {
|
|
"path": str(canonical_path(path)),
|
|
"role": role,
|
|
"lane": lane,
|
|
"valid_sentinel": not errors,
|
|
"sentinel_errors": errors,
|
|
"sentinel": document,
|
|
"usage": asdict(usage),
|
|
}
|
|
|
|
|
|
def _discover_lanes(root: Path, workflow: Mapping[str, Any]) -> tuple[list[dict[str, Any]], Path, Path]:
|
|
storage = workflow["build_storage"]
|
|
external_partition = cache_root(root, workflow) / workspace_hash(root)
|
|
exceptional_root = canonical_path(
|
|
root / str(storage.get("exceptional_target_root", "target/lanes"))
|
|
)
|
|
records: list[dict[str, Any]] = []
|
|
if external_partition.is_dir() and not external_partition.is_symlink():
|
|
for child in sorted(external_partition.iterdir()):
|
|
if child.is_dir() and not child.is_symlink():
|
|
records.append(_marker_record(child, root, "build"))
|
|
if exceptional_root.is_dir() and not exceptional_root.is_symlink():
|
|
for child in sorted(exceptional_root.iterdir()):
|
|
if child.is_dir() and not child.is_symlink():
|
|
records.append(_marker_record(child, root, "target"))
|
|
return records, external_partition, exceptional_root
|
|
|
|
|
|
def _group_lanes(records: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for record in records:
|
|
grouped[str(record["lane"])].append(record)
|
|
result: list[dict[str, Any]] = []
|
|
for lane, artifacts in sorted(grouped.items()):
|
|
total = sum(item["usage"]["allocated_bytes"] for item in artifacts)
|
|
last_use_values = [
|
|
item["sentinel"].get("last_use")
|
|
for item in artifacts
|
|
if isinstance(item.get("sentinel"), Mapping)
|
|
and item["sentinel"].get("last_use")
|
|
]
|
|
result.append(
|
|
{
|
|
"lane": lane,
|
|
"allocated_bytes": total,
|
|
"last_use": max(last_use_values) if last_use_values else None,
|
|
"valid": all(item["valid_sentinel"] for item in artifacts),
|
|
"artifacts": artifacts,
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def storage_snapshot(root: Path, workflow: Mapping[str, Any]) -> dict[str, Any]:
|
|
root = canonical_path(root)
|
|
policy = _policy(workflow)
|
|
repository_target = canonical_path(
|
|
root / str(policy.get("dev_target_dir", "target"))
|
|
)
|
|
target_usage = measure_tree(repository_target)
|
|
records, external_partition, exceptional_root = _discover_lanes(root, workflow)
|
|
external_usage = measure_tree(external_partition)
|
|
lanes = _group_lanes(records)
|
|
sccache = _sccache_status(root)
|
|
|
|
total = (
|
|
target_usage.allocated_bytes
|
|
+ external_usage.allocated_bytes
|
|
+ int(sccache.get("allocated_bytes", 0))
|
|
)
|
|
watched = [repository_target, external_partition, exceptional_root]
|
|
filesystems = _filesystem_status(
|
|
[repository_target, cache_root(root, workflow)], policy
|
|
)
|
|
|
|
debug = repository_target / "debug"
|
|
legacy_paths = {
|
|
"deps": debug / "deps",
|
|
"incremental": debug / "incremental",
|
|
"build": debug / "build",
|
|
"fingerprint": debug / ".fingerprint",
|
|
}
|
|
legacy_intermediates = {
|
|
name: asdict(measure_tree(path))
|
|
for name, path in legacy_paths.items()
|
|
if path.exists()
|
|
}
|
|
|
|
return {
|
|
"schema_version": 1,
|
|
"timestamp": utc_now(),
|
|
"workspace": str(root),
|
|
"workspace_hash": workspace_hash(root),
|
|
"toolchain": toolchain_info(),
|
|
"policy": policy,
|
|
"repository_target": asdict(target_usage),
|
|
"external_partition": asdict(external_usage),
|
|
"exceptional_target_root": str(exceptional_root),
|
|
"lanes": lanes,
|
|
"legacy_intermediates": legacy_intermediates,
|
|
"sccache": sccache,
|
|
"filesystems": filesystems,
|
|
"active_processes": active_processes(root, watched),
|
|
"total_allocated_bytes": total,
|
|
}
|
|
|
|
|
|
def _parse_simulated_lanes(values: Sequence[str]) -> dict[str, int]:
|
|
result: dict[str, int] = {}
|
|
for value in values:
|
|
lane, separator, amount = value.partition("=")
|
|
if not separator:
|
|
raise StorageError("--simulate-lane-gib expects LANE=GIB")
|
|
validate_lane_name(lane)
|
|
result[lane] = int(float(amount) * GIB)
|
|
return result
|
|
|
|
|
|
def apply_simulation(snapshot: dict[str, Any], arguments: argparse.Namespace) -> bool:
|
|
simulated = False
|
|
if arguments.simulate_total_gib is not None:
|
|
snapshot["total_allocated_bytes"] = int(arguments.simulate_total_gib * GIB)
|
|
simulated = True
|
|
if arguments.simulate_free_gib is not None:
|
|
for filesystem in snapshot["filesystems"]:
|
|
filesystem["free_bytes"] = int(arguments.simulate_free_gib * GIB)
|
|
simulated = True
|
|
lane_sizes = _parse_simulated_lanes(arguments.simulate_lane_gib or [])
|
|
if lane_sizes:
|
|
simulated = True
|
|
existing = {lane["lane"]: lane for lane in snapshot["lanes"]}
|
|
for lane, size in lane_sizes.items():
|
|
if lane in existing:
|
|
existing[lane]["allocated_bytes"] = size
|
|
else:
|
|
snapshot["lanes"].append(
|
|
{
|
|
"lane": lane,
|
|
"allocated_bytes": size,
|
|
"last_use": None,
|
|
"valid": True,
|
|
"artifacts": [],
|
|
"simulated": True,
|
|
}
|
|
)
|
|
snapshot["simulated"] = simulated
|
|
return simulated
|
|
|
|
|
|
def evaluate_budget(snapshot: Mapping[str, Any]) -> dict[str, list[str]]:
|
|
policy = snapshot["policy"]
|
|
total = int(snapshot["total_allocated_bytes"])
|
|
warnings: list[str] = []
|
|
violations: list[str] = []
|
|
soft = int(float(policy["soft_total_gib"]) * GIB)
|
|
hard = int(float(policy["hard_total_gib"]) * GIB)
|
|
if total > hard:
|
|
violations.append(
|
|
f"total build artifacts {human_bytes(total)} exceed hard budget {human_bytes(hard)}"
|
|
)
|
|
elif total > soft:
|
|
warnings.append(
|
|
f"total build artifacts {human_bytes(total)} exceed soft budget {human_bytes(soft)}"
|
|
)
|
|
|
|
persistent = str(policy.get("persistent_lane", "dev"))
|
|
for lane in snapshot["lanes"]:
|
|
maximum_gib = (
|
|
policy["persistent_lane_max_gib"]
|
|
if lane["lane"] == persistent
|
|
else policy["disposable_lane_max_gib"]
|
|
)
|
|
maximum = int(float(maximum_gib) * GIB)
|
|
if int(lane["allocated_bytes"]) > maximum:
|
|
warnings.append(
|
|
f"lane {lane['lane']} uses {human_bytes(lane['allocated_bytes'])}, "
|
|
f"above {human_bytes(maximum)}"
|
|
)
|
|
|
|
for filesystem in snapshot["filesystems"]:
|
|
free = int(filesystem["free_bytes"])
|
|
floor = int(filesystem["minimum_free_bytes"])
|
|
if free < floor:
|
|
violations.append(
|
|
f"filesystem {filesystem['mount']} has {human_bytes(free)} free; "
|
|
f"minimum is {human_bytes(floor)}"
|
|
)
|
|
return {"warnings": warnings, "violations": violations}
|
|
|
|
|
|
def _parse_timestamp(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed
|
|
|
|
|
|
def _retention_hours(lane: str, policy: Mapping[str, Any]) -> float:
|
|
if lane == str(policy.get("candidate_lane", "candidate")):
|
|
return float(policy["candidate_retention_hours"])
|
|
if lane == str(policy.get("hot_reload_lane", "hot-reload")):
|
|
return float(policy["hot_reload_retention_hours"])
|
|
if lane == str(policy.get("full_debug_lane", "full-debug")):
|
|
return float(policy["full_debug_retention_hours"])
|
|
return float(policy["failed_lane_retention_hours"])
|
|
|
|
|
|
def prune_plan(snapshot: Mapping[str, Any]) -> list[dict[str, Any]]:
|
|
policy = snapshot["policy"]
|
|
persistent = str(policy.get("persistent_lane", "dev"))
|
|
now = datetime.now(timezone.utc)
|
|
plans: list[dict[str, Any]] = []
|
|
for lane in snapshot["lanes"]:
|
|
if lane["lane"] == persistent:
|
|
continue
|
|
last_use = _parse_timestamp(lane.get("last_use"))
|
|
age_hours = (
|
|
(now - last_use).total_seconds() / 3600 if last_use is not None else None
|
|
)
|
|
retention = _retention_hours(lane["lane"], policy)
|
|
eligible = bool(lane["valid"] and age_hours is not None and age_hours >= retention)
|
|
plans.append(
|
|
{
|
|
"lane": lane["lane"],
|
|
"allocated_bytes": lane["allocated_bytes"],
|
|
"age_hours": age_hours,
|
|
"retention_hours": retention,
|
|
"eligible": eligible,
|
|
"reason": (
|
|
"expired disposable lane"
|
|
if eligible
|
|
else "invalid sentinel"
|
|
if not lane["valid"]
|
|
else "missing last-use timestamp"
|
|
if age_hours is None
|
|
else "retention window active"
|
|
),
|
|
"artifacts": [artifact["path"] for artifact in lane["artifacts"]],
|
|
}
|
|
)
|
|
return plans
|
|
|
|
|
|
def _contains_symlink(path: Path) -> bool:
|
|
if path.is_symlink():
|
|
return True
|
|
for directory, names, filenames in os.walk(path, topdown=True, followlinks=False):
|
|
current = Path(directory)
|
|
for name in [*names, *filenames]:
|
|
try:
|
|
if (current / name).is_symlink():
|
|
return True
|
|
except OSError:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _tracked_files(root: Path, path: Path) -> list[str]:
|
|
if not _path_within(path, root):
|
|
return []
|
|
relative = canonical_path(path).relative_to(root)
|
|
completed = subprocess.run(
|
|
["git", "ls-files", "--", str(relative)],
|
|
cwd=root,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
)
|
|
return [line for line in completed.stdout.splitlines() if line]
|
|
|
|
|
|
def _candidate_lane_name(workflow: Mapping[str, Any]) -> str:
|
|
verification = workflow.get("verification", {})
|
|
if isinstance(verification, Mapping):
|
|
value = verification.get("candidate_lane", "candidate")
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
return "candidate"
|
|
|
|
|
|
def _candidate_preservation_path(root: Path) -> Path:
|
|
return canonical_path(root) / CANDIDATE_PRESERVATION_PATH
|
|
|
|
|
|
def _lane_generation(lane: Mapping[str, Any]) -> str | None:
|
|
"""Bind preservation evidence to the exact last-used lane generation."""
|
|
|
|
sentinels: list[dict[str, Any]] = []
|
|
artifacts = sorted(
|
|
lane.get("artifacts", []),
|
|
key=lambda item: (str(item.get("role", "")), str(item.get("path", ""))),
|
|
)
|
|
for artifact in artifacts:
|
|
marker = read_sentinel(Path(str(artifact["path"])))
|
|
if marker is None:
|
|
return None
|
|
sentinels.append(marker)
|
|
if not sentinels:
|
|
return None
|
|
encoded = json.dumps(
|
|
sentinels, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(1024 * 1024):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _preserved_path(root: Path, raw: object) -> Path | None:
|
|
if not isinstance(raw, str) or not raw:
|
|
return None
|
|
path = Path(os.path.expandvars(os.path.expanduser(raw)))
|
|
if not path.is_absolute():
|
|
path = root / path
|
|
return path
|
|
|
|
|
|
def _preserved_record_errors(
|
|
record: object,
|
|
*,
|
|
label: str,
|
|
root: Path,
|
|
lane_paths: Sequence[Path],
|
|
) -> list[str]:
|
|
if not isinstance(record, Mapping):
|
|
return [f"{label} must be an object with path, bytes, and sha256"]
|
|
raw_path = _preserved_path(root, record.get("path"))
|
|
if raw_path is None:
|
|
return [f"{label} has no valid path"]
|
|
if raw_path.is_symlink():
|
|
return [f"{label} is a symlink: {raw_path}"]
|
|
path = canonical_path(raw_path)
|
|
errors: list[str] = []
|
|
if any(_path_within(path, lane_path) for lane_path in lane_paths):
|
|
errors.append(f"{label} remains inside the disposable candidate lane: {path}")
|
|
if not path.is_file():
|
|
errors.append(f"{label} is not a preserved file: {path}")
|
|
return errors
|
|
|
|
expected_bytes = record.get("bytes")
|
|
if not isinstance(expected_bytes, int) or expected_bytes < 0:
|
|
errors.append(f"{label} has an invalid byte count")
|
|
else:
|
|
actual_bytes = path.stat().st_size
|
|
if actual_bytes != expected_bytes:
|
|
errors.append(
|
|
f"{label} byte count changed: expected {expected_bytes}, found {actual_bytes}"
|
|
)
|
|
|
|
expected_digest = record.get("sha256")
|
|
if not isinstance(expected_digest, str) or not re.fullmatch(
|
|
r"[0-9a-f]{64}", expected_digest
|
|
):
|
|
errors.append(f"{label} has an invalid sha256")
|
|
else:
|
|
try:
|
|
actual_digest = _sha256_file(path)
|
|
except OSError as error:
|
|
errors.append(f"{label} cannot be hashed: {error}")
|
|
else:
|
|
if actual_digest != expected_digest:
|
|
errors.append(f"{label} sha256 no longer matches: {path}")
|
|
return errors
|
|
|
|
|
|
def _candidate_preservation_errors(
|
|
lane: Mapping[str, Any], *, root: Path, workflow: Mapping[str, Any]
|
|
) -> list[str]:
|
|
if str(lane.get("lane")) != _candidate_lane_name(workflow):
|
|
return []
|
|
|
|
marker_path = _candidate_preservation_path(root)
|
|
if marker_path.is_symlink():
|
|
return [f"candidate preservation marker is a symlink: {marker_path}"]
|
|
if not marker_path.is_file():
|
|
return [
|
|
"candidate preservation marker is missing; preserve the candidate evidence "
|
|
"and artifacts, then run mark-candidate-preserved"
|
|
]
|
|
try:
|
|
marker = json.loads(marker_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
return [f"candidate preservation marker is unreadable: {error}"]
|
|
if not isinstance(marker, Mapping):
|
|
return ["candidate preservation marker must be a JSON object"]
|
|
|
|
root = canonical_path(root)
|
|
generation = _lane_generation(lane)
|
|
expected = {
|
|
"schema_version": CANDIDATE_PRESERVATION_SCHEMA_VERSION,
|
|
"workspace": str(root),
|
|
"workspace_hash": workspace_hash(root),
|
|
"lane": _candidate_lane_name(workflow),
|
|
"lane_generation": generation,
|
|
"prune_safe": True,
|
|
}
|
|
errors: list[str] = []
|
|
for key, value in expected.items():
|
|
if marker.get(key) != value:
|
|
errors.append(
|
|
f"candidate preservation {key}: expected {value!r}, found {marker.get(key)!r}"
|
|
)
|
|
|
|
lane_paths = [canonical_path(Path(str(item["path"]))) for item in lane["artifacts"]]
|
|
evidence = marker.get("evidence")
|
|
errors.extend(
|
|
_preserved_record_errors(
|
|
evidence, label="candidate evidence", root=root, lane_paths=lane_paths
|
|
)
|
|
)
|
|
artifacts = marker.get("preserved_artifacts")
|
|
if not isinstance(artifacts, list) or not artifacts:
|
|
errors.append("candidate preservation must name at least one preserved artifact")
|
|
else:
|
|
for index, record in enumerate(artifacts):
|
|
errors.extend(
|
|
_preserved_record_errors(
|
|
record,
|
|
label=f"preserved candidate artifact {index}",
|
|
root=root,
|
|
lane_paths=lane_paths,
|
|
)
|
|
)
|
|
evidence_path = (
|
|
canonical_path(_preserved_path(root, evidence.get("path")))
|
|
if isinstance(evidence, Mapping)
|
|
and _preserved_path(root, evidence.get("path")) is not None
|
|
else None
|
|
)
|
|
artifact_paths = {
|
|
canonical_path(path)
|
|
for record in artifacts
|
|
if isinstance(record, Mapping)
|
|
and (path := _preserved_path(root, record.get("path"))) is not None
|
|
}
|
|
if evidence_path is not None and artifact_paths == {evidence_path}:
|
|
errors.append("candidate evidence alone is not a preserved candidate artifact")
|
|
return errors
|
|
|
|
|
|
def deletion_safety(
|
|
artifact: Mapping[str, Any],
|
|
*,
|
|
root: Path,
|
|
workflow: Mapping[str, Any],
|
|
processes: Sequence[Mapping[str, Any]],
|
|
) -> list[str]:
|
|
path = Path(str(artifact["path"]))
|
|
lane = str(artifact["lane"])
|
|
role = str(artifact["role"])
|
|
errors: list[str] = []
|
|
if not path.exists():
|
|
return ["path no longer exists"]
|
|
if path.is_symlink():
|
|
errors.append("lane root is a symlink")
|
|
|
|
path = canonical_path(path)
|
|
root = canonical_path(root)
|
|
policy = _policy(workflow)
|
|
external_parent = canonical_path(cache_root(root, workflow) / workspace_hash(root))
|
|
exceptional_parent = canonical_path(
|
|
root / str(policy.get("exceptional_target_root", "target/lanes"))
|
|
)
|
|
expected_parent = external_parent if role == "build" else exceptional_parent
|
|
if path.parent != expected_parent:
|
|
errors.append(f"path is not a direct child of managed {role} root")
|
|
|
|
protected = [Path("/"), Path.home(), cargo_home(), root, root / ".git"]
|
|
for protected_path in protected:
|
|
protected_path = canonical_path(protected_path)
|
|
if path == protected_path or _path_within(protected_path, path):
|
|
errors.append(f"path is or contains protected path {protected_path}")
|
|
|
|
# Never recursively inspect a path that already failed containment or
|
|
# protected-root checks. This keeps even a malicious dry run bounded.
|
|
if errors:
|
|
return errors
|
|
|
|
marker = read_sentinel(path)
|
|
if marker is None:
|
|
errors.append(f"missing or malformed {SENTINEL_NAME}")
|
|
else:
|
|
errors.extend(
|
|
validate_sentinel(path, marker, root=root, lane=lane, role=role)
|
|
)
|
|
if _contains_symlink(path):
|
|
errors.append("lane contains a symlink")
|
|
tracked = _tracked_files(root, path)
|
|
if tracked:
|
|
errors.append(f"lane contains tracked files: {', '.join(tracked[:5])}")
|
|
|
|
for process in processes:
|
|
build_activity = bool(
|
|
process.get("build_tool")
|
|
or process.get("packager")
|
|
or process.get("workspace_related")
|
|
)
|
|
direct_use = any(
|
|
_path_within(Path(used), path) or _path_within(path, Path(used))
|
|
for used in process.get("uses", [])
|
|
)
|
|
if direct_use or (build_activity and process.get("workspace_related")):
|
|
errors.append(
|
|
f"active process {process.get('pid')} ({process.get('comm')}) may use the lane"
|
|
)
|
|
return errors
|
|
|
|
|
|
def _lane_by_name(snapshot: Mapping[str, Any], lane: str) -> dict[str, Any] | None:
|
|
return next((item for item in snapshot["lanes"] if item["lane"] == lane), None)
|
|
|
|
|
|
def _delete_lane(
|
|
lane: Mapping[str, Any],
|
|
*,
|
|
root: Path,
|
|
workflow: Mapping[str, Any],
|
|
apply: bool,
|
|
) -> dict[str, Any]:
|
|
paths = [Path(item["path"]) for item in lane["artifacts"]]
|
|
processes = active_processes(root, paths)
|
|
checks: list[dict[str, Any]] = []
|
|
for artifact in lane["artifacts"]:
|
|
errors = deletion_safety(
|
|
artifact, root=root, workflow=workflow, processes=processes
|
|
)
|
|
checks.append({"path": artifact["path"], "errors": errors})
|
|
if str(lane.get("lane")) == _candidate_lane_name(workflow):
|
|
checks.append(
|
|
{
|
|
"path": str(_candidate_preservation_path(root)),
|
|
"kind": "candidate-preservation",
|
|
"errors": _candidate_preservation_errors(
|
|
lane, root=root, workflow=workflow
|
|
),
|
|
}
|
|
)
|
|
blockers = [error for check in checks for error in check["errors"]]
|
|
result = {
|
|
"lane": lane["lane"],
|
|
"apply": apply,
|
|
"allocated_bytes": lane["allocated_bytes"],
|
|
"checks": checks,
|
|
"deleted": [],
|
|
"blocked": blockers,
|
|
}
|
|
if blockers or not apply:
|
|
return result
|
|
|
|
# Delete only complete, independently marked lane roots. rmtree does not
|
|
# follow directory symlinks, and the preflight above rejects every symlink.
|
|
for artifact in sorted(lane["artifacts"], key=lambda item: item["role"] == "build"):
|
|
path = Path(artifact["path"])
|
|
shutil.rmtree(path)
|
|
result["deleted"].append(str(path))
|
|
return result
|
|
|
|
|
|
def _execute_checked_deletions(
|
|
lanes: Sequence[Mapping[str, Any]],
|
|
*,
|
|
root: Path,
|
|
workflow: Mapping[str, Any],
|
|
action: str,
|
|
apply: bool,
|
|
output: Any | None = None,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
"""Print the complete checked plan before any requested deletion occurs."""
|
|
|
|
checked = [
|
|
_delete_lane(lane, root=root, workflow=workflow, apply=False)
|
|
for lane in lanes
|
|
]
|
|
plan = {
|
|
"timestamp": utc_now(),
|
|
"action": f"{action}-checked-plan",
|
|
"apply_requested": apply,
|
|
"total_allocated_bytes": sum(
|
|
int(result["allocated_bytes"]) for result in checked
|
|
),
|
|
"total": human_bytes(
|
|
sum(int(result["allocated_bytes"]) for result in checked)
|
|
),
|
|
"lanes": checked,
|
|
}
|
|
if not apply:
|
|
return plan, checked
|
|
|
|
# `flush=True` is part of the safety contract: the complete byte-counted,
|
|
# sentinel/process/candidate-evidence checked plan is externally visible
|
|
# before the first recursive deletion can begin.
|
|
print(
|
|
json.dumps(plan, indent=2, sort_keys=True),
|
|
file=output if output is not None else sys.stdout,
|
|
flush=True,
|
|
)
|
|
results = [
|
|
_delete_lane(lane, root=root, workflow=workflow, apply=True)
|
|
for lane in lanes
|
|
]
|
|
return plan, results
|
|
|
|
|
|
def _ledger_path(root: Path) -> Path:
|
|
return root / ".codex" / "session" / "build-storage.json"
|
|
|
|
|
|
def _atomic_json(path: Path, document: Mapping[str, Any]) -> None:
|
|
if path.is_symlink():
|
|
raise StorageError(f"refusing to replace symlinked JSON state: {path}")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
descriptor, temporary = tempfile.mkstemp(
|
|
prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
|
|
)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
|
json.dump(document, handle, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def write_ledger(root: Path, event: Mapping[str, Any]) -> None:
|
|
path = _ledger_path(root)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
document: dict[str, Any] = {"schema_version": LEDGER_SCHEMA_VERSION, "events": []}
|
|
if path.is_file() and not path.is_symlink():
|
|
try:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
existing = json.load(handle)
|
|
if isinstance(existing, dict) and isinstance(existing.get("events"), list):
|
|
document = existing
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
events = list(document.get("events", []))[-499:]
|
|
events.append(dict(event))
|
|
document["schema_version"] = LEDGER_SCHEMA_VERSION
|
|
document["events"] = events
|
|
document["updated_at"] = utc_now()
|
|
|
|
_atomic_json(path, document)
|
|
|
|
|
|
def _summary(snapshot: Mapping[str, Any], evaluation: Mapping[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"timestamp": snapshot["timestamp"],
|
|
"workspace": snapshot["workspace"],
|
|
"total_allocated_bytes": snapshot["total_allocated_bytes"],
|
|
"total": human_bytes(snapshot["total_allocated_bytes"]),
|
|
"repository_target": {
|
|
"path": snapshot["repository_target"]["path"],
|
|
"allocated_bytes": snapshot["repository_target"]["allocated_bytes"],
|
|
"human": human_bytes(snapshot["repository_target"]["allocated_bytes"]),
|
|
"largest": [
|
|
{"name": name, "bytes": size, "human": human_bytes(size)}
|
|
for name, size in list(
|
|
(snapshot["repository_target"].get("top_level") or {}).items()
|
|
)[:10]
|
|
],
|
|
},
|
|
"external_partition": {
|
|
"path": snapshot["external_partition"]["path"],
|
|
"allocated_bytes": snapshot["external_partition"]["allocated_bytes"],
|
|
"human": human_bytes(snapshot["external_partition"]["allocated_bytes"]),
|
|
},
|
|
"lanes": [
|
|
{
|
|
"lane": lane["lane"],
|
|
"bytes": lane["allocated_bytes"],
|
|
"human": human_bytes(lane["allocated_bytes"]),
|
|
"last_use": lane["last_use"],
|
|
"valid": lane["valid"],
|
|
}
|
|
for lane in snapshot["lanes"]
|
|
],
|
|
"legacy_intermediates": {
|
|
name: {
|
|
"path": usage["path"],
|
|
"bytes": usage["allocated_bytes"],
|
|
"human": human_bytes(usage["allocated_bytes"]),
|
|
}
|
|
for name, usage in snapshot["legacy_intermediates"].items()
|
|
},
|
|
"filesystems": [
|
|
{
|
|
**filesystem,
|
|
"free": human_bytes(filesystem["free_bytes"]),
|
|
"minimum_free": human_bytes(filesystem["minimum_free_bytes"]),
|
|
}
|
|
for filesystem in snapshot["filesystems"]
|
|
],
|
|
"active_processes": snapshot["active_processes"],
|
|
"sccache": snapshot["sccache"],
|
|
"warnings": evaluation["warnings"],
|
|
"violations": evaluation["violations"],
|
|
"simulated": snapshot.get("simulated", False),
|
|
}
|
|
|
|
|
|
def _print_human(summary: Mapping[str, Any]) -> None:
|
|
print(f"Build artifacts: {summary['total']}")
|
|
repository = summary["repository_target"]
|
|
print(f"Repository target: {repository['human']} ({repository['path']})")
|
|
for item in repository["largest"][:6]:
|
|
print(f" {item['name']}: {item['human']}")
|
|
external = summary["external_partition"]
|
|
print(f"Managed intermediates: {external['human']} ({external['path']})")
|
|
if summary["lanes"]:
|
|
print("Lanes:")
|
|
for lane in summary["lanes"]:
|
|
print(
|
|
f" {lane['lane']}: {lane['human']} "
|
|
f"({'valid' if lane['valid'] else 'INVALID SENTINEL'})"
|
|
)
|
|
if summary["legacy_intermediates"]:
|
|
print("Legacy intermediates still in repository target:")
|
|
for name, usage in summary["legacy_intermediates"].items():
|
|
print(f" {name}: {usage['human']}")
|
|
for filesystem in summary["filesystems"]:
|
|
print(
|
|
f"Free on {filesystem['mount']}: {filesystem['free']} "
|
|
f"(floor {filesystem['minimum_free']})"
|
|
)
|
|
if summary["active_processes"]:
|
|
print("Active build/artifact users:")
|
|
for process in summary["active_processes"]:
|
|
print(f" PID {process['pid']} {process['comm']}: {process['command']}")
|
|
for warning in summary["warnings"]:
|
|
print(f"WARNING: {warning}")
|
|
for violation in summary["violations"]:
|
|
print(f"BLOCK: {violation}")
|
|
|
|
|
|
def _snapshot_with_simulation(
|
|
root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace
|
|
) -> tuple[dict[str, Any], dict[str, list[str]]]:
|
|
snapshot = storage_snapshot(root, workflow)
|
|
apply_simulation(snapshot, arguments)
|
|
return snapshot, evaluate_budget(snapshot)
|
|
|
|
|
|
def _make_preserved_record(
|
|
path: Path, *, root: Path, lane_paths: Sequence[Path], label: str
|
|
) -> dict[str, Any]:
|
|
original = path.expanduser()
|
|
if not original.is_absolute():
|
|
original = root / original
|
|
if original.is_symlink():
|
|
raise StorageError(f"{label} must not be a symlink: {original}")
|
|
resolved = canonical_path(original)
|
|
if any(_path_within(resolved, lane_path) for lane_path in lane_paths):
|
|
raise StorageError(
|
|
f"{label} must be copied outside the disposable candidate lane: {resolved}"
|
|
)
|
|
if not resolved.is_file():
|
|
raise StorageError(f"{label} is not a file: {resolved}")
|
|
return {
|
|
"path": str(resolved),
|
|
"bytes": resolved.stat().st_size,
|
|
"sha256": _sha256_file(resolved),
|
|
}
|
|
|
|
|
|
def _mark_candidate_preserved(
|
|
root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace
|
|
) -> int:
|
|
snapshot = storage_snapshot(root, workflow)
|
|
lane_name = _candidate_lane_name(workflow)
|
|
lane = _lane_by_name(snapshot, lane_name)
|
|
if lane is None:
|
|
raise StorageError(f"managed candidate lane does not exist: {lane_name}")
|
|
if not lane.get("valid"):
|
|
raise StorageError("candidate lane has an invalid sentinel")
|
|
generation = _lane_generation(lane)
|
|
if generation is None:
|
|
raise StorageError("candidate lane generation cannot be established")
|
|
lane_paths = [canonical_path(Path(item["path"])) for item in lane["artifacts"]]
|
|
evidence = _make_preserved_record(
|
|
arguments.evidence,
|
|
root=root,
|
|
lane_paths=lane_paths,
|
|
label="candidate evidence manifest",
|
|
)
|
|
artifacts = [
|
|
_make_preserved_record(
|
|
path,
|
|
root=root,
|
|
lane_paths=lane_paths,
|
|
label=f"preserved candidate artifact {index}",
|
|
)
|
|
for index, path in enumerate(arguments.artifact)
|
|
]
|
|
evidence_path = evidence["path"]
|
|
if all(record["path"] == evidence_path for record in artifacts):
|
|
raise StorageError(
|
|
"candidate evidence alone is insufficient; preserve at least one distinct artifact"
|
|
)
|
|
|
|
document = {
|
|
"schema_version": CANDIDATE_PRESERVATION_SCHEMA_VERSION,
|
|
"workspace": str(canonical_path(root)),
|
|
"workspace_hash": workspace_hash(root),
|
|
"lane": lane_name,
|
|
"lane_generation": generation,
|
|
"prune_safe": True,
|
|
"preserved_at": utc_now(),
|
|
"evidence": evidence,
|
|
"preserved_artifacts": artifacts,
|
|
}
|
|
marker_path = _candidate_preservation_path(root)
|
|
_atomic_json(marker_path, document)
|
|
errors = _candidate_preservation_errors(lane, root=root, workflow=workflow)
|
|
if errors:
|
|
raise StorageError("candidate preservation marker failed validation: " + "; ".join(errors))
|
|
write_ledger(
|
|
root,
|
|
{
|
|
"timestamp": utc_now(),
|
|
"action": "mark-candidate-preserved",
|
|
"lane": lane_name,
|
|
"lane_generation": generation,
|
|
"marker": str(marker_path),
|
|
"evidence": evidence,
|
|
"preserved_artifacts": artifacts,
|
|
},
|
|
)
|
|
print(json.dumps(document, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
def _status(root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace) -> int:
|
|
snapshot, evaluation = _snapshot_with_simulation(root, workflow, arguments)
|
|
summary = _summary(snapshot, evaluation)
|
|
if arguments.json:
|
|
print(json.dumps(summary, indent=2, sort_keys=True))
|
|
else:
|
|
_print_human(summary)
|
|
return 0
|
|
|
|
|
|
def _plan(root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace) -> int:
|
|
snapshot, evaluation = _snapshot_with_simulation(root, workflow, arguments)
|
|
document = {
|
|
"summary": _summary(snapshot, evaluation),
|
|
"prune": prune_plan(snapshot),
|
|
"legacy_migration": {
|
|
"required": bool(snapshot["legacy_intermediates"]),
|
|
"automatic": False,
|
|
"reason": (
|
|
"unmanaged repository intermediates require the separately authorized "
|
|
"one-time migration; they are never removed by prune"
|
|
),
|
|
},
|
|
}
|
|
if arguments.json:
|
|
print(json.dumps(document, indent=2, sort_keys=True))
|
|
else:
|
|
_print_human(document["summary"])
|
|
print("Prune plan:")
|
|
for item in document["prune"]:
|
|
disposition = "eligible" if item["eligible"] else "keep"
|
|
print(f" {item['lane']}: {disposition} ({item['reason']})")
|
|
if document["legacy_migration"]["required"]:
|
|
print("Legacy target migration is required and will not be auto-pruned.")
|
|
return 0
|
|
|
|
|
|
def _prune(root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace) -> int:
|
|
snapshot, evaluation = _snapshot_with_simulation(root, workflow, arguments)
|
|
if snapshot.get("simulated") and arguments.apply:
|
|
raise StorageError("--apply is forbidden with simulated inputs")
|
|
candidates = {item["lane"]: item for item in prune_plan(snapshot) if item["eligible"]}
|
|
lanes: list[dict[str, Any]] = []
|
|
for lane_name in sorted(candidates):
|
|
lane = _lane_by_name(snapshot, lane_name)
|
|
if lane is not None:
|
|
lanes.append(lane)
|
|
checked_plan, results = _execute_checked_deletions(
|
|
lanes,
|
|
root=root,
|
|
workflow=workflow,
|
|
action="prune",
|
|
apply=bool(arguments.apply),
|
|
)
|
|
event = {
|
|
"timestamp": utc_now(),
|
|
"action": "prune",
|
|
"apply": bool(arguments.apply),
|
|
"before_bytes": snapshot["total_allocated_bytes"],
|
|
"checked_plan": checked_plan,
|
|
"results": results,
|
|
"violations": evaluation["violations"],
|
|
}
|
|
if arguments.apply:
|
|
write_ledger(root, event)
|
|
print(json.dumps(event, indent=2, sort_keys=True))
|
|
return 2 if any(result["blocked"] for result in results) else 0
|
|
|
|
|
|
def _reset_lane(root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace) -> int:
|
|
lane_name = validate_lane_name(arguments.lane)
|
|
snapshot, _ = _snapshot_with_simulation(root, workflow, arguments)
|
|
if snapshot.get("simulated") and arguments.apply:
|
|
raise StorageError("--apply is forbidden with simulated inputs")
|
|
lane = _lane_by_name(snapshot, lane_name)
|
|
if lane is None:
|
|
raise StorageError(f"managed lane does not exist: {lane_name}")
|
|
checked_plan, results = _execute_checked_deletions(
|
|
[lane],
|
|
root=root,
|
|
workflow=workflow,
|
|
action="reset-lane",
|
|
apply=bool(arguments.apply),
|
|
)
|
|
result = results[0]
|
|
event = {
|
|
"timestamp": utc_now(),
|
|
"action": "reset-lane",
|
|
"apply": bool(arguments.apply),
|
|
"checked_plan": checked_plan,
|
|
"result": result,
|
|
}
|
|
if arguments.apply:
|
|
write_ledger(root, event)
|
|
print(json.dumps(event, indent=2, sort_keys=True))
|
|
return 2 if result["blocked"] else 0
|
|
|
|
|
|
def _enforce(root: Path, workflow: Mapping[str, Any], arguments: argparse.Namespace) -> int:
|
|
snapshot, evaluation = _snapshot_with_simulation(root, workflow, arguments)
|
|
before_bytes = int(snapshot["total_allocated_bytes"])
|
|
before_warnings = list(evaluation["warnings"])
|
|
before_violations = list(evaluation["violations"])
|
|
pressure = bool(evaluation["violations"] or evaluation["warnings"])
|
|
eligible = [item for item in prune_plan(snapshot) if item["eligible"]]
|
|
checked_plan: dict[str, Any] | None = None
|
|
prune_results: list[dict[str, Any]] = []
|
|
policy = _policy(workflow)
|
|
auto_prune = bool(policy.get("auto_prune_disposable", True))
|
|
|
|
if pressure and eligible and auto_prune and not snapshot.get("simulated"):
|
|
lanes = [
|
|
lane
|
|
for item in eligible
|
|
if (lane := _lane_by_name(snapshot, str(item["lane"]))) is not None
|
|
]
|
|
checked_plan, prune_results = _execute_checked_deletions(
|
|
lanes,
|
|
root=root,
|
|
workflow=workflow,
|
|
action=f"enforce-{arguments.phase}",
|
|
apply=True,
|
|
# Keep --json stdout machine-readable while still making the
|
|
# mandatory pre-delete plan visible before recursive deletion.
|
|
output=sys.stderr if arguments.json else sys.stdout,
|
|
)
|
|
snapshot = storage_snapshot(root, workflow)
|
|
evaluation = evaluate_budget(snapshot)
|
|
|
|
event = {
|
|
"timestamp": utc_now(),
|
|
"action": "enforce",
|
|
"phase": arguments.phase,
|
|
"before_bytes": before_bytes,
|
|
"total_allocated_bytes": snapshot["total_allocated_bytes"],
|
|
"reclaimed_bytes": max(
|
|
0, before_bytes - int(snapshot["total_allocated_bytes"])
|
|
),
|
|
"warnings_before_prune": before_warnings,
|
|
"violations_before_prune": before_violations,
|
|
"warnings": evaluation["warnings"],
|
|
"violations": evaluation["violations"],
|
|
"auto_prune_enabled": auto_prune,
|
|
"auto_prune_would_apply": bool(
|
|
pressure and eligible and auto_prune and snapshot.get("simulated")
|
|
),
|
|
"destructive": any(result["deleted"] for result in prune_results),
|
|
"checked_plan": checked_plan,
|
|
"prune_results": prune_results,
|
|
"eligible_prune_bytes": sum(
|
|
int(item["allocated_bytes"]) for item in eligible
|
|
),
|
|
"eligible_prune_lanes": [item["lane"] for item in eligible],
|
|
"prune_recommended": pressure and bool(eligible),
|
|
"simulated": snapshot.get("simulated", False),
|
|
}
|
|
if not snapshot.get("simulated"):
|
|
write_ledger(root, event)
|
|
if arguments.json:
|
|
print(json.dumps(event, indent=2, sort_keys=True))
|
|
else:
|
|
_print_human(_summary(snapshot, evaluation))
|
|
if pressure and eligible:
|
|
if auto_prune and snapshot.get("simulated"):
|
|
print("Simulation only: expired disposable lanes would be pruned.")
|
|
elif not auto_prune:
|
|
print(
|
|
"Automatic pruning is disabled. Review `build_storage.py prune "
|
|
"--dry-run` before any explicit apply."
|
|
)
|
|
return 2 if evaluation["violations"] else 0
|
|
|
|
|
|
def _write_test_marker(
|
|
path: Path,
|
|
root: Path,
|
|
lane: str,
|
|
role: str,
|
|
*,
|
|
last_use: str | None = None,
|
|
) -> None:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
document = {
|
|
"schema_version": 1,
|
|
"workspace": str(canonical_path(root)),
|
|
"workspace_hash": workspace_hash(root),
|
|
"lane": lane,
|
|
"role": role,
|
|
"path": str(canonical_path(path)),
|
|
"created_at": utc_now(),
|
|
"last_use": last_use or utc_now(),
|
|
}
|
|
sentinel_path(path).write_text(json.dumps(document), encoding="utf-8")
|
|
|
|
|
|
def _self_test() -> dict[str, Any]:
|
|
checks: list[str] = []
|
|
fake = {
|
|
"policy": dict(DEFAULT_POLICY),
|
|
"total_allocated_bytes": 81 * GIB,
|
|
"lanes": [],
|
|
"filesystems": [
|
|
{"mount": "/test", "free_bytes": 100 * GIB, "minimum_free_bytes": 40 * GIB}
|
|
],
|
|
}
|
|
assert evaluate_budget(fake)["violations"]
|
|
checks.append("simulated-hard-budget-crossing")
|
|
fake["total_allocated_bytes"] = 56 * GIB
|
|
soft = evaluate_budget(fake)
|
|
assert not soft["violations"] and soft["warnings"]
|
|
checks.append("simulated-soft-budget-crossing")
|
|
fake["total_allocated_bytes"] = 1 * GIB
|
|
fake["filesystems"][0]["free_bytes"] = 39 * GIB
|
|
assert evaluate_budget(fake)["violations"]
|
|
checks.append("simulated-free-space-floor")
|
|
|
|
with tempfile.TemporaryDirectory(prefix="blacksite-storage-self-test-") as temporary:
|
|
base = Path(temporary)
|
|
root = base / "workspace"
|
|
(root / ".git").mkdir(parents=True)
|
|
(root / "Cargo.toml").write_text("[workspace]\nmembers=[]\n", encoding="utf-8")
|
|
cache = base / "cache"
|
|
workflow = {
|
|
"build_storage": {
|
|
**DEFAULT_POLICY,
|
|
"cache_root": str(cache),
|
|
"workspace_partition": "canonical-path-hash",
|
|
"exceptional_target_root": "target/lanes",
|
|
}
|
|
}
|
|
lane_path = cache / workspace_hash(root) / "candidate"
|
|
_write_test_marker(lane_path, root, "candidate", "build")
|
|
artifact = {
|
|
"path": str(lane_path),
|
|
"lane": "candidate",
|
|
"role": "build",
|
|
}
|
|
assert not deletion_safety(
|
|
artifact, root=root, workflow=workflow, processes=[]
|
|
)
|
|
checks.append("valid-sentinel-path-accepted")
|
|
|
|
link = lane_path / "escape"
|
|
link.symlink_to(root)
|
|
errors = deletion_safety(
|
|
artifact, root=root, workflow=workflow, processes=[]
|
|
)
|
|
assert any("symlink" in error for error in errors)
|
|
checks.append("internal-symlink-rejected")
|
|
link.unlink()
|
|
|
|
errors = deletion_safety(
|
|
artifact,
|
|
root=root,
|
|
workflow=workflow,
|
|
processes=[
|
|
{
|
|
"pid": 42,
|
|
"comm": "cargo",
|
|
"uses": [str(lane_path)],
|
|
"build_tool": True,
|
|
"packager": False,
|
|
"workspace_related": True,
|
|
}
|
|
],
|
|
)
|
|
assert any("active process" in error for error in errors)
|
|
checks.append("active-process-rejected")
|
|
|
|
unmarked = cache / workspace_hash(root) / "unmarked"
|
|
unmarked.mkdir(parents=True)
|
|
errors = deletion_safety(
|
|
{"path": str(unmarked), "lane": "unmarked", "role": "build"},
|
|
root=root,
|
|
workflow=workflow,
|
|
processes=[],
|
|
)
|
|
assert any("missing or malformed" in error for error in errors)
|
|
checks.append("unmarked-lane-rejected")
|
|
|
|
for protected_path, label in (
|
|
(root, "workspace-root"),
|
|
(root.parent, "workspace-parent"),
|
|
(Path.home(), "home"),
|
|
(cargo_home(), "cargo-home"),
|
|
):
|
|
errors = deletion_safety(
|
|
{
|
|
"path": str(protected_path),
|
|
"lane": "candidate",
|
|
"role": "build",
|
|
},
|
|
root=root,
|
|
workflow=workflow,
|
|
processes=[],
|
|
)
|
|
assert errors
|
|
checks.append(f"{label}-rejected")
|
|
|
|
# Candidate deletion requires a generation-bound marker whose evidence
|
|
# and preserved artifacts still exist outside the disposable lane.
|
|
dev_path = cache / workspace_hash(root) / "dev"
|
|
_write_test_marker(dev_path, root, "dev", "build")
|
|
evidence = base / "evidence" / "candidate-manifest.json"
|
|
evidence.parent.mkdir(parents=True)
|
|
evidence.write_text('{"candidate":"test"}\n', encoding="utf-8")
|
|
preserved_artifact = base / "preserved" / "editor-package.tar.zst"
|
|
preserved_artifact.parent.mkdir(parents=True)
|
|
preserved_artifact.write_bytes(b"preserved candidate package")
|
|
lane = {
|
|
"lane": "candidate",
|
|
"allocated_bytes": measure_tree(lane_path).allocated_bytes,
|
|
"artifacts": [artifact],
|
|
}
|
|
|
|
blocked_output = io.StringIO()
|
|
_, blocked_results = _execute_checked_deletions(
|
|
[lane],
|
|
root=root,
|
|
workflow=workflow,
|
|
action="self-test-candidate",
|
|
apply=True,
|
|
output=blocked_output,
|
|
)
|
|
assert blocked_results[0]["blocked"] and lane_path.exists()
|
|
assert any(
|
|
"preservation marker is missing" in error
|
|
for error in blocked_results[0]["blocked"]
|
|
)
|
|
checks.append("candidate-without-preservation-marker-rejected")
|
|
|
|
mark_arguments = argparse.Namespace(
|
|
evidence=evidence, artifact=[preserved_artifact]
|
|
)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
assert _mark_candidate_preserved(root, workflow, mark_arguments) == 0
|
|
assert not _candidate_preservation_errors(
|
|
lane, root=root, workflow=workflow
|
|
)
|
|
checks.append("candidate-preservation-marker-validated")
|
|
|
|
class FlushProbe(io.StringIO):
|
|
def __init__(self, guarded_path: Path) -> None:
|
|
super().__init__()
|
|
self.guarded_path = guarded_path
|
|
self.flushed_before_delete = False
|
|
|
|
def flush(self) -> None:
|
|
assert self.guarded_path.exists()
|
|
self.flushed_before_delete = True
|
|
super().flush()
|
|
|
|
output = FlushProbe(lane_path)
|
|
checked_plan, results = _execute_checked_deletions(
|
|
[lane],
|
|
root=root,
|
|
workflow=workflow,
|
|
action="self-test-candidate",
|
|
apply=True,
|
|
output=output,
|
|
)
|
|
deletion = results[0]
|
|
assert output.flushed_before_delete
|
|
assert checked_plan["total_allocated_bytes"] == lane["allocated_bytes"]
|
|
assert "self-test-candidate-checked-plan" in output.getvalue()
|
|
assert not deletion["blocked"] and deletion["deleted"]
|
|
assert not lane_path.exists()
|
|
assert dev_path.exists() and evidence.is_file() and preserved_artifact.is_file()
|
|
checks.append("checked-plan-flushed-before-candidate-prune")
|
|
checks.append("candidate-pruned-artifact-evidence-and-dev-preserved")
|
|
|
|
# Recreating the candidate lane invalidates the old generation-bound
|
|
# marker, so a stale preservation decision cannot authorize deletion.
|
|
_write_test_marker(
|
|
lane_path,
|
|
root,
|
|
"candidate",
|
|
"build",
|
|
last_use="2099-01-01T00:00:00+00:00",
|
|
)
|
|
lane["allocated_bytes"] = measure_tree(lane_path).allocated_bytes
|
|
stale = _delete_lane(lane, root=root, workflow=workflow, apply=False)
|
|
assert any("lane_generation" in error for error in stale["blocked"])
|
|
checks.append("stale-candidate-preservation-marker-rejected")
|
|
|
|
# Enforcement reports pressure and eligible lanes but never performs a
|
|
# deletion. Auto-prune must emit its complete checked plan first.
|
|
hot_path = cache / workspace_hash(root) / "hot-reload"
|
|
_write_test_marker(
|
|
hot_path,
|
|
root,
|
|
"hot-reload",
|
|
"build",
|
|
last_use="2000-01-01T00:00:00+00:00",
|
|
)
|
|
workflow["build_storage"].update(
|
|
{
|
|
"soft_total_gib": 0,
|
|
"hard_total_gib": 100000,
|
|
"min_free_gib": 0,
|
|
"min_free_percent": 0,
|
|
"hot_reload_retention_hours": 0,
|
|
"auto_prune_disposable": True,
|
|
}
|
|
)
|
|
enforce_arguments = argparse.Namespace(
|
|
phase="pre",
|
|
json=True,
|
|
simulate_total_gib=None,
|
|
simulate_free_gib=None,
|
|
simulate_lane_gib=[],
|
|
)
|
|
enforce_output = io.StringIO()
|
|
enforce_plan_output = io.StringIO()
|
|
with contextlib.redirect_stdout(enforce_output), contextlib.redirect_stderr(
|
|
enforce_plan_output
|
|
):
|
|
assert _enforce(root, workflow, enforce_arguments) == 0
|
|
enforce_event = json.loads(enforce_output.getvalue())
|
|
enforce_plan = json.loads(enforce_plan_output.getvalue())
|
|
assert enforce_plan["action"] == "enforce-pre-checked-plan"
|
|
assert enforce_plan["total_allocated_bytes"] > 0
|
|
assert enforce_event["destructive"] is True
|
|
assert "hot-reload" in enforce_event["eligible_prune_lanes"]
|
|
assert not hot_path.exists()
|
|
checks.append("enforce-pressure-prunes-after-checked-plan")
|
|
|
|
return {"ok": True, "checks": checks}
|
|
|
|
|
|
def _add_simulation_options(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--simulate-total-gib", type=float)
|
|
parser.add_argument("--simulate-free-gib", type=float)
|
|
parser.add_argument(
|
|
"--simulate-lane-gib",
|
|
action="append",
|
|
default=[],
|
|
metavar="LANE=GIB",
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--workspace", type=Path, help="workspace root override")
|
|
subparsers = parser.add_subparsers(dest="action", required=True)
|
|
|
|
for name in ("status", "plan"):
|
|
command = subparsers.add_parser(name)
|
|
command.add_argument("--json", action="store_true")
|
|
_add_simulation_options(command)
|
|
|
|
enforce = subparsers.add_parser("enforce")
|
|
enforce.add_argument("--phase", choices=("pre", "post"), required=True)
|
|
enforce.add_argument("--json", action="store_true")
|
|
_add_simulation_options(enforce)
|
|
|
|
prune = subparsers.add_parser("prune")
|
|
mode = prune.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--dry-run", action="store_true")
|
|
mode.add_argument("--apply", action="store_true")
|
|
_add_simulation_options(prune)
|
|
|
|
reset = subparsers.add_parser("reset-lane")
|
|
reset.add_argument("lane")
|
|
mode = reset.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--dry-run", action="store_true")
|
|
mode.add_argument("--apply", action="store_true")
|
|
_add_simulation_options(reset)
|
|
|
|
preserved = subparsers.add_parser("mark-candidate-preserved")
|
|
preserved.add_argument(
|
|
"--evidence",
|
|
type=Path,
|
|
required=True,
|
|
help="preserved candidate evidence manifest outside the disposable lane",
|
|
)
|
|
preserved.add_argument(
|
|
"--artifact",
|
|
type=Path,
|
|
action="append",
|
|
required=True,
|
|
help="preserved candidate binary/package file; repeat for multiple artifacts",
|
|
)
|
|
|
|
subparsers.add_parser("self-test")
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
arguments = build_parser().parse_args(argv)
|
|
if arguments.action == "self-test":
|
|
print(json.dumps(_self_test(), indent=2, sort_keys=True))
|
|
return 0
|
|
try:
|
|
root = workspace_root(arguments.workspace)
|
|
workflow = load_workflow(root)
|
|
if arguments.action == "status":
|
|
return _status(root, workflow, arguments)
|
|
if arguments.action == "plan":
|
|
return _plan(root, workflow, arguments)
|
|
if arguments.action == "enforce":
|
|
return _enforce(root, workflow, arguments)
|
|
if arguments.action == "prune":
|
|
return _prune(root, workflow, arguments)
|
|
if arguments.action == "reset-lane":
|
|
return _reset_lane(root, workflow, arguments)
|
|
if arguments.action == "mark-candidate-preserved":
|
|
return _mark_candidate_preserved(root, workflow, arguments)
|
|
except (LaneError, StorageError, OSError, ValueError) as error:
|
|
print(f"build-storage: {error}", file=sys.stderr)
|
|
return 2
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|