Replication & High Availability

# Open a psql session inside the primary container. Everything below the
# next line is SQL typed at the resulting "demo=#" prompt. Use \q to exit.
docker exec -it pg-primary psql -U demo -d demo

-- The containers start empty, so create the table this example writes to.
-- It reaches the standby through replication, like any other change.
CREATE TABLE IF NOT EXISTS users (id serial PRIMARY KEY, username text);

-- PostgreSQL replicates asynchronously by default: sync_state reads
-- "async" and COMMIT never waits for a standby.
SELECT application_name, sync_state FROM pg_stat_replication;

-- Name the standbys that must acknowledge before COMMIT returns. The name
-- is matched against the standby's application_name, which defaults to
-- "walreceiver" unless primary_conninfo sets one.
ALTER SYSTEM SET synchronous_standby_names = 'walreceiver';
SELECT pg_reload_conf();

SELECT application_name, sync_state, sync_priority FROM pg_stat_replication;

-- Durability is per transaction, so the expensive guarantee can apply only
-- where it is worth paying for. This session opts out and does not wait:
SET synchronous_commit = off;
INSERT INTO users (username) VALUES ('analytics_import');

\q

# sync_failure.sh - what a single synchronous standby costs when it dies.
# Self-contained: it makes the pair synchronous and builds its own table, so
# the numbers below are the same on every run and do not depend on anything
# else in this lesson you have already run.

echo "--- setup: make the pair synchronous, with exactly one committed row"
docker exec pg-primary psql -U demo -d demo -qtAc \
  "ALTER SYSTEM SET synchronous_standby_names = 'walreceiver';"
docker exec pg-primary psql -U demo -d demo -qtAc "SELECT pg_reload_conf();"
docker exec pg-primary psql -U demo -d demo -q \
  -c "SET client_min_messages = warning;" \
  -c "DROP TABLE IF EXISTS sync_demo;" \
  -c "CREATE TABLE sync_demo (id serial PRIMARY KEY, note text);" \
  -c "INSERT INTO sync_demo (note) VALUES ('committed while the standby was up');"

docker stop pg-replica

echo "--- write (10s timeout, because it will not return on its own):"
timeout 10 docker exec pg-primary psql -U demo -d demo \
  -c "INSERT INTO sync_demo (note) VALUES ('written while the standby is down');"
echo "exit status: $?"

echo "--- read, same server, same moment:"
timeout 10 docker exec pg-primary psql -U demo -d demo \
  -tAc "SELECT count(*) FROM sync_demo;"
echo "exit status: $?"

echo "--- restoring: bring the standby back"
docker start pg-replica

# Careful with the quorum form. synchronous_standby_names is matched against
# each standby's application_name, and a list that matches NOTHING behaves
# exactly like the outage above: every write blocks, forever. This pair has
# one standby, called "walreceiver", so setting
#     'ANY 1 (replica1, replica2)'
# here would wedge the primary for the rest of the lesson. That form belongs
# in production, where replica1 and replica2 actually exist. Go back to
# asynchronous instead.
docker exec pg-primary psql -U demo -d demo -qtAc \
  "ALTER SYSTEM SET synchronous_standby_names = '';"
docker exec pg-primary psql -U demo -d demo -qtAc "SELECT pg_reload_conf();"

echo "--- rows now that the wait is over:"
docker exec pg-primary psql -U demo -d demo -tAc "SELECT count(*) FROM sync_demo;"

# On the PRIMARY. Three server settings, one auth rule, one role.

# 1. wal_level, max_wal_senders and hot_standby need a RESTART, so the
#    container at the top of this lesson passes them on its command line:
#      -c wal_level=replica -c max_wal_senders=10 -c hot_standby=on
#    On a real server the same three go in postgresql.conf:
#      wal_level = replica     WAL detailed enough to rebuild a replica from
#      max_wal_senders = 10    One sender per connected replica, plus one
#                              MORE for each running pg_basebackup that uses
#                              "-X stream". Set it above your replica count.
#      hot_standby = on        Let replicas serve read-only queries

