PostgreSQL Read Replicas — Lag Caused 400 Double Charges

-- io.thecodeforge.postgres.replication — Production setup
-- ============================================================
-- PRIMARY configuration (postgresql.conf)
-- ============================================================

-- Enable WAL archiving level — 'replica' is sufficient for streaming replication
ALTER SYSTEM SET wal_level = 'replica';

-- Allow at least as many WAL senders as you have replicas + 1 for pg_basebackup
ALTER SYSTEM SET max_wal_senders = 5;

-- Keep enough WAL segments to allow replica to catch up after a brief disconnect
-- At 16MB per segment, this is 1GB of WAL retained — tune based on your write volume
ALTER SYSTEM SET wal_keep_size = '1GB';

-- Enable hot_standby on primary (needed so replica can serve reads)
-- Technically a replica setting, but set it on primary too for when roles flip
ALTER SYSTEM SET hot_standby = on;

-- Reload config — no restart needed for most of these
SELECT pg_reload_conf();

-- ============================================================
-- Create replication user (PRIMARY)
-- Never use a superuser for replication — least privilege matters
-- ============================================================

CREATE USER replicator
    WITH REPLICATION        -- grants REPLICATION privilege specifically
    LOGIN
    PASSWORD 'use_a_real_secret_from_vault_not_this';

-- ============================================================
-- pg_hba.conf on PRIMARY — allow replica to connect
-- Add this line to pg_hba.conf, then SELECT pg_reload_conf();
-- ============================================================

-- host  replication  replicator  10.0.1.50/32  scram-sha-256

-- ============================================================
-- On the REPLICA host — take base backup from primary
-- Run as postgres OS user on the REPLICA machine
-- ============================================================

-- pg_basebackup \
--   --host=10.0.1.10 \
--   --username=replicator \
--   --pgdata=/var/lib/postgresql/15/main \
--   --wal-method=stream \
--   --write-recovery-conf \
--   --checkpoint=fast \
--   --progress
--
-- --write-recovery-conf automatically creates standby.signal
-- and writes primary_conninfo into postgresql.auto.conf

-- ============================================================
-- REPLICA configuration (postgresql.conf)
-- ============================================================

-- Connection string to primary — vault this credential properly
ALTER SYSTEM SET primary_conninfo = 
    'host=10.0.1.10 port=5432 user=replicator password=your_secret application_name=replica_01';

-- hot_standby = on allows read queries while in recovery mode
-- Without this the replica is dark — no reads, just applying WAL
ALTER SYSTEM SET hot_standby = on;

-- Send feedback to primary about which transactions are still active on replica
-- This prevents primary from vacuuming rows the replica still needs for queries
-- Trade-off: primary retains dead tuples longer — monitor table bloat
ALTER SYSTEM SET hot_standby_feedback = on;

-- How long to wait before cancelling a conflicting query on the replica
-- -1 means wait indefinitely — use this only if your replica queries are short
-- In practice, set to 30s-120s depending on your reporting query SLAs
ALTER SYSTEM SET max_standby_streaming_delay = '120s';

-- ============================================================
-- Monitoring replication health — run on PRIMARY
-- ============================================================

SELECT
    application_name,                          -- identifies which replica
    client_addr,                               -- replica's IP
    state,                                     -- 'streaming' is healthy; 'catchup' means it's behind
    sent_lsn,                                  -- how far primary has sent WAL
    write_lsn,                                 -- how far replica has written WAL to disk
    flush_lsn,                                 -- how far replica has fsync'd WAL
    replay_lsn,                                -- how far replica has APPLIED WAL
    pg_wal_lsn_diff(sent_lsn, replay_lsn) AS replication_lag_bytes,
    sync_state                                 -- 'async' or 'sync'
FROM pg_stat_replication;

-- ============================================================
-- Monitoring lag in seconds — run on REPLICA
-- ============================================================

