Freezing and transaction ID wraparound

freeze_session.py

"""Freezing and transaction ID wraparound on PostgreSQL 18.6: freeze.sql, what VACUUM (FREEZE) changes in a row's
header, in the visibility map and in the ages of the table and the database."""

from pg_session import Server

# Autovacuum off: nothing but the session takes transaction IDs, so every ID and age is the same on every run.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'"}
# VACUUM VERBOSE ends with the read and write rates, the buffer and WAL counts and the CPU and elapsed times of the
# run, which differ every time: the record shows one fixed line in their place. Every other line is as printed.
RUN_FIGURES = (r"^avg read rate: .*\nbuffer usage: .*\nWAL usage: .*\nsystem usage: .*$",
               "(rates, buffer and WAL counts and times of this run: they differ on every run)")

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("spend_xids.sql", flags=("-q",))
    pg.psql("freeze.sql", mask=[RUN_FIGURES])

$ psql -X -q -f spend_xids.sql

$ psql -X -a -f freeze.sql
-- 10,000 accounts written by one transaction and read once, then 5,000 transactions that write nothing.
CREATE EXTENSION pageinspect;
CREATE EXTENSION
CREATE EXTENSION pg_visibility;
CREATE EXTENSION
CREATE TABLE accounts (id int PRIMARY KEY, balance int NOT NULL);
CREATE TABLE
INSERT INTO accounts SELECT n, 1000 FROM generate_series(1, 10000) AS n;
INSERT 0 10000
SELECT count(*) FROM accounts;
 count 
-------
 10000
(1 row)

CALL spend_xids(5000);
CALL
-- How old is the table, and what does a row's header say?
SELECT relfrozenxid, age(relfrozenxid) FROM pg_class WHERE relname = 'accounts';
 relfrozenxid | age  
--------------+------
          756 | 5002
(1 row)

SELECT lp, t_xmin, f.combined_flags FROM heap_page_items(get_raw_page('accounts', 0)),
  LATERAL heap_tuple_infomask_flags(t_infomask, t_infomask2) AS f WHERE lp <= 2;
 lp | t_xmin | combined_flags 
----+--------+----------------
  1 |    757 | {}
  2 |    757 | {}
(2 rows)

SELECT * FROM pg_visibility_map_summary('accounts');
 all_visible | all_frozen 
-------------+------------
           0 |          0
(1 row)

VACUUM (FREEZE, VERBOSE) accounts;
psql:freeze.sql:13: INFO:  aggressively vacuuming "postgres.public.accounts"
psql:freeze.sql:13: INFO:  finished vacuuming "postgres.public.accounts": index scans: 0
pages: 0 removed, 45 remain, 45 scanned (100.00% of total), 0 eagerly scanned
tuples: 0 removed, 10000 remain, 0 are dead but not yet removable
removable cutoff: 5758, which was 0 XIDs old when operation ended
new relfrozenxid: 5758, which is 5002 XIDs ahead of previous value
frozen: 45 pages from table (100.00% of total) had 10000 tuples frozen
visibility map: 45 pages set all-visible, 45 pages set all-frozen (0 were all-visible)
index scan not needed: 0 pages from table (0.00% of total) had 0 dead item identifiers removed
(rates, buffer and WAL counts and times of this run: they differ on every run)
VACUUM
SELECT relfrozenxid, age(relfrozenxid) FROM pg_class WHERE relname = 'accounts';
 relfrozenxid | age 
--------------+-----
         5758 |   0
(1 row)

SELECT lp, t_xmin, f.combined_flags FROM heap_page_items(get_raw_page('accounts', 0)),
  LATERAL heap_tuple_infomask_flags(t_infomask, t_infomask2) AS f WHERE lp <= 2;
 lp | t_xmin |   combined_flags   
----+--------+--------------------
  1 |    757 | {HEAP_XMIN_FROZEN}
  2 |    757 | {HEAP_XMIN_FROZEN}
(2 rows)

SELECT * FROM pg_visibility_map_summary('accounts');
 all_visible | all_frozen 
-------------+------------
          45 |         45
(1 row)

-- One change clears the bits of the pages it touches.
UPDATE accounts SET balance = 900 WHERE id = 1;
UPDATE 1
SELECT * FROM pg_visibility_map_summary('accounts');
 all_visible | all_frozen 
-------------+------------
          43 |         43
(1 row)

-- A database is as old as its oldest table: freeze them all.
SELECT datname, age(datfrozenxid) FROM pg_database ORDER BY datname;
  datname  | age  
-----------+------
 postgres  | 5014
 template0 | 5014
 template1 | 5014
(3 rows)

VACUUM (FREEZE);
VACUUM
SELECT datname, age(datfrozenxid) FROM pg_database ORDER BY datname;
  datname  | age  
-----------+------
 postgres  |    0
 template0 | 5014
 template1 | 5014
(3 rows)

freeze.sql