# 2. wal_keep_size needs only a reload, so change it the way you would change
#    any setting on a server you cannot restart: ALTER SYSTEM, then reload.
#    It is fallback WAL retention for replicas that fall behind; replication
#    slots (next block) do the same job precisely. If you use slots, set
#    max_slot_wal_keep_size too, so a replica that never comes back cannot
#    fill the primary's disk.
docker exec pg-primary psql -U demo -d demo -qtAc \
  "ALTER SYSTEM SET wal_keep_size = '1GB';"
docker exec pg-primary psql -U demo -d demo -qtAc "SELECT pg_reload_conf();"

# 3. A dedicated role. REPLICATION is a privilege that has to be granted
#    explicitly, and this role needs no access to any table. Written so that
#    re-running this script is harmless.
docker exec -i pg-primary psql -U demo -d demo -q <<'SQL'
DO $$
BEGIN
  IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'replicator') THEN
    CREATE ROLE replicator WITH REPLICATION LOGIN PASSWORD 'secure_password';
  END IF;
END $$;
SQL

# 4. pg_hba.conf. "replication" is a pseudo-database: a line with "all" in
#    the DATABASE column does NOT match replication connections. That is the
#    single most common reason pg_basebackup fails with "no pg_hba.conf entry
#    for replication connection". The setup at the top of this lesson already
#    added a permissive version of this line; in production scope it to the
#    replication role and your subnet:
#      # TYPE  DATABASE     USER        ADDRESS         METHOD
#      host    replication  replicator  192.168.1.0/24  scram-sha-256

# 5. Confirm what the server actually ended up using. The "source" column is
#    the one to read, not just "setting".
docker exec pg-primary psql -U demo -d demo \
  -c "SELECT name, setting, source FROM pg_settings
      WHERE name IN ('wal_level','max_wal_senders','wal_keep_size','hot_standby')
      ORDER BY name;"

# On the REPLICA. The quick-start container at the top of this lesson cloned
# the primary as "demo" with no replication slot, which is fine for getting a
# pair running. Rebuild it the way you would in production: as the dedicated
# replicator role, with a slot of its own.

echo "--- replacing the quick-start replica"
docker rm -f pg-replica > /dev/null

# Slots outlive the replica that used them. That is exactly what they are
# for (the primary keeps WAL for a standby that is temporarily gone) and
# also their danger: an orphaned slot retains WAL forever and will fill the
# primary's disk. Since we are rebuilding, drop the old one, or the "-C"
# below fails with 'replication slot "replica1" already exists'.
docker exec pg-primary psql -U demo -d demo -qtAc \
  "SELECT pg_drop_replication_slot('replica1')
   WHERE EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'replica1');" \
  > /dev/null

# 1. pg_basebackup clones the primary's whole data directory. You do not
#    create a schema or restore a dump; the replica then catches up by
#    replaying WAL. There is no "systemctl start" afterwards: the server is
#    the container's main process, so the command ends by exec'ing it.
#    "su-exec postgres" drops from root first, because the server refuses to
#    run as root and this sh -c DOES run as root: passing a command other
#    than "postgres" makes the image entrypoint skip its own directory setup
#    and user switch, which is also why chown/chmod are needed above.
#    On a real server: the same pg_basebackup, then
#    "sudo systemctl start postgresql".
#
#    Flags:
#      -R          Write standby.signal AND the connection settings (step 3)
#      -X stream   Stream WAL during the backup, so the copy is consistent
#                  by the time the backup finishes
#      -C -S name  Create replication slot 'replica1' on the primary, so the
#                  primary retains WAL for this replica specifically instead
#                  of relying on wal_keep_size guesswork
docker run -d --name pg-replica --network pgrepl -p 5433:5432 \
  -e PGPASSWORD=secure_password postgres:16-alpine sh -c \
  "rm -rf /var/lib/postgresql/data/* && \
   chown postgres:postgres /var/lib/postgresql/data && \
   chmod 700 /var/lib/postgresql/data && \
   su-exec postgres pg_basebackup -h pg-primary -U replicator \
     -D /var/lib/postgresql/data -P -v -R -X stream -C -S replica1 && \
   exec su-exec postgres postgres" > /dev/null

sleep 10