SELECT
    CASE WHEN pg_is_in_recovery() THEN 'replica' ELSE 'primary' END AS role,
    EXTRACT(EPOCH FROM now() - pg_last_xact_replay_timestamp()) AS lag_seconds,
    pg_last_xact_replay_timestamp() AS last_applied_at
FROM pg_stat_wal_receiver;

# io.thecodeforge.postgres.routing — Production routing layer

import contextlib
from enum import Enum, auto
from typing import Generator
import psycopg2
import psycopg2.pool
from threading import local


class ReadPolicy(Enum):
    EVENTUAL = auto()   # replica pool — accepts lag
    STRONG = auto()     # primary pool — fresh data


class DatabaseRouter:
    """
    Routes database connections to primary or replica based on context.
    Thread-local state tracks whether we're inside a write transaction.
    
    PgBouncer endpoints:
      - PRIMARY:  pgbouncer-primary:6432  (transaction mode)
      - REPLICA:  pgbouncer-replica:6432  (session mode for replicas)
    """

    def __init__(self, primary_dsn: str, replica_dsn: str, pool_size: int = 20):
        self._local = local()
        self._primary_pool = psycopg2.pool.ThreadedConnectionPool(
            minconn=2, maxconn=pool_size, dsn=primary_dsn,
            application_name='order_service_primary'
        )
        self._replica_pool = psycopg2.pool.ThreadedConnectionPool(
            minconn=5, maxconn=pool_size * 2, dsn=replica_dsn,
            application_name='order_service_replica'
        )

    def _in_write_transaction(self) -> bool:
        return getattr(self._local, 'write_transaction_active', False)

    def _get_connection(self, policy: ReadPolicy):
        # If inside a write transaction, MUST use primary — consistency overrides
        if self._in_write_transaction() or policy == ReadPolicy.STRONG:
            return self._primary_pool.getconn(), self._primary_pool
        return self._replica_pool.getconn(), self._replica_pool

    @contextlib.contextmanager
    def write_transaction(self) -> Generator:
        """All reads inside this block go to primary — prevents stale-read bugs."""
        conn = self._primary_pool.getconn()
        self._local.write_transaction_active = True
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self._local.write_transaction_active = False
            self._primary_pool.putconn(conn)

    @contextlib.contextmanager
    def read_query(self, policy: ReadPolicy = ReadPolicy.EVENTUAL) -> Generator:
        """Default to replica — caller must opt into STRONG for fresh reads."""
        conn, pool = self._get_connection(policy)
        try:
            yield conn
            conn.rollback()  # ends implicit transaction, returns to pool cleanly
        except Exception:
            conn.rollback()
            raise
        finally:
            pool.putconn(conn)


# ============================================================
# Checkout flow with proper routing
# ============================================================

def process_checkout(router: DatabaseRouter, cart_id: str, user_id: str) -> dict:
    # Inventory check — replica is fine, lag acceptable
    with router.read_query(ReadPolicy.EVENTUAL) as conn:
        cursor = conn.cursor()
        cursor.execute(
            'SELECT product_id, quantity FROM inventory WHERE cart_id = %s',
            (cart_id,)
        )
        inventory = cursor.fetchall()

    if not inventory:
        raise ValueError('Cart empty')

    # Write transaction — everything inside hits primary
    with router.write_transaction() as conn:
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO orders (cart_id, user_id, status) VALUES (%s, %s, %s) RETURNING id',
            (cart_id, user_id, 'pending')
        )
        order_id = cursor.fetchone()[0]

        cursor.execute(
            'INSERT INTO payments (order_id, status, created_at) VALUES (%s, %s, NOW())',
            (order_id, 'initiated')
        )

    # Order summary read — replica is fine, display only
    with router.read_query(ReadPolicy.EVENTUAL) as conn:
        cursor = conn.cursor()
        cursor.execute(
            'SELECT id, status, created_at FROM orders WHERE id = %s',
            (order_id,)
        )
        order = cursor.fetchone()

    return {'order_id': order_id, 'status': order[1] if order else 'pending'}

