Replicant

A 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.

Status: v1 is complete and production-hardened. Replicant owns the replication slot via Postgrex.ReplicationConnection, acks only after the sink durably commits (ack-after-checkpoint), halts fail-closed on slot invalidation, and is proven by a real-PG16 crash-injection suite (loss = 0, effect-dup = 0). As of v0.2.0, initial snapshot/backfill (incl. a resumable incremental mode), a lib-owned checkpoint store for non-transactional sinks, batched checkpointing, sink-owned atomic batch delivery, in-progress-transaction streaming, consumer-side disk spill for oversized transactions, multi-publication per pipeline, and logical-decoding messages have all shipped. See "How it streams" below.

Highlights

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:

Replicant.lsn_to_string(0x16E3778)

#=> "0/16E3778"

Use Replicant.lsn_to_string/1 for display; LSNs are WAL positions, not row data, so they are permitted in telemetry metadata. The exactly-once watermark check is plain integer comparison: txn.commit_lsn <= checkpoint.

How it streams

A running pipeline is two processes under a :one_for_all supervisor:

Because the ack reports the durable checkpoint while a transaction is in flight (advancing over filtered WAL only when idle, which carries no publication data), a crash between dispatch and persist re-delivers from the durable confirmed_flush and the idempotent sink dedups — the exactly-once seam that walex's fire-and-forget wal_end + 1 ack does not have.

PostgreSQL version support

Replicant targets PostgreSQL 16 as the tested baseline and is forward-compatible with 17+. On PG17+ it reads the authoritative invalidation_reason slot column (a superset of the PG16 wal_status/conflicting signals) and supports failover slots for HA.

Failover slots (PG17+)

Pass failover: true to Replicant.start_link/1 to create the replication slot with the FAILOVER option, so PostgreSQL syncs it to physical standbys:

Replicant.start_link(

  connection: [hostname: "primary.internal", ...],

  slot_name: "replicant_orders",

  publication: "orders_pub",

  sink: MyApp.OrdersSink,

  failover: true            # PG17+ only; on PG16 the pipeline halts {:config, :failover_unsupported}

)

After a failover, repoint the connection at the promoted primary — the slot already exists there with its confirmed_flush position, so replication resumes with zero loss. Slot sync (sync_replication_slots), promotion, and DNS/endpoint changes are operator concerns. Do not point Replicant at an unpromoted standby's synced slot — it cannot be consumed until promotion, and Replicant halts fail-closed ({:slot_synced_unpromoted}) rather than looping.

The 5 critical rules (see AGENTS.md for the full text)

  1. No row value in an error, log, or telemetry event.
  2. Validate identifiers before they reach SQL.
  3. Exactly-once is at-least-once + a transaction-watermark-idempotent sink — never claim a naked exactly-once.
  4. Unchanged TOAST is a sentinel, not a value — never overwrite it.
  5. Stay tenant-blind — multitenancy and classification live in ash_replicant, never here.

Installation

def deps do

  [

    {:replicant, "~> 0.2"}

  ]

end

Interactive tour (Livebook)

Run in Livebook

notebooks/getting_started.livemd is a runnable, self-verifying tour: it starts a live pipeline against a local Postgres, streams INSERT/UPDATE/DELETE through a tiny sink you can watch, then demonstrates the unchanged-TOAST sentinel, transaction-granularity exactly-once, snapshot/backfill, and logical-decoding messages. Click the badge to open it in Livebook, or read it rendered on HexDocs. The notebook's code is executed against a live PG16/PG17 on every CI run (test/integration/livebook_getting_started_test.exs), so it never drifts from the library.

Usage

Start a pipeline against a standby with Replicant.start_link/1, pointing it at a sink that implements checkpoint/0 + handle_transaction/1:

Replicant.start_link(

  connection: [hostname: "standby.internal", port: 5432, username: "u",

               password: "p", database: "orders", ssl: true],

  slot_name: "replicant_orders",

  publication: "orders_pub",

  sink: MyApp.OrdersSink,

  go_forward_only: false

)



defmodule MyApp.OrdersSink do

  @behaviour Replicant.Sink



  @impl true

  def checkpoint, do: {:ok, MyApp.Repo.last_committed_lsn()}



  @impl true

  def handle_transaction(%Replicant.Transaction{commit_lsn: lsn} = txn) do

    # In ONE DB transaction: skip if lsn <= checkpoint, else upsert txn.changes

    # by table PK and persist lsn as the new checkpoint. Then:

    {:ok, lsn}

  end