# 2. -R created standby.signal. Its presence is what makes the server start
#    in standby mode. The file is empty; only the filename matters.
echo "--- standby.signal (its presence is what starts the server as a standby):"
docker exec pg-replica ls -l /var/lib/postgresql/data/standby.signal

# 3. -R ALSO wrote the connection settings, into postgresql.auto.conf. Do not
#    hand-write this file: ALTER SYSTEM rewrites it wholesale, and it says so
#    on its first line. Read it to confirm; change it with ALTER SYSTEM.
echo "--- postgresql.auto.conf (written by -R, not by you):"
docker exec pg-replica cat /var/lib/postgresql/data/postgresql.auto.conf

# 4. No start step: the container is already running as a standby. Confirm
#    it from both ends rather than assuming.
echo "--- is the replica in recovery?"
docker exec pg-replica psql -U demo -d demo -tAc "SELECT pg_is_in_recovery();"
echo "--- what the primary sees, and the slot it is now retaining WAL for:"
docker exec pg-primary psql -U demo -d demo \
  -c "SELECT application_name, state, sync_state FROM pg_stat_replication;" \
  -c "SELECT slot_name, slot_type, active FROM pg_replication_slots;"

# The two queries below run on DIFFERENT servers, so open a session on each.
# Both report on REPLAYED transactions, so on a pair that has never taken a
# write they come back empty: write something first, or you will misread a
# perfectly healthy idle pair as a broken one.

docker exec -it pg-primary psql -U demo -d demo

CREATE TABLE IF NOT EXISTS users (id serial PRIMARY KEY, username text);
INSERT INTO users (username) VALUES ('alice');

-- Which replicas are connected, and how far behind are they?
SELECT client_addr, state, sync_state, replay_lag FROM pg_stat_replication;


docker exec -it pg-replica psql -U demo -d demo

-- How long since this replica last replayed a transaction? Ask twice,
-- 15 seconds apart, with nobody writing to the primary in between.
SELECT now() - pg_last_xact_replay_timestamp() AS replication_lag;
SELECT pg_sleep(15);
SELECT now() - pg_last_xact_replay_timestamp() AS replication_lag;


# Then go back to the primary and ask IT the same question, after all that
# idling, to see how differently the two metrics behave.
docker exec -it pg-primary psql -U demo -d demo

SELECT client_addr, state, sync_state, replay_lag FROM pg_stat_replication;

import random

import psycopg

# Both URLs point at the containers started at the top of this lesson.
# psycopg connects with libpq directly, so the plain "postgresql://" scheme
# is correct here. (The "postgresql+psycopg://" form is a SQLAlchemy-only
# spelling, needed there to pick psycopg 3 over psycopg2.)
PRIMARY = "postgresql://demo:demo@localhost:5432/demo"
REPLICAS = [
    "postgresql://demo:demo@localhost:5433/demo",
    # In production you list every replica here:
    #   "postgresql://demo:demo@replica2.example.com:5432/demo",
    #   "postgresql://demo:demo@replica3.example.com:5432/demo",
]


def get_write_conn():
    """Writes always go to the primary."""
    return psycopg.connect(PRIMARY)


def get_read_conn():
    """Reads go to a replica, picked at random to spread the load."""
    return psycopg.connect(random.choice(REPLICAS))


# pg_is_in_recovery() is the ground truth: it returns True on a standby and
# False on a primary. This is how you prove your routing actually works,
# rather than assuming it does.
with get_write_conn() as conn:
    in_recovery = conn.execute("SELECT pg_is_in_recovery()").fetchone()[0]
    print(f"get_write_conn() -> in recovery: {in_recovery}  (False = primary)")

with get_read_conn() as conn:
    in_recovery = conn.execute("SELECT pg_is_in_recovery()").fetchone()[0]
    print(f"get_read_conn()  -> in recovery: {in_recovery}   (True  = replica)")

# A replica is genuinely read-only. The error is worth seeing once, because
# it is exactly what a mis-routed write looks like in production.
try:
    with get_read_conn() as conn:
        conn.execute("CREATE TABLE should_not_work (id int)")
except psycopg.errors.ReadOnlySqlTransaction as exc:
    print(f"write sent to a replica -> {type(exc).__name__}: "
          f"{str(exc).strip().splitlines()[0]}")

