pgx Logical Replication Client in Go for Change Capture

A database knows what changed. Most systems that need to know too, a search index, a cache, an analytics warehouse, another service, find out by asking: poll a table, scan an updated_at column, or have the application publish an event next to every write and hope the two never disagree. PostgreSQL offers a better source of truth: its own write ahead log, decoded into row level changes and streamed in commit order. This is logical replication, and the same stream feeds PostgreSQL's built in replication to other servers and change data capture tools such as Debezium. From Go, the pglogrepl package, written by pgx's author, gives you the protocol. This article builds a client that connects, creates a slot, streams a table's changes and decodes them, against a PostgreSQL 18.6 server started with wal_level=logical on Go 1.25.6.

It belongs to the database cornerstone. The simpler alternatives for the same job are the outbox pattern and LISTEN and NOTIFY, and the physical form of replication, which copies pages and not rows, is in read replicas.

Moving parts

WAL level. The server must run with wal_level = logical, which makes it record enough information in the log to reconstruct row changes. It needs a restart to change, and it increases the volume of WAL a little.

A publication says which tables' changes are available: create publication pub for table orders (or for all tables, with optional filters on operations and, since PostgreSQL 15, on rows and columns).

A replication slot is a named position in the log that the server keeps for a consumer. As long as the slot exists, the server retains all WAL since the slot's position, so that the consumer can come back and continue where it stopped. That is the guarantee and the risk, as discussed below.

An output plugin decides the format. pgoutput is the built in binary format used by native replication and by most clients.

A replication connection is an ordinary connection with the parameter replication=database, which makes it speak the replication protocol: IDENTIFY_SYSTEM, CREATE_REPLICATION_SLOT, START_REPLICATION. It does not run normal SQL, so changes to the table in the lab are made on a second, normal connection.

The client

The lab starts a PostgreSQL container with logical decoding enabled, creates a table and a publication, and runs the program.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
	"github.com/jackc/pgx/v5/pgtype"
)

const (
	replDSN = "postgres://postgres:[email protected]:55435/lab?replication=database"
	dsn     = "postgres://postgres:[email protected]:55435/lab"
)

func must(err error) {
	if err != nil {
		log.Fatal(err)
	}
}