end

Start modes. A :state_mirror sink starting from an empty checkpoint must declare its intent — go_forward_only: true (stream only new changes), or snapshot: true (backfill the current state, then hand off to streaming at the snapshot LSN with zero gap and zero duplication). A non-empty checkpoint simply resumes. snapshot: true requires the sink to also implement handle_snapshot/2 (batch upsert; clear the table on first_for_table?) and handle_snapshot_complete/1 (durably persist the handoff checkpoint); a mid-snapshot crash halts fail-closed (:snapshot_incomplete) for an operator to drop the slot and retry.

Resumable incremental snapshot. For large tables where an all-or-nothing snapshot: true (a crash at 99% of a multi-hour COPY redoes everything) is the adoption blocker, opt into the incremental mode instead:

snapshot: [mode: :incremental, chunk_rows: 1000, max_pending_chunks: 4]

It creates a durable slot and streams immediately, while a linked reader backfills the publication's tables in PK-ordered keyset chunks on its own connection — each chunk consistency-bracketed by read-only LSN watermarks and collision-corrected against the live stream (a concurrent write to a backfilling row wins; the stale chunk row is dropped). Progress is durable per chunk, so a crash, halt, or reconnect resumes from the last applied chunk rather than restarting. Chunks arrive through the same handle_snapshot/2 callback; sink-owned mode gives effect-once chunks, lib mode dup ≤ 1 chunk (never loss). PK-less tables fall back to a bounded whole-table redo. snapshot: true remains the point-in-time option and is untouched.

Lib-owned checkpoint (non-transactional sinks). Pass a :checkpoint_store ([connection: <postgrex opts>, table: "replicant_checkpoints"]) to flip the pipeline into lib mode: the library writes the checkpoint to a durable Postgres table after the sink persists (checkpoint-after-persist), so a non-transactional sink (files, S3, Kafka, external APIs) needs to implement only handle_transaction/1. The guarantee is at-least-once, duplicate bounded to one transaction, never loss — not effect-once (a non-transactional sink cannot dedup). A store outage (connect-read or mid-stream write) is bounded: the pipeline retries max_retries times (default 5) retry_backoff_ms apart (default 1000 ms — ~5s of outage tolerated) then halts fail-closed; a permanent fault (schema mismatch / invalid config) halts immediately.

Sink-owned atomic batch delivery (transactional sinks). For a transactional sink that can persist multiple rows + checkpoint in one database transaction, pass a top-level batch_delivery: [max_transactions: 100, max_delay_ms: 1000] to accumulate committed transactions and deliver them as a batch:

Replicant.start_link(

  connection: [hostname: "standby.internal", port: 5432, username: "u",

               password: "p", database: "orders", ssl: true],

  slot_name: "replicant_orders",

  publication: "orders_pub",

  sink: MyApp.OrdersSink,

  batch_delivery: [max_transactions: 100, max_delay_ms: 1000],

  go_forward_only: false

)



defmodule MyApp.OrdersSink do

  @behaviour Replicant.Sink



  @impl true

  def checkpoint, do: {:ok, MyApp.Repo.last_committed_lsn()}



  @impl true

  def handle_batch(transactions) do

    # In ONE DB transaction: skip any commit_lsn <= checkpoint, else upsert all

    # rows from all transactions by table PK, and persist the batch's highest

    # commit_lsn as the new checkpoint. Then:

    {:ok, List.last(transactions).commit_lsn}

  end

end

The batch flushes when it reaches max_transactions transactions, after max_delay_ms milliseconds idle, or when the batch's WAL span (LSN-span lag) hits an auto-derived safety cap (derived from :max_inflight_lag). Because the rows + checkpoint write is atomic, effect-once is preserved (dup=0, loss=0) — stronger than lib-mode's checkpoint_store: [batch: …] which is per-transaction delivery with batched checkpointing. The sink must implement both checkpoint/0 (resume) and handle_batch/1; it cannot use checkpoint_store, and any handle_transaction/1 implementation is ignored. Emits [:replicant, :sink, :batch_committed] telemetry once per flush.

Consumer-side disk spill (oversized transactions). By default a single in-progress streamed transaction is bounded by the in-flight window: one larger than max_inflight_lag halts fail-closed. Opt into disk spill to reassemble such a transaction partly on disk and still deliver it effect-once:

