9  Replicating PostgreSQL

Read replicas and logical replication between RDS for PostgreSQL instances.

9.1 Before you start

oblako up rds
export AWS_PROFILE=oblako

uv add boto3 "psycopg[binary]"

The examples use a table of shop orders that the listings create. Because each listing creates or uses RDS instances by name, run the listings in order; the last section removes everything they create.

9.2 Physical and logical replication

PostgreSQL writes every change to its write-ahead log (WAL) before it changes the data files, and both kinds of replication in this chapter start from that log.

Physical replication sends the WAL itself to a standby server, which replays it into an exact copy of the primary’s data files: every database, table, index and role, at the same PostgreSQL major version. A standby that serves read-only queries while it replays is a hot standby, and it can be promoted to a primary of its own. An RDS read replica is a physical standby fed by streaming replication. The stream is asynchronous, which means the primary commits without waiting for the replica, so the replica can lag behind.

Logical replication decodes the WAL into row changes for the tables in a publication, the set of tables a server makes available to others. A subscription on another server receives those changes and applies them to its own tables. The subscriber is an ordinary, writable database that can hold other tables and indexes and can run a newer major version. That makes logical replication the usual way to upgrade with little downtime or to feed an analytics database. It copies rows only, so the tables must exist on both sides, and CREATE and ALTER TABLE statements are made on each side separately.

On RDS you create a read replica with a control-plane call, a request to the RDS API, whereas logical replication is SQL you run yourself once a parameter group sets rds.logical_replication.

What oblako runs

Each PostgreSQL instance you create through the RDS API on oblako is its own PostgreSQL 16 container with its own data, built from the same image as Chapter 8. The RDS API on port 8014 forwards every call to moto, which keeps the metadata, and then starts, restarts, promotes and removes the containers to match. An instance’s endpoint is <id>.<region>.rds.localhost on a port of its own. Your operating system resolves any name under localhost to your own machine. The containers share a Docker network on which the same name reaches the same instance, so one endpoint works both from your code and between instances (Figure 9.1). Aurora clusters keep the shared database of Chapter 8.

Figure 9.1: Replication on oblako. The RDS API starts one PostgreSQL container per instance, and each is reached at its own endpoint. The replica streams the primary’s WAL; the subscriber receives decoded changes.

9.3 The primary

The primary is a standalone instance, so it needs no cluster. The examples connect to instances by name and look up the endpoint each time; save that lookup as db.py:

import boto3
import psycopg

rds = boto3.client("rds")
USER, PASSWORD = "shop", "shop-local-1"


def describe(instance_id):
    found = rds.describe_db_instances(
        DBInstanceIdentifier=instance_id
    )
    return found["DBInstances"][0]


def wait(instance_id):
    waiter = rds.get_waiter("db_instance_available")
    waiter.wait(DBInstanceIdentifier=instance_id)
    return describe(instance_id)


def endpoint(instance_id):
    return describe(instance_id)["Endpoint"]


def connect(instance_id):
    e = endpoint(instance_id)
    return psycopg.connect(
        host=e["Address"],
        port=e["Port"],
        dbname="shop",
        user=USER,
        password=PASSWORD,
        autocommit=True,
    )


def one(conn, sql, params=None):
    # the first column of the first row
    return conn.execute(sql, params).fetchone()[0]

Create the primary and wait for it:

from db import PASSWORD, USER, rds, wait

rds.create_db_instance(
    DBInstanceIdentifier="shop-primary",
    Engine="postgres",
    EngineVersion="16",
    DBInstanceClass="db.t4g.micro",
    AllocatedStorage=20,
    MasterUsername=USER,
    MasterUserPassword=PASSWORD,
    DBName="shop",
    BackupRetentionPeriod=1,
)
db = wait("shop-primary")
print(db["DBInstanceStatus"], db["Engine"], db["EngineVersion"])
print(db["Endpoint"]["Address"])

available postgres 16.15
shop-primary.us-east-1.rds.localhost

