replicant usage rules

Copy Markdown View Source

A framework-agnostic Elixir CDC consumer for Postgres logical replication (pgoutput).

What replicant is (and is not)

  • Is: a reliable CDC consumer. Decodes the pgoutput logical replication stream, assembles committed transactions, validates schema changes, and delivers each transaction to a pluggable sink with sink-owned, transaction-granularity exactly-once semantics.
  • Is not: Ash-aware, tenant-aware, or classification-aware. Never put multitenancy or sensitive-data logic here — that is ash_replicant's job.
  • Is: a live streaming client. Owns the replication slot via Postgrex.ReplicationConnection, acks only after the sink durably commits (ack-after-checkpoint), and halts fail-closed on slot invalidation — proven by a real-PG16 crash-injection suite (loss = 0, effect-dup = 0).

Public surface

  • Replicant — the facade module. start_link/1 starts a supervised streaming pipeline (validates opts, enforces the go-forward guard) and stop/1 tears one down; t:lsn/0 (a non_neg_integer 64-bit LSN, (file <<< 32) ||| offset) and lsn_to_string/1 (uppercase "file/offset" hex display, matching Postgres pg_lsn).
  • Replicant.Transaction — an assembled, committed transaction: ordered changes plus the transaction's single commit_lsn, and (when messages: true is enabled) any transactional logical-decoding messages in messages (a [Replicant.Decoder.Messages.Message.t()] — these ride the txn and inherit its commit_lsn effect-once dedup). Ordinarily changes is a List; for an oversized spilled streamed transaction (opt-in streaming: [spill: [dir: …, max_spill_bytes: …]]) it is a lazy, single-pass, disk-backed Enumerable (Replicant.Spill.Reader) valid only during the handle_transaction/1 / handle_batch/1 call — iterate it with Enum/Stream, never length/1 / Enum.to_list/1 (which force the whole transaction back into RAM, defeating spill), and do not retain it past the call. Spill emits value-free [:replicant, :stream, :spilled] (a tail flushed to disk) and, when disk usage would exceed max_spill_bytes, [:replicant, :stream, :spill_exhausted] before the fail-closed halt — attach to alert operators on exhaustion.
  • Replicant.Change — a single row change (insert/update/delete), the decoded record, and the unchanged list of TOASTed columns the source UPDATE did not touch (never a value — sinks must leave those columns alone).
  • Replicant.SchemaChange — a detected DDL-shape change (column add/drop, type change, replica-identity change). Destructive changes halt fail-closed.
  • Replicant.Sink — the behaviour a consumer implements. In the default sink-owned transaction mode (batch_delivery not set), receives one Replicant.Transaction at a time and must durably persist it (or raise) before the slot advances past its commit_lsn. When sink-owned batch delivery is enabled (batch_delivery: [max_transactions: N, max_delay_ms: T]), receives N committed transactions in a single handle_batch/1 call and must persist all rows
    • checkpoint atomically — the transaction is atomic and the effect-once guarantee (dup=0, loss=0) rests on it. Transactions arrive in ascending commit_lsn order; the sink must skip any commit_lsn <= checkpoint and upsert rows by table PK. A non-{:ok, _} return (or a raise/throw/exit) halts the pipeline fail-closed; the batch is discarded un-acked and re-delivered on resume, deduped to zero net effect by the idempotent sink. Optional snapshot callbacks (handle_snapshot/2 + handle_snapshot_complete/1) let a sink be bootstrapped from a populated source: batches of %Change{op: :snapshot} rows (upsert by PK; clear the table when first_for_table? is true — a hard redo-safety obligation), then a durable handoff checkpoint at the snapshot's consistent point. In lib mode (:checkpoint_store configured) the library owns the checkpoint, so a sink implements only handle_transaction/1 — checkpoint/0 is optional there; its returned LSN is ignored. Batch delivery is mutually exclusive with lib mode.
  • handle_message/2 (opt-in, messages: true) — delivers a non-transactional logical-decoding message (pg_logical_emit_message with transactional => false). The guarantee is at-least-once, NOT effect-once (see the rule below): a reconnect between the {:ok} return and the checkpoint advancing can re-deliver the same message. On :ok/{:ok, _} the library advances the checkpoint to the message's LSN; a non-:ok return (or raise/throw/exit) halts fail-closed. context carries %{lsn: lsn}. A messages: true config whose sink lacks this callback is rejected at start (:messages_unsupported). Transactional messages do NOT route here — they ride %Transaction.messages and inherit the txn path's effect-once.
  • Incremental snapshot (snapshot: [mode: :incremental]) — a resumable, chunked backfill for large tables, interleaved with the live stream. Chunks arrive through the SAME handle_snapshot/2 (same first_for_table? redo-safety obligation; handle_snapshot_complete/1 is NOT used — completion rides a dedicated final handle_snapshot/2 call). A sink-owned incremental sink must ALSO implement snapshot_progress/0 (return the opaque ctx.progress token it persisted atomically with each chunk); lib mode carries progress in the checkpoint store, so only handle_snapshot/2 is required. A concurrent write to a backfilling row wins over its stale chunk row (collision-corrected); PK-less tables fall back to a bounded whole-table redo.
  • Replicant.Decoder — decode/1 wraps the vendored pgoutput byte parser; catches and redacts any raise into a value-free Replicant.Error.
  • Replicant.Assembler — groups decoded messages into Replicant.Transactions by commit_lsn.
  • Replicant.Connection — the Postgrex.ReplicationConnection that owns the replication slot: ack-after-checkpoint keepalive replies, async ack, slot-invalidation fail-closed halt, the bounded in-flight window, and the replication-command-error watchdog (below).
  • Replicant.AssemblerServer — the serial process that applies the sink synchronously off the keepalive path.
  • Replicant.Pipeline — the per-slot :one_for_all supervisor pairing a Connection with an AssemblerServer; started by Replicant.start_link/1.
  • Replicant.Config — validates pipeline options + the start-mode guard (go_forward_only / snapshot / resume; go_forward_only + snapshot both set is refused as :conflicting_start_mode). publication: accepts a single validated name or a list (["p1", "p2"]) for multi-publication; every name is identifier-validated, the connect chain fails closed if any requested pub is absent (:publication_check), and messages: true is the opt-in for logical-decoding messages (rejected :messages_unsupported if the sink lacks handle_message/2).
  • Replicant.Snapshotter — reads a consistent snapshot of the publication's tables at the EXPORT_SNAPSHOT LSN (a REPEATABLE READ cursor on a separate connection) and pushes %Change{op: :snapshot} batches to the sink, behind a value-free error boundary; the Connection hands off to streaming at that LSN.
  • Replicant.CheckpointStore — in lib mode (a :checkpoint_store option is present), a supervised GenServer owning one replicant_checkpoints row per slot: the library writes the checkpoint (commit_lsn bigint) to this durable Postgres table after the sink persists, so a non-transactional sink (files, S3, Kafka, external APIs) needs no atomic data+checkpoint unit. Value-free boundary; lazy table create + shape-probe. A store fault is bounded by two :checkpoint_store knobs — max_retries (default 5) and retry_backoff_ms (default 1000) — shared by both fault sites: a transient connect-read fault paces N fresh reconnects, a transient mid-stream write fault retries N (blocking the applier, so dup-bounded-to-one holds), then both halt fail-closed on exhaustion (Supervisor.halt, loss = 0). A permanent fault (schema mismatch / :config_invalid) halts immediately; max_retries: 0 opts out of retry (halt-now). Each retry emits [:replicant, :checkpoint_store, :retrying].
  • Replication-command-error watchdog — the top-level max_command_retries option (default 5) bounds a persistent PRE-FRAME command error — CREATE_REPLICATION_SLOT failing because the server's replication slots are exhausted, a slot already active for another consumer, or a forward-incompatible result shape — which otherwise reconnects forever via auto_reconnect. After max_command_retries failed connect cycles without the stream establishing, the pipeline halts fail-closed and stays idle, emitting value-free [:replicant, :connection, :command_error_halt] (attempt/max_retries/ slot_name); max_command_retries: 0 halts on the first fault. The bound is a CYCLE count, not a wall-clock time. Sibling of the store-retry bound above but a separate budget (a store outage never trips it). Only PRE-FRAME errors are bounded: once the stream is flowing the counter resets on the first replication frame, so a later transient outage self-heals, and a server that is simply down (connection refused, never establishes) keeps retrying untouched.
  • Replicant.QueryBuilder — builds the identifier-validated SQL used to create/manage slots and publications.
  • Replicant.Identifier — allowlist validation for slot, publication, and other identifiers that reach SQL.
  • Replicant.Telemetry — value-free :telemetry spans (LSNs, table names, counts, durations, error classes — never row values).
  • Replicant.Error — the typed, value-free error struct raised/returned at decode and validation boundaries.