-- 10,000 accounts written by one transaction and read once, then 5,000 transactions that write nothing.
CREATE EXTENSION pageinspect;
CREATE EXTENSION pg_visibility;
CREATE TABLE accounts (id int PRIMARY KEY, balance int NOT NULL);
INSERT INTO accounts SELECT n, 1000 FROM generate_series(1, 10000) AS n;
SELECT count(*) FROM accounts;
CALL spend_xids(5000);
-- How old is the table, and what does a row's header say?
SELECT relfrozenxid, age(relfrozenxid) FROM pg_class WHERE relname = 'accounts';
SELECT lp, t_xmin, f.combined_flags FROM heap_page_items(get_raw_page('accounts', 0)),
  LATERAL heap_tuple_infomask_flags(t_infomask, t_infomask2) AS f WHERE lp <= 2;
SELECT * FROM pg_visibility_map_summary('accounts');
VACUUM (FREEZE, VERBOSE) accounts;
SELECT relfrozenxid, age(relfrozenxid) FROM pg_class WHERE relname = 'accounts';
SELECT lp, t_xmin, f.combined_flags FROM heap_page_items(get_raw_page('accounts', 0)),
  LATERAL heap_tuple_infomask_flags(t_infomask, t_infomask2) AS f WHERE lp <= 2;
SELECT * FROM pg_visibility_map_summary('accounts');
-- One change clears the bits of the pages it touches.
UPDATE accounts SET balance = 900 WHERE id = 1;
SELECT * FROM pg_visibility_map_summary('accounts');
-- A database is as old as its oldest table: freeze them all.
SELECT datname, age(datfrozenxid) FROM pg_database ORDER BY datname;
VACUUM (FREEZE);
SELECT datname, age(datfrozenxid) FROM pg_database ORDER BY datname;

spend_xids.sql

-- spend_xids(n): n transactions in a row, each of which takes a transaction ID and commits without writing a row.
-- It ages every table of the cluster by n transactions in a second or two.
CREATE PROCEDURE spend_xids(n int) LANGUAGE plpgsql AS $$
BEGIN
  FOR i IN 1..n LOOP
    PERFORM pg_current_xact_id();
    COMMIT;
  END LOOP;
END $$;

pg_session.py

"""Throwaway PostgreSQL 18.6 servers for the examples of this lesson.

Every session creates a cluster of its own with initdb in a new folder inside the examples folder, starts it, runs
psql scripts and server programs on it, prints what they printed, and stops and deletes it again, also when something
fails. The server is the pinned PostgreSQL 18.6 from the mise installations next to the Python that runs this file.
It 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 it asks for a password (scram-sha-256) made up for the run and never printed.
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 four substitutions so that a record never shows the computer
it ran on and is the same on every run: the run's 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", and the dots pg_ctl prints while it waits
(one about every tenth of a second) as three. A session may name more patterns (mask=) for numbers that change on
every run, such as the rates and times ANALYZE VERBOSE prints or the random key of pg_dump's \\restrict line; the
session's own file says which, and its record shows a fixed word in their place. The sessions never print the port.
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


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"):
        self.settings = dict(settings or {})
        self.initdb_args = list(initdb)
        self.name = name
        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")
        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)
        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):
        folder = folder or self.name
        self._expect(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"], show=show)
        self.running.append((folder, port or self.port))

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

    def standby(self, folder="standby"):
        """A streaming standby of this server, made with pg_basebackup -R, started on a port of its own."""
        self._expect(["pg_basebackup", "-D", folder, "-R", "-X", "stream", "-c", "fast"])
        port = free_port()
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {port}\n")
        self.start(folder=folder, port=port)
        for _ in range(300):
            if self.query("SELECT count(*) FROM pg_stat_replication WHERE state = 'streaming'") == "1":
                return port
            time.sleep(0.1)
        sys.exit("the standby did not start streaming")

    # ---- 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=()):
        code, out = self._call(args)
        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):
        """The run's folder as ".", the user as learner, pg_ctl's waiting dots as three (see the docstring)."""
        text = text.replace(str(BIN) + "/", "").replace(str(self.base) + "/", "").replace(str(self.base), ".")
        text = re.sub(r"(waiting for server to (?:start|shut down))\.+", r"\1...", text)
        return text.replace(f'"{OS_USER}"', '"learner"')

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

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

    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):
        """Runs a statement without printing it (used only to give a role the run's password)."""
        self.query(sql)

    def psql(self, script, errors=0, status=None, user=None, db=None, port=None, flags=("-a",), note=None, mask=()):
        """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."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        if user:
            env["PGUSER"] = user
        if db:
            env["PGDATABASE"] = db
        args = ["psql", "-X", *flags, "-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=()):
        """Runs one of PostgreSQL's programs and prints the command and its output."""
        return self._expect(list(args), show=True, status=status, mask=mask)

    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 (a rate, a time, a random key) 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)

ages_session.py

"""Freezing and transaction ID wraparound on PostgreSQL 18.6: ages.sql, the ages of three tables and of the databases,
measured against autovacuum_freeze_max_age and against the points where warnings start and new IDs are refused."""

