Logical replication and change data capture

replicate_session.py

"""Logical replication between two PostgreSQL 18.6 clusters: a publisher with wal_level = logical, and a subscriber.

publish.sql creates the orders table on the publisher, fills it and publishes it. subscribe.sql creates the same table on
the subscriber, which receives no schema from the publisher, and subscribes to the publication, which copies the three
rows first. changes.sql then updates, deletes and inserts a row on the publisher and adds a column there;
publisher_side.sql and subscriber_side.sql read the rows, the sequence behind id and the replication slot on each side.

The subscription's connection string reaches psql as the variable pub, so the record shows CONNECTION :'pub' and never a
host, a port or a password; the publisher is shown on port 5432 and the subscriber on port 5434. The session waits until
the subscriber has applied every change and the initial copy's temporary slot is gone, so nothing in the record differs
between runs and nothing is masked.
"""

from pg_session import Server

COMMON = {"autovacuum": "off", "jit": "off", "timezone": "'UTC'", "log_line_prefix": "''",
          "max_parallel_workers_per_gather": 0}
# wal_level = logical is needed on the publisher only, and it takes a server restart to change.
PUBLISHER = {**COMMON, "wal_level": "logical"}
# --no-sync: initdb does not flush the new clusters to disk, which only saves time; they are deleted after the run.
FAST = ("--no-sync",)

with Server(settings=PUBLISHER, initdb=FAST) as pub, Server(settings=COMMON, initdb=FAST, shown_port=5434) as sub:
    pub.up()
    sub.up()
    pub.psql("publish.sql", note="on the publisher, port 5432")
    sub.psql("subscribe.sql", variables=[("pub", pub.conninfo(db="postgres"))], note="on the subscriber, port 5434")
    sub.wait_for("SELECT count(*) FROM orders", "3")
    pub.psql("changes.sql", note="on the publisher again")
    sub.wait_for("SELECT count(*) FROM orders WHERE region = 'east'", "1")
    sub.wait_for("SELECT count(*) FROM pg_subscription_rel WHERE srsubstate <> 'r'", "0")
    pub.wait_for("SELECT count(*) FROM pg_replication_slots", "1")
    pub.psql("publisher_side.sql", note="the publisher's side")
    sub.psql("subscriber_side.sql", note="the subscriber's side")

-- on the publisher, port 5432
$ psql -X -a -f publish.sql
-- The publisher must write WAL at the logical level.
SHOW wal_level;
 wal_level 
-----------
 logical
(1 row)

CREATE TABLE orders (
    id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    region text NOT NULL,
    net    numeric(12,2) NOT NULL,
    tax    numeric(12,2) GENERATED ALWAYS AS (round(net * 0.18, 2)) STORED
);
CREATE TABLE
INSERT INTO orders (region, net) VALUES ('south', 1000), ('north', 2500), ('west', 400);
INSERT 0 3
-- A publication names what to send. Creating it sends nothing yet.
CREATE PUBLICATION shop_pub FOR TABLE orders;
CREATE PUBLICATION

-- on the subscriber, port 5434
$ psql -X -a -f subscribe.sql
-- No schema comes from the publisher: the table is created here by hand, column for column.
CREATE TABLE orders (
    id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    region text NOT NULL,
    net    numeric(12,2) NOT NULL,
    tax    numeric(12,2) GENERATED ALWAYS AS (round(net * 0.18, 2)) STORED
);
CREATE TABLE
-- The connection string arrives as a psql variable, so this file holds no host and no password.
CREATE SUBSCRIPTION shop_sub CONNECTION :'pub' PUBLICATION shop_pub;
psql:subscribe.sql:9: NOTICE:  created replication slot "shop_sub" on publisher
CREATE SUBSCRIPTION

-- on the publisher again
$ psql -X -a -f changes.sql
-- Three row changes: each one is sent when its transaction commits.
UPDATE orders SET net = 1200 WHERE id = 1;
UPDATE 1
DELETE FROM orders WHERE id = 3;
DELETE 1
INSERT INTO orders (region, net) VALUES ('east', 800);
INSERT 0 1
-- A schema change stays on this side.
ALTER TABLE orders ADD COLUMN note text;
ALTER TABLE

-- the publisher's side
$ psql -X -a -f publisher_side.sql
SELECT * FROM orders ORDER BY id;
 id | region |   net   |  tax   | note 
----+--------+---------+--------+------
  1 | south  | 1200.00 | 216.00 | 
  2 | north  | 2500.00 | 450.00 | 
  4 | east   |  800.00 | 144.00 | 
(3 rows)

-- The sequence behind the identity column, and the slot the subscription created here.
SELECT last_value FROM pg_sequences WHERE sequencename = 'orders_id_seq';
 last_value 
------------
          4
(1 row)

SELECT slot_name, plugin, slot_type, active FROM pg_replication_slots;
 slot_name |  plugin  | slot_type | active 
-----------+----------+-----------+--------
 shop_sub  | pgoutput | logical   | t
(1 row)

-- the subscriber's side
$ psql -X -a -f subscriber_side.sql
SELECT * FROM orders ORDER BY id;
 id | region |   net   |  tax   
----+--------+---------+--------
  1 | south  | 1200.00 | 216.00
  2 | north  | 2500.00 | 450.00
  4 | east   |  800.00 | 144.00
(3 rows)

-- The subscriber's own sequence: null means it has never handed out a value.
SELECT last_value FROM pg_sequences WHERE sequencename = 'orders_id_seq';
 last_value 
------------
           
(1 row)

publish.sql

-- The publisher must write WAL at the logical level.
SHOW wal_level;
CREATE TABLE orders (
    id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    region text NOT NULL,
    net    numeric(12,2) NOT NULL,
    tax    numeric(12,2) GENERATED ALWAYS AS (round(net * 0.18, 2)) STORED
);
INSERT INTO orders (region, net) VALUES ('south', 1000), ('north', 2500), ('west', 400);
-- A publication names what to send. Creating it sends nothing yet.
CREATE PUBLICATION shop_pub FOR TABLE orders;

subscribe.sql

-- No schema comes from the publisher: the table is created here by hand, column for column.
CREATE TABLE orders (
    id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    region text NOT NULL,
    net    numeric(12,2) NOT NULL,
    tax    numeric(12,2) GENERATED ALWAYS AS (round(net * 0.18, 2)) STORED
);
-- The connection string arrives as a psql variable, so this file holds no host and no password.
CREATE SUBSCRIPTION shop_sub CONNECTION :'pub' PUBLICATION shop_pub;

changes.sql

-- Three row changes: each one is sent when its transaction commits.
UPDATE orders SET net = 1200 WHERE id = 1;
DELETE FROM orders WHERE id = 3;
INSERT INTO orders (region, net) VALUES ('east', 800);
-- A schema change stays on this side.
ALTER TABLE orders ADD COLUMN note text;

publisher_side.sql

SELECT * FROM orders ORDER BY id;
-- The sequence behind the identity column, and the slot the subscription created here.
SELECT last_value FROM pg_sequences WHERE sequencename = 'orders_id_seq';
SELECT slot_name, plugin, slot_type, active FROM pg_replication_slots;

subscriber_side.sql

SELECT * FROM orders ORDER BY id;
-- The subscriber's own sequence: null means it has never handed out a value.
SELECT last_value FROM pg_sequences WHERE sequencename = 'orders_id_seq';

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)