diff --git a/pg_llm_batch/postgres_recovery_replay_observation.py b/pg_llm_batch/postgres_recovery_replay_observation.py new file mode 100644 index 000000000..58e4fa694 --- /dev/null +++ b/pg_llm_batch/postgres_recovery_replay_observation.py @@ -0,0 +1,221 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Observe bounded WAL replay progress on a paused PostgreSQL recovery target.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from weakref import WeakKeyDictionary + + +_LSN_RE = re.compile(r"(?:0|[1-9A-F][0-9A-F]{0,7})/[0-9A-F]{1,8}\Z") +_REPLAY_SQL = """ +SELECT + pg_catalog.pg_is_in_recovery(), + pg_catalog.pg_get_wal_replay_pause_state(), + pg_catalog.pg_last_wal_replay_lsn()::pg_catalog.text +""".strip() +_REPLAY_OBSERVATION_MARK = object() +_OBSERVED_REPLAY: WeakKeyDictionary[ + PostgresRecoveryReplayObservation, tuple[str, str] +] = WeakKeyDictionary() + + +class PostgresRecoveryReplayObservationError(ValueError): + """Report a fail-closed PostgreSQL recovery replay observation violation.""" + + +@dataclass(frozen=True, eq=False) +class PostgresRecoveryReplayObservation: + """Represent one content-free paused recovery replay observation. + + Only ``observe_postgres_recovery_replay`` registers an object as observed. + Public construction, copying, or post-construction field mutation therefore + cannot be reused as package inspection provenance. The record proves only + what the caller-owned connection returned: recovery was still active, + replay was actually paused, and the last replayed WAL location was at or + beyond the requested LSN. It does not prove exact recovery-target semantics, + archive completeness, timeline correctness, application readiness, or an + achieved RPO/RTO objective. + """ + + target_lsn: str + replay_lsn: str + _observation_mark: object = field(default=None, repr=False, compare=False) + + @property + def recovery_in_progress(self) -> bool: + """Return the recovery-state predicate from a live observed record.""" + _require_observed(self) + return True + + @property + def replay_paused(self) -> bool: + """Return the actual pause-state predicate from a live observed record.""" + _require_observed(self) + return True + + @property + def target_reached(self) -> bool: + """Return the bounded replay-progress predicate from a live record.""" + _require_observed(self) + return True + + def as_dict(self) -> dict[str, object]: + """Return the stable content-free machine-readable observation schema.""" + target_lsn, replay_lsn = _require_observed(self) + return { + "target_lsn": target_lsn, + "replay_lsn": replay_lsn, + "recovery_in_progress": True, + "replay_paused": True, + "target_reached": True, + } + + +def postgres_recovery_replay_observation_was_observed(evidence: object) -> bool: + """Return whether one exact live object still matches its observed snapshot.""" + if type(evidence) is not PostgresRecoveryReplayObservation: + return False + observed = _OBSERVED_REPLAY.get(evidence) + if observed is None: + return False + observed_target_lsn, observed_replay_lsn = observed + return ( + evidence._observation_mark, + evidence.target_lsn, + evidence.replay_lsn, + ) == ( + _REPLAY_OBSERVATION_MARK, + observed_target_lsn, + observed_replay_lsn, + ) + + +def _require_observed( + evidence: PostgresRecoveryReplayObservation, +) -> tuple[str, str]: + """Return the validated immutable observation snapshot or fail closed.""" + if not postgres_recovery_replay_observation_was_observed(evidence): + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay observation provenance is invalid" + ) + observed = _OBSERVED_REPLAY.get(evidence) + if observed is None: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay observation provenance is invalid" + ) + return observed + + +def _record_observed_replay(evidence: PostgresRecoveryReplayObservation) -> None: + """Remember the exact live observation object and its immutable field snapshot.""" + _OBSERVED_REPLAY[evidence] = (evidence.target_lsn, evidence.replay_lsn) + + +def _lsn_position(value: object) -> int | None: + """Return a normalized nonzero PostgreSQL LSN as a uint64 position.""" + if type(value) is not str: + return None + if _LSN_RE.fullmatch(value) is None: + return None + high_text, low_text = value.split("/", 1) + position = (int(high_text, 16) << 32) | int(low_text, 16) + if position == 0: + return None + return position + + +def _evaluate_replay_row( + row: object, + *, + target_lsn: str, + target_position: int, +) -> PostgresRecoveryReplayObservation: + """Validate one finite PostgreSQL recovery observation row.""" + if type(row) is not tuple: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay evidence is invalid" + ) + if len(row) != 3: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay evidence is invalid" + ) + recovery_in_progress, pause_state, replay_lsn = row + if type(recovery_in_progress) is not bool: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay evidence is invalid" + ) + if type(pause_state) is not str: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay evidence is invalid" + ) + replay_position = _lsn_position(replay_lsn) + if replay_position is None: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay evidence is invalid" + ) + if not recovery_in_progress: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery target is not in recovery" + ) + if pause_state != "paused": + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery target is not paused" + ) + if replay_position < target_position: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery target has not been replayed" + ) + evidence = PostgresRecoveryReplayObservation( + target_lsn=target_lsn, + replay_lsn=replay_lsn, + _observation_mark=_REPLAY_OBSERVATION_MARK, + ) + _record_observed_replay(evidence) + return evidence + + +def observe_postgres_recovery_replay( + connection: object, + *, + target_lsn: str, +) -> PostgresRecoveryReplayObservation: + """Observe that an isolated recovery target replayed at least one target LSN. + + ``connection`` is caller-owned and already connected to the isolated + PostgreSQL target. ``target_lsn`` must be normalized uppercase nonzero + PostgreSQL LSN text with at most eight hexadecimal digits per segment. The + function performs one fixed, catalog-qualified read-only query and requires + three conditions in the same returned row: recovery is still in progress, + ``pg_get_wal_replay_pause_state()`` reports ``paused`` rather than merely a + pause request, and ``pg_last_wal_replay_lsn()`` is at or beyond the requested + LSN. + + PostgreSQL documents LSNs as monotonically increasing WAL byte positions and + documents the ``paused`` state as the state in which no further database + changes are applied until replay resumes. Consequently this is a bounded + replay-progress/acceptance-window observation only. It does not start or + configure PostgreSQL, create ``recovery.signal``, install ``restore_command``, + validate WAL bytes or timeline ancestry, prove exact stop-target semantics, + resume or promote recovery, prove application readiness, recover external + secrets, or establish RPO/RTO, HA/DR, CSAP, SOC 2, or certification claims. + """ + target_position = _lsn_position(target_lsn) + if target_position is None: + raise PostgresRecoveryReplayObservationError( + "invalid PostgreSQL recovery replay observation inputs" + ) + try: + with connection.cursor() as cursor: + cursor.execute(_REPLAY_SQL) + row = cursor.fetchone() + except Exception: + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay state could not be inspected" + ) from None + return _evaluate_replay_row( + row, + target_lsn=target_lsn, + target_position=target_position, + ) diff --git a/tests/test_postgres_recovery_replay_observation.py b/tests/test_postgres_recovery_replay_observation.py new file mode 100644 index 000000000..010d8bcc9 --- /dev/null +++ b/tests/test_postgres_recovery_replay_observation.py @@ -0,0 +1,302 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Recovery-replay observation regressions for an isolated PostgreSQL target.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest + +import pg_llm_batch.postgres_recovery_replay_observation as replay_observation +from pg_llm_batch.postgres_recovery_replay_observation import ( + PostgresRecoveryReplayObservation, + PostgresRecoveryReplayObservationError, + observe_postgres_recovery_replay, + postgres_recovery_replay_observation_was_observed, +) + + +class _ReplayCursor: + """Return one caller-owned recovery observation row.""" + + def __init__(self, row: object) -> None: + self.row = row + self.executed_sql: str | None = None + self.execute_calls = 0 + + def __enter__(self) -> _ReplayCursor: + return self + + def __exit__(self, *_exc: object) -> bool: + return False + + def execute(self, sql: object) -> None: + if type(sql) is not str: + raise AssertionError("recovery SQL must be an exact built-in string") + self.executed_sql = sql + self.execute_calls += 1 + + def fetchone(self) -> object: + return self.row + + +class _ReplayConnection: + """Expose one caller-owned PostgreSQL connection without DSN authority.""" + + def __init__(self, row: object) -> None: + self.cursor_handle = _ReplayCursor(row) + self.cursor_calls = 0 + + def cursor(self) -> _ReplayCursor: + self.cursor_calls += 1 + return self.cursor_handle + + +class _FailingCursor(_ReplayCursor): + """Leak deployment text if the package reflects lower-layer diagnostics.""" + + def execute(self, sql: object) -> None: + del sql + raise RuntimeError("password=supersecret host=restore.internal") + + +class _FailingConnection(_ReplayConnection): + """Return a cursor whose fixed inspection query fails.""" + + def cursor(self) -> _ReplayCursor: + self.cursor_calls += 1 + return _FailingCursor(None) + + +class _HostileString(str): + """Refuse rendering if rejected database text reaches diagnostics.""" + + def __str__(self) -> str: + raise AssertionError("hostile database text must not be rendered") + + +def test_observe_recovery_replay_accepts_paused_target_at_or_beyond_lsn() -> None: + connection = _ReplayConnection((True, "paused", "1/00000020")) + + evidence = observe_postgres_recovery_replay( + connection, + target_lsn="1/00000010", + ) + + assert postgres_recovery_replay_observation_was_observed(evidence) is True + assert evidence.target_lsn == "1/00000010" + assert evidence.replay_lsn == "1/00000020" + assert evidence.recovery_in_progress is True + assert evidence.replay_paused is True + assert evidence.target_reached is True + assert evidence.as_dict() == { + "target_lsn": "1/00000010", + "replay_lsn": "1/00000020", + "recovery_in_progress": True, + "replay_paused": True, + "target_reached": True, + } + + +def test_observe_recovery_replay_accepts_exact_target_lsn() -> None: + evidence = observe_postgres_recovery_replay( + _ReplayConnection((True, "paused", "A/000000FF")), + target_lsn="A/000000FF", + ) + + assert evidence.replay_lsn == evidence.target_lsn + assert evidence.target_reached is True + + +def test_observe_recovery_replay_accepts_postgresql_printed_short_lsn_segments() -> None: + evidence = observe_postgres_recovery_replay( + _ReplayConnection((True, "paused", "0/16B1971")), + target_lsn="0/16B1970", + ) + + assert evidence.target_lsn == "0/16B1970" + assert evidence.replay_lsn == "0/16B1971" + assert evidence.target_reached is True + + +def test_manual_observation_is_not_inspection_provenance() -> None: + fabricated = PostgresRecoveryReplayObservation( + target_lsn="1/00000010", + replay_lsn="1/00000020", + ) + + assert postgres_recovery_replay_observation_was_observed(fabricated) is False + assert postgres_recovery_replay_observation_was_observed(object()) is False + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery replay observation provenance is invalid", + ): + fabricated.as_dict() + + +def test_copied_or_mutated_observation_loses_inspection_provenance() -> None: + evidence = observe_postgres_recovery_replay( + _ReplayConnection((True, "paused", "1/00000020")), + target_lsn="1/00000010", + ) + + copied = replace(evidence) + assert postgres_recovery_replay_observation_was_observed(copied) is False + + object.__setattr__(evidence, "replay_lsn", "1/00000030") + assert postgres_recovery_replay_observation_was_observed(evidence) is False + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery replay observation provenance is invalid", + ): + evidence.target_reached + + +def test_observation_registry_eviction_between_checks_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evidence = observe_postgres_recovery_replay( + _ReplayConnection((True, "paused", "1/00000020")), + target_lsn="1/00000010", + ) + snapshot = (evidence.target_lsn, evidence.replay_lsn) + + class _EvictingRegistry: + """Model concurrent provenance eviction between the guarded reads.""" + + def __init__(self) -> None: + self.calls = 0 + + def get(self, candidate: object) -> tuple[str, str] | None: + self.calls += 1 + if candidate is evidence and self.calls == 1: + return snapshot + return None + + registry = _EvictingRegistry() + monkeypatch.setattr(replay_observation, "_OBSERVED_REPLAY", registry) + + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery replay observation provenance is invalid", + ): + evidence.as_dict() + + assert registry.calls == 2 + + +@pytest.mark.parametrize( + "target_lsn", + [ + object(), + _HostileString("1/00000010"), + "", + "01/00000010", + "1/0000000a", + "0/00000000", + "1/000000010", + "100000000/1", + ], +) +def test_observe_recovery_replay_rejects_invalid_target_before_database_io( + target_lsn: object, +) -> None: + connection = _ReplayConnection((True, "paused", "1/00000020")) + + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="invalid PostgreSQL recovery replay observation inputs", + ): + observe_postgres_recovery_replay(connection, target_lsn=target_lsn) # type: ignore[arg-type] + + assert connection.cursor_calls == 0 + + +def test_observe_recovery_replay_uses_fixed_catalog_qualified_query() -> None: + connection = _ReplayConnection((True, "paused", "1/00000010")) + + observe_postgres_recovery_replay(connection, target_lsn="1/00000010") + + sql = connection.cursor_handle.executed_sql + assert sql is not None + assert connection.cursor_handle.execute_calls == 1 + assert "pg_catalog.pg_is_in_recovery()" in sql + assert "pg_catalog.pg_get_wal_replay_pause_state()" in sql + assert "pg_catalog.pg_last_wal_replay_lsn()" in sql + assert "%s" not in sql + + +def test_observe_recovery_replay_hides_database_diagnostics() -> None: + connection = _FailingConnection(None) + + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery replay state could not be inspected", + ) as raised: + observe_postgres_recovery_replay(connection, target_lsn="1/00000010") + + assert "supersecret" not in str(raised.value) + assert "restore.internal" not in str(raised.value) + assert raised.value.__cause__ is None + + +@pytest.mark.parametrize( + "row", + [ + None, + [True, "paused", "1/00000010"], + (True, "paused"), + (1, "paused", "1/00000010"), + (True, _HostileString("paused"), "1/00000010"), + (True, "paused", _HostileString("1/00000010")), + (True, "paused", "01/00000010"), + (True, "paused", "0/00000000"), + ], +) +def test_observe_recovery_replay_rejects_malformed_database_evidence( + row: object, +) -> None: + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery replay evidence is invalid", + ): + observe_postgres_recovery_replay( + _ReplayConnection(row), + target_lsn="1/00000010", + ) + + +def test_observe_recovery_replay_requires_recovery_to_still_be_in_progress() -> None: + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery target is not in recovery", + ): + observe_postgres_recovery_replay( + _ReplayConnection((False, "paused", "1/00000010")), + target_lsn="1/00000010", + ) + + +@pytest.mark.parametrize("pause_state", ["not paused", "pause requested"]) +def test_observe_recovery_replay_requires_actual_paused_state( + pause_state: str, +) -> None: + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery target is not paused", + ): + observe_postgres_recovery_replay( + _ReplayConnection((True, pause_state, "1/00000010")), + target_lsn="1/00000010", + ) + + +def test_observe_recovery_replay_rejects_replay_position_before_target() -> None: + with pytest.raises( + PostgresRecoveryReplayObservationError, + match="PostgreSQL recovery target has not been replayed", + ): + observe_postgres_recovery_replay( + _ReplayConnection((True, "paused", "1/0000000F")), + target_lsn="1/00000010", + ) diff --git a/tests/test_postgres_recovery_replay_serialization_snapshot.py b/tests/test_postgres_recovery_replay_serialization_snapshot.py new file mode 100644 index 000000000..37aedc26f --- /dev/null +++ b/tests/test_postgres_recovery_replay_serialization_snapshot.py @@ -0,0 +1,61 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Serialization-snapshot regression for recovery replay observations.""" + +from __future__ import annotations + +import pytest + +import pg_llm_batch.postgres_recovery_replay_observation as replay_observation + + +class _ReplayCursor: + """Return one fixed paused-recovery observation row.""" + + def __enter__(self) -> _ReplayCursor: + return self + + def __exit__(self, *_exc: object) -> bool: + return False + + def execute(self, sql: object) -> None: + if type(sql) is not str: + raise AssertionError("recovery SQL must remain an exact built-in string") + + def fetchone(self) -> tuple[bool, str, str]: + return True, "paused", "1/00000020" + + +class _ReplayConnection: + """Expose one caller-owned connection seam without DSN authority.""" + + def cursor(self) -> _ReplayCursor: + return _ReplayCursor() + + +def test_as_dict_serializes_the_already_validated_observation_snapshot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Serialization must not re-read mutable public fields after provenance validation.""" + evidence = replay_observation.observe_postgres_recovery_replay( + _ReplayConnection(), + target_lsn="1/00000010", + ) + original_require = replay_observation._require_observed + + def validate_then_mutate( + candidate: replay_observation.PostgresRecoveryReplayObservation, + ) -> tuple[str, str] | None: + snapshot = original_require(candidate) + object.__setattr__(candidate, "target_lsn", "1/00000030") + object.__setattr__(candidate, "replay_lsn", "1/00000040") + return snapshot + + monkeypatch.setattr(replay_observation, "_require_observed", validate_then_mutate) + + assert evidence.as_dict() == { + "target_lsn": "1/00000010", + "replay_lsn": "1/00000020", + "recovery_in_progress": True, + "replay_paused": True, + "target_reached": True, + }