Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file.

### Fixed

- Rotate bounded byte-retention size walks across invocations with a persisted
advisory cursor so bundles beyond the first scan budget are eventually seen.
- Preserve run bundles while their lease is held through terminal metadata
writing and cleanup; recheck lease state before deletion.
- Preserve explicit application identities losslessly while using
Expand Down
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -897,7 +897,9 @@ Recovery work is bounded on the foreground command path. Count- and age-only
policies inspect metadata without recursively sizing bundle contents. A byte
policy performs at most 512 recursive size walks and removes at most 256
bundles per pass; any remaining policy debt is retained safely and reported as
a warning for a later invocation. The diagnostic index records at most 512
a warning for a later invocation. An atomic advisory cursor rotates the size
walk across invocations, so repeated passes eventually inspect the full set;
the cursor never authorizes deletion. The diagnostic index records at most 512
entries and sets `complete: false` plus `omitted_bundles` when a cache is
larger, so a stale, corrupt, or missing index is always reconciled from the
filesystem rather than trusted for deletion.
Expand Down
5 changes: 3 additions & 2 deletions docs/performance.md
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,9 @@ foreground pass (protected bundles and unreadable entries are retained):
When a bound prevents a complete reconciliation, base-cli leaves the
unprocessed bundles intact, writes a partial index with `complete: false`, and
emits a warning describing the remaining policy debt. A later invocation
continues from the filesystem; the index is an observation aid, never an
authorization to delete a path. The retention regression suite covers count,
continues from the filesystem. An atomic advisory cursor rotates the bounded
byte-size walk across invocations, including after process restart; the index
is an observation aid, never an authorization to delete a path. The retention regression suite covers count,
age, byte limits, deep trees, corrupt metadata/index files, unreadable files,
concurrent invocations, and live-run lease protection.

