Replicant (Replicant v0.1.0)

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 and ash_age. Multitenancy, classification, and Ash resources live one layer up, in a future ash_replicant sink adapter.

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 and advances it only after the sink has durably persisted a transaction (ack-after-checkpoint); the Replicant.AssemblerServer applies the sink synchronously off the keepalive path.

Summary

Types

lsn()

A Postgres LSN as a single 64-bit integer (file <<< 32) ||| offset.

Functions

lsn_from_string(string)

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

lsn_to_string(lsn)

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_link(opts)

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(slot_name)

Stop a running pipeline by slot name (idempotent).

Types

lsn()

@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

lsn_from_string(string)

@spec lsn_from_string(String.t()) :: lsn()

lsn_to_string(lsn)

@spec lsn_to_string(lsn()) :: String.t()

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_link(opts)

@spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}

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

Returns {:ok, pipeline_pid} or {:error, reason} where reason is :invalid_identifier / :invalid_sink / :config_invalid / :go_forward_required.

stop(slot_name)

@spec stop(String.t()) :: :ok

Stop a running pipeline by slot name (idempotent).