From PostgreSQL WAL to Search

Logical Replication, Publications, Slots, and LSNs

Applications end up needing the same thing over and over: some other system has to find out when a row changes. A search index needs to know when an article's body changes. An analytics warehouse needs every insert. A cache needs to know when to invalidate. The obvious first instinct is to poll: run SELECT * FROM articles WHERE updated_at > ? every few seconds. This works until it doesn't. Polling misses anything that changes twice between polls, it can't see deletes without a soft-delete column, and it puts a steady query load on the database whether anything changed or not.

The other obvious instinct is triggers: write an AFTER INSERT OR UPDATE OR DELETE trigger that pushes a row into an outbox table or calls out to another service. This works, but now the application's write path is coupled to whatever the trigger does. A slow or failing trigger slows down or fails the write itself.

Change Data Capture, CDC, is a third option. Instead of asking the database "did anything change," you ask it to hand you the ordered stream of changes it already produces internally, as a side effect of just being a database. PostgreSQL already writes down every change durably, in commit order, for its own crash recovery. Logical replication is PostgreSQL exposing that stream to you.

This is the first part of a series that builds a real CDC pipeline in Go: read PostgreSQL's change stream, turn it into typed events, checkpoint progress durably, and project it into a small search index over an articles table. By the end, the system backfills existing rows, survives crashes without losing or duplicating writes it shouldn't, and fails over between two instances. This part does none of that yet. It gets a Go program connected to PostgreSQL's replication stream and reading the raw messages it sends. That connection is the foundation everything else in the series sits on.

If you've read the WAL post on this blog, you already know half the story. PostgreSQL keeps a write-ahead log for the same reason that post's toy WAL does: durability. Every change is written and fsynced to the log before it's considered committed, so a crash can replay the log and recover. Logical replication reuses that exact log. It doesn't add a second log for replication purposes. It reads the same WAL that already exists for crash recovery and reinterprets it.

Physical WAL vs. Logical Decoding

PostgreSQL's WAL, in its native form, is physical. A physical WAL record says something like "at this byte offset on this disk page, these bytes changed." That's exactly enough information for PostgreSQL itself to replay changes onto an identical copy of the same database files, which is how physical streaming replication (the kind that builds a byte-identical standby) works. It is not useful to anything outside PostgreSQL. You can't hand a Go program "bytes 40 through 52 of page 9 changed" and expect it to reconstruct that an article's title changed from one string to another. Physical WAL records don't carry table names, column names, or old and new values in a form external code can interpret without effectively reimplementing PostgreSQL's storage layer.

Logical decoding solves this by running the physical WAL back through PostgreSQL itself. A background process on the server reads physical WAL records the normal way, but instead of applying them to disk pages, it feeds them through an output plugin that turns them into logical changes: table articles, row with id = 12, UPDATE, old values, new values. That translation happens inside PostgreSQL, using PostgreSQL's own knowledge of table schemas and row formats. The Go program never touches raw WAL bytes or page formats. It receives already-decoded messages over a normal connection.

This project uses the built-in output plugin, pgoutput. It's the same plugin PostgreSQL's own built-in logical replication (the CREATE SUBSCRIPTION feature, for replicating between two Postgres instances) uses internally, so it's well maintained and doesn't require installing anything on the server. Other plugins exist, wal2json is common and emits JSON instead of a binary protocol, but pgoutput is the one every serious Go CDC client ends up building around, since there's no server-side extension to install.

Publications: What Gets Decoded

A publication is a named, server-side declaration of which tables, and optionally which operations and which rows via WHERE, are eligible to be decoded. It's a permission and scope boundary, not a subscriber. Creating one doesn't start streaming anything by itself.

Table and publication names can't be parameterized as SQL arguments the way values can, so creating one from Go means building the statement string directly. That's fine here because the name comes from hardcoded config, never from user input; it would not be fine if table were something a caller could influence.

// internal/cdc/setup.go
package cdc

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5"
)

