The reference pipeline — replicant as a durable Postgres→Postgres replicator

Copy Markdown View Source

One docker compose up brings up the whole reference stack:

Postgres 18 (source, wal_level=logical + example_pub) ──pgoutput──> [this app]
                                                                     ├─ idempotent orders replica (upsert by PK)
                                                                     └─ durable commit-LSN checkpoint
                                                                  ──> Postgres 18 (destination)

This is a go-forward change replicator: it streams changes that commit from the moment the pipeline starts. Pre-existing source rows are not backfilled — snapshot: true is the one-flag alternative for that (it additionally requires handle_snapshot/2 + handle_snapshot_complete/1 on the sink; see the repo README's "Start modes").

All credentials are throwaway local-example values, not secrets. Ports bind to 127.0.0.1 only (15432 source / 15433 destination).

Run it

cd examples/replication_pipeline
docker compose up -d --build          # source + destination + pipeline

First boot is a full OTP-release build — a few minutes. The pipeline creates its replication slot a few seconds after boot; an insert committed BEFORE the slot exists is never streamed (go-forward starts at the slot's creation point), so confirm the slot is active first:

docker compose exec source-pg psql -U postgres -d example_src \
  -c "SELECT slot_name, active FROM pg_replication_slots"
# one row, active = t — ready

# write a row on the source
docker compose exec source-pg psql -U postgres -d example_src \
  -c "INSERT INTO orders (id, note) VALUES (1, 'hello')"

# see it on the destination, with its checkpoint
docker compose exec dest-pg psql -U postgres -d example_dst \
  -c "SELECT * FROM orders; SELECT commit_lsn FROM pipeline_checkpoint"

Teardown: docker compose down -v.

Prove the durability

docker compose restart pipeline                     # kill/restart mid-life
docker compose exec source-pg psql -U postgres -d example_src \
  -c "INSERT INTO orders (id, note) VALUES (3, 'after restart')"
docker compose exec dest-pg psql -U postgres -d example_dst \
  -c "SELECT id, note FROM orders ORDER BY id" \
  -c "SELECT count(*) FROM cdc_receipts"            # no re-delivery of prior rows

The durable checkpoint row made that resume gap-free and effect-once: the post-restart row arrives alongside the earlier ones, and the receipts for prior transactions are unchanged — a re-delivery would have doubled them.

What each piece teaches

PieceThe lesson
ReplicationPipeline.Sinkat-least-once delivery made effect-once by the commit-LSN watermark: data + checkpoint in ONE destination transaction, and the commit_lsn <= checkpoint skip IS the dedup (Critical Rule 3)
ReplicationPipeline.Sink (receipts)the value-free cdc_receipts ledger (commit_lsn/schema/table/op, never a value) exists exactly when its transaction took effect — re-delivery is skipped whole
ReplicationPipeline.Sink (session identity)the first connect BINDS the source's {system_identifier, database} into the checkpoint row; every later connect COMPARES — a rebuilt source container HALTS the pipeline instead of silently resuming a different database (ADR-0007)
ReplicationPipeline.Sink (upsert)unchanged-TOAST columns are OMITTED from the upsert SET — the sentinel never appears in record (Critical Rule 4)
docker-compose.yml (source flags)wal_level=logical is the load-bearing flag; the publication is the operator's SQL and a missing one fails closed

The demo is intentionally minimal (one table, id PK). Real deployments extend the sink per table — the seam is the point, not the schema.

CI

The reference-example CI job builds this stack every push and drives the full sequence: the example's own static gates (format, compile-warnings, credo --strict, dialyzer), then insert → replica + receipts + checkpoint assertions, TOAST-sentinel survival, and restart-resume with zero duplicate receipts — so the example can never silently rot out of sync with replicant's public sink API.