Replicant.start_link(

  connection: [...], slot_name: "replicant_orders", publication: "orders_pub",

  sink: MyApp.OrdersSink, go_forward_only: false,

  max_inflight_lag: 64 * 1024 * 1024,

  streaming: [

    max_concurrent_txns: 64,

    spill: [dir: "/var/lib/replicant/spill", max_spill_bytes: 1024 * 1024 * 1024]

  ]

)

A transaction whose resident bytes cross max_inflight_lag spills its oldest changes to a per-txn file under dir; at commit it is delivered as a lazy, single-pass, disk-backed%Transaction{} whose changes streams the spilled frames + the resident tail. There are two ceilings: the resident RAM bound max_inflight_lag (the spill trigger) and the disk bound max_spill_bytes (a transaction exceeding it halts :spill_exhausted). Defaults: max_spill_bytes is 16 × max_inflight_lag; dir is a 0700 subdir of the OS temp dir.

Delivery obligation. A spilled transaction's changes is a single-passEnumerable valid only during the handle_transaction/1 (or handle_batch/1) call — iterate it with Enum/Stream and do not call length/1, Enum.to_list/1, or re-iterate it (any of which forces the whole transaction back into RAM, defeating spill), and do not retain it past the call. The usual List-backed changes still works exactly as before; only an oversized spilled transaction delivers the lazy form.

Operator guidance. Spill files are ephemeral non-fsync'd scratch (0600, value-free on fault), deleted on commit/abort/reset/halt and swept per-slot on (re)connect. Replicant does not encrypt them — if the source rows are sensitive, point dir at an encrypted/secure volume; a custom persistent dir is yours to clean on decommission (the default OS temp dir is cleared by the OS).

Multi-publication per pipeline.publication: accepts a single name (the default) or a list. Pass a list to stream the union of several publications through one slot — every name is validated and deduplicated, and overlapping tables are de-duped by pgoutput on the wire:

Replicant.start_link(

  connection: [...],

  slot_name: "replicant_orders",

  publication: ["orders_pub", "audit_pub"],   # was: publication: "orders_pub"

  sink: MyApp.OrdersSink,

  go_forward_only: false

)

Every requested publication must exist; a missing one halts fail-closed at connect time rather than silently streaming the subset (a START_REPLICATION that names a missing publication would otherwise stream only the found subset).

Logical-decoding messages (pg_logical_emit_message). Opt into Postgres logical-decoding messages with messages: true to receive outbox-pattern / heartbeat rows emitted via pg_logical_emit_message. The durability guarantee depends on the message's transactional flag (stated honestly in the handle_message/2 docs):

Replicant.start_link(

  connection: [...],

  slot_name: "replicant_orders",

  publication: "orders_pub",

  sink: MyApp.OrdersSink,

  messages: true,                            # opt-in; sink MUST implement handle_message/2

  go_forward_only: false

)



defmodule MyApp.OrdersSink do

  @behaviour Replicant.Sink



  @impl true

  def checkpoint, do: {:ok, MyApp.Repo.last_committed_lsn()}



  @impl true

  def handle_transaction(%Replicant.Transaction{commit_lsn: lsn, messages: msgs} = txn) do

    # msgs :: [Replicant.Decoder.Messages.Message.t()] — transactional messages ride the

    # txn and inherit its commit_lsn effect-once dedup. Persist them with the txn's rows.

    {:ok, lsn}

  end



  @impl true

  # context :: %{lsn: lsn}; non-transactional messages ONLY (transactional ride handle_transaction/1).

  # At-least-once: a reconnect can re-deliver — your effect must be idempotent. The message's

  # `content`/`prefix` are USER BYTES (Critical Rule 1: never log or surface them in telemetry).

  def handle_message(%Replicant.Decoder.Messages.Message{} = msg, %{lsn: lsn}) do

    :ok

  end

end

messages: true requires the sink to implement handle_message/2; a sink missing it is rejected at start (:messages_unsupported) rather than silently dropping non-transactional messages later.

Development

mix deps.get

mix test

mix quality   # format --check-formatted + credo --strict + dialyzer

Contributor and agent working rules — including the redaction, identifier-validation, and tenant-blind invariants — live in AGENTS.md.

Roadmap

The v1 CDC core and every delivery slice have shipped and are closeout-reviewed against a real-PG16 crash-injection suite:

The sibling consumer library has also shipped, one layer up from this core:

Credits

License

MIT — see LICENSE. Third-party attributions in NOTICE.