From fb04977925743f0130baf7f991cae2642e237537 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 17:12:12 -0700 Subject: [PATCH 1/9] test(recovery): define paused replay observation contract --- ...st_postgres_recovery_replay_observation.py | 218 ++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 tests/test_postgres_recovery_replay_observation.py diff --git a/tests/test_postgres_recovery_replay_observation.py b/tests/test_postgres_recovery_replay_observation.py new file mode 100644 index 000000000..1b6e2776f --- /dev/null +++ b/tests/test_postgres_recovery_replay_observation.py @@ -0,0 +1,218 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Recovery-replay observation regressions for an isolated PostgreSQL target.""" + +from __future__ import annotations + +import pytest + +from pg_llm_batch.postgres_recovery_replay_observation import ( + PostgresRecoveryReplayObservationError, + observe_postgres_recovery_replay, +) + + +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 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 + + +@pytest.mark.parametrize( + "target_lsn", + [ + object(), + _HostileString("1/00000010"), + "", + "01/00000010", + "1/10", + "1/0000000a", + "0/00000000", + ], +) +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", + ) From e16ddb11e51573ff794b937c491d7c62088194b8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 17:12:53 -0700 Subject: [PATCH 2/9] feat(recovery): observe paused WAL replay progress --- .../postgres_recovery_replay_observation.py | 165 ++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 pg_llm_batch/postgres_recovery_replay_observation.py 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..950652cec --- /dev/null +++ b/pg_llm_batch/postgres_recovery_replay_observation.py @@ -0,0 +1,165 @@ +# 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 + + +_LSN_RE = re.compile(r"(?:0|[1-9A-F][0-9A-F]{0,7})/[0-9A-F]{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() + + +class PostgresRecoveryReplayObservationError(ValueError): + """Report a fail-closed PostgreSQL recovery replay observation violation.""" + + +@dataclass(frozen=True, slots=True) +class PostgresRecoveryReplayObservation: + """Represent one content-free paused recovery replay observation. + + The record proves only what ``observe_postgres_recovery_replay`` observed + through the caller-owned connection: 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 that the configured recovery target was + exact, that the archive is complete, that the timeline is correct, that the + target is application-ready, or that any RPO/RTO objective is achieved. + """ + + target_lsn: str + replay_lsn: str + + @property + def recovery_in_progress(self) -> bool: + """Return the recovery-state predicate required by this observation.""" + return True + + @property + def replay_paused(self) -> bool: + """Return the actual pause-state predicate required by this observation.""" + return True + + @property + def target_reached(self) -> bool: + """Return the bounded replay-progress predicate established on creation.""" + return True + + def as_dict(self) -> dict[str, object]: + """Return the stable content-free machine-readable observation schema.""" + return { + "target_lsn": self.target_lsn, + "replay_lsn": self.replay_lsn, + "recovery_in_progress": True, + "replay_paused": True, + "target_reached": True, + } + + +def _lsn_position(value: object) -> int | None: + """Return a canonical 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" + ) + return PostgresRecoveryReplayObservation( + target_lsn=target_lsn, + replay_lsn=replay_lsn, + ) + + +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 canonical nonzero PostgreSQL LSN + text. 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, + ) From b2b8bce912a7d4d2614cec97a3d4388cd507f368 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 17:15:18 -0700 Subject: [PATCH 3/9] test(recovery): reject fabricated replay observations --- ...st_postgres_recovery_replay_observation.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_postgres_recovery_replay_observation.py b/tests/test_postgres_recovery_replay_observation.py index 1b6e2776f..0234ad4ae 100644 --- a/tests/test_postgres_recovery_replay_observation.py +++ b/tests/test_postgres_recovery_replay_observation.py @@ -3,11 +3,15 @@ from __future__ import annotations +from dataclasses import replace + import pytest from pg_llm_batch.postgres_recovery_replay_observation import ( + PostgresRecoveryReplayObservation, PostgresRecoveryReplayObservationError, observe_postgres_recovery_replay, + postgres_recovery_replay_observation_was_observed, ) @@ -78,6 +82,7 @@ def test_observe_recovery_replay_accepts_paused_target_at_or_beyond_lsn() -> Non 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 @@ -102,6 +107,39 @@ def test_observe_recovery_replay_accepts_exact_target_lsn() -> None: 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 + + @pytest.mark.parametrize( "target_lsn", [ From eac027f738970614b500313f22d8c52064a38764 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 17:16:07 -0700 Subject: [PATCH 4/9] fix(recovery): retain replay observation provenance --- .../postgres_recovery_replay_observation.py | 71 +++++++++++++++---- 1 file changed, 59 insertions(+), 12 deletions(-) diff --git a/pg_llm_batch/postgres_recovery_replay_observation.py b/pg_llm_batch/postgres_recovery_replay_observation.py index 950652cec..c8dcc4745 100644 --- a/pg_llm_batch/postgres_recovery_replay_observation.py +++ b/pg_llm_batch/postgres_recovery_replay_observation.py @@ -4,7 +4,8 @@ from __future__ import annotations import re -from dataclasses import dataclass +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]{8}\Z") @@ -14,44 +15,55 @@ 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, slots=True) +@dataclass(frozen=True, eq=False) class PostgresRecoveryReplayObservation: """Represent one content-free paused recovery replay observation. - The record proves only what ``observe_postgres_recovery_replay`` observed - through the caller-owned connection: 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 that the configured recovery target was - exact, that the archive is complete, that the timeline is correct, that the - target is application-ready, or that any RPO/RTO objective is achieved. + 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 required by this observation.""" + """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 required by this observation.""" + """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 established on creation.""" + """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.""" + _require_observed(self) return { "target_lsn": self.target_lsn, "replay_lsn": self.replay_lsn, @@ -61,6 +73,38 @@ def as_dict(self) -> dict[str, object]: } +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) -> None: + """Fail closed when a record is fabricated, copied, or mutated after observation.""" + if not postgres_recovery_replay_observation_was_observed(evidence): + raise PostgresRecoveryReplayObservationError( + "PostgreSQL recovery replay observation provenance is invalid" + ) + + +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 canonical nonzero PostgreSQL LSN as a uint64 position.""" if type(value) is not str: @@ -115,10 +159,13 @@ def _evaluate_replay_row( raise PostgresRecoveryReplayObservationError( "PostgreSQL recovery target has not been replayed" ) - return PostgresRecoveryReplayObservation( + 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( From 00c876773161009922beeb56b4c47d3ead021825 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 17:23:25 -0700 Subject: [PATCH 5/9] test(recovery): accept PostgreSQL printed LSN widths --- tests/test_postgres_recovery_replay_observation.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/tests/test_postgres_recovery_replay_observation.py b/tests/test_postgres_recovery_replay_observation.py index 0234ad4ae..e53720d4e 100644 --- a/tests/test_postgres_recovery_replay_observation.py +++ b/tests/test_postgres_recovery_replay_observation.py @@ -107,6 +107,17 @@ def test_observe_recovery_replay_accepts_exact_target_lsn() -> None: 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", @@ -147,9 +158,10 @@ def test_copied_or_mutated_observation_loses_inspection_provenance() -> None: _HostileString("1/00000010"), "", "01/00000010", - "1/10", "1/0000000a", "0/00000000", + "1/000000010", + "100000000/1", ], ) def test_observe_recovery_replay_rejects_invalid_target_before_database_io( From 897515a023c2fd5499ff3ad7aea965f157c471b5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 20 Aug 2026 18:08:42 -0700 Subject: [PATCH 6/9] fix(recovery): accept bounded PostgreSQL LSN widths --- .../postgres_recovery_replay_observation.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/pg_llm_batch/postgres_recovery_replay_observation.py b/pg_llm_batch/postgres_recovery_replay_observation.py index c8dcc4745..45c3e4db1 100644 --- a/pg_llm_batch/postgres_recovery_replay_observation.py +++ b/pg_llm_batch/postgres_recovery_replay_observation.py @@ -8,7 +8,7 @@ from weakref import WeakKeyDictionary -_LSN_RE = re.compile(r"(?:0|[1-9A-F][0-9A-F]{0,7})/[0-9A-F]{8}\Z") +_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(), @@ -106,7 +106,7 @@ def _record_observed_replay(evidence: PostgresRecoveryReplayObservation) -> None def _lsn_position(value: object) -> int | None: - """Return a canonical nonzero PostgreSQL LSN as a uint64 position.""" + """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: @@ -176,12 +176,13 @@ def observe_postgres_recovery_replay( """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 canonical nonzero PostgreSQL LSN - text. 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 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 From 559b7d600f0fbd34ecc435bc35bbae109f3defb8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 21 Aug 2026 04:35:24 -0700 Subject: [PATCH 7/9] test(recovery): require snapshot-bound replay serialization --- ..._recovery_replay_serialization_snapshot.py | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 tests/test_postgres_recovery_replay_serialization_snapshot.py 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, + } From 80742ec0265e2cd0d708ca1dcb52d3d7ac95c4b2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 21 Aug 2026 04:37:03 -0700 Subject: [PATCH 8/9] fix(recovery): serialize validated replay observation snapshot --- .../postgres_recovery_replay_observation.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/pg_llm_batch/postgres_recovery_replay_observation.py b/pg_llm_batch/postgres_recovery_replay_observation.py index 45c3e4db1..58e4fa694 100644 --- a/pg_llm_batch/postgres_recovery_replay_observation.py +++ b/pg_llm_batch/postgres_recovery_replay_observation.py @@ -63,10 +63,10 @@ def target_reached(self) -> bool: def as_dict(self) -> dict[str, object]: """Return the stable content-free machine-readable observation schema.""" - _require_observed(self) + target_lsn, replay_lsn = _require_observed(self) return { - "target_lsn": self.target_lsn, - "replay_lsn": self.replay_lsn, + "target_lsn": target_lsn, + "replay_lsn": replay_lsn, "recovery_in_progress": True, "replay_paused": True, "target_reached": True, @@ -92,12 +92,20 @@ def postgres_recovery_replay_observation_was_observed(evidence: object) -> bool: ) -def _require_observed(evidence: PostgresRecoveryReplayObservation) -> None: - """Fail closed when a record is fabricated, copied, or mutated after observation.""" +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: From 16a03699d069ae424b9bacf16ddd9ba0db5b1c9f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 21 Aug 2026 06:49:39 -0700 Subject: [PATCH 9/9] test(recovery): cover provenance eviction guard --- ...st_postgres_recovery_replay_observation.py | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/test_postgres_recovery_replay_observation.py b/tests/test_postgres_recovery_replay_observation.py index e53720d4e..010d8bcc9 100644 --- a/tests/test_postgres_recovery_replay_observation.py +++ b/tests/test_postgres_recovery_replay_observation.py @@ -7,6 +7,7 @@ import pytest +import pg_llm_batch.postgres_recovery_replay_observation as replay_observation from pg_llm_batch.postgres_recovery_replay_observation import ( PostgresRecoveryReplayObservation, PostgresRecoveryReplayObservationError, @@ -151,6 +152,39 @@ def test_copied_or_mutated_observation_loses_inspection_provenance() -> None: 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", [