// EnsurePublication creates the given publication for the given table if
// it does not already exist. Publications are not automatically created
// by CREATE_REPLICATION_SLOT, so this must run before slot creation.
func EnsurePublication(ctx context.Context, conn *pgx.Conn, pubName, table string) error {
	var exists bool
	err := conn.QueryRow(ctx,
		`SELECT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = $1)`,
		pubName,
	).Scan(&exists)
	if err != nil {
		return fmt.Errorf("check publication: %w", err)
	}
	if exists {
		return nil
	}

	sql := fmt.Sprintf("CREATE PUBLICATION %s FOR TABLE %s", pubName, table)
	if _, err := conn.Exec(ctx, sql); err != nil {
		return fmt.Errorf("create publication %s: %w", pubName, err)
	}
	return nil
}

One detail that matters for correctness later: a publication only covers changes made after it exists. It says nothing about rows already sitting in the table beforehand. Getting those into the search index is a separate problem called backfill, and it's genuinely tricky to do without racing the live stream. That's Part 5's entire subject. For this part, the table starts empty and stays that way except for what we insert ourselves.

Replication Slots: The Part That Can Bite You

A replication slot is server-side state that tracks how far a specific consumer has read PostgreSQL's WAL, and it exists specifically so PostgreSQL knows it's safe to reclaim, delete, WAL segments once every slot that needs them has confirmed it's past them.

Here's the invariant a slot enforces: PostgreSQL will not delete a WAL segment that any active slot still needs, no matter how long that takes. Normally, WAL segments are recycled quickly once a checkpoint has happened and nothing needs them. A slot changes that. If the consumer reading it is slow, disconnected, or never started, the slot sitting there un-advanced tells PostgreSQL "don't delete this WAL yet," and PostgreSQL obeys that indefinitely.

This is the most common way people break a production PostgreSQL instance with logical replication: create a slot, have the consumer crash or get decommissioned, and forget the slot exists. WAL accumulates on disk forever because nothing is confirming progress against that slot. Disk fills up. This isn't a rare edge case, it's the default failure mode, which is why it's worth internalizing here in Part 1 rather than as a footnote later.

A slot is created bound to a specific decoding plugin, pgoutput here, and it remembers its own position in the WAL stream as a restart LSN: the earliest point PostgreSQL still needs to keep around for this slot. As the consumer confirms progress, PostgreSQL is allowed to move that watermark forward and reclaim what's behind it.

// internal/cdc/slot.go
package cdc