import random
import time

import psycopg

PRIMARY = "postgresql://demo:demo@localhost:5432/demo"
REPLICAS = ["postgresql://demo:demo@localhost:5433/demo"]


def get_write_conn():
    return psycopg.connect(PRIMARY)


def get_read_conn():
    return psycopg.connect(random.choice(REPLICAS))


def wait_for_replica(timeout=10.0):
    """Block until the replica has replayed everything the primary has written.

    Compares the primary's current WAL position against the replica's last
    replayed position. This is the honest way to wait: sleeping for a fixed
    number of milliseconds only hides the race, it does not remove it.
    """
    with get_write_conn() as conn:
        target = conn.execute("SELECT pg_current_wal_lsn()").fetchone()[0]
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        with get_read_conn() as conn:
            replayed = conn.execute(
                "SELECT pg_last_wal_replay_lsn() >= %s", (target,)
            ).fetchone()[0]
        if replayed:
            return
        time.sleep(0.01)
    raise TimeoutError(f"replica did not reach {target} within {timeout}s")


def setup():
    """DDL is a write, so it goes to the primary. It reaches the replica the
    same way every other change does: through replication."""
    with get_write_conn() as conn:
        conn.execute("DROP TABLE IF EXISTS users")
        conn.execute("""
            CREATE TABLE users (
                id       serial PRIMARY KEY,
                username text NOT NULL UNIQUE,
                email    text NOT NULL
            )""")
        conn.commit()


def create_user(username, email):
    """Write operation -> primary."""
    with get_write_conn() as conn:
        conn.execute(
            "INSERT INTO users (username, email) VALUES (%s, %s)",
            (username, email),
        )
        conn.commit()


def get_users():
    """Read operation -> replica."""
    with get_read_conn() as conn:
        return conn.execute(
            "SELECT username, email FROM users ORDER BY id"
        ).fetchall()


setup()
create_user("alice", "alice@example.com")
create_user("bob", "bob@example.com")

wait_for_replica()
for username, email in get_users():
    print(f"read from replica: {username:6} {email}")

import psycopg

PRIMARY = "postgresql://demo:demo@localhost:5432/demo"
REPLICA = "postgresql://demo:demo@localhost:5433/demo"


def setup():
    with psycopg.connect(PRIMARY) as conn:
        conn.execute("DROP TABLE IF EXISTS posts")
        conn.execute("""
            CREATE TABLE posts (
                id      serial PRIMARY KEY,
                user_id int  NOT NULL,
                content text NOT NULL
            )""")
        conn.commit()


def create_post_and_show(user_id, content):
    """Write to the primary, then immediately read from the replica."""
    with psycopg.connect(PRIMARY) as conn:
        post_id = conn.execute(
            "INSERT INTO posts (user_id, content) VALUES (%s, %s) RETURNING id",
            (user_id, content),
        ).fetchone()[0]
        conn.commit()

    with psycopg.connect(REPLICA) as conn:
        # Could be None: the replica may not have replayed this INSERT yet.
        return post_id, conn.execute(
            "SELECT id, user_id, content FROM posts WHERE id = %s", (post_id,)
        ).fetchone()


setup()

# On a loopback network the replica is usually a millisecond behind, so the
# race below almost never loses and the bug ships to production undetected.
# pg_wal_replay_pause() stops WAL replay on the replica, which turns that
# rare race into a guaranteed failure you can actually see and test against.
# It is the standby-side equivalent of a slow network or an overloaded disk.
with psycopg.connect(REPLICA) as conn:
    conn.execute("SELECT pg_wal_replay_pause()")
    conn.commit()
    print("replay state:", conn.execute(
        "SELECT pg_get_wal_replay_pause_state()").fetchone()[0])

post_id, post = create_post_and_show(1, "my first post")
print(f"primary wrote post id={post_id}")
print(f"replica returned:  {post}   <-- the user does not see their own post")

with psycopg.connect(REPLICA) as conn:
    conn.execute("SELECT pg_wal_replay_resume()")
    conn.commit()

