Modules
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.
Assembles a stream of decoded pgoutput messages into a Replicant.Transaction
and applies the sink synchronously per transaction (spec §4/§6).
The serial process shell over the pure Replicant.Assembler (spec §4). Receives
decoded pgoutput messages from Replicant.Connection (which decodes behind
the value-free boundary and never applies the sink), assembles transactions, and
applies the sink synchronously — blocking THIS process, off the Connection's
keepalive path. On a durable sink commit (or a watermark skip) it messages the
Connection so the ack advances asynchronously; on a fail-closed condition
(destructive schema change, sink WRITE fault, an unidentifiable-relation row) it
halts the whole pipeline permanently (spec §6/§9).
Parser for PostgreSQL array literals
Cast from Postgres to Elixir types
One row change within a Replicant.Transaction.
Per-column metadata carried on a Change (mirrors the Relation message).
The lib-owned checkpoint store (spec §7) for non-transactional sinks. A GenServer
over a normal Postgrex connection that owns one replicant_checkpoints row per
slot and reads/writes commit_lsn behind the value-free boundary (Critical Rule
1 — the store carries slot_name + LSN only; a Postgrex fault is scrubbed to a
structural %Replicant.Error{}, never echoing a parameter).
Validates Replicant.start_link/1 options and enforces the go-forward-only
start guard: a :state_mirror sink resuming from an empty
checkpoint without go_forward_only: true would silently deliver partial data
from the slot's creation point, so it is refused at start.
The Postgrex.ReplicationConnection that owns the replication slot and closes
the exactly-once seam (spec §2/§4/§8). It
Binary decoder for the Postgres logical-replication pgoutput stream.
Decoded representations of each pgoutput replication message.
B — start of a transaction; carries the final LSN, commit timestamp, and xid.
C — end of a transaction; carries the commit LSN and timestamp.
D — a row delete; carries either the key or the full old tuple, depending on REPLICA IDENTITY.
I — a row insert; tuple_data holds the new values. xid is the (sub)transaction id when streamed (spec §5), else nil.
O — origin marker emitted when a transaction has a logical replication origin.
R — relation (table) metadata: identifier, schema, name, replica identity, columns.
Per-column metadata within a Relation message.
A — aborts a streamed (sub)transaction (spec §5); subxid == xid is a whole-transaction abort, else a savepoint rollback.
c — commits a streamed transaction (spec §5); carries the commit LSN.
S — begins/continues a streamed transaction's segment (spec §5); carries the top-level xid and first-segment flag.
E — ends a streamed transaction's segment (spec §5); no fields.
T — a TRUNCATE of one or more relations; options indicate CASCADE / RESTART IDENTITY.
Y — a type registration for a non-built-in OID.
Catch-all for replication messages Replicant does not yet decode; carries the raw binary.
U — a row update; carries new values and, if available, the key or old tuple.
Maps a numeric PostgreSQL type ID to a descriptive string.
A typed, value-free error.
Allowlist validation for Postgres identifiers replicant interpolates into slot
and publication SQL (Critical Rule 2). A failure carries the invalid-SHAPE fact
only, never the offending string — it may be attacker-controlled.
Per-pipeline Supervisor (spec §4): supervises the Replicant.AssemblerServer
and the Replicant.Connection under :one_for_all — a crash of either
restarts both together, so a fresh Connection (resuming from the sink checkpoint)
is never paired with a stale in-memory assembler buffer. Registered by slot in
Replicant.Registry so Replicant.Supervisor.halt/2 can terminate the whole
pipeline permanently on a fail-closed condition.
Slot and publication SQL/command strings built from validated identifiers
(Critical Rule 2). Hardens walex's raw '#{publication}' interpolation: every
name passes through Replicant.Identifier.validate/1 before it reaches the
string; an invalid name returns {:error, :invalid_identifier} and builds
nothing. START_REPLICATION/CREATE_REPLICATION_SLOT are replication commands
(no bind parameters), so validated interpolation is the gate.
Classification of a Relation diff (spec §9): additive (new column —
auto-apply) or destructive (dropped column, type change, replica-identity
change — delegate to the sink or halt fail-closed). A Truncate is delivered
as a change, not a schema change.
The pluggable sink contract. A sink durably persists a transaction
AND its checkpoint atomically, returns the commit LSN, and is idempotent on the
transaction-granularity watermark (skip any txn.commit_lsn <= checkpoint;
upsert rows by table PK).
Reads a consistent snapshot of the publication's tables (spec §4) on its own normal
Postgrex connection, at the LSN a durable slot exported via EXPORT_SNAPSHOT, and
pushes the rows to the sink as %Change{op: :snapshot} batches — then durably sets the
sink checkpoint to the snapshot's consistent point (the handoff commit). Spawned and
LINKED to Replicant.Connection, which holds the exported snapshot valid by staying
idle (proven safe past wal_sender_timeout, spec §11). The link binds the snapshotter's
lifetime to the pipeline: a Connection/pipeline teardown mid-snapshot tears the
snapshotter down with it, so an orphan can never mutate the sink after the pipeline is
gone. Graceful completion exits :normal (which never propagates over the link), so the
{:snapshot_done, lsn} / {:snapshot_failed, err} messages still drive the handoff.
The single File.* boundary for consumer disk spill (spec §5/§6). Owns the per-slot spill
directory (0700), the per-top-xid spill file (0600), append of length-prefixed
:erlang.term_to_binary({subxid, %Change{}}) frames, file discard, and the per-slot startup
sweep. Every I/O or serialization fault is scrubbed to a value-free
{:error, %Replicant.Error{reason: :spill_io_failed}} here (Critical Rule 1) — no path, frame,
or row value ever escapes. Spill files are scratch: never fsync'd, deleted on
commit/abort/reset/halt, and (for a prior abnormal exit) swept per-slot on that slot's next start.
A value-free spill exception. Raised by Replicant.Spill.Reader on any File/binary_to_term
fault so the lazy read (which runs inside the sink call, spec §11) never surfaces a raw byte.
Carries ONLY the fixed reason — no path, no frame, no row value (Critical Rule 1).
A lazy, SINGLE-PASS Enumerable over a spilled transaction's changes (spec §5): it streams the
on-disk frames (in commit order) followed by the still-resident in-memory tail, rejects any change
whose subxid is in the aborted set, and stamps commit_lsn + an ascending ordinal at replay
(streamed changes have no LSN before commit). It is value-free BY CONSTRUCTION — a File read or
binary_to_term fault raises Replicant.Spill.Error (fixed reason, no bytes), never a raw-byte
exception, even though the sink forces this enumeration inside its own DB transaction (spec §11).
The top DynamicSupervisor — one child per running pipeline (a
Replicant.Pipeline supervisor). Pipelines run as :temporary children:
a fail-closed halt terminates one permanently (no restart), while transient
crashes are recovered inside each Pipeline's own :one_for_all strategy.
Value-free :telemetry.span/3 wrapper. Owns the metadata allowlist — the single
enforcement point for "no row values in telemetry" (Critical Rule 1). LSNs,
table names, counts, durations, and error classes are permitted; an
off-allowlist key raises rather than shipping a value downstream. Mirrors
arcadic's allowlist pattern.
A decoded, committed transaction: an ordered list of Replicant.Change rows
plus the transaction-granularity commit LSN (every row in a
pgoutput proto-v1 transaction shares one commit LSN — a per-row LSN would
collapse an N-row transaction into one row).