EngineVersion="16" asks for the newest PostgreSQL 16 release, and RDS reports the one it chose. BackupRetentionPeriod must be above zero on an instance you want to replicate, because RDS builds a read replica from the source’s backups and refuses to create one without them. The waiter polls DescribeDBInstances until the status is available, which takes several minutes on AWS and a few seconds on oblako.

Load a thousand orders:

from db import connect

with connect("shop-primary") as conn:
    conn.execute("""
        CREATE TABLE orders (
            id        bigint PRIMARY KEY,
            customer  text NOT NULL,
            total     numeric(10, 2) NOT NULL,
            placed_at timestamptz NOT NULL DEFAULT now()
        )""")
    conn.execute("""
        INSERT INTO orders (id, customer, total)
        SELECT g, 'customer-' || (g % 50),
               round((g % 97) * 1.25, 2)
        FROM generate_series(1, 1000) AS g""")
    totals = "SELECT count(*), sum(total) FROM orders"
    print(conn.execute(totals).fetchone())

(1000, Decimal('58781.25'))

9.4 Read replicas

One call creates a replica:

from db import rds, wait

rds.create_db_instance_read_replica(
    DBInstanceIdentifier="shop-replica",
    SourceDBInstanceIdentifier="shop-primary",
)
db = wait("shop-replica")
source = db["ReadReplicaSourceDBInstanceIdentifier"]
print(db["DBInstanceStatus"], "replica of", source)
print(db["Endpoint"]["Address"])

available replica of shop-primary
shop-replica.us-east-1.rds.localhost

The replica has its own endpoint. The primary lists it in pg_stat_replication, which has one row per connected standby, and the replica reports that it is a standby:

import psycopg

from db import connect, one

RECOVERY = "SELECT pg_is_in_recovery()"

with connect("shop-primary") as primary:
    standbys = primary.execute(
        "SELECT state, sync_state FROM pg_stat_replication"
    )
    print("primary sends to:", standbys.fetchall())

with connect("shop-replica") as replica:
    print("in recovery:", one(replica, RECOVERY))
    print("orders:", one(replica, "SELECT count(*) FROM orders"))
    try:
        replica.execute("DELETE FROM orders")
    except psycopg.errors.ReadOnlySqlTransaction as e:
        print("refused:", e)

primary sends to: [('streaming', 'async')]
in recovery: True
orders: 1000
refused: cannot execute DELETE in a read-only transaction

pg_is_in_recovery() returns true on a standby, because the server is still replaying WAL. All 1,000 orders are there, copied when the replica was created, and the replica refuses any write.

9.5 Read-your-writes consistency

Because the stream is asynchronous, a row you have just committed on the primary may not be on the replica yet. An application that writes and then reads from a replica, say to show a customer the order they placed, can get an empty result. The fix uses the position in the WAL. After a write, ask the primary for its current log sequence number (LSN), and before reading, wait until the replica has replayed past that position:

import time

from db import connect, one

with connect("shop-primary") as primary:
    primary.execute(
        "INSERT INTO orders (id, customer, total)"
        " VALUES (1001, 'customer-7', 42.00)"
    )
    lsn = one(primary, "SELECT pg_current_wal_lsn()")

caught_up = "SELECT pg_last_wal_replay_lsn() >= %s::pg_lsn"
with connect("shop-replica") as replica:
    polls = 0
    while not one(replica, caught_up, (lsn,)):
        time.sleep(0.01)
        polls += 1
    row = replica.execute(
        "SELECT customer, total FROM orders WHERE id = 1001"
    ).fetchone()
    print("replayed", lsn, "after", polls, "polls:", row)

replayed 0/303DA58 after 0 polls: ('customer-7', Decimal('42.00'))

On oblako the replica had caught up before the first poll; on AWS, expect a few milliseconds of polling within a region, and more under heavy writes.

9.6 Replication lag

Both servers report lag:

from db import connect, one

with connect("shop-primary") as primary:
    bytes_behind, replay_lag = primary.execute("""
        SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn),
               replay_lag
        FROM pg_stat_replication""").fetchone()
    print("bytes behind:", bytes_behind)
    print("replay lag:", replay_lag)