-- ============================================================
-- Comprehensive lag monitoring view — create on PRIMARY
-- ============================================================

CREATE OR REPLACE VIEW v_replication_health AS
SELECT
    r.application_name,
    r.client_addr,
    r.state,
    r.sync_state,
    -- Byte-level lag breakdown
    pg_wal_lsn_diff(r.sent_lsn, r.write_lsn)   AS network_lag_bytes,
    pg_wal_lsn_diff(r.write_lsn, r.flush_lsn)  AS disk_lag_bytes,
    pg_wal_lsn_diff(r.flush_lsn, r.replay_lsn) AS apply_lag_bytes,
    pg_wal_lsn_diff(r.sent_lsn, r.replay_lsn)  AS total_lag_bytes,
    r.replay_lag,                               -- human time
    rs.slot_name,
    rs.active AS slot_active,
    pg_wal_lsn_diff(pg_current_wal_lsn(), rs.confirmed_flush_lsn) AS slot_lag_bytes
FROM
    pg_stat_replication r
    LEFT JOIN pg_replication_slots rs ON rs.active_pid = r.pid
ORDER BY total_lag_bytes DESC;

-- ============================================================
-- Circuit breaker: check if replica is fresh enough for reads
-- Run on REPLICA — embed this in your connection pool health check
-- ============================================================

CREATE OR REPLACE FUNCTION fn_replica_is_fresh(
    max_lag_seconds NUMERIC DEFAULT 30
) RETURNS BOOLEAN AS $$
DECLARE
    lag_seconds NUMERIC;
BEGIN
    SELECT COALESCE(
        EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp())),
        999999  -- if NULL, replica hasn't replayed anything yet
    ) INTO lag_seconds;
    RETURN lag_seconds <= max_lag_seconds;
END;
$$ LANGUAGE plpgsql STABLE;

-- Usage: SELECT fn_replica_is_fresh(30); -- TRUE if lag < 30s

-- ============================================================
-- Find what's causing lag spikes: conflicting queries on replica
-- Run on REPLICA
-- ============================================================

SELECT
    pid,
    now() - query_start AS duration,
    wait_event_type,
    wait_event,
    LEFT(query, 120) AS query_preview
FROM pg_stat_activity
WHERE state != 'idle'
  AND (wait_event_type = 'Lock' OR wait_event LIKE '%recovery%')
ORDER BY duration DESC;

-- ============================================================
-- Pre-failover health check — run on PRIMARY
-- ============================================================

SELECT
    application_name,
    state,
    pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) AS unconfirmed_bytes,
    replay_lag,
    sync_state
FROM pg_stat_replication
ORDER BY unconfirmed_bytes ASC;

-- For zero-RPO failover: only promote if unconfirmed_bytes = 0

-- ============================================================
-- Manual promotion (run on REPLICA being promoted)
-- ============================================================

-- Shell: pg_ctl promote -D /var/lib/postgresql/15/main
-- OR if trigger_file is configured: touch /tmp/postgresql.trigger.5432

-- After promotion, confirm:
SELECT
    pg_is_in_recovery() AS is_still_replica,     -- should be false
    pg_current_wal_lsn() AS current_lsn,
    current_setting('transaction_read_only') AS read_only;

-- ============================================================
-- After promotion: re-sync old primary using pg_rewind
-- Run on OLD PRIMARY (now stopped)
-- ============================================================

-- pg_rewind \
--   --target-pgdata=/var/lib/postgresql/15/main \
--   --source-server='host=10.0.1.50 user=replicator dbname=postgres' \
--   --progress
--
-- Then: echo 'primary_conninfo = ...' >> postgresql.auto.conf
-- touch standby.signal
-- pg_ctl start

-- ============================================================
-- Verify topology after failover — run on new PRIMARY
-- ============================================================

SELECT
    application_name,
    client_addr,
    state,
    pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) AS lag_bytes
FROM pg_stat_replication;