with psycopg.connect(REPLICA) as conn:
    print("after resuming replay:",
          conn.execute("SELECT id, user_id, content FROM posts WHERE id = %s",
                       (post_id,)).fetchone())

import psycopg

PRIMARY = "postgresql://demo:demo@localhost:5432/demo"
REPLICA = "postgresql://demo:demo@localhost:5433/demo"


def setup():
    with psycopg.connect(PRIMARY) as conn:
        conn.execute("DROP TABLE IF EXISTS posts")
        conn.execute("""
            CREATE TABLE posts (
                id      serial PRIMARY KEY,
                user_id int  NOT NULL,
                content text NOT NULL
            )""")
        conn.commit()


def create_post_and_show(user_id, content):
    """Read-your-writes: the author reads back from the same primary
    connection that performed the write, so replication lag cannot apply."""
    with psycopg.connect(PRIMARY) as conn:
        post_id = conn.execute(
            "INSERT INTO posts (user_id, content) VALUES (%s, %s) RETURNING id",
            (user_id, content),
        ).fetchone()[0]
        conn.commit()

        # Same connection, same server. Nothing to be stale about.
        return conn.execute(
            "SELECT id, user_id, content FROM posts WHERE id = %s", (post_id,)
        ).fetchone()


def get_feed():
    """Everyone else reads from the replica and tolerates a little staleness."""
    with psycopg.connect(REPLICA) as conn:
        return conn.execute(
            "SELECT id, user_id, content FROM posts ORDER BY id").fetchall()


setup()

# Pause replay again, to prove the fix holds under the exact conditions that
# broke the previous version rather than under ideal ones.
with psycopg.connect(REPLICA) as conn:
    conn.execute("SELECT pg_wal_replay_pause()")
    conn.commit()

print("author sees:", create_post_and_show(1, "my first post"))
print("replica feed while replay is paused:", get_feed())

with psycopg.connect(REPLICA) as conn:
    conn.execute("SELECT pg_wal_replay_resume()")
    conn.commit()

import json

# A real response from Patroni's REST API, GET http://<node>:8008/cluster.
# Keeping it as a literal makes the parsing below runnable and testable
# without a live cluster; swap in the requests call at the bottom for real use.
SAMPLE = """
{
  "members": [
    {"name": "node1", "role": "leader",  "state": "running",
     "api_url": "http://192.168.1.10:8008/patroni",
     "host": "192.168.1.10", "port": 5432, "timeline": 7},
    {"name": "node2", "role": "replica", "state": "streaming",
     "api_url": "http://192.168.1.11:8008/patroni",
     "host": "192.168.1.11", "port": 5432, "timeline": 7, "lag": 0},
    {"name": "node3", "role": "replica", "state": "streaming",
     "api_url": "http://192.168.1.12:8008/patroni",
     "host": "192.168.1.12", "port": 5432, "timeline": 7, "lag": 12582912}
  ]
}
"""

MAX_LAG_BYTES = 8 * 1024 * 1024  # alert above 8 MB of replication lag


def check_cluster(cluster_info):
    """Turn one /cluster payload into a list of alerts. A pure function, so
    it can be unit tested against recorded payloads like the one above."""
    alerts = []
    for member in cluster_info["members"]:
        # Only replicas report "lag"; the leader has no lag key at all, so
        # member["lag"] would raise KeyError on the leader.
        lag = member.get("lag")
        lag_text = "n/a" if lag is None else f"{lag / 1024 / 1024:.1f} MB"
        print(f"  {member['name']:6} role={member['role']:8} "
              f"state={member['state']:10} lag={lag_text}")
        if lag is not None and lag > MAX_LAG_BYTES:
            alerts.append(f"{member['name']} is {lag_text} behind the leader")

    leaders = [m for m in cluster_info["members"] if m["role"] == "leader"]
    if not leaders:
        alerts.append("no leader: failover in progress or the cluster is down")
    elif len(leaders) > 1:
        alerts.append(f"CRITICAL split-brain: {len(leaders)} leaders")

    # Patroni bumps the timeline on every promotion, so members left on an
    # older timeline have not caught up to the most recent failover.
    timelines = {m.get("timeline") for m in cluster_info["members"]}
    if len(timelines) > 1:
        alerts.append(f"members disagree on timeline: {sorted(timelines)}")
    return alerts