func main() {
	ctx := context.Background()

	// The replication connection is a normal connection that has asked to speak the replication protocol.
	conn, err := pgconn.Connect(ctx, replDSN)
	must(err)
	defer conn.Close(ctx)

	sys, err := pglogrepl.IdentifySystem(ctx, conn)
	must(err)
	fmt.Printf("1. IdentifySystem: timeline %d, database %q, a log position was returned: %v\n", sys.Timeline, sys.DBName, sys.XLogPos > 0)

	// A temporary slot lives as long as this connection; a permanent one keeps WAL on the server until it is consumed.
	const slot = "slot679"
	res, err := pglogrepl.CreateReplicationSlot(ctx, conn, slot, "pgoutput", pglogrepl.CreateReplicationSlotOptions{Temporary: true})
	must(err)
	fmt.Printf("2. slot %s created with the %s plugin, consistent from a log position: %v\n", res.SlotName, "pgoutput", res.ConsistentPoint != "")
	start, _ := pglogrepl.ParseLSN(res.ConsistentPoint)
	must(pglogrepl.StartReplication(ctx, conn, slot, start, pglogrepl.StartReplicationOptions{
		PluginArgs: []string{"proto_version '1'", "publication_names 'pub679'"},
	}))

	// Make changes on another connection while the stream is open.
	go func() {
		time.Sleep(500 * time.Millisecond)
		c, err := pgx.Connect(ctx, dsn)
		if err != nil {
			return
		}
		defer c.Close(ctx)
		_, _ = c.Exec(ctx, "insert into orders679 values (1, 'bolt', 10), (2, 'nut', 20)")
		_, _ = c.Exec(ctx, "update orders679 set qty = 15 where id = 1")
		_, _ = c.Exec(ctx, "delete from orders679 where id = 2")
		tx, _ := c.Begin(ctx)
		_, _ = tx.Exec(ctx, "insert into orders679 values (3, 'washer', 30)")
		_ = tx.Rollback(ctx) // a rolled back transaction never appears in the stream
	}()

	relations := map[uint32]*pglogrepl.RelationMessage{}
	typeMap := pgtype.NewMap()
	decode := func(rel *pglogrepl.RelationMessage, t *pglogrepl.TupleData) string {
		if t == nil {
			return "-"
		}
		out := ""
		for i, col := range t.Columns {
			val := "<null>"
			switch col.DataType {
			case 't':
				val = string(col.Data)
			case 'u':
				val = "<unchanged>"
			}
			out += fmt.Sprintf("%s=%s ", rel.Columns[i].Name, val)
		}
		return out
	}

	fmt.Println("3. changes received from the stream:")
	received := 0
	deadline := time.Now().Add(4 * time.Second)
	for received < 4 && time.Now().Before(deadline) {
		rctx, cancel := context.WithTimeout(ctx, time.Second)
		raw, err := conn.ReceiveMessage(rctx)
		cancel()
		if err != nil {
			if pgconn.Timeout(err) {
				continue
			}
			log.Fatal(err)
		}
		msg, ok := raw.(*pgproto3.CopyData)
		if !ok {
			continue
		}
		switch msg.Data[0] {
		case pglogrepl.PrimaryKeepaliveMessageByteID:
			ka, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
			must(err)
			if ka.ReplyRequested {
				must(pglogrepl.SendStandbyStatusUpdate(ctx, conn, pglogrepl.StandbyStatusUpdate{WALWritePosition: ka.ServerWALEnd}))
			}
		case pglogrepl.XLogDataByteID:
			xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
			must(err)
			lm, err := pglogrepl.Parse(xld.WALData)
			must(err)
			switch m := lm.(type) {
			case *pglogrepl.RelationMessage:
				relations[m.RelationID] = m
			case *pglogrepl.BeginMessage:
				fmt.Println("   BEGIN")
			case *pglogrepl.InsertMessage:
				fmt.Printf("   INSERT %s %s\n", relations[m.RelationID].RelationName, decode(relations[m.RelationID], m.Tuple))
				received++
			case *pglogrepl.UpdateMessage:
				fmt.Printf("   UPDATE %s new: %s\n", relations[m.RelationID].RelationName, decode(relations[m.RelationID], m.NewTuple))
				received++
			case *pglogrepl.DeleteMessage:
				fmt.Printf("   DELETE %s old key: %s\n", relations[m.RelationID].RelationName, decode(relations[m.RelationID], m.OldTuple))
				received++
			case *pglogrepl.CommitMessage:
				fmt.Println("   COMMIT")
			}
		}
	}
	_ = typeMap
	fmt.Println("4. four changes arrived: two inserts, an update and a delete; the insert of id 3 was rolled back and never appeared")
}

set -e
trap 'docker rm -f gh-lab-logical >/dev/null 2>&1' EXIT
docker rm -f gh-lab-logical >/dev/null 2>&1 || true
docker run -d --name gh-lab-logical -p 127.0.0.1:55435:5432 -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=lab postgres:18-alpine -c wal_level=logical -c max_replication_slots=4 -c max_wal_senders=4 >/dev/null
for i in $(seq 1 60); do
  docker exec gh-lab-logical pg_isready -U postgres -d lab >/dev/null 2>&1 && break
  sleep 1
done
sleep 2
docker exec gh-lab-logical psql -U postgres -d lab -q -c "create table orders679 (id int primary key, item text, qty int)" -c "create publication pub679 for table orders679"
go run .