with connect("shop-replica") as replica:
    since = "SELECT now() - pg_last_xact_replay_timestamp()"
    print("last replayed commit:", one(replica, since), "ago")

bytes behind: 0
replay lag: 0:00:00.000472
last replayed commit: 0:00:00.295249 ago

The primary reports how many bytes of WAL the replica has yet to replay and, as replay_lag, how long recent WAL took to be replayed there. The replica reports how long ago the last transaction it replayed was committed, a number that keeps growing on an idle primary even when the replica is fully caught up, because there is nothing new to replay. RDS computes its ReplicaLag CloudWatch metric the same way, and its documentation warns that an idle source can show up to five minutes of lag. The byte count is therefore the reliable measure of how far behind a replica is.

9.7 Promotion

Promotion turns a replica into a standalone instance that accepts writes, either to recover from losing the primary or to split off a copy of a database. Promotion is permanent:

from db import connect, one, rds, wait

rds.promote_read_replica(DBInstanceIdentifier="shop-replica")
db = wait("shop-replica")
source = db.get("ReadReplicaSourceDBInstanceIdentifier")
print(db["DBInstanceStatus"], "source:", source)

COUNT = "SELECT count(*) FROM orders"
RECOVERY = "SELECT pg_is_in_recovery()"
with connect("shop-replica") as promoted:
    print("in recovery:", one(promoted, RECOVERY))
    promoted.execute("DELETE FROM orders WHERE id = 1001")
    print("orders:", one(promoted, COUNT))

with connect("shop-primary") as primary:
    print("primary still has:", one(primary, COUNT))

available source: None
in recovery: False
orders: 1000
primary still has: 1001

The promoted instance is out of recovery and has no source. From the moment of promotion the two databases diverge, so the order deleted on the promoted instance is still on the primary.

9.8 Enabling logical replication

Logical decoding needs wal_level = logical, which RDS lets you change only through its own parameter. Set rds.logical_replication = 1 in a parameter group, attach the group to the instance and reboot it, since the setting is read only at startup:

from db import connect, one, rds, wait

rds.create_db_parameter_group(
    DBParameterGroupName="shop-logical",
    DBParameterGroupFamily="postgres16",
    Description="PostgreSQL 16 with logical replication",
)
rds.modify_db_parameter_group(
    DBParameterGroupName="shop-logical",
    Parameters=[
        {
            "ParameterName": "rds.logical_replication",
            "ParameterValue": "1",
            "ApplyMethod": "pending-reboot",
        }
    ],
)
rds.modify_db_instance(
    DBInstanceIdentifier="shop-primary",
    DBParameterGroupName="shop-logical",
    ApplyImmediately=True,
)
wait("shop-primary")
rds.reboot_db_instance(DBInstanceIdentifier="shop-primary")
wait("shop-primary")

with connect("shop-primary") as primary:
    print("wal_level:", one(primary, "SHOW wal_level"))

wal_level: logical

The first wait is needed on AWS, where an instance must finish being modified before it can be rebooted. A higher wal_level writes more WAL, so turn it on only on instances that publish.

9.9 Subscriptions

The subscriber is a second, ordinary instance:

from db import PASSWORD, USER, rds, wait

rds.create_db_instance(
    DBInstanceIdentifier="shop-analytics",
    Engine="postgres",
    EngineVersion="16",
    DBInstanceClass="db.t4g.micro",
    AllocatedStorage=20,
    MasterUsername=USER,
    MasterUserPassword=PASSWORD,
    DBName="shop",
)
db = wait("shop-analytics")
print(db["DBInstanceIdentifier"], "is", db["DBInstanceStatus"])

shop-analytics is available

The next listing publishes the orders table on the primary, then creates the same table on the subscriber and subscribes to the publication, with the primary’s endpoint as the connection string:

import time

from db import PASSWORD, USER, connect, endpoint, one

COUNT = "SELECT count(*) FROM orders"
ORDERS = """
    CREATE TABLE orders (
        id        bigint PRIMARY KEY,
        customer  text NOT NULL,
        total     numeric(10, 2) NOT NULL,
        placed_at timestamptz NOT NULL DEFAULT now()
    )"""

