Streaming replication: standbys, slots and lag

standby_session.py

"""A hot standby on PostgreSQL 18.6, built with one command: standby_setup.sql writes 100 orders on a primary, and
pg_basebackup -R -C -S copies the running primary into the folder standby, creates the replication slot standby1 for
it and writes the settings that make the copy start as a standby. The copy starts on a port of its own (shown as
5433). The run lists the file that puts it in standby mode, prints the settings -R added, reads from the standby,
tries a write there, and reads the primary's two replication views.

The standby gets cluster_name = 'standby1', which is the name it gives the primary. Its connection string names the
run's password file (shown as pgpass), never a password. standby_primary.sql shows how far replay is behind in whole
kB, because a background record of a few dozen bytes (one the server writes about running transactions) may be on
its way at the moment of reading. Every value in the record is as the servers printed it, with the ports shown as
5432 and 5433. initdb runs with --no-sync, which only skips flushing its new files to disk.
"""

from pg_session import Server

# The settings of this track's sessions: no background work that would make two runs differ.
SETTINGS = {"autovacuum": "off", "jit": "off", "timezone": "'UTC'", "log_line_prefix": "''",
            "max_parallel_workers_per_gather": 0}
CAUGHT_UP = "SELECT pg_wal_lsn_diff(sent_lsn, replay_lsn) FROM pg_stat_replication"

with Server(settings=SETTINGS, initdb=("--no-sync",)) as pg:
    pg.up()
    pg.psql("standby_setup.sql", note="on the primary: 100 orders")
    port = pg.standby(slot="standby1", show=True, settings={"cluster_name": "'standby1'"})
    pg.listing("standby", prefix="standby")
    pg.show_file("standby/postgresql.auto.conf", note="what -R added to the standby's settings")
    pg.psql("standby_check.sql", port=port, note="on the standby")
    pg.psql("standby_write.sql", port=port, errors=1, status=3, note="a write on the standby")
    # Unprinted: wait until the standby has replayed everything the primary sent, as it does within moments.
    pg.wait_for(CAUGHT_UP, "0")
    pg.psql("standby_primary.sql", note="on the primary")

-- on the primary: 100 orders
$ psql -X -a -f standby_setup.sql
CREATE TABLE orders (id bigserial PRIMARY KEY, total numeric(12,2) NOT NULL);
CREATE TABLE
INSERT INTO orders (total) SELECT 100 + g FROM generate_series(1, 100) AS g;
INSERT 0 100

$ pg_basebackup -D standby -R -X stream -c fast -C -S standby1

-- in standby, the names that start with standby:
   standby.signal

-- what -R added to the standby's settings
$ cat standby/postgresql.auto.conf
# Do not edit this file manually!
# It will be overwritten by the ALTER SYSTEM command.
primary_conninfo = 'user=postgres passfile=''pgpass'' channel_binding=prefer connect_timeout=10 host=127.0.0.1 port=5432 sslmode=prefer sslnegotiation=postgres sslcompression=0 sslcertmode=allow sslsni=1 ssl_min_protocol_version=TLSv1.2 gssencmode=disable krbsrvname=postgres gssdelegation=0 target_session_attrs=any load_balance_hosts=disable dbname=postgres'
primary_slot_name = 'standby1'

-- on the standby
$ psql -X -a -f standby_check.sql
SELECT pg_is_in_recovery();
 pg_is_in_recovery 
-------------------
 t
(1 row)

SELECT count(*) AS orders, sum(total) AS total FROM orders;
 orders |  total   
--------+----------
    100 | 15050.00
(1 row)

-- a write on the standby
$ psql -X -a -f standby_write.sql
\set ON_ERROR_STOP on
INSERT INTO orders (total) VALUES (1);
psql:standby_write.sql:2: ERROR:  cannot execute INSERT in a read-only transaction

-- on the primary
$ psql -X -a -f standby_primary.sql
SELECT application_name, state, sync_state,
       round(pg_wal_lsn_diff(sent_lsn, replay_lsn) / 1024) AS replay_behind_kb
  FROM pg_stat_replication;
 application_name |   state   | sync_state | replay_behind_kb 