1. IdentifySystem: timeline 1, database "lab", a log position was returned: true
2. slot slot679 created with the pgoutput plugin, consistent from a log position: true
3. changes received from the stream:
   BEGIN
   INSERT orders679 id=1 item=bolt qty=10 
   INSERT orders679 id=2 item=nut qty=20 
   COMMIT
   BEGIN
   UPDATE orders679 new: id=1 item=bolt qty=15 
   COMMIT
   BEGIN
   DELETE orders679 old key: id=2 item=<null> qty=<null> 
4. four changes arrived: two inserts, an update and a delete; the insert of id 3 was rolled back and never appeared

The program does five things.

It identifies the system. pglogrepl.IdentifySystem returns the timeline, the current log position and the database name: timeline 1, database lab, and a position. It is a handshake that proves the connection is in replication mode.

It creates a slot. CreateReplicationSlot with the pgoutput plugin and Temporary: true makes a slot that disappears when the connection closes. That is right for an experiment. A real consumer uses a permanent slot with a stable name, so it can stop and resume. The call returns the slot's consistent point, the log position from which changes will be delivered, and that is the point to start streaming from.

It starts streaming. StartReplication takes the slot, the starting position and plugin arguments: proto_version '1' and publication_names 'pub679'. From here the connection delivers a stream of messages.

It makes changes on another connection in a goroutine: two inserts in one statement, an update, a delete, and an insert in a transaction that is rolled back.