Non-negotiable rules

  • No row value in an error, log, or telemetry event. Assume every value is PII or a secret. Column names are strings, never atoms (String.to_atom on a wide or attacker-influenced schema exhausts the atom table).
  • Validate identifiers. Slot and publication names go through Replicant.Identifier.validate/1 before reaching SQL. A failure carries the invalid-shape fact only, never the offending string.
  • Exactly-once is at-least-once + a transaction-watermark-idempotent sink. The watermark is the commit LSN at transaction granularity — skip any transaction whose commit_lsn <= checkpoint; upsert rows by table PK. There is no naked exactly-once without two-phase commit or an idempotent sink; never claim one.
  • Lib mode is at-least-once, never effect-once. With a :checkpoint_store, the library writes the checkpoint after the sink persists (checkpoint-after-persist), so a crash between persist and checkpoint re-delivers exactly one transaction on resume: duplicate bounded to one transaction, never loss. A non-transactional sink cannot dedup — do not claim effect-once for it.
  • Batching is opt-in and lib-mode only. checkpoint_store: [batch: [max_transactions: N, max_delay_ms: T]]. It batches the checkpoint write + ack, NOT sink delivery (handle_transaction/1 is still per-transaction). A crash or graceful stop mid-batch re-delivers up to one batch — size max_transactions for your dup tolerance. Do not set :batch at the top level (it belongs under :checkpoint_store; a misplaced top-level :batch is rejected at start).
  • Sink-owned atomic batch delivery (handle_batch/1) preserves effect-once. The batch_delivery: [max_transactions: N, max_delay_ms: T] config (top-level, sink-owned only; mutually exclusive with :checkpoint_store) routes delivery through handle_batch/1 instead of handle_transaction/1. The HARD OBLIGATION is that the data + checkpoint write is ATOMIC — the effect-once guarantee (dup=0 across mid-batch teardown) rests on it. Transactions arrive in ascending commit_lsn order; the sink must skip any commit_lsn <= checkpoint and upsert rows by table PK, exactly as handle_transaction/1 does. A non-{:ok, _} return (or a raise/throw/exit) halts the pipeline fail-closed; the batch is discarded un-acked and re-delivered on resume, deduped to zero net effect by the idempotent sink.
  • Failover slots are opt-in and PG17+. failover: true creates the replication slot with Postgres's FAILOVER option so it syncs to physical standbys, letting a pipeline resume against a promoted standby with zero loss (the slot's confirmed_flush position carries over). PG16 does not support FAILOVER slots — passing failover: true there halts fail-closed ({:config, :failover_unsupported}) instead of silently ignoring the option. Never point Replicant at an unpromoted standby's synced slot — it halts fail-closed ({:slot_synced_unpromoted}) instead of retry-looping against a slot it cannot yet consume.
  • Multi-publication is opt-in via a list. publication: "p" stays the byte-unchanged default; publication: ["p1", "p2"] streams the union. Every requested publication must exist — a missing one halts fail-closed at connect (:publication_check) rather than silently streaming the subset (a START_REPLICATION that names a missing pub streams only the found set). pgoutput de-dupes overlapping tables across pubs on the wire.
  • Logical-decoding messages: state the guarantee honestly. messages: true is opt-in. A transactional message (transactional => true to pg_logical_emit_message) rides %Transaction.messages and is effect-once (inherits the txn commit_lsn dedup). A non-transactional message routes to handle_message/2 and is at-least-once — duplicates are possible on reconnect (no dedup key). Never claim effect-once for the non-transactional path. A message's content and prefix are user bytes (Critical Rule 1) — never log them or surface them in telemetry.
  • Unchanged TOAST is a sentinel, not a value. It surfaces only as Replicant.Change's unchanged list of column names, never in record. Sinks must leave those columns untouched on upsert.
  • Stay tenant-blind. No multitenancy, scope, or classification logic belongs here — that boundary is the whole reason replicant and ash_replicant are separate libraries.

See AGENTS.md for the full working rules.