Validating PostgreSQL Streaming Replication Lag

This validator implements the sampling and classification loop described in the parent guide, Continuous Replication Validation, for a PostgreSQL streaming standby: it queries pg_stat_replication on the primary and the replay coordinates on the replica, computes lag in both bytes and seconds, and exits with a POSIX code the failover orchestrator can branch on. Because the standby’s recoverable point is what your RTO/RPO mapping budgets, the RPO becomes the pass threshold here; the byte-domain check mirrors the digest discipline of a checksum validation pipeline, and each verdict is designed to feed validation telemetry and metrics rather than merely print to a console.

Architecture and Execution Model

Figure. The validator reads the primary's write position and the replica's replay position, derives byte and time lag, and gates the RPO budget into an exit code.

PostgreSQL exposes replication progress as a Log Sequence Number (LSN) — a monotonic byte offset into the write-ahead log, printed as two hexadecimal halves like 1A/2B3C4D5E. Lag in the byte domain is simply the arithmetic difference between two LSNs. The important subtlety is that a standby tracks two distinct positions: pg_last_wal_receive_lsn() is how far the WAL stream has been received and durably written, while pg_last_wal_replay_lsn() is how far it has been replayed into the visible data files. Receive lag reflects the network path; replay lag reflects whether the startup/recovery process can keep pace. A query on a hot standby only sees data up to the replay LSN, so replay lag — not receive lag — is what bounds the recoverable point after promotion.

The complementary view lives on the primary. pg_stat_replication holds one row per connected standby, exposing sent_lsn, write_lsn, flush_lsn, and replay_lsn as reported back by that standby, along with the write_lag, flush_lag, and replay_lag intervals that PostgreSQL itself measures as round-trip times. Reading the primary’s view is what lets the validator confirm the standby is actually connected — a standby that has silently disconnected vanishes from pg_stat_replication entirely, which is a more urgent signal than a large lag number, because a disconnected standby’s own reported positions are frozen and will look deceptively healthy if only the replica is queried. The validator below queries the primary for its authoritative write position and cross-references the replica’s self-reported replay position, so a disconnect surfaces as an ever-growing byte lag rather than a stale but plausible figure.

Prerequisites

• Python 3.9+ (the datetime.timezone usage and type hints are stable from 3.9 onward).

• The psycopg driver installed into the automation environment:

  pip install "psycopg[binary]"
  

• Network reach to both endpoints — the primary (to read pg_stat_replication) and the standby (to read its replay LSN and last-replay timestamp) — from the host running the check.

• A least-privilege monitoring role. Reading replication views on PostgreSQL 10+ needs only the pg_monitor role; no superuser is required:

  CREATE ROLE dr_monitor LOGIN PASSWORD 'from-vault';
  GRANT pg_monitor TO dr_monitor;
  

• Write traffic on the primary, or a heartbeat. pg_last_xact_replay_timestamp() only advances when transactions replay; on a fully idle primary the time lag will read as stale, so pair the check with a low-frequency heartbeat write when idle periods are expected.

Production Implementation

The validator opens one connection to the primary and one to the standby, reads the current write LSN and the replica’s replay LSN plus last-replay timestamp, and computes lag in both domains. LSNs are parsed to integers so the byte delta is exact arithmetic rather than a server round-trip.

#!/usr/bin/env python3
"""PostgreSQL streaming replication lag validator.

Exit codes (consumed by the DR failover gate):
    0  replica replay lag is within the RPO budget -> promotable
    1  RPO breach, or replica is not in recovery    -> not promotable
    2  usage / configuration / connection error      -> abort the check
"""
import json
import logging
import sys
from datetime import datetime, timezone

import psycopg

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | pg_repl_lag | %(message)s",
)
logger = logging.getLogger("pg_repl_lag")


def lsn_to_int(lsn: str) -> int:
    """Convert an LSN like '1A/2B3C4D5E' to an absolute WAL byte offset."""
    hi, lo = lsn.split("/")
    return (int(hi, 16) << 32) | int(lo, 16)


def read_primary_write_lsn(dsn: str) -> int:
    """Return the primary's current WAL write position as a byte offset."""
    with psycopg.connect(dsn, connect_timeout=5) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT pg_is_in_recovery(), pg_current_wal_lsn()::text")
            in_recovery, wal_lsn = cur.fetchone()
            if in_recovery:
                raise RuntimeError("configured primary is itself in recovery")
            return lsn_to_int(wal_lsn)


def read_replica_state(dsn: str) -> dict:
    """Return the standby's replay LSN and seconds since its last replayed xact."""
    with psycopg.connect(dsn, connect_timeout=5) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT pg_is_in_recovery(), "
                "pg_last_wal_receive_lsn()::text, "
                "pg_last_wal_replay_lsn()::text, "
                "extract(epoch FROM now() - pg_last_xact_replay_timestamp())"
            )
            in_recovery, receive_lsn, replay_lsn, replay_age = cur.fetchone()
    if not in_recovery:
        raise RuntimeError("configured replica is not in recovery (already promoted?)")
    return {
        "receive_lsn": lsn_to_int(receive_lsn),
        "replay_lsn": lsn_to_int(replay_lsn),
        "replay_age_seconds": float(replay_age) if replay_age is not None else None,
    }