It receives and decodes. conn.ReceiveMessage returns protocol messages. A CopyData message is either a keepalive (the server's way of saying it is alive and, if ReplyRequested is set, asking for a status update) or XLogData, which carries a logical message. pglogrepl.Parse turns the payload into a typed value.

What arrives

The output shows the stream in order:

• BEGIN, then INSERT orders679 id=1 item=bolt qty=10, INSERT ... id=2 item=nut qty=20, then COMMIT: the two rows inserted by one statement arrive as one transaction.
• BEGIN, UPDATE orders679 new: id=1 item=bolt qty=15, COMMIT.
• BEGIN, DELETE orders679 old key: id=2 item=<null> qty=<null>.
• Nothing for the rolled back insert of id 3.

Four properties of this stream are the reason to use it. It is ordered by commit: changes appear in the order transactions committed, grouped by transaction, so a consumer that applies them in order reproduces the database's history. It contains only committed data: the insert in the rolled back transaction never appeared, and a consumer never has to undo anything. It is complete: unlike polling an updated_at column, it sees deletes, and it sees every intermediate update, not just the latest state. And it does not depend on the application: any writer, a migration, a psql session, produces the same events.

Decoding details

Relation messages. The first time a table appears in the stream (and again after a schema change), the server sends a RelationMessage with the relation id, the table name and the list of columns with their types. Data messages refer to the table by relation id only, so the client keeps relations[id] and looks up column names when decoding, as the lab does. A client that ignores relation messages cannot name its columns.

Tuples in text form. Each column of a tuple arrives as t (text, with the value as bytes), n (NULL) or u (unchanged TOAST value, meaning a large value that was not modified and is not resent). The lab prints the text. To convert, use the column's type OID with a pgtype.Map, which Decodes text values into Go types.

Replica identity. The delete carried id=2 and NULLs for the other columns. By default (replica identity default) the old row is identified by its primary key only, so the delete message holds the key columns and nothing else. If a consumer needs the old values of all columns for updates and deletes, set alter table ... replica identity full, at the price of more WAL. A table with no primary key and the default identity cannot publish updates and deletes at all.

Transactions. With protocol version 1 a large transaction is sent only after it commits. Protocol versions 2 and later can stream in-progress transactions, enabled with the streaming option, which matters for very large transactions and requires handling stream start, stop, commit and abort messages.

Acknowledging, and why it matters

A consumer must tell the server how far it has processed, using SendStandbyStatusUpdate with the log position (LSN) it has durably handled. The server uses it to decide how much WAL it may discard. The lab replies to keepalives that request an answer and does not otherwise track positions, since its slot is temporary. A real consumer follows a strict order: apply the change to the destination durably first, then acknowledge its position. If it acknowledges first and crashes, those changes are lost. If it applies first and crashes before acknowledging, the server resends them on restart, so the destination must tolerate duplicates: make writes idempotent, or store the last applied LSN with the data in the same transaction and skip anything at or below it.

The cost of a slot

A permanent slot makes the server keep WAL until the consumer acknowledges it. A consumer that stops, is deleted, or falls far behind makes the WAL accumulate until the disk fills and the primary database stops accepting writes. This is the most serious operational risk of logical replication, and it has happened to many teams. Defences:

• Monitor slot lag (pg_replication_slots, columns restart_lsn and confirmed_flush_lsn, and pg_wal_lsn_diff against the current position) and alert on it.
• Set max_slot_wal_keep_size, which makes the server drop a slot that holds back more than a given amount of WAL, protecting the database at the cost of the consumer.
• Drop slots you no longer use (pg_drop_replication_slot), including from environments that were shut down.
• Run the consumer as a supervised service that restarts, and alert when it is not running.

Initial state and schema changes

A slot delivers changes from its start. The existing rows of a table are not in the stream, so a consumer that needs them must copy a snapshot consistent with the slot's start. The protocol supports this: creating a slot with the export snapshot option returns a snapshot name that a second connection can use in set transaction snapshot to read the table as of exactly the slot's start, and then the stream continues without a gap. Tools that do this for you, such as Debezium and PostgreSQL's own subscriber, handle the dance; if you write your own, test the boundary with writes happening during the copy.

Schema changes arrive as new relation messages, but the stream does not carry DDL: a alter table add column is visible only through the changed column list in the next relation message. Plan how your consumer maps columns, and deploy consumers that tolerate new columns before the migration.

Structure of a real consumer

The lab prints messages. A consumer you would run in production has a slightly different shape, and the differences are where bugs hide. It reads messages in one goroutine and hands decoded changes, with their LSNs, to a second that writes to the destination in batches, so a slow destination applies backpressure instead of growing memory without bound. It tracks the LSN of the last commit message, not of the last data message, because only a commit is a safe place to resume: resuming in the middle of a transaction would deliver half of it. It sends a standby status update periodically, every ten seconds or so even when idle, because the server drops a connection that goes quiet for longer than wal_sender_timeout, 60 seconds by default. And it reconnects with backoff when the connection breaks, restarting from the last acknowledged LSN, which the server remembers in the slot, so a restart needs no state of its own beyond the slot name.

On the destination side, treat the change stream as at least once. For a key value destination such as a cache or a search index, an upsert by primary key is idempotent and an apply of a delete is too. For a relational destination, store the last applied LSN in a table, update it in the same transaction as the changes, and skip any change at or below it after a restart. That single table turns at least once delivery into effectively once processing.

Testing a consumer

The lab is the shape of the test. Start a real server with wal_level=logical, create the table and publication, run the consumer, make the changes the consumer must handle, and assert on what the destination ends up holding. The cases that find bugs are the unpleasant ones: a transaction that touches several tables, a rolled back transaction, an update that changes the primary key, a very large transaction, a table without a primary key, a column added mid stream, and a consumer killed between applying a change and acknowledging it. Run the last by killing the process at each step of the apply and acknowledge sequence and checking that the restart converges on the same destination state, as the crash tests in the tiny database project do for a log. Check also that the slot is dropped when the test ends, since a leaked slot on a shared test server fills its disk.

When to use it

Logical replication fits change data capture into search indexes, caches, warehouses and other services; keeping several PostgreSQL databases in sync; zero downtime major version upgrades, as in upgrading Postgres under a running service; and audit streams. Prefer simpler tools first: an outbox table and a relay is easier to operate and reason about for service to service events, since it needs no slot, no special server setting and no decoding. Prefer logical replication when you need every change to every row without modifying the writers, and can run the consumer reliably. The package is documented in the pglogrepl repository, and the server side in the PostgreSQL manual's chapters on logical replication and the streaming replication protocol.