Replication & High Availability

# On MASTER server (primary)

# 1. Configure postgresql.conf
wal_level = replica               # Write-Ahead Log includes replication data
max_wal_senders = 3               # Max 3 replicas can connect
wal_keep_size = 1GB               # Keep 1GB of WAL for slow replicas

# 2. Allow replica connections in pg_hba.conf
# TYPE  DATABASE  USER        ADDRESS         METHOD
host    replication  replicator  192.168.1.0/24  md5

# 3. Create replication user
CREATE USER replicator WITH REPLICATION PASSWORD 'secure_password';

# 4. Restart PostgreSQL
sudo systemctl restart postgresql

# On REPLICA server (standby)

# 1. Create base backup from master
pg_basebackup -h master.example.com -D /var/lib/postgresql/data               -U replicator -P -v -R -X stream -C -S replica1

# Flags:
#   -R: Create standby.signal (marks as replica)
#   -X stream: Stream WAL during backup
#   -C -S replica1: Create replication slot 'replica1'

# 2. Verify standby.signal exists
ls /var/lib/postgresql/data/standby.signal  # Should exist

# 3. Configure primary connection in postgresql.auto.conf
primary_conninfo = 'host=master.example.com port=5432 user=replicator password=secure_password'
primary_slot_name = 'replica1'

# 4. Start PostgreSQL (automatically becomes replica)
sudo systemctl start postgresql

# On MASTER: Check connected replicas
SELECT client_addr, state, sync_state, replay_lag
FROM pg_stat_replication;

# Output:
#  client_addr  |   state   | sync_state | replay_lag
# --------------+-----------+------------+------------
#  192.168.1.10 | streaming | async      | 00:00:00.05
#  192.168.1.11 | streaming | async      | 00:00:00.12

# On REPLICA: Check replication lag
SELECT
    now() - pg_last_xact_replay_timestamp() AS replication_lag;

# Output:
#  replication_lag
# -----------------
#  00:00:00.087654  -- 87ms behind master (excellent!)

import psycopg2
import random

# Database connections
MASTER = "postgresql://user:pass@master.example.com:5432/mydb"
REPLICAS = [
    "postgresql://user:pass@replica1.example.com:5432/mydb",
    "postgresql://user:pass@replica2.example.com:5432/mydb",
    "postgresql://user:pass@replica3.example.com:5432/mydb",
]

def get_write_conn():
    """Get connection to master for writes."""
    return psycopg2.connect(MASTER)

def get_read_conn():
    """Get connection to random replica for reads."""
    replica_url = random.choice(REPLICAS)  # Random selection
    return psycopg2.connect(replica_url)

def create_user(username, email):
    """Write operation → Master"""
    conn = get_write_conn()
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO users (username, email) VALUES (%s, %s)",
        (username, email)
    )
    conn.commit()
    conn.close()
    # Result: Written to master, async replicated to replicas

def get_users():
    """Read operation → Replica"""
    conn = get_read_conn()
    cur = conn.cursor()
    cur.execute("SELECT username, email FROM users")
    results = cur.fetchall()
    conn.close()
    return results
    # Result: Read from replica, master not touched

# Usage
create_user('alice', 'alice@example.com')  # Writes to master
users = get_users()  # Reads from random replica

import time

def create_post_and_show(user_id, content):
    """Problem: Read replica may not have the post yet!"""

    # Write to master
    conn_master = get_write_conn()
    cur = conn_master.cursor()
    cur.execute(
        "INSERT INTO posts (user_id, content) VALUES (%s, %s) RETURNING id",
        (user_id, content)
    )
    post_id = cur.fetchone()[0]
    conn_master.commit()
    conn_master.close()

    # Immediate read from replica - MIGHT FAIL!
    conn_replica = get_read_conn()
    cur_replica = conn_replica.cursor()
    cur_replica.execute("SELECT * FROM posts WHERE id = %s", (post_id,))
    post = cur_replica.fetchone()  # Could be None due to replication lag!

    return post  # User doesn't see their own post! Bad UX

def create_post_and_show_fixed(user_id, content):
    """Solution: Read from master after write (read-your-writes consistency)"""

    # Write to master
    conn_master = get_write_conn()
    cur = conn_master.cursor()
    cur.execute(
        "INSERT INTO posts (user_id, content) VALUES (%s, %s) RETURNING id",
        (user_id, content)
    )
    post_id = cur.fetchone()[0]
    conn_master.commit()

    # Read from SAME master connection (not replica)
    cur.execute("SELECT * FROM posts WHERE id = %s", (post_id,))
    post = cur.fetchone()
    conn_master.close()

    return post  # Guaranteed to exist!

# Result: User always sees their own writes, other users tolerate slight delay

# Scenario: Two users update same record on different masters simultaneously