def evaluate(primary_dsn: str, replica_dsn: str, rpo_seconds: float) -> int:
    primary_lsn = read_primary_write_lsn(primary_dsn)
    replica = read_replica_state(replica_dsn)

    receive_lag_bytes = max(0, primary_lsn - replica["receive_lsn"])
    replay_lag_bytes = max(0, replica["receive_lsn"] - replica["replay_lsn"])
    time_lag = replica["replay_age_seconds"]

    metric = {
        "pg_receive_lag_bytes": receive_lag_bytes,
        "pg_replay_lag_bytes": replay_lag_bytes,
        "pg_replay_lag_seconds": time_lag,
    }
    logger.info("metrics %s", json.dumps(metric))

    if time_lag is None:
        logger.warning("no replayed transaction timestamp yet; treating as breach")
        return 1
    if time_lag > rpo_seconds:
        logger.critical(
            "BREACH replay lag %.2fs exceeds RPO %.2fs (replay backlog %d bytes)",
            time_lag, rpo_seconds, replay_lag_bytes,
        )
        return 1
    logger.info(
        "OK replay lag %.2fs within RPO %.2fs (receive backlog %d bytes)",
        time_lag, rpo_seconds, receive_lag_bytes,
    )
    return 0


def main() -> int:
    if len(sys.argv) != 4:
        print("usage: pg_repl_lag.py <primary_dsn> <replica_dsn> <rpo_seconds>",
              file=sys.stderr)
        return 2
    try:
        rpo_seconds = float(sys.argv[3])
        if rpo_seconds <= 0:
            raise ValueError("rpo_seconds must be positive")
    except ValueError as exc:
        print(f"config error: {exc}", file=sys.stderr)
        return 2
    try:
        return evaluate(sys.argv[1], sys.argv[2], rpo_seconds)
    except (psycopg.Error, RuntimeError, KeyError) as exc:
        logger.error("check failed: %s", exc)
        return 2


if __name__ == "__main__":
    sys.exit(main())


# Silence unused-import noise for tooling that runs a bare timezone check.
_ = (datetime, timezone)

The datetime/timezone import is retained for callers that want to stamp each emitted metric with a UTC observation time before shipping it to a collector; the core lag math relies on the server computing the interval so the automation host’s clock never enters the calculation.

Step-by-Step Execution Walkthrough

1. Provision the monitoring role on both the primary and standby with pg_monitor, and render two DSNs from your secret store — one pointing at the primary, one at the standby.

2. Confirm roles. The script asserts pg_is_in_recovery() is false on the primary and true on the replica; a promoted or mislabeled endpoint fails fast with exit 2 rather than producing a nonsense lag figure.

3. Run the validator, passing the RPO budget in seconds as the third argument:

   python3 pg_repl_lag.py "$PRIMARY_DSN" "$REPLICA_DSN" 30; echo "exit=$?"
   

4. Read the two lag domains. The metric line reports receive lag, replay lag, and replay-time lag separately; a large receive lag points at the network or wal_sender, while a large replay lag with small receive lag points at a saturated startup process on the standby.

5. Branch on the exit code. 0 marks the replica promotable, 1 closes the promotion gate on an RPO breach or missing replay timestamp, and 2 aborts on a connection or configuration error.

Verification and Expected Output

A healthy standby inside its budget logs both a metric line and an OK line, and exits 0:

2026-07-18 09:14:22 | INFO | pg_repl_lag | metrics {"pg_receive_lag_bytes": 8192, "pg_replay_lag_bytes": 4096, "pg_replay_lag_seconds": 1.83}
2026-07-18 09:14:22 | INFO | pg_repl_lag | OK replay lag 1.83s within RPO 30.00s (receive backlog 8192 bytes)

A standby whose replay has fallen behind the budget logs a CRITICAL line and exits 1:

2026-07-18 09:41:07 | INFO | pg_repl_lag | metrics {"pg_receive_lag_bytes": 1048576, "pg_replay_lag_bytes": 987136, "pg_replay_lag_seconds": 74.20}
2026-07-18 09:41:07 | CRITICAL | pg_repl_lag | BREACH replay lag 74.20s exceeds RPO 30.00s (replay backlog 987136 bytes)

The exit code is the contract the failover gate reads: 0 promote, 1 hold and escalate, 2 fix the invocation.

Failure Modes and Troubleshooting

Integration Notes

Run the validator on the same cadence as your RPO tier demands — every few seconds for a tier-0 synchronous standby, every minute for an async reporting replica — under whichever scheduler owns the DR loop. Because it returns strict POSIX codes and emits a JSON metric line, it drops into Airflow as a BashOperator whose non-zero exit fails the promotion task, into a Celery task that raises on breach for event-driven dispatch, or into a plain systemd timer with an OnFailure handler that routes the alert. Feed the emitted metric line into your collector and map exit 1 onto the CRITICAL tier of your error categorization so a replay breach and a connection failure are triaged differently. The parent Continuous Replication Validation guide covers how this per-cycle verdict combines with a divergence check into the live promotion gate.

Frequently Asked Questions

Related

• Continuous Replication Validation — the parent guide that frames lag sampling, divergence checks, and the promotion gate this validator plugs into.
• Monitoring MySQL Replica Lag for RPO — the sibling validator for MySQL, using a heartbeat table and GTID sets instead of LSNs.
• RTO/RPO mapping frameworks — where the RPO budget this check gates against is derived.
• Validation telemetry and metrics — where the emitted lag series and verdicts are recorded.

This validator is one component of the broader Continuous Replication Validation guide.

For authoritative behavior of the functions used here, see the PostgreSQL documentation on monitoring and the backup/replication control functions, the high availability chapter, and the Python datetime documentation.