with connect("shop-primary") as primary:
    primary.execute(
        "CREATE PUBLICATION shop_orders FOR TABLE orders"
    )

source = endpoint("shop-primary")
dsn = (
    f"host={source['Address']} port={source['Port']} "
    f"dbname=shop user={USER} password={PASSWORD}"
)
copying = """
    SELECT bool_or(srsubstate <> 'r') FROM pg_subscription_rel"""
with connect("shop-analytics") as analytics:
    analytics.execute(ORDERS)
    analytics.execute(
        f"CREATE SUBSCRIPTION shop_orders CONNECTION '{dsn}'"
        " PUBLICATION shop_orders"
    )
    while one(analytics, copying):
        time.sleep(0.2)
    print("copied:", one(analytics, COUNT))

copied: 1001

CREATE SUBSCRIPTION creates a replication slot on the publisher, which records how far the subscriber has received changes, then copies the table’s current rows and streams every later change. pg_subscription_rel tracks each table’s state, and r (ready) means the initial copy is done. All 1,001 orders arrived, including the one deleted from the promoted instance, because that delete was made on a separate database.

After the initial copy, changes replicate as they commit. The next listing inserts, updates and deletes rows on the primary, waits until the slot confirms that the subscriber has received the changes, and then reads the subscriber:

import time

from db import connect, one

confirmed = """
    SELECT confirmed_flush_lsn >= %s::pg_lsn
    FROM pg_replication_slots WHERE slot_name = 'shop_orders'"""

with connect("shop-primary") as primary:
    primary.execute(
        "INSERT INTO orders (id, customer, total)"
        " VALUES (2001, 'customer-3', 19.99)"
    )
    primary.execute(
        "UPDATE orders SET total = total * 2 WHERE id = 1"
    )
    primary.execute("DELETE FROM orders WHERE id = 2")
    lsn = one(primary, "SELECT pg_current_wal_lsn()")
    while not one(primary, confirmed, (lsn,)):
        time.sleep(0.1)

with connect("shop-analytics") as analytics:
    rows = analytics.execute(
        "SELECT id, total FROM orders"
        " WHERE id IN (1, 2, 2001) ORDER BY id"
    )
    print(rows.fetchall())
    count = one(analytics, "SELECT count(*) FROM orders")
    print("orders:", count)

[(1, Decimal('2.50')), (2001, Decimal('19.99'))]
orders: 1001

Order 1 doubled to 2.50, order 2 is gone and order 2001 arrived. On the subscriber, updates and deletes find their row by its replica identity, the columns that identify a row in the change stream, which here is the table’s primary key. A published table without a primary key needs a replica identity set by hand before the primary accepts updates and deletes on it.

9.10 Adding tables

Because logical replication copies only rows, a new table is created on both sides, added to the publication, and picked up by refreshing the subscription:

import time

from db import connect, one

CUSTOMERS = """
    CREATE TABLE customers (
        name    text PRIMARY KEY,
        segment text NOT NULL
    )"""

with connect("shop-primary") as primary:
    primary.execute(CUSTOMERS)
    primary.execute("""
        INSERT INTO customers
        SELECT 'customer-' || g,
               CASE WHEN g % 5 = 0 THEN 'wholesale'
                    ELSE 'retail' END
        FROM generate_series(0, 49) AS g""")
    primary.execute(
        "ALTER PUBLICATION shop_orders ADD TABLE customers"
    )

copying = """
    SELECT bool_or(srsubstate <> 'r') FROM pg_subscription_rel"""
with connect("shop-analytics") as analytics:
    analytics.execute(CUSTOMERS)
    analytics.execute(
        "ALTER SUBSCRIPTION shop_orders REFRESH PUBLICATION"
    )
    while one(analytics, copying):
        time.sleep(0.2)
    segments = analytics.execute("""
        SELECT segment, count(*) FROM customers
        GROUP BY segment ORDER BY segment""")
    print(segments.fetchall())

[('retail', 40), ('wholesale', 10)]