import (
	"context"
	"errors"
	"fmt"
	"strings"

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

// EnsureSlot creates a durable pgoutput slot with the given name if it does
// not already exist, and returns the LSN it should start streaming from.
//
// The slot is durable (Temporary: false) on purpose: it needs to survive
// the Go process restarting, so a restart resumes from where it left off
// instead of silently jumping to the current WAL position and skipping
// whatever happened while it was down. A durable slot is also the thing
// that can accumulate WAL forever if this program stops running and
// nobody notices, which the demo further down makes concrete.
func EnsureSlot(ctx context.Context, conn *pgconn.PgConn, slotName string) (pglogrepl.LSN, error) {
	result, err := pglogrepl.CreateReplicationSlot(
		ctx, conn, slotName, "pgoutput",
		pglogrepl.CreateReplicationSlotOptions{Temporary: false},
	)
	if err == nil {
		return result.ConsistentPoint, nil
	}

	// PostgreSQL returns a specific error message when the slot already
	// exists; there's no separate pglogrepl helper for checking existence
	// up front, so this checks the error text. An already-existing slot is
	// expected on every run after the first.
	if strings.Contains(err.Error(), "already exists") {
		return pglogrepl.LSN(0), errAlreadyExists
	}
	return pglogrepl.LSN(0), fmt.Errorf("create replication slot %s: %w", slotName, err)
}

var errAlreadyExists = errors.New("slot already exists")

errAlreadyExists gets handled in main.go rather than inside this function, because looking up an existing slot's restart position means querying pg_replication_slots, a catalog table, and that has to go through the normal SQL connection, not the replication one. A replication-mode connection can't run arbitrary catalog queries; more on that distinction next.

LSNs: The Position Type

An LSN, log sequence number, is PostgreSQL's byte-offset position within the WAL stream, globally increasing for the lifetime of the database. It's structurally the same idea as the LSN in the WAL post on this blog, where LSN was defined as the byte offset of a record in a single log file. PostgreSQL's version generalizes that across many WAL segment files: a 64-bit value, printed conventionally as two hex numbers separated by a slash, like 0/1634C50. The two halves are the high and low 32 bits; pglogrepl parses and compares LSNs as a normal ordered value, so there's no reason to decode that split by hand.

Everything about resuming and checkpointing in this series comes down to LSN comparisons: have we definitely applied everything up to LSN X, and can PostgreSQL now forget anything before LSN Y. Part 3 builds the actual checkpoint persistence. For now, an LSN is just a position, not a timestamp or a row count, and the slot above already hands one back as ConsistentPoint, the point streaming should resume from.

Connections: Replication Mode vs. Normal SQL

A replication connection to PostgreSQL is a regular libpq-protocol TCP connection, but opened with replication=database set in the connection parameters. That one flag changes what the connection is allowed to do. A replication-mode connection can't run arbitrary SQL like SELECT * FROM articles. It can run a small allowed set of replication commands, IDENTIFY_SYSTEM, CREATE_REPLICATION_SLOT, START_REPLICATION, and a few others, and once START_REPLICATION has run, it switches into a streaming mode where the server pushes messages continuously instead of waiting for queries.

Practically, this means the project needs two separate connections to the same database: a normal pgx.Conn for ordinary SQL, creating the publication now, later reading rows for backfill, and a replication-mode pgconn.PgConn dedicated to the streaming protocol. Trying to do both over one connection doesn't work; they're different protocol modes.

// internal/cdc/client.go
package cdc

import (
	"context"
	"fmt"

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

// Connect opens a replication-mode connection to PostgreSQL. connString
// should not already include replication=database; this function adds it,
// so the same base connection string can be reused for a normal SQL
// connection elsewhere without duplicating the parameter.
func Connect(ctx context.Context, connString string) (*pgconn.PgConn, error) {
	config, err := pgconn.ParseConfig(connString)
	if err != nil {
		return nil, fmt.Errorf("parse connection string: %w", err)
	}
	config.RuntimeParams["replication"] = "database"

	conn, err := pgconn.ConnectConfig(ctx, config)
	if err != nil {
		return nil, fmt.Errorf("connect in replication mode: %w", err)
	}
	return conn, nil
}

// SystemInfo wraps IDENTIFY_SYSTEM, mainly useful to confirm the connection
// really is in replication mode and to get a starting LSN when no slot
// position exists yet.
func SystemInfo(ctx context.Context, conn *pgconn.PgConn) (pglogrepl.IdentifySystemResult, error) {
	return pglogrepl.IdentifySystem(ctx, conn)
}

The replication user also needs specific privileges. It's tempting to reuse a superuser or the application's normal role, but a CDC consumer only needs to read the change stream, not modify data, so it should get the narrowest role that can do the job:

CREATE ROLE cdc_reader WITH LOGIN PASSWORD 'change_me' REPLICATION;
GRANT SELECT ON articles TO cdc_reader;

REPLICATION is the privilege that allows opening a replication-mode connection and creating or using slots at all. The SELECT grant isn't needed yet, this part never runs a SELECT, but it's included here because Part 5's backfill will need it, and there's no reason to widen the role's privileges mid-series when it can be scoped correctly from the start.

Starting the Stream and Reading What Comes Back

START_REPLICATION takes a slot name, a starting LSN, and, for pgoutput, plugin arguments: a protocol version and the publication name to filter by. From that point on, the connection stops behaving like a request/response protocol and starts pushing messages.

Two kinds of messages arrive, both wrapped in CopyData. XLogData carries an actual change, still encoded in pgoutput's own binary format, opaque to us for now. PrimaryKeepaliveMessage is the server checking the connection is alive, sent periodically even with nothing to report, and it can optionally set a flag asking for an immediate reply.

The other direction matters just as much: the client has to periodically send a standby status update telling PostgreSQL how far it's gotten. This is what actually advances the slot's restart LSN. Without it, PostgreSQL has no idea the consumer is making progress, and the WAL retention problem described above starts immediately, even with a perfectly healthy consumer, simply because nobody told the server anything.

// internal/cdc/stream.go
package cdc

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

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

// standbyUpdateInterval controls how often we tell PostgreSQL where we've
// gotten to, independent of how often messages actually arrive. PostgreSQL
// expects to hear from the client periodically even when nothing changed,
// or it will eventually decide the connection is dead.
const standbyUpdateInterval = 10 * time.Second

// StreamRaw starts replication from startLSN and logs every message it
// receives until ctx is canceled. It does not persist progress and does
// not decode pgoutput payloads — see Part 2 and Part 3 for those.
func StreamRaw(ctx context.Context, conn *pgconn.PgConn, slotName string, startLSN pglogrepl.LSN) error {
	err := pglogrepl.StartReplication(ctx, conn, slotName, startLSN, pglogrepl.StartReplicationOptions{
		PluginArgs: []string{
			"proto_version '1'",
			"publication_names 'articles_pub'",
		},
	})
	if err != nil {
		return fmt.Errorf("start replication: %w", err)
	}
	log.Printf("cdc: streaming from %s", startLSN)

	lastReported := startLSN
	nextStandby := time.Now().Add(standbyUpdateInterval)

	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		if time.Now().After(nextStandby) {
			if err := pglogrepl.SendStandbyStatusUpdate(ctx, conn, pglogrepl.StandbyStatusUpdate{
				WALWritePosition: lastReported,
			}); err != nil {
				return fmt.Errorf("send standby status: %w", err)
			}
			nextStandby = time.Now().Add(standbyUpdateInterval)
		}

		recvCtx, cancel := context.WithTimeout(ctx, standbyUpdateInterval)
		msg, err := conn.ReceiveMessage(recvCtx)
		cancel()
		if err != nil {
			if pgconn.Timeout(err) {
				// Nothing arrived within the timeout; loop back around so
				// the standby status check above still runs on schedule.
				continue
			}
			return fmt.Errorf("receive message: %w", err)
		}

		cd, ok := msg.(*pgproto3.CopyData)
		if !ok {
			log.Printf("cdc: unexpected message type %T", msg)
			continue
		}

		switch cd.Data[0] {
		case pglogrepl.XLogDataByteID:
			xld, err := pglogrepl.ParseXLogData(cd.Data[1:])
			if err != nil {
				return fmt.Errorf("parse XLogData: %w", err)
			}
			log.Printf("cdc: XLogData at LSN %s, %d bytes of pgoutput payload",
				xld.WALStart, len(xld.WALData))
			lastReported = xld.WALStart

		case pglogrepl.PrimaryKeepaliveMessageByteID:
			pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(cd.Data[1:])
			if err != nil {
				return fmt.Errorf("parse keepalive: %w", err)
			}
			log.Printf("cdc: keepalive, server at %s, reply requested=%v",
				pkm.ServerWALEnd, pkm.ReplyRequested)
			if pkm.ReplyRequested {
				nextStandby = time.Now() // force the update on the next loop iteration
			}

		default:
			log.Printf("cdc: unknown CopyData message type %q", cd.Data[0])
		}
	}
}

lastReported is deliberately updated from xld.WALStart on every XLogData message rather than left at whatever startLSN was. This part doesn't yet distinguish "received" from "safely applied," so it's reporting optimistically, as soon as a message arrives, not after anything has been done with it. That distinction between received, flushed, and applied positions is exactly what Part 3 has to get right; here it would be premature to build machinery for a guarantee this part doesn't provide.

Wiring It Together

main.go owns both connections and decides what to do with errAlreadyExists from EnsureSlot: look the slot's restart position up through the normal SQL connection, since that's a catalog query the replication connection can't run.

// cmd/server/main.go
package main

import (
	"context"
	"errors"
	"log"
	"os/signal"
	"syscall"

	"github.com/amrrdev/cdc/internal/cdc"
	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5"
)

const (
	connString = "postgres://cdc:cdc@localhost:5432/cdcdb"
	pubName    = "articles_pub"
	slotName   = "articles_slot"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	sqlConn, err := pgx.Connect(ctx, connString)
	if err != nil {
		log.Fatalf("connect (sql): %v", err)
	}
	defer sqlConn.Close(ctx)

	if _, err := sqlConn.Exec(ctx, `CREATE TABLE IF NOT EXISTS articles (
		id BIGSERIAL PRIMARY KEY,
		title TEXT NOT NULL,
		body TEXT NOT NULL,
		author TEXT NOT NULL,
		published_at TIMESTAMPTZ NOT NULL DEFAULT now(),
		updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
	)`); err != nil {
		log.Fatalf("create table: %v", err)
	}

	if err := cdc.EnsurePublication(ctx, sqlConn, pubName, "articles"); err != nil {
		log.Fatalf("ensure publication: %v", err)
	}

	replConn, err := cdc.Connect(ctx, connString)
	if err != nil {
		log.Fatalf("connect (replication): %v", err)
	}
	defer replConn.Close(ctx)

	sysInfo, err := cdc.SystemInfo(ctx, replConn)
	if err != nil {
		log.Fatalf("identify system: %v", err)
	}
	log.Printf("cdc: connected to system %s, current WAL position %s", sysInfo.SystemID, sysInfo.XLogPos)

	startLSN, err := cdc.EnsureSlot(ctx, replConn, slotName)
	if err != nil {
		if !errors.Is(err, cdc.ErrAlreadyExists) {
			log.Fatalf("ensure slot: %v", err)
		}
		var restartLSN string
		if err := sqlConn.QueryRow(ctx,
			`SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = $1`,
			slotName,
		).Scan(&restartLSN); err != nil {
			log.Fatalf("look up slot restart_lsn: %v", err)
		}
		parsed, err := pglogrepl.ParseLSN(restartLSN)
		if err != nil {
			log.Fatalf("parse restart_lsn: %v", err)
		}
		startLSN = parsed
	}

	if err := cdc.StreamRaw(ctx, replConn, slotName, startLSN); err != nil {
		log.Printf("cdc: stream ended: %v", err)
	}
}

(errAlreadyExists from the slot file is exported as ErrAlreadyExists for this to compile against; a one-line rename in slot.go.)

Running this against a fresh database, docker compose up -d followed by go run ./cmd/server, connects, creates the table and publication if they're missing, creates the slot, and starts logging keepalives every ten seconds or so. Inserting a row from a second terminal with psql,

docker compose exec postgres psql -U cdc -d cdcdb -c \
  "INSERT INTO articles (title, body, author) VALUES ('Hello CDC', 'first row', 'amr')"

produces an XLogData line in the Go program's output within about a second, carrying an LSN past whatever sysInfo.XLogPos was at startup. Nothing in that message is human-readable yet, it's raw pgoutput bytes, but the plumbing between an application-level INSERT and a Go process is now live end to end, which is the actual milestone for this part.

Watching the Disk Risk Happen

The WAL retention behavior described earlier is worth seeing directly rather than taking on faith. Stop the Go program with Ctrl-C, leaving the slot in place, then generate WAL churn with nothing consuming it:

docker compose exec postgres psql -U cdc -d cdcdb -c \
  "INSERT INTO articles (title, body, author) SELECT 'x', repeat('a', 5000), 'amr' FROM generate_series(1, 20000)"

Then check retention directly:

SELECT slot_name, active, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

With the consumer stopped, active reads false and retained_wal grows with every insert above, instead of staying near zero the way it would if something were confirming progress. That number is disk PostgreSQL is refusing to reclaim, specifically because articles_slot exists and hasn't advanced. Restart the Go program and rerun the query: as soon as standby status updates resume, retained_wal shrinks back down. This is also the tradeoff in creating a durable, not temporary, slot: a temporary slot disappears when its connection closes, avoiding this problem entirely, but also meaning a real restart can't resume from where it left off, it would have to start over from the current WAL position and silently skip whatever happened while it was down. Part 3 is what actually makes leaving a durable slot running safe long-term.

What This Part Doesn't Do

This program can't tell you what changed yet, only that something changed and where. XLogData.WALData is opaque pgoutput-encoded bytes right now. There's no relation metadata being tracked, so even once those bytes are parsed, there's no way yet to know which columns they correspond to. Nothing is durable in the sense that matters: the LSN this program starts from on the next run comes from the slot's own already-recorded restart position, not from anything this program tracked itself, and that position only advanced because a standby update happened to go out before the process died. If the process crashes between applying a change and reporting that LSN, nothing here has even defined yet what "applying a change" means. That's Part 3's problem to solve.

Next

Part 2 opens XLogData.WALData and decodes it: Relation messages that describe a table's columns, Begin and Commit messages that bound a transaction, and Insert, Update, and Delete messages that carry actual tuple data, keyed by the relation ID a prior Relation message assigned. The milestone there is turning the byte blob this part logs into something like INSERT articles: {id: 1, title: "Hello CDC", body: "first row", author: "amr"}, readable in Go, with the tricky parts, replica identity for updates and deletes, NULL handling, and the unchanged TOAST values PostgreSQL omits from the wire format to save bandwidth, called out explicitly as they come up.