Expand Down
78 changes: 58 additions & 20 deletions lib/python/base_cli/_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -414,13 +414,15 @@ def prune_run_bundles(
# outside the lock. The destructive phase revalidates each candidate
# under the lock so another invocation can never turn a live bundle into a
# deletion candidate while discovery is in progress.
bundles = _discover_run_bundles(
size_scan_cursor = _read_size_scan_cursor(runs_root)
bundles, size_scan_cursor = _discover_run_bundles(
runs_root,
protected=protected,
max_age_seconds=effective.max_age_seconds,
now=clock,
measure_sizes=effective.max_total_bytes is not None,
size_budget=_RETENTION_SIZE_MEASUREMENT_BUDGET,
size_scan_cursor=size_scan_cursor,
)
try:
with _retention_lock(runs_root):
Expand All @@ -439,6 +441,7 @@ def prune_run_bundles(
log,
current_run_root=current_run_root,
now=clock,
size_scan_cursor=size_scan_cursor,
)
except (OSError, RuntimeError) as exc:
# Retention is maintenance. An unavailable lock or a transient
Expand All @@ -459,7 +462,7 @@ def refresh_run_bundle_index(
if not runs_root.exists() or runs_root.is_symlink():
return
try:
bundles = _discover_run_bundles(
bundles, _size_scan_cursor = _discover_run_bundles(
runs_root,
protected=set(),
max_age_seconds=None,
Expand All @@ -481,13 +484,13 @@ def _discover_run_bundles(
now: float,
measure_sizes: bool,
size_budget: int,
) -> list[dict[str, Any]]:
size_scan_cursor: str | None = None,
) -> tuple[list[dict[str, Any]], str | None]:
bundles: list[dict[str, Any]] = []
measured_sizes = 0
try:
children = sorted(runs_root.iterdir(), key=lambda path: path.name)
except OSError:
return bundles
return bundles, size_scan_cursor
for child in children:
if child.name.startswith(".") or child.is_symlink() or not child.is_dir():
continue
Expand Down Expand Up @@ -517,18 +520,6 @@ def _discover_run_bundles(
if status not in {"running", "ok", "aborted", "error"}:
continue
resolved = _safe_resolved_path(child)
size = 0
size_known = False
if measure_sizes and measured_sizes < size_budget:
try:
size = _bundle_size(child)
size_known = True
measured_sizes += 1
except OSError:
# A file that disappears or becomes unreadable remains a
# retention candidate for count/age policy, but its byte
# contribution is unknown and must be reported below.
pass
retention_metadata = metadata.get("retention")
preserve = bool(metadata.get("preserve")) or (
isinstance(retention_metadata, dict) and retention_metadata.get("preserve") is True
Expand All @@ -541,14 +532,59 @@ def _discover_run_bundles(
"status": status,
"started_at": started_at,
"age": age,
"size": size,
"size_known": size_known,
"size": 0,
"size_known": False,
"preserve": preserve,
"protected": resolved in protected,
}
)
bundles.sort(key=lambda bundle: (float(bundle["started_at"]), str(bundle["path"])))
return bundles
if measure_sizes and bundles and size_budget > 0:
# The run index's cursor affects only which discovered bundles receive
# an expensive size walk. It never authorizes deletion; every candidate
# is re-read and revalidated before the destructive phase.
scan_order = sorted(bundles, key=lambda bundle: Path(bundle["path"]).name)
if size_scan_cursor is not None:
start_index = next(
(index for index, bundle in enumerate(scan_order) if Path(bundle["path"]).name > size_scan_cursor),
0,
)
scan_order = scan_order[start_index:] + scan_order[:start_index]
attempted = 0
for bundle in scan_order:
if attempted >= size_budget:
break
attempted += 1
path = Path(bundle["path"])
size_scan_cursor = path.name
try:
bundle["size"] = _bundle_size(path)
bundle["size_known"] = True
except OSError:
# A file that disappears or becomes unreadable remains a
# retention candidate for count/age policy, but its byte
# contribution is unknown and reported below. Advancing the
# cursor prevents one unreadable entry from starving others.
pass
return bundles, size_scan_cursor


def _read_size_scan_cursor(runs_root: Path) -> str | None:
"""Read the advisory byte-scan cursor; never use it to select deletions."""

index_path = runs_root / _RUN_INDEX_NAME
try:
if index_path.is_symlink() or not index_path.is_file() or index_path.stat().st_size > 1_048_576:
return None
payload = json.loads(index_path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
return None
if not isinstance(payload, dict):
return None
cursor = payload.get("byte_scan_cursor")
if not isinstance(cursor, str) or not cursor or len(cursor) > 1024 or "/" in cursor or "\\" in cursor:
return None
return cursor


def _apply_bundle_retention(
Expand Down Expand Up @@ -683,6 +719,7 @@ def _write_run_index(
*,
current_run_root: Path | None = None,
now: float | None = None,
size_scan_cursor: str | None = None,
) -> None:
indexed = list(bundles)
if current_run_root is not None and current_run_root.exists():
Expand All @@ -708,6 +745,7 @@ def _write_run_index(
"version": 1,
"complete": omitted_bundles == 0,
"omitted_bundles": omitted_bundles,
"byte_scan_cursor": size_scan_cursor if size_scan_cursor is not None else _read_size_scan_cursor(runs_root),
"bundles": [
{
"path": str(bundle["path"]),
Expand Down
3 changes: 3 additions & 0 deletions tests/test_adversarial_regressions.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,9 @@ def test_run_bundle_retention_remains_bounded_across_processes(self) -> None:
"preserve": False,
},
)
# Retention now fails closed if a bundle has no lease record,
# because missing liveness cannot prove that it is inactive.
(bundle / ".base-cli-run-lease").write_bytes(b"0")

_run_processes(_prune_worker, [(str(runs_root),) for _seed in SEEDS])

Expand Down
3 changes: 2 additions & 1 deletion tests/test_platform_edge_paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,13 +146,14 @@ def test_retention_scan_and_apply_are_portable_without_recursive_sizes(self) ->
for index in range(3):
bundle = root / f"run-{index}"
bundle.mkdir()
(bundle / ".base-cli-run-lease").write_bytes(b"0")
(bundle / "run.json").write_text(
f'{{"run_id": "run-{index}", "status": "ok", '
'"started_at": "2020-01-01T00:00:00Z", "preserve": false}',
encoding="utf-8",
)
with mock.patch.object(runtime, "_bundle_size", side_effect=AssertionError("unexpected size walk")):
bundles = runtime._discover_run_bundles( # pylint: disable=protected-access
bundles, _size_scan_cursor = runtime._discover_run_bundles( # pylint: disable=protected-access
root,
protected=set(),
max_age_seconds=None,
Expand Down
34 changes: 34 additions & 0 deletions tests/test_run_bundle_retention.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,40 @@ def test_byte_retention_bounds_recursive_size_work(self) -> None:
self.assertLessEqual(bundle_size.call_count, 512)
self.assertTrue(any("size walk(s)" in str(call) for call in logger.warning.call_args_list))

def test_byte_retention_cursor_advances_across_restarted_bounded_passes(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "runs"
root.mkdir()
expected_names = {f"run-{index:05d}" for index in range(1_025)}
for index in range(1_025):
_bundle(
root,
f"run-{index:05d}",
preserve=index < 1_024,
size=1_048_576 if index == 1_024 else 1,
)

scanned: set[str] = set()
policy = RetentionPolicy(max_total_bytes=400_000)
for pass_number in range(1, 4):
with mock.patch.object(runtime, "_bundle_size", wraps=runtime._bundle_size) as bundle_size:
prune_run_bundles(
root,
policy=policy,
logger=logging.getLogger(__name__),
now=1_600_000_000,
)
scanned.update(Path(call.args[0]).name for call in bundle_size.call_args_list)
self.assertLessEqual(bundle_size.call_count, 512)
index = json.loads((root / ".base-cli-run-index.json").read_text(encoding="utf-8"))
self.assertIn("byte_scan_cursor", index)
if pass_number < 3:
self.assertTrue((root / "run-01024").exists())

self.assertTrue(expected_names <= scanned)
self.assertFalse((root / "run-01024").exists())
self.assertTrue(all((root / name).exists() for name in expected_names if name != "run-01024"))

def test_corrupt_index_is_reconciled_without_trusting_paths(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "runs"
Expand Down
Loading