------------------+-----------+------------+------------------
 standby1         | streaming | async      |                0
(1 row)

SELECT slot_name, slot_type, active, wal_status
  FROM pg_replication_slots;
 slot_name | slot_type | active | wal_status 
-----------+-----------+--------+------------
 standby1  | physical  | t      | reserved
(1 row)

standby_setup.sql

CREATE TABLE orders (id bigserial PRIMARY KEY, total numeric(12,2) NOT NULL);
INSERT INTO orders (total) SELECT 100 + g FROM generate_series(1, 100) AS g;

standby_check.sql

SELECT pg_is_in_recovery();
SELECT count(*) AS orders, sum(total) AS total FROM orders;

standby_write.sql

\set ON_ERROR_STOP on
INSERT INTO orders (total) VALUES (1);

standby_primary.sql

SELECT application_name, state, sync_state,
       round(pg_wal_lsn_diff(sent_lsn, replay_lsn) / 1024) AS replay_behind_kb
  FROM pg_stat_replication;
SELECT slot_name, slot_type, active, wal_status
  FROM pg_replication_slots;

pg_session.py

"""Throwaway PostgreSQL 18.6 clusters for the examples of this lesson, including standbys and a second cluster.

Every session creates its clusters with initdb in new folders inside the examples folder, starts them, runs psql
scripts and server programs on them, prints what they printed, and stops and deletes them again, also when something
fails. The servers are the pinned PostgreSQL 18.6 from the mise installations next to the Python that runs this file.
Each listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system reports as
unused, with no Unix socket, and asks for a password (scram-sha-256) made up for the run and never printed. A server
that another server must reach (a standby, a subscriber) is given a password file of its own, mode 0600, so no
connection string in the output carries a password.
learn:verify's run sandbox cannot read the home folder, where mise keeps PostgreSQL, so these sessions are recorded
with LEARN_VERIFY_NETWORK=allow; they read only this folder and the toolchain and connect to nothing but 127.0.0.1.

The output is exactly what the programs printed, with five substitutions so that a record never shows the computer it
ran on and is the same on every run: each run folder is shown as ".", the folder of PostgreSQL's programs is left out
of their paths, the operating-system user's name is shown as "learner", the dots pg_ctl prints while it waits (one
about every tenth of a second) as three, and the port the operating system picked for each cluster as a fixed number
(5432 for the first cluster, 5433 for its standby, 5434 for a second cluster), the numbers a reader would see with
PostgreSQL's own defaults. A session may name more patterns (mask=) for values that change on every run, such as a
log sequence number, a timing or a process id; the session's own file says which, and its record shows fixed text in
their place. A session stops with status 1 when a command fails that should have worked, or reports a different
number of errors than the session expects, so a run that went wrong is never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

PG_VERSION = "18.6"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
BIN = INSTALLS / "postgres" / PG_VERSION / "bin"
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
# Every cluster of the run, so that each one's output can be cleaned of every run folder and port.
BASES = []
PORTS = {}
# The passwords of this run: no output of a session ever shows one, whatever a program decides to write.
PASSWORDS = []


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway cluster: initdb, settings, start; stopped and deleted when the with block ends."""

    def __init__(self, settings=None, initdb=(), name="data", shown_port=5432):
        self.settings = dict(settings or {})
        self.initdb_args = list(initdb)
        self.name = name
        self.shown_port = shown_port
        self.running = []

    def __enter__(self):
        if not (BIN / "postgres").exists():
            sys.exit(f"PostgreSQL {PG_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="pg", dir=".")).resolve()
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        (self.base / "pwfile").write_text(self.password + "\n")
        BASES.append(self.base)
        PORTS[self.port] = self.shown_port
        PASSWORDS.append(self.password)
        self.env = {**os.environ, "PGHOST": "127.0.0.1", "PGPORT": str(self.port), "PGUSER": "postgres",
                    "PGPASSWORD": self.password, "PGDATABASE": "postgres", "PGCONNECT_TIMEOUT": "10"}
        for key in ("PGSERVICE", "PGSERVICEFILE", "PGOPTIONS", "PSQLRC"):
            self.env.pop(key, None)
        # The server itself must not inherit this cluster's password: a replication connection to ANOTHER cluster
        # would use it instead of the password file its connection string names.
        self.server_env = {k: v for k, v in self.env.items() if k != "PGPASSWORD"}
        for script in Path(".").glob("*.sql"):
            shutil.copy(script, self.base / script.name)
        for data in Path(".").glob("*.csv"):
            shutil.copy(data, self.base / data.name)
        return self

    def __exit__(self, *exc):
        for folder, _port in reversed(self.running):
            self._call(["pg_ctl", "-D", folder, "-m", "fast", "-w", "-t", "60", "stop"])
        self.running = []
        shutil.rmtree(self.base, ignore_errors=True)

    # ---- the cluster ---------------------------------------------------------------------------------------------
    def initdb(self, show=False):
        args = ["initdb", "-D", self.name, "-U", "postgres", "--auth=scram-sha-256", "--pwfile=pwfile",
                *self.initdb_args]
        if not show:
            args.append("--no-instructions")
        self._expect(args, show=show)
        with open(self.base / self.name / "postgresql.conf", "a") as conf:
            conf.write("\n# Added by the session: 127.0.0.1 only, no Unix socket.\n")
            conf.write(f"listen_addresses = '127.0.0.1'\nport = {self.port}\nunix_socket_directories = ''\n")
            for key, value in self.settings.items():
                conf.write(f"{key} = {value}\n")

    def start(self, show=False, folder=None, port=None, note=None):
        folder = folder or self.name
        if note:
            print(f"-- {note}")
        code, out = self._call(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"],
                               env=self.server_env)
        if show:
            _show(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"], out)
        if code != 0:
            if not show:
                print(out)
            log = self.base / f"{folder}.log"
            if log.exists():
                print(self.clean("\n".join(log.read_text().splitlines()[-25:])))
            sys.exit(f"pg_ctl start of {folder} ended with status {code}")
        self.running.append((folder, port or self.port))

    def stop(self, folder=None, show=False, mode="fast"):
        """Stops one cluster of this session (the standby, or the primary of a failover drill)."""
        folder = folder or self.name
        self._expect(["pg_ctl", "-D", folder, "-m", mode, "-w", "-t", "60", "stop"], show=show)
        self.running = [r for r in self.running if r[0] != folder]

    def up(self):
        """initdb and start, quietly."""
        self.initdb()
        self.start()
        return self

    def passfile(self):
        """A libpq password file for this cluster, so a connection string never carries the password."""
        path = self.base / "pgpass"
        if not path.exists():
            path.write_text(f"127.0.0.1:*:*:postgres:{self.password}\n")
            path.chmod(0o600)
        return path

    def conninfo(self, db="postgres", user="postgres", application_name=None, port=None, hosts=None):
        """A connection string another server can use to reach this one (password in a file, never in the text).

        port names another cluster of this run (a promoted standby); hosts is a list of ports for a multi-host
        connection string (host=127.0.0.1,127.0.0.1 port=a,b).
        """
        if hosts:
            where = f"host={','.join('127.0.0.1' for _ in hosts)} port={','.join(str(p) for p in hosts)}"
        else:
            where = f"host=127.0.0.1 port={port or self.port}"
        parts = [f"{where} user={user} dbname={db} passfile={self.passfile()}"]
        if application_name:
            parts.append(f"application_name={application_name}")
        return " ".join(parts)

    def standby(self, folder="standby", slot=None, shown_port=5433, wait=True, extra=(), show=False, settings=None):
        """A streaming standby of this cluster, made with pg_basebackup -R, started on a port of its own.

        slot names a physical replication slot pg_basebackup creates and the standby then uses (-C -S). extra adds
        arguments to pg_basebackup; settings are lines appended to the standby's postgresql.conf. Returns its port.
        """
        args = ["pg_basebackup", "-D", folder, "-R", "-X", "stream", "-c", "fast"]
        if slot:
            args += ["-C", "-S", slot]
        args += list(extra)
        # PGPASSFILE (and no PGPASSWORD) so that the primary_conninfo pg_basebackup -R writes names the password
        # file: the standby then authenticates without a password in any file the record shows.
        self._expect(args, show=show, env={**self.server_env, "PGPASSFILE": str(self.passfile())})
        port = free_port()
        PORTS[port] = shown_port
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {port}\n")
            for key, value in dict(settings or {}).items():
                conf.write(f"{key} = {value}\n")
        self.start(folder=folder, port=port)
        if wait:
            for _ in range(300):
                if self.query("SELECT count(*) FROM pg_stat_replication WHERE state = 'streaming'") != "0":
                    return port
                time.sleep(0.1)
            sys.exit("the standby did not start streaming")
        return port

    def promote(self, folder="standby", show=True, port=None):
        """Promotes a standby to a primary with pg_ctl promote and waits until it accepts writes."""
        self._expect(["pg_ctl", "-D", folder, "-w", "-t", "60", "promote"], show=show)
        port = port or dict((f, p) for f, p in self.running).get(folder)
        for _ in range(300):
            if self.query("SELECT pg_is_in_recovery()", port=port) == "f":
                return port
            time.sleep(0.1)
        sys.exit(f"{folder} did not leave recovery")

    def rewind(self, folder, source=None, port=None, show=True, extra=(), target_port=None):
        """pg_rewind: resynchronises the stopped cluster in `folder` with a running cluster (a Server or a port).

        It runs without this cluster's password in the environment and with PGPASSFILE set, so a -R run writes a
        primary_conninfo that names the password file instead of a password. pg_rewind copies the source cluster's
        postgresql.conf, which on one machine carries the source's own port, so the target's port is written back
        afterwards (one machine runs both clusters here; on two machines both would listen on 5432).
        """
        who = source or self
        args = ["pg_rewind", "--target-pgdata", folder, f"--source-server={who.conninfo(port=port)}", *extra]
        out = self._expect(args, show=show, env={**self.server_env, "PGPASSFILE": str(who.passfile())})
        mine = target_port or dict((f, p) for f, p in self.running).get(folder) or self.port
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {mine}\n")
        return out

    # ---- running things ------------------------------------------------------------------------------------------
    def _call(self, args, env=None):
        args = [str(a) for a in args]
        done = subprocess.run([str(BIN / args[0]), *args[1:]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              text=True, cwd=self.base, env=env or self.env)
        return done.returncode, self.clean(done.stdout)

    def _expect(self, args, show=False, status=0, mask=(), env=None):
        code, out = self._call(args, env=env)
        out = _masked(out, mask)
        if show:
            _show(args, out)
        if code != status:
            if not show:
                print(out)
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """Every run folder as ".", the user as learner, pg_ctl's dots as three, each port as a fixed number.

        A password the run made up is never shown, even when a program (pg_rewind -R, for one) writes it into a file
        the session prints.
        """
        text = text.replace(str(BIN) + "/", "")
        for base in BASES:
            text = text.replace(str(base) + "/", "").replace(str(base), ".")
        for real, shown in PORTS.items():
            text = re.sub(rf"\b{real}\b", str(shown), text)
        for password in PASSWORDS:
            text = text.replace(password, "(the password of this run, not shown)")
        text = re.sub(r"(waiting for server to (?:start|shut down|promote))\.+", r"\1...", text)
        return text.replace(f'"{OS_USER}"', '"learner"')

    def wait_for(self, sql, value, port=None):
        """Waits (unprinted) until a query returns the value, for example after a reload."""
        for _ in range(600):
            if self.query(sql, port=port) == value:
                return
            time.sleep(0.1)
        sys.exit(f"{sql} never returned {value}")

    def show_file(self, path, note=None):
        """Prints a text file of the cluster, as `cat` would."""
        if note:
            print(f"-- {note}")
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def log_lines(self, path, match, count=1, note=None, mask=()):
        """Waits for and prints the server-log lines that contain `match` (the server log of one cluster)."""
        log = self.base / path
        for _ in range(600):
            found = [ln for ln in log.read_text().splitlines() if match in ln] if log.exists() else []
            if len(found) >= count:
                break
            time.sleep(0.1)
        else:
            sys.exit(f"{path} never showed {count} line(s) with {match!r}")
        print(f"-- {note or f'{path}, the lines about {match}'}:")
        for line in found[:count]:
            print(_masked(self.clean(line), mask))
        print()
        return found[:count]

    def query(self, sql, port=None):
        """One value from one query, unprinted (the session uses it to wait or to decide)."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        code, out = self._call(["psql", "-X", "-At", "-c", sql], env=env)
        if code != 0:
            sys.exit(f"the query failed: {out}")
        return out.strip()

    def silent(self, sql, port=None):
        """Runs a statement without printing it (used only to give a role the run's password)."""
        self.query(sql, port=port)

    def psql(self, script, errors=0, status=None, user=None, db=None, port=None, flags=("-a",), note=None, mask=(),
             variables=()):
        """Runs a psql script with -X (no ~/.psqlrc) and the given flags, and prints the command and its output.
        `errors` is how many ERROR lines the script must print; `status` the exit status it must end with;
        `variables` are (name, value) pairs the script reads as :'name' (a connection string, for example)."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        if user:
            env["PGUSER"] = user
        if db:
            env["PGDATABASE"] = db
        passed = []
        for name, value in variables:
            passed += ["-v", f"{name}={value}"]
        args = ["psql", "-X", *flags, *passed, "-f", script]
        code, out = self._call(args, env=env)
        out = _masked(out, mask)
        shown = ["psql", "-X", *flags, "-f", script]
        if user:
            shown[1:1] = ["-U", user]
        if db:
            shown[1:1] = ["-d", db]
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if "ERROR:" in line or ": error:" in line)
        expected_status = 0 if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")

    def sessions(self, count):
        """Opens `count` idle psql connections (each one a client backend) until the with block ends."""
        return _Idle(self, count)

    def shell(self, *args, status=0, mask=(), note=None, timeout=None, stdin=None):
        """Runs one of PostgreSQL's programs and prints the command and its output."""
        if note:
            print(f"-- {note}")
        if timeout is None:
            return self._expect(list(args), show=True, status=status, mask=mask)
        return self._timed(list(args), status=status, mask=mask, timeout=timeout, stdin=stdin)

    def _timed(self, args, status, mask, timeout, stdin=None):
        """Runs a program that would not stop by itself (pg_recvlogical) and prints what it printed until timeout."""
        args = [str(a) for a in args]
        proc = subprocess.Popen([str(BIN / args[0]), *args[1:]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                stdin=subprocess.PIPE if stdin is not None else None, text=True, cwd=self.base,
                                env=self.env)
        try:
            out, _ = proc.communicate(input=stdin, timeout=timeout)
        except subprocess.TimeoutExpired:
            proc.terminate()
            out, _ = proc.communicate()
        out = _masked(self.clean(out), mask)
        _show(args, out)
        return out

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the cluster (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"-- in {folder}" + (f", the names that start with {prefix}" if prefix else "") + ":")
        print(textwrap.fill("  ".join(names), width=78, initial_indent="   ", subsequent_indent="   ",
                            break_on_hyphens=False))
        print()
        return names


def _masked(text, mask):
    """Replaces what differs on every run (an LSN, a timing, a process id) with fixed text (see the docstring)."""
    for pattern, replacement in mask:
        text = re.sub(pattern, replacement, text, flags=re.M)
    return text


def _arg(a):
    """A command-line argument as a reader would type it: double quotes around text with single quotes in it."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()


class _Idle:
    """Idle psql connections, closed again when the with block ends."""

    def __init__(self, server, count):
        self.server, self.count, self.procs = server, count, []

    def __enter__(self):
        before = int(self.server.query("SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"))
        for _ in range(self.count):
            self.procs.append(subprocess.Popen([str(BIN / "psql"), "-X", "-q"], stdin=subprocess.PIPE,
                                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                               env=self.server.env, cwd=self.server.base))
        sql = "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"
        for _ in range(300):
            # the query's own connection is one more client backend
            if int(self.server.query(sql)) == before + self.count:
                return self
            time.sleep(0.1)
        sys.exit("the idle connections did not open")

    def __exit__(self, *exc):
        for p in self.procs:
            p.stdin.close()
            p.wait(timeout=30)