print("steady state, one replica falling behind:")
for alert in check_cluster(json.loads(SAMPLE)):
    print(f"  ALERT: {alert}")

print("\nsame cluster during a bad failover:")
broken = json.loads(SAMPLE)
broken["members"][1]["role"] = "leader"   # a second leader appears
broken["members"][2]["timeline"] = 6      # and node3 is on an old timeline
for alert in check_cluster(broken):
    print(f"  ALERT: {alert}")

print("\nand mid-failover, with no leader at all:")
headless = json.loads(SAMPLE)
for member in headless["members"]:
    member["role"] = "replica"
for alert in check_cluster(headless):
    print(f"  ALERT: {alert}")

# In production, replace the literal with a poll of the real endpoint:
#
#   import requests, time
#   while True:
#       payload = requests.get("http://localhost:8008/cluster", timeout=5).json()
#       for alert in check_cluster(payload):
#           send_to_pagerduty(alert)
#       time.sleep(5)

def quorum(cluster_size):
    """Smallest group that is guaranteed to be unique in any partition.

    Integer division floors, so this is floor(n/2) + 1: strictly more than
    half. Two disjoint groups can never both satisfy it.
    """
    return cluster_size // 2 + 1


def no_quorum_rule(group_size, cluster_size):
    """"I cannot reach the leader, so the leader is dead and I am promoting."
    Perfectly reasonable, and every isolated group reaches it independently.
    """
    return group_size >= 1


def quorum_rule(group_size, cluster_size):
    """Promote only with a strict majority of the WHOLE cluster behind you,
    not just a majority of the nodes you can still see."""
    return group_size >= quorum(cluster_size)


def simulate(cluster_size, partition):
    """partition is the sizes of the isolated groups, e.g. [2, 1]. Runs both
    rules over the same split so the difference is the rule, nothing else."""
    assert sum(partition) == cluster_size, "every node lands in some group"
    print(f"cluster of {cluster_size}, partitioned into groups of "
          f"{' and '.join(str(s) for s in partition)}"
          f"   (quorum = {quorum(cluster_size)})")
    for name, rule in (("no quorum", no_quorum_rule), ("quorum", quorum_rule)):
        outcomes = ["leader" if rule(s, cluster_size) else "read-only"
                    for s in partition]
        leaders = outcomes.count("leader")
        detail = ", ".join(f"{s}->{o}" for s, o in zip(partition, outcomes))
        flag = "   <-- SPLIT BRAIN" if leaders > 1 else ""
        print(f"  {name:9}  {detail:34} leaders: {leaders}{flag}")
    print()


# The classic 3-node cluster. Under the naive rule both sides promote and the
# data diverges; under quorum only the side holding two nodes may lead.
simulate(3, [2, 1])

# 5 nodes tolerate two simultaneous failures instead of one.
simulate(5, [3, 2])

# The two-node trap, from both directions: the naive rule gives you two
# leaders, and quorum gives you none. A pair bought for availability is
# either corrupt or fully read-only, and there is no third option.
simulate(2, [1, 1])

# Adding a node to an odd cluster buys nothing. 3 and 4 both tolerate exactly
# one failure, so the fourth node adds cost and coordination, not resilience.
for n in range(2, 8):
    print(f"n={n}  quorum={quorum(n)}  tolerates {n - quorum(n)} node failure(s)")

import shlex
import subprocess

# STONITH: Shoot The Other Node In The Head. Before promoting a replica, the
# cluster makes sure the old primary cannot possibly still be serving writes.
# Quorum stops a minority from ELECTING a leader; it does not stop a node that
# already believes it is leader from accepting writes from clients that can
# still reach it. Fencing is what closes that gap.

DRY_RUN = True            # flip to False only against hardware you intend to kill
UNREACHABLE = {"192.168.1.20"}   # stands in for a controller that does not answer


def run(argv, node_ip):
    """Run a fencing command, or print it when DRY_RUN is set."""
    if DRY_RUN:
        failed = node_ip in UNREACHABLE
        print(f"  would run: {shlex.join(argv)}"
              f"{'   -> no response' if failed else ''}")
        return 1 if failed else 0
    return subprocess.run(argv, check=False).returncode


def fence_node_power(node_ip, user="admin", password="password"):
    """Power fencing: cut power via the node's out-of-band controller (IPMI,
    iLO, DRAC). The strongest option, because a powered-off machine cannot
    write to shared storage no matter what it believes about itself."""
    return run(["ipmitool", "-H", node_ip, "-U", user, "-P", password,
                "power", "off"], node_ip)


def fence_node_network(node_ip, port=5432):
    """Network fencing: drop the node's database traffic at the switch or
    firewall. Weaker than power fencing, because the node keeps running and
    can still reach anything you forgot to block, such as shared storage."""
    return run(["iptables", "-A", "INPUT", "-s", node_ip, "-p", "tcp",
                "--dport", str(port), "-j", "DROP"], node_ip)


def promote_with_fencing(old_primary_ip, new_primary_ip):
    print(f"failover: {old_primary_ip} -> {new_primary_ip}")
    print(" 1. fence the old primary BEFORE promoting anything")
    if fence_node_power(old_primary_ip) != 0:
        # Never promote on a failed fence. An unfenced old primary plus a new
        # one is the split-brain you were trying to avoid.
        raise RuntimeError("fencing failed; refusing to promote")
    print(" 2. fence confirmed, safe to promote")
    print(f" 3. promote {new_primary_ip} and repoint the remaining replicas")


promote_with_fencing("192.168.1.10", "192.168.1.11")

# The case that actually decides whether your cluster is safe: the fence does
# not succeed. Staying down is the correct outcome, and it has to be the
# default, because the alternative is two primaries taking writes.
print()
try:
    promote_with_fencing("192.168.1.20", "192.168.1.21")
except RuntimeError as exc:
    print(f" !! {exc}")
    print("    The cluster stays read-only until an operator intervenes.")
    print("    Promoting here would produce exactly the split-brain that")
    print("    fencing exists to prevent.")

print()
print("network fencing, as a fallback when there is no out-of-band controller:")
fence_node_network("192.168.1.10")

import time

import psycopg
from psycopg_pool import ConnectionPool

DSN = "postgresql://demo:demo@localhost:5432/demo"

# Create the pool once, at application startup. psycopg 3's pool already gives
# you a context manager that checks a connection out and always returns it, so
# no hand-rolled getconn/putconn wrapper is needed and no connection leaks on
# an exception path.
pool = ConnectionPool(DSN, min_size=5, max_size=50, open=True)
pool.wait(timeout=10)   # block until min_size connections are actually up

with pool.connection() as conn:
    conn.execute("DROP TABLE IF EXISTS users")
    conn.execute("CREATE TABLE users (id serial PRIMARY KEY, username text)")
    conn.execute("INSERT INTO users (username) VALUES ('alice')")


def get_user(user_id):
    # pool.connection() also wraps the block in a transaction: it commits on
    # clean exit and rolls back if the body raises.
    with pool.connection() as conn:
        return conn.execute(
            "SELECT id, username FROM users WHERE id = %s", (user_id,)
        ).fetchone()
    # the connection is back in the pool here, however the block exited


def timed(label, fn, n=200):
    start = time.perf_counter()
    for _ in range(n):
        fn()
    elapsed = (time.perf_counter() - start) / n * 1000
    print(f"{label:28} {elapsed:6.2f} ms per query")


def without_pool():
    with psycopg.connect(DSN) as conn:
        conn.execute("SELECT id, username FROM users WHERE id = 1").fetchone()


timed("new connection per query", without_pool)
timed("pooled connection", lambda: get_user(1))

# An exception inside the block still returns the connection to the pool.
try:
    with pool.connection() as conn:
        conn.execute("SELECT 1 / 0")
except psycopg.errors.DivisionByZero:
    print("query raised, connection returned anyway")
print("pool stats:", {k: v for k, v in pool.get_stats().items()
                      if k in ("pool_size", "pool_available", "requests_waiting")})

pool.close()   # at shutdown