# Master 1 (US East)                    # Master 2 (EU West)
UPDATE users                            UPDATE users
SET email = 'alice@usa.com'             SET email = 'alice@europe.com'
WHERE id = 1;                           WHERE id = 1;

# Both commit successfully (on their local master)
# Replication propagates changes to all masters

# Result: CONFLICT! Which email is correct?
# Master 1 sees: alice@usa.com → alice@europe.com (overwritten by M2)
# Master 2 sees: alice@europe.com → alice@usa.com (overwritten by M1)

# Without conflict resolution, databases diverge!

# Patroni: HA solution using etcd/Consul for leader election

# patroni.yml configuration
scope: postgres-cluster
name: node1

restapi:
  listen: 0.0.0.0:8008
  connect_address: 192.168.1.10:8008

etcd:
  hosts: etcd1:2379,etcd2:2379,etcd3:2379  # Distributed consensus

bootstrap:
  dcs:
    ttl: 30                    # Leader lease time
    loop_wait: 10              # Check interval
    retry_timeout: 10          # Retry before giving up
    maximum_lag_on_failover: 1048576  # Max 1MB lag for promotion

postgresql:
  listen: 0.0.0.0:5432
  data_dir: /var/lib/postgresql/data
  parameters:
    max_connections: 100
    shared_buffers: 256MB

import requests
import time

def monitor_cluster_health():
    """Monitor Patroni cluster via REST API."""
    patroni_api = "http://localhost:8008"

    while True:
        try:
            response = requests.get(f"{patroni_api}/cluster")
            cluster_info = response.json()

            for member in cluster_info['members']:
                print(f"Node: {member['name']}")
                print(f"  Role: {member['role']}")  # leader or replica
                print(f"  State: {member['state']}")  # running or stopped
                print(f"  Lag: {member.get('lag', 0)} bytes")

            # Alert if no leader
            leaders = [m for m in cluster_info['members'] if m['role'] == 'leader']
            if len(leaders) == 0:
                print("⚠️  ALERT: No leader! Failover in progress...")
            elif len(leaders) > 1:
                print("🚨 CRITICAL: Split-brain detected! Multiple leaders!")

        except Exception as e:
            print(f"Error monitoring cluster: {e}")

        time.sleep(5)

# Prevent split-brain with quorum (majority voting)

Cluster with 3 nodes:
  Master, Replica 1, Replica 2

Quorum = (3 / 2) + 1 = 2 nodes minimum

Network Partition:
  Group A: Master + Replica 1 (2 nodes) ✓ Has quorum, keeps leadership
  Group B: Replica 2 (1 node)           ✗ No quorum, goes read-only

Result:
  - Group A continues operating (has majority)
  - Group B rejects writes (minority, can't elect leader)
  - No split-brain!

When partition heals:
  - Replica 2 rejoins, replicates from Master
  - Cluster returns to normal automatically

# STONITH: Shoot The Other Node In The Head
# When promoting new master, forcibly shut down old master

# Option 1: Power fencing (IPMI/iLO)
def fence_node_power(node_ip):
    """Power off node via IPMI interface."""
    subprocess.run([
        "ipmitool", "-H", node_ip, "-U", "admin", "-P", "password",
        "power", "off"
    ])
    # Result: Old master powered off, can't accept writes

# Option 2: Network fencing (firewall rules)
def fence_node_network(node_ip):
    """Block node's database port at network level."""
    subprocess.run([
        "iptables", "-A", "INPUT", "-s", node_ip, "-p", "tcp",
        "--dport", "5432", "-j", "DROP"
    ])
    # Result: Old master can't communicate with clients or replicas

# PgBouncer: External connection pooler (sits between app and database)

# pgbouncer.ini configuration
[databases]
mydb = host=localhost port=5432 dbname=mydb

[pgbouncer]
listen_addr = *
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt

pool_mode = transaction        # Connection returned after each transaction
max_client_conn = 1000         # Max connections from applications
default_pool_size = 25         # Actual connections to database
reserve_pool_size = 5          # Extra connections for spikes

# Result:
# 1000 app connections → 25-30 database connections
# Database sees consistent, low connection count
# Apps get fast connection acquisition

from psycopg2 import pool
from contextlib import contextmanager

# Create global pool at application startup
connection_pool = pool.ThreadedConnectionPool(
    minconn=5,
    maxconn=50,
    host='localhost',
    database='mydb',
    user='postgres',
    password='password'
)

@contextmanager
def get_db_connection():
    """Context manager for safe connection handling."""
    conn = connection_pool.getconn()
    try:
        yield conn
        conn.commit()  # Commit transaction
    except Exception:
        conn.rollback()  # Rollback on error
        raise
    finally:
        connection_pool.putconn(conn)  # Always return to pool

# Usage
def get_user(user_id):
    with get_db_connection() as conn:
        cur = conn.cursor()
        cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cur.fetchone()
    # Connection automatically returned to pool when context exits