From b3262874eb9c0364e5bb508786e38dffe3270574 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:58:09 +0530 Subject: [PATCH] fix: report reconciliation publication state accurately --- src/base_cli_demo/cli.py | 44 ++++++++++++++++--- tests/test_cli.py | 92 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 6 deletions(-) diff --git a/src/base_cli_demo/cli.py b/src/base_cli_demo/cli.py index 095e5cd..2ba6c8d 100644 --- a/src/base_cli_demo/cli.py +++ b/src/base_cli_demo/cli.py @@ -111,6 +111,10 @@ def _render( ) +class _PublishedSnapshotDurabilityError(OSError): + """The snapshot was replaced, but the directory update was not durable.""" + + def _persist_reconciliation( context: base_cli.Context[Any, Any, Any], record: Mapping[str, Any] ) -> None: @@ -142,8 +146,20 @@ def _persist_reconciliation( temporary_input = context.temp_dir / "reconciliation-input.json" temporary_input.write_text(serialized, encoding="utf-8") context.on_cleanup(lambda: temporary_input.unlink(missing_ok=True)) - _replace_state(staged_path, state_path) + durability_confirmed = _replace_state(staged_path, state_path) + staged_path = None + if not durability_confirmed: + context.log.warning( + "Reconciliation snapshot was published, but directory durability could not be confirmed." + ) + except _PublishedSnapshotDurabilityError as exc: + # os.replace is the commit point. Never describe the previous bytes as + # surviving a failure that happened while confirming durability after it. staged_path = None + raise click.ClickException( + "Could not confirm reconciliation snapshot durability; the new snapshot was published, " + "but durability could not be confirmed." + ) from exc except (OSError, TypeError, ValueError) as exc: snapshot_message = ( "the previous snapshot was left unchanged." @@ -167,20 +183,36 @@ def _serialize_reconciliation(record: Mapping[str, Any]) -> str: return json.dumps(dict(record), sort_keys=True) + "\n" -def _replace_state(staged_path: Path, state_path: Path) -> None: - """Atomically publish a complete snapshot from the same filesystem.""" +def _replace_state(staged_path: Path, state_path: Path) -> bool: + """Atomically publish a snapshot and report directory durability. + + ``True`` means the parent directory was opened and fsynced after the + replacement. Windows does not provide the same directory-fsync contract, + so an unavailable directory handle is treated as a published snapshot + with unconfirmed durability; the caller emits a warning and continues. + """ os.replace(staged_path, state_path) try: directory_fd = os.open(state_path.parent, os.O_RDONLY) except OSError: if os.name != "nt": - raise - return + raise _PublishedSnapshotDurabilityError("parent directory durability could not be confirmed") + return False + + durability_error: OSError | None = None try: os.fsync(directory_fd) - finally: + except OSError as exc: + durability_error = exc + try: os.close(directory_fd) + except OSError as exc: + if durability_error is None: + durability_error = exc + if durability_error is not None: + raise _PublishedSnapshotDurabilityError(str(durability_error)) from durability_error + return True def _preserve_state_mode(staged_path: Path, state_path: Path) -> None: diff --git a/tests/test_cli.py b/tests/test_cli.py index 25ccfc8..57d142d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -10,6 +10,7 @@ import base_cli import base_cli_demo.cli as cli_module +import pytest from base_cli_demo.cli import command @@ -184,6 +185,97 @@ def fail_replace(_staged_path: Path, _state_path: Path) -> None: assert not list(state_path.parent.glob(f".{state_path.name}.*.tmp")) +def test_directory_open_failure_reports_published_snapshot( + monkeypatch: Any, +) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + first = invoke(["release", "reconcile", "--version", "2.6.0"], root) + assert first.exit_code == 0, first.output + state_path = next(root.rglob("last-reconciliation.json")) + + def fail_after_replace(staged_path: Path, target_path: Path) -> bool: + cli_module.os.replace(staged_path, target_path) + raise cli_module._PublishedSnapshotDurabilityError("simulated directory open failure") + + monkeypatch.setattr(cli_module, "_replace_state", fail_after_replace) + failed = invoke(["release", "reconcile", "--version", "2.7.0"], root) + + assert failed.exit_code == 1 + assert "new snapshot was published" in failed.output + assert "previous snapshot was left unchanged" not in failed.output + assert json.loads(state_path.read_text(encoding="utf-8"))["target_version"] == "2.7.0" + assert not list(state_path.parent.glob(f".{state_path.name}.*.tmp")) + + +def test_directory_fsync_failure_reports_published_snapshot( + monkeypatch: Any, +) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + first = invoke(["release", "reconcile", "--version", "2.6.0"], root) + assert first.exit_code == 0, first.output + state_path = next(root.rglob("last-reconciliation.json")) + def fail_after_replace(staged_path: Path, target_path: Path) -> bool: + cli_module.os.replace(staged_path, target_path) + raise cli_module._PublishedSnapshotDurabilityError("simulated directory fsync failure") + + monkeypatch.setattr(cli_module, "_replace_state", fail_after_replace) + failed = invoke(["release", "reconcile", "--version", "2.7.0"], root) + + assert failed.exit_code == 1 + assert "new snapshot was published" in failed.output + assert json.loads(state_path.read_text(encoding="utf-8"))["target_version"] == "2.7.0" + assert not list(state_path.parent.glob(f".{state_path.name}.*.tmp")) + + +def test_replace_state_directory_open_failure_occurs_after_publication( + monkeypatch: Any, + tmp_path: Path, +) -> None: + staged_path = tmp_path / "staged.json" + state_path = tmp_path / "last-reconciliation.json" + staged_path.write_text("new\n", encoding="utf-8") + state_path.write_text("old\n", encoding="utf-8") + real_open = cli_module.os.open + + def fail_directory_open(path: Any, flags: int, *args: Any, **kwargs: Any) -> int: + if Path(path) == tmp_path: + raise OSError("simulated directory open failure") + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(cli_module.os, "open", fail_directory_open) + with pytest.raises(cli_module._PublishedSnapshotDurabilityError): + cli_module._replace_state(staged_path, state_path) + + assert state_path.read_text(encoding="utf-8") == "new\n" + + +def test_file_fsync_failure_preserves_previous_snapshot( + monkeypatch: Any, +) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + first = invoke(["release", "reconcile", "--version", "2.6.0"], root) + assert first.exit_code == 0, first.output + state_path = next(root.rglob("last-reconciliation.json")) + previous = state_path.read_bytes() + real_fsync = cli_module.os.fsync + + def fail_file_fsync(fd: int) -> None: + if not stat.S_ISDIR(cli_module.os.fstat(fd).st_mode): + raise OSError("simulated file fsync failure") + real_fsync(fd) + + monkeypatch.setattr(cli_module.os, "fsync", fail_file_fsync) + failed = invoke(["release", "reconcile", "--version", "2.7.0"], root) + + assert failed.exit_code == 1 + assert "previous snapshot was left unchanged" in failed.output + assert state_path.read_bytes() == previous + assert not list(state_path.parent.glob(f".{state_path.name}.*.tmp")) + + def test_serialization_failure_preserves_the_previous_snapshot(monkeypatch: Any) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory)