Framework-agnostic Elixir CDC consumer for Postgres logical replication
(pgoutput), delivering committed row changes to a pluggable sink with
zero data loss: the replication slot advances only after the sink has
durably persisted the transaction.
replicant is tenant-blind and classification-blind — the reliable CDC
consumer sibling to arcadic. Multitenancy, classification, and Ash resources
live one layer up, in the ash_replicant sink adapter (published on Hex).
LSN representation
A Postgres LSN is exposed as a single non_neg_integer — the 64-bit value
(xlog_file <<< 32) ||| xlog_offset — so that ordinary integer comparison is
correct WAL ordering, and the same value feeds the wire-level standby status
update. Use lsn_to_string/1 for display ("0/16E3778"); LSNs are WAL
positions, not row data, so they are permitted in telemetry metadata.
Starting a pipeline
Configure each pipeline explicitly at start_link/1 (no global mutable state):
Replicant.start_link(
connection: [hostname: "standby.internal", port: 5432, username: "u",
password: "p", database: "orders", ssl: true], # point at a STANDBY (R-ISO)
slot_name: "replicant_orders",
publication: "orders_pub",
sink: MyApp.OrdersSink, # implements Replicant.Sink
go_forward_only: false # must be true to start a :state_mirror sink from empty
)The Replicant.Connection owns the slot, reports the exact replication-session
identity to the sink before checkpoint lookup, and advances the slot only after
the sink has durably persisted a transaction (ack-after-checkpoint); the
Replicant.AssemblerServer applies the sink synchronously off the keepalive path.
Summary
Functions
Parse a Postgres pg_lsn display string ("0/16E3778") into the uint64 LSN —
the inverse of lsn_to_string/1. The two hex halves are file/offset;
String.to_integer/2 accepts either case (Postgres renders uppercase).
The hex display form of an LSN, e.g. lsn_to_string(0x16E3778) == "0/16E3778".
Postgres displays pg_lsn as uppercase file/offset hex with no padding.
Start a CDC pipeline. Validates opts, enforces the go-forward-only start guard,
and supervises a Replicant.Connection + Replicant.AssemblerServer under the
pipeline DynamicSupervisor.
Stop a running pipeline by slot name (idempotent).
Types
@type lsn() :: non_neg_integer()
A Postgres LSN as a single 64-bit integer (file <<< 32) ||| offset.
Integer comparison (<=, >) is correct WAL ordering, so the exactly-once
watermark is txn.commit_lsn <= checkpoint.
Functions
Parse a Postgres pg_lsn display string ("0/16E3778") into the uint64 LSN —
the inverse of lsn_to_string/1. The two hex halves are file/offset;
String.to_integer/2 accepts either case (Postgres renders uppercase).
Returns {:ok, lsn} or the value-free {:error, :invalid_lsn} (1.3.0: a public
API never raises, so a caller's input can never reach a crash report — Critical
Rule 1; each half must fit 32 bits).
The hex display form of an LSN, e.g. lsn_to_string(0x16E3778) == "0/16E3778".
Postgres displays pg_lsn as uppercase file/offset hex with no padding.
Start a CDC pipeline. Validates opts, enforces the go-forward-only start guard,
and supervises a Replicant.Connection + Replicant.AssemblerServer under the
pipeline DynamicSupervisor.
Options: :connection (a Postgrex connection keyword list — point at a STANDBY
per R-ISO), :slot_name, :publication (both allowlist-validated identifiers),
:sink (a module implementing Replicant.Sink), and :go_forward_only
(default false; must be true to start a :state_mirror sink from an empty
checkpoint — see Replicant.Config).
:snapshot bootstraps the sink from a populated source: snapshot: true is the
point-in-time backfill (EXPORT_SNAPSHOT → COPY → hand off; requires
handle_snapshot/2 + handle_snapshot_complete/1). For large tables, the resumable
form snapshot: [mode: :incremental, chunk_rows: 1000, max_pending_chunks: 4] streams
from a durable slot while a linked reader backfills in PK-ordered keyset chunks through
handle_snapshot/2, resuming from the last applied chunk after a crash/halt/reconnect
(sink-owned mode requires snapshot_progress/0; lib mode carries progress in the
checkpoint store). A sink-owned adapter returns {:ok, :backfill_pending} after it
durably arms a backfill but before the first chunk token exists; Replicant then resumes
discovery from the live slot origin after a pre-chunk crash. Both snapshot modes are
mutually exclusive with go_forward_only: true.
Both modes read only published columns in physical order: pgoutput column lists
on PostgreSQL 15+, pglogical replication-set column lists, or all configured-table
columns for wal2json. Different lists for one table halt
:snapshot_column_list_conflict before snapshot delivery. Unpublished keys are
never added; incremental backfills use a whole-table window if the complete PK
is unavailable. Snapshots select the same columns and rows as the stream,
including generated-column policy and server row filters. PG15+ publication
predicates combine with OR among INSERT publications (an unfiltered INSERT
publication wins); snapshots are INSERT-equivalent. No INSERT publication means
no snapshot rows. pglogical 2.4.8 non-NULL set predicates combine with AND, even
beside an unfiltered set. PG12–15 excludes generated columns; PG16+ uses the
effective publication list. A table with columns but no publishable projection
halts :snapshot_no_publishable_columns naming the table. Zero-column and
all-dropped tables retain the empty reset. ADR-0002 documents PG15's additional
dropped-column stream refusal.
Returns {:ok, pipeline_pid} or {:error, reason} where reason includes
:invalid_identifier / :invalid_sink / :config_invalid / :go_forward_required,
plus the start-gate rejections :conflicting_start_mode, :snapshot_unsupported,
:batch_unsupported, and :messages_unsupported (e.g. messages: true without a
handle_message/2 callback in the sink).
@spec stop(String.t()) :: :ok
Stop a running pipeline by slot name (idempotent).