diff --git a/CHANGELOG.md b/CHANGELOG.md index ec20c2e..2361750 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ 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 collision-resistant, path-safe runtime namespace components. - Give `BatteriesIncludedConfigLoader.cli_name` a documented identity role by diff --git a/README.md b/README.md index 23dc8dd..0429834 100644 --- a/README.md +++ b/README.md @@ -883,8 +883,9 @@ app = base_cli.App( Retention runs during startup after the current run's default log file is resolved. The active invocation, inherited parent bundle, and bundles marked `preserve` (including `--keep-temp`) are never removed. Each lifecycle-owned -running bundle also holds an advisory `.base-cli-run-lease` for its lifetime; -retention never removes a bundle whose lease is active. A stale `running` +bundle holds an advisory `.base-cli-run-lease` through final cleanup, including +the brief period after metadata becomes terminal; retention never removes a +bundle whose lease is active or whose liveness cannot be established. A stale `running` bundle is eligible for crash recovery only when an age bound is configured and its lease can be acquired, proving that the original process has exited. Missing, unreadable, or unsupported leases fail closed and remain retained for @@ -896,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 46de122..778f9c2 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -410,20 +410,20 @@ def prune_run_bundles( protected.add(_safe_resolved_path(current_run_root)) clock = time.time() if now is None else now - # Filesystem discovery and recursive size accounting are deliberately - # 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( - 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, - ) try: with _retention_lock(runs_root): + # Keep cursor read, size walk, and index update in one critical + # section so concurrent pruners cannot overwrite scan progress. + 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, + ) _apply_bundle_retention( runs_root, bundles, @@ -439,6 +439,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 +460,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 +482,14 @@ 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 + scan_order: list[dict[str, Any]] = [] 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 @@ -506,28 +508,17 @@ def _discover_run_bundles( continue age = max(0.0, now - started_at) running = status == "running" + # The owner keeps its lease through cleanup, which occurs after the + # run metadata has been made terminal. Liveness therefore protects + # every state, not only the transient "running" state. + if _run_lease_state(child) != "inactive": + continue if running: - # A running record is removable only when the lease proves that - # its owner has exited. Missing or unreadable leases fail closed. - if _run_lease_state(child) != "inactive": - continue if max_age_seconds is None or age < max_age_seconds: continue 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 @@ -540,14 +531,60 @@ 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, } ) + if measure_sizes: + scan_order.append(bundles[-1]) 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. + if size_scan_cursor is not None: + start_index = next( + (index for index, bundle in enumerate(scan_order) if 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 = 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( @@ -652,9 +689,9 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa if metadata is None: return False status = str(metadata.get("status", "")) + if _run_lease_state(path) != "inactive": + return False if status == "running": - if _run_lease_state(path) != "inactive": - return False if policy.max_age_seconds is None: return False started_at = _timestamp_to_epoch(metadata.get("started_at")) @@ -682,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(): @@ -707,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..81e902c 100644 --- a/tests/test_adversarial_regressions.py +++ b/tests/test_adversarial_regressions.py @@ -87,7 +87,7 @@ def _write_log_worker(path_text: str, seed: int, count: int) -> None: def _prune_worker(runs_root_text: str) -> None: prune_run_bundles( Path(runs_root_text), - policy=base_cli.RetentionPolicy(max_bundles=2), + policy=base_cli.RetentionPolicy(max_bundles=2, max_total_bytes=2), ) @@ -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_app_run_metadata.py b/tests/test_app_run_metadata.py index bd4c860..1e7bb24 100644 --- a/tests/test_app_run_metadata.py +++ b/tests/test_app_run_metadata.py @@ -5,7 +5,10 @@ import json import logging import os +import subprocess +import sys import tempfile +import time import unittest from contextlib import redirect_stderr from dataclasses import replace @@ -16,8 +19,9 @@ import base_cli import base_cli._lifecycle as lifecycle_module import base_cli.app as app_module +from base_cli import RetentionPolicy from base_cli._lifecycle import RunRecorder -from base_cli._runtime import runtime_layout +from base_cli._runtime import prune_run_bundles, runtime_layout def _run(app: base_cli.App, home: Path, args: list[str] | None = None) -> tuple[int, str]: @@ -78,6 +82,118 @@ def _assert_terminal_metadata( @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") class AppRunMetadataTests(unittest.TestCase): + def test_terminal_run_lease_survives_a_concurrent_invocation_during_cleanup(self) -> None: + import click + + child_program = "\n".join( + ( + "import json, sys, time", + "from pathlib import Path", + "import click, base_cli", + "from base_cli import RetentionPolicy", + "mode, app_name, cache_text, ready_text, release_text = sys.argv[1:]", + "cache, ready, release = Path(cache_text), Path(ready_text), Path(release_text)", + "profile = base_cli.CliProfile.generic(cache_root=cache)", + "app = base_cli.App(name=app_name, profile=profile, retention=RetentionPolicy(max_bundles=1))", + "def block_cleanup(ctx):", + " metadata = json.loads((ctx.run_root / 'run.json').read_text(encoding='utf-8'))", + " (ctx.run_root / 'cleanup-marker').write_text('held', encoding='utf-8')", + " ready.write_text(json.dumps({'run_root': str(ctx.run_root), 'status': metadata['status']}), encoding='utf-8')", + " while not release.exists(): time.sleep(0.01)", + "if mode == 'native':", + " @app.command()", + " def main(ctx: base_cli.Context): ctx.on_cleanup(lambda: block_cleanup(ctx))", + " target = app", + "else:", + " @click.command(name=app_name)", + " def command(): base_cli.get_current_context().on_cleanup(lambda: block_cleanup(base_cli.get_current_context()))", + " target = app.attach(command)", + "base_cli.run_app(target, [])", + ) + ) + + for mode in ("native", "attached"): + with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + cache = root / "cache" + home = root / "home" + home.mkdir() + ready = root / "ready.json" + release = root / "release" + app_name = f"terminal-lease-{mode}" + child_env = { + key: value for key, value in os.environ.items() if not key.startswith(("COV_CORE_", "COVERAGE_")) + } + child_env.update(HOME=str(home), BASE_CLI_CACHE_DIR=str(cache)) + child = subprocess.Popen( + [sys.executable, "-c", child_program, mode, app_name, str(cache), str(ready), str(release)], + cwd=Path(__file__).resolve().parents[1], + env=child_env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.monotonic() + 10 + while not ready.exists() and child.poll() is None and time.monotonic() < deadline: + time.sleep(0.01) + if not ready.exists(): + stdout, stderr = child.communicate(timeout=5) + self.fail(f"cleanup hook did not become ready (exit={child.returncode}): {stdout}\n{stderr}") + + ready_payload = json.loads(ready.read_text(encoding="utf-8")) + run_root = Path(ready_payload["run_root"]) + self.assertEqual(ready_payload["status"], "ok") + self.assertEqual(json.loads((run_root / "run.json").read_text())["status"], "ok") + self.assertTrue((run_root / "cleanup-marker").is_file()) + + profile = base_cli.CliProfile.generic(cache_root=cache) + concurrent_app = base_cli.App( + name=app_name, + profile=profile, + retention=RetentionPolicy(max_bundles=1), + ) + if mode == "native": + + @concurrent_app.command() + def concurrent_main(ctx: base_cli.Context) -> None: + del ctx + + target = concurrent_app + else: + + @click.command(name=app_name) + def concurrent_command() -> None: + pass + + target = concurrent_app.attach(concurrent_command) + + result = base_cli.testing.invoke(target, [], home=home) + self.assertEqual(result.exit_code, 0, result.output) + self.assertTrue( + run_root.is_dir(), "retention deleted a run whose cleanup hook still held its lease" + ) + self.assertTrue((run_root / "cleanup-marker").is_file()) + finally: + release.touch() + try: + child.wait(timeout=10) + except subprocess.TimeoutExpired: + child.kill() + child.wait(timeout=5) + if child.stdout is not None: + child.stdout.close() + if child.stderr is not None: + child.stderr.close() + + prune_run_bundles( + run_root.parent, + policy=RetentionPolicy(max_age_seconds=60), + logger=logging.getLogger(__name__), + now=time.time() + 3_600, + ) + self.assertFalse(run_root.exists(), "eligible terminal run was not pruned after its lease was released") + def test_normal_returns_finalize_core_owned_metadata(self) -> None: cases = ( ("none", None, 0, "ok", "success"), 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 30eb8dc..e0957e1 100644 --- a/tests/test_run_bundle_retention.py +++ b/tests/test_run_bundle_retention.py @@ -31,6 +31,7 @@ def _bundle( path = root / name (path / "logs").mkdir(parents=True) (path / "logs" / "primary.log").write_bytes(b"x" * size) + (path / ".base-cli-run-lease").write_bytes(b"0") write_private_json( path / "run.json", { @@ -93,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" @@ -230,6 +265,64 @@ def test_live_running_bundle_lease_survives_from_another_process(self) -> None: child.kill() child.wait(timeout=5) + def test_live_terminal_bundle_lease_survives_from_another_process(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) / "runs" + root.mkdir() + live = _bundle(root, "live-terminal", status="ok", started_at="2020-01-01T00:00:00Z") + ready = live / "ready" + release = live / "release" + child = subprocess.Popen( + [ + sys.executable, + "-c", + "\n".join( + ( + "import sys, time", + "from pathlib import Path", + "from base_cli._runtime import acquire_run_lease, close_run_lease", + "run_root, ready_path, release_path = map(Path, sys.argv[1:])", + "lease = acquire_run_lease(run_root)", + "ready_path.touch()", + "while not release_path.exists(): time.sleep(0.01)", + "close_run_lease(lease)", + ) + ), + str(live), + str(ready), + str(release), + ], + env={key: value for key, value in os.environ.items() if not key.startswith(("COV_CORE_", "COVERAGE_"))}, + stdin=subprocess.DEVNULL, + ) + try: + deadline = time.monotonic() + 5 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(0.01) + self.assertTrue(ready.exists(), "lease holder did not start") + prune_run_bundles( + root, + policy=RetentionPolicy(max_age_seconds=60), + logger=logging.getLogger(__name__), + now=1_600_000_000, + ) + self.assertTrue(live.exists()) + finally: + release.touch() + try: + child.wait(timeout=5) + except subprocess.TimeoutExpired: + child.kill() + child.wait(timeout=5) + + prune_run_bundles( + root, + policy=RetentionPolicy(max_age_seconds=60), + logger=logging.getLogger(__name__), + now=1_600_000_000, + ) + self.assertFalse(live.exists(), "finished bundles remain eligible after the lease is released") + def test_aborted_bundle_is_indexed_as_terminal(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) / "runs"