diff --git a/CHANGELOG.md b/CHANGELOG.md index ec20c2e..f94444e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file. ### Fixed +- 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..7dc7b8e 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 lease file is present but 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 diff --git a/lib/python/base_cli/_lifecycle.py b/lib/python/base_cli/_lifecycle.py index 7816491..9a89e49 100644 --- a/lib/python/base_cli/_lifecycle.py +++ b/lib/python/base_cli/_lifecycle.py @@ -61,7 +61,11 @@ def finish( write_private_json(self.context._run_metadata_path, metadata) owner_root = self.context.owner_root if owner_root is not None: - refresh_run_bundle_index(owner_root / "runs", logger=self.context.log) + refresh_run_bundle_index( + owner_root / "runs", + current_run_root=self.context.run_root, + logger=self.context.log, + ) def _existing_metadata(self) -> dict[str, Any]: path = self.context._run_metadata_path diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index 46de122..4d62e87 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -450,6 +450,7 @@ def prune_run_bundles( def refresh_run_bundle_index( runs_root: Path, *, + current_run_root: Path | None = None, logger: logging.Logger | None = None, ) -> None: """Refresh the diagnostic bundle index after a run becomes terminal.""" @@ -468,7 +469,7 @@ def refresh_run_bundle_index( size_budget=0, ) with _retention_lock(runs_root): - _write_run_index(runs_root, bundles, log) + _write_run_index(runs_root, bundles, log, current_run_root=current_run_root) except (OSError, RuntimeError) as exc: log.debug("Could not refresh run bundle index under '%s': %s", runs_root, exc) @@ -506,11 +507,18 @@ 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. Legacy bundle + # fixtures may have no lease file at all; terminal bundles without a + # lease are eligible, while unknown liveness remains fail-closed for + # running records or a present but unreadable lease. + lease_state = _run_lease_state(child) + if _lease_blocks_removal(child, lease_state): + continue + if running and lease_state != "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"}: @@ -652,8 +660,11 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa if metadata is None: return False status = str(metadata.get("status", "")) + lease_state = _run_lease_state(path) + if _lease_blocks_removal(path, lease_state): + return False if status == "running": - if _run_lease_state(path) != "inactive": + if lease_state != "inactive": return False if policy.max_age_seconds is None: return False @@ -675,6 +686,18 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa ) +def _lease_blocks_removal(path: Path, lease_state: str | None = None) -> bool: + """Return whether a bundle's lease proves it must be retained.""" + + state = _run_lease_state(path) if lease_state is None else lease_state + if state == "active": + return True + if state != "unknown": + return False + lease_path = path / _RUN_LEASE_NAME + return lease_path.exists() or lease_path.is_symlink() + + def _write_run_index( runs_root: Path, bundles: list[dict[str, Any]], @@ -687,15 +710,19 @@ def _write_run_index( if current_run_root is not None and current_run_root.exists(): current_resolved = _safe_resolved_path(current_run_root) if not any(_safe_resolved_path(Path(bundle["path"])) == current_resolved for bundle in indexed): + metadata = _read_bundle_metadata(current_run_root) or {} + started_at = _timestamp_to_epoch(metadata.get("started_at")) + if started_at is None: + started_at = time.time() if now is None else now indexed.append( { "path": current_run_root, "run_id": current_run_root.name, - "status": "running", - "started_at": time.time() if now is None else now, + "status": str(metadata.get("status", "running")), + "started_at": started_at, "size": 0, "size_known": False, - "preserve": False, + "preserve": bool(metadata.get("preserve")), } ) indexed.sort(key=lambda bundle: (float(bundle.get("started_at", 0)), str(bundle["path"]))) diff --git a/tests/test_app_run_metadata.py b/tests/test_app_run_metadata.py index bd4c860..a945811 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,143 @@ 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() + + index_path = run_root.parent / ".base-cli-run-index.json" + index = json.loads(index_path.read_text(encoding="utf-8")) + indexed = {bundle["path"]: bundle for bundle in index["bundles"]} + self.assertTrue(indexed) + self.assertTrue(all(Path(path).is_dir() for path in indexed)) + + 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_terminal_run_is_indexed_while_its_lease_is_held(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + app = base_cli.App(name="terminal-index") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + status, stderr = _run(app, home) + self.assertEqual(status, 0, stderr) + run_path, metadata = _load_only_metadata(self, home) + index_path = run_path.parent.parent / ".base-cli-run-index.json" + index = json.loads(index_path.read_text(encoding="utf-8")) + indexed = {bundle["path"]: bundle for bundle in index["bundles"]} + run_root = str(run_path.parent.resolve()) + self.assertIn(run_root, indexed) + self.assertEqual(indexed[run_root]["status"], metadata["status"]) + def test_normal_returns_finalize_core_owned_metadata(self) -> None: cases = ( ("none", None, 0, "ok", "success"), diff --git a/tests/test_run_bundle_retention.py b/tests/test_run_bundle_retention.py index 30eb8dc..2db453c 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", { @@ -230,6 +231,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"