from pg_session import Server

# Autovacuum off: nothing but the session takes transaction IDs, so every ID and age is the same on every run.
SETTINGS = {"autovacuum": "off", "max_parallel_workers_per_gather": 0, "jit": "off", "timezone": "'UTC'"}

with Server(settings=SETTINGS) as pg:
    pg.up()
    pg.psql("spend_xids.sql", flags=("-q",))
    pg.psql("ages.sql")

$ psql -X -q -f spend_xids.sql

$ psql -X -a -f ages.sql
-- Three tables: audit_log is never vacuumed, orders is frozen 20,000 transactions later, carts 30,000 later.
CREATE TABLE audit_log (id int PRIMARY KEY, note text NOT NULL);
CREATE TABLE
CREATE TABLE orders (id int PRIMARY KEY, total int NOT NULL);
CREATE TABLE
CREATE TABLE carts (id int PRIMARY KEY, items int NOT NULL);
CREATE TABLE
INSERT INTO audit_log SELECT n, 'login ' || n FROM generate_series(1, 1000) AS n;
INSERT 0 1000
INSERT INTO orders SELECT n, n % 900 FROM generate_series(1, 1000) AS n;
INSERT 0 1000
INSERT INTO carts SELECT n, 1 FROM generate_series(1, 1000) AS n;
INSERT 0 1000
CALL spend_xids(20000);
CALL
VACUUM (FREEZE) orders;
VACUUM
CALL spend_xids(10000);
CALL
VACUUM (FREEZE) carts;
VACUUM
-- The manual's age query, oldest first, with the transactions left before autovacuum must vacuum each table.
SELECT c.oid::regclass AS table_name,
       greatest(age(c.relfrozenxid), age(t.relfrozenxid)) AS xid_age,
       current_setting('autovacuum_freeze_max_age')::int
         - greatest(age(c.relfrozenxid), age(t.relfrozenxid)) AS left_before_forced_vacuum
FROM pg_class c LEFT JOIN pg_class t ON t.oid = c.reltoastrelid
WHERE c.relkind IN ('r', 'm') AND c.relnamespace = 'public'::regnamespace
ORDER BY xid_age DESC;
 table_name | xid_age | left_before_forced_vacuum 
------------+---------+---------------------------
 audit_log  |   30006 |                 199969994
 orders     |   10000 |                 199990000
 carts      |       0 |                 200000000
(3 rows)

-- A database is as old as its oldest table, catalogs included: the distance to the warnings and to the stop.
SELECT datname, age(datfrozenxid) AS xid_age,
       2147483647 - 40000000 - age(datfrozenxid) AS left_before_warnings,
       2147483647 - 3000000 - age(datfrozenxid) AS left_before_stop
FROM pg_database ORDER BY datname;
  datname  | xid_age | left_before_warnings | left_before_stop 
-----------+---------+----------------------+------------------
 postgres  |   30015 |           2107453632 |       2144453632
 template0 |   30015 |           2107453632 |       2144453632
 template1 |   30015 |           2107453632 |       2144453632
(3 rows)

ages.sql

-- Three tables: audit_log is never vacuumed, orders is frozen 20,000 transactions later, carts 30,000 later.
CREATE TABLE audit_log (id int PRIMARY KEY, note text NOT NULL);
CREATE TABLE orders (id int PRIMARY KEY, total int NOT NULL);
CREATE TABLE carts (id int PRIMARY KEY, items int NOT NULL);
INSERT INTO audit_log SELECT n, 'login ' || n FROM generate_series(1, 1000) AS n;
INSERT INTO orders SELECT n, n % 900 FROM generate_series(1, 1000) AS n;
INSERT INTO carts SELECT n, 1 FROM generate_series(1, 1000) AS n;
CALL spend_xids(20000);
VACUUM (FREEZE) orders;
CALL spend_xids(10000);
VACUUM (FREEZE) carts;
-- The manual's age query, oldest first, with the transactions left before autovacuum must vacuum each table.
SELECT c.oid::regclass AS table_name,
       greatest(age(c.relfrozenxid), age(t.relfrozenxid)) AS xid_age,
       current_setting('autovacuum_freeze_max_age')::int
         - greatest(age(c.relfrozenxid), age(t.relfrozenxid)) AS left_before_forced_vacuum
FROM pg_class c LEFT JOIN pg_class t ON t.oid = c.reltoastrelid
WHERE c.relkind IN ('r', 'm') AND c.relnamespace = 'public'::regnamespace
ORDER BY xid_age DESC;
-- A database is as old as its oldest table, catalogs included: the distance to the warnings and to the stop.
SELECT datname, age(datfrozenxid) AS xid_age,
       2147483647 - 40000000 - age(datfrozenxid) AS left_before_warnings,
       2147483647 - 3000000 - age(datfrozenxid) AS left_before_stop
FROM pg_database ORDER BY datname;