fix: report reconciliation publication state accurately

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:

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)