# pgbouncer_up.sh - writes both config files, then starts PgBouncer.
#
# Write the files rather than assuming they exist. Docker does not fail when
# a bind-mount source is missing: it silently creates a DIRECTORY at that
# path, and PgBouncer then dies with the baffling
#   can't create /etc/pgbouncer/pgbouncer.ini: Is a directory
# If you have already hit that, delete the stray directories first:
#   docker rm -f pgbouncer; rmdir pgbouncer.ini userlist.txt

cat > pgbouncer.ini <<'INI'
[databases]
demo = host=pg-primary port=5432 dbname=demo

[pgbouncer]
listen_addr = *
listen_port = 6432
auth_type = scram-sha-256
auth_file = /etc/pgbouncer/userlist.txt

pool_mode = transaction
max_client_conn = 1000
default_pool_size = 25
reserve_pool_size = 5
INI

# userlist.txt holds the SCRAM verifier, which you copy from the server
# rather than write by hand. Postgres keeps it in pg_shadow. The verifier is
# salted, so regenerate this file whenever you recreate the role or the
# container: a stale one fails with "password authentication failed".
docker exec pg-primary psql -U demo -d demo -tAc \
  "SELECT '\"'||usename||'\" \"'||passwd||'\"' FROM pg_shadow WHERE usename='demo'" \
  > userlist.txt

docker rm -f pgbouncer > /dev/null 2>&1
docker run -d --name pgbouncer --network pgrepl -p 6432:6432 \
  -v "$PWD/pgbouncer.ini:/etc/pgbouncer/pgbouncer.ini:ro" \
  -v "$PWD/userlist.txt:/etc/pgbouncer/userlist.txt:ro" \
  edoburu/pgbouncer:latest > /dev/null

sleep 3
# Timestamps stripped so the output is the same on every run.
docker logs pgbouncer 2>&1 | sed -E 's/^[0-9-]{10} [0-9:.]+ UTC //' | tail -5

# Clean up:  docker rm -f pgbouncer

import threading
import time

import psycopg

# Straight to Postgres, used only to observe how many backends exist.
DIRECT = "postgresql://demo:demo@localhost:5432/demo"
# Through PgBouncer, which is what the application would actually use.
VIA_PGBOUNCER = "postgresql://demo:demo@localhost:6432/demo"


def server_backends():
    """How many real Postgres backends are serving this database right now."""
    with psycopg.connect(DIRECT) as conn:
        return conn.execute("""
            SELECT count(*) FROM pg_stat_activity
            WHERE datname = 'demo' AND backend_type = 'client backend'
        """).fetchone()[0] - 1        # minus this observing connection


# Run this against a freshly started PgBouncer. It keeps server connections
# warm after a burst (see the last line), so a second run starts at whatever
# the first run grew the pool to rather than at zero.
clients = [psycopg.connect(VIA_PGBOUNCER) for _ in range(100)]
print(f"client connections held open  : {len(clients)}")
print(f"postgres backends, all idle   : {server_backends()}")

# pool_mode = transaction means a server connection is only assigned for the
# duration of a transaction. Run the queries one after another and a single
# backend serves all 100 clients.
for conn in clients:
    conn.execute("SELECT 1").fetchone()
    conn.commit()
print(f"after 100 serial transactions : {server_backends()}")

# Concurrency is what actually drives the backend count: PgBouncer needs one
# server connection per IN-FLIGHT transaction, and default_pool_size is the
# ceiling. Ask for 40 at once and 25 run while 15 queue for a free slot.
peak = 0


def busy_transaction(conn):
    conn.execute("SELECT pg_sleep(0.4)")
    conn.commit()


threads = [threading.Thread(target=busy_transaction, args=(c,))
           for c in clients[:40]]
for t in threads:
    t.start()
for _ in range(12):
    peak = max(peak, server_backends())
    time.sleep(0.05)
for t in threads:
    t.join()

print(f"peak backends, 40 concurrent  : {peak}  (default_pool_size = 25)")
# Not back to 1: PgBouncer holds grown server connections until
# server_idle_timeout (600s by default) rather than reconnecting per burst.
print(f"after the burst, still warm   : {server_backends()}")

for conn in clients:
    conn.close()