The refresh copies the table’s existing rows, and from then on its changes stream with the orders. To add a column, add it on the subscriber first and then on the publisher, so that the stream only ever carries columns the subscriber already has.

9.11 Replication slots

A replication slot makes the publisher keep every WAL segment the subscriber has not yet confirmed. This protects the subscriber from missing changes. It is also the commonest way to fill a database’s disk, because a subscriber that goes away without dropping its subscription leaves the slot behind and the publisher keeps WAL for it indefinitely, as AWS’s documentation warns. Check the slots, and drop subscriptions you no longer need:

from db import connect

SLOTS = """
    SELECT slot_name, slot_type, active
    FROM pg_replication_slots"""

with connect("shop-primary") as primary:
    print("before:", primary.execute(SLOTS).fetchall())

with connect("shop-analytics") as analytics:
    analytics.execute("DROP SUBSCRIPTION shop_orders")

with connect("shop-primary") as primary:
    print("after:", primary.execute(SLOTS).fetchall())
    primary.execute("DROP PUBLICATION shop_orders")

before: [('shop_orders', 'logical', True)]
after: []

DROP SUBSCRIPTION connected to the publisher and dropped the slot with it, so the drop fails when the publisher cannot be reached. In that case, detach the slot first with ALTER SUBSCRIPTION ... SET (slot_name = NONE), drop the subscription, and remove the slot on the publisher later with pg_drop_replication_slot. Investigate any slot whose active column stays false for a long time.

9.12 Cleaning up

from db import rds

for name in ("shop-analytics", "shop-replica", "shop-primary"):
    try:
        rds.delete_db_instance(
            DBInstanceIdentifier=name, SkipFinalSnapshot=True
        )
    except rds.exceptions.DBInstanceNotFoundFault:
        pass
try:
    rds.delete_db_parameter_group(
        DBParameterGroupName="shop-logical"
    )
except rds.exceptions.DBParameterGroupNotFoundFault:
    pass

On oblako, deleting an instance removes its container and its data.

9.13 Where oblako differs from AWS

• Instances are containers on your machine. Instance classes, storage size, IOPS and Multi-AZ are recorded and ignored, and every instance runs PostgreSQL 16, whatever version you request. Other engines, and the members of Aurora clusters, stay metadata on the shared database of Chapter 8.

• The master user is a PostgreSQL superuser. On RDS it is a member of rds_superuser, which includes rds_replication and pg_monitor, so the listings here work on both. A statement that needs true superuser rights works on oblako and fails on RDS.

• All instances share one network. Every instance can reach every other one, whereas on AWS the subscriber reaches the publisher only if a security group allows it.

• Replicas copy from the primary directly with pg_basebackup, whereas RDS builds them from a backup snapshot. Because oblako keeps no backups, it checks BackupRetentionPeriod but does not act on it.

• Lag is measured in SQL, as above, because oblako publishes no CloudWatch metrics such as ReplicaLag.

• State lives in moto’s memory. If moto restarts, the containers stay but the RDS API forgets the instances.

9.14 Running it on AWS

The listings run on AWS unchanged, against three db.t4g.micro instances that bill by the hour; delete them with the cleanup listing when you finish. Your machine must reach the instances: add PubliclyAccessible=True to the create calls and use a security group that admits your address on port 5432, or run the listings from inside the VPC. The subscriber connects to the publisher over the VPC, so the publisher’s security group must also admit the subscriber. Creating the replica takes longer than on oblako, because RDS snapshots the source first, and the reboot that applies the parameter group is a short outage for the primary.

9.15 Summary

A read replica replays the primary’s WAL into an exact, read-only copy that can lag behind. To read your own writes from it, wait until it has replayed past the primary’s LSN, and measure its lag in bytes. Promotion makes the copy a permanently independent database. Logical replication publishes chosen tables as row changes to any writable PostgreSQL database. On RDS it needs rds.logical_replication and a reboot, the tables must exist on both sides, and every subscription holds a slot that keeps WAL on the publisher until the subscription is dropped. On oblako each instance is a real PostgreSQL container with its own endpoint, and all of the listings run on your laptop.