diff --git a/CHANGELOG.md b/CHANGELOG.md index f94444e..2361750 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md index e080b6a..0429834 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/docs/performance.md b/docs/performance.md index a46b881..acddf56 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -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. diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index 86f587c..79ad545 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -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): @@ -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 @@ -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, @@ -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 @@ -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 @@ -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( @@ -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(): @@ -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"]), diff --git a/tests/test_adversarial_regressions.py b/tests/test_adversarial_regressions.py index 4b1fc32..306fcc7 100644 --- a/tests/test_adversarial_regressions.py +++ b/tests/test_adversarial_regressions.py @@ -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]) diff --git a/tests/test_platform_edge_paths.py b/tests/test_platform_edge_paths.py index e83fe3b..9735b38 100644 --- a/tests/test_platform_edge_paths.py +++ b/tests/test_platform_edge_paths.py @@ -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, diff --git a/tests/test_run_bundle_retention.py b/tests/test_run_bundle_retention.py index 2db453c..e0957e1 100644 --- a/tests/test_run_bundle_retention.py +++ b/tests/test_run_bundle_retention.py @@ -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"