Connectors / Change data capture

PostgreSQL CDC

source

PostgreSQL CDC source (logical replication)

Install

cargo install faucet-cli --features source-postgres-cdc
faucet init --source postgres-cdc

# Cargo.toml — use it as a library
cargo add faucet-source-postgres-cdc

Starter config

source:
  type: postgres-cdc
  config:
    connection_url: ""    # REQUIRED — Connection URL pointing at the database whose WAL we want to read.
    slot_name: ""    # REQUIRED — Logical replication slot name.
    publication_name: ""    # REQUIRED — Publication name on the server.
    # create_slot_if_missing: true    # If the slot does not exist, create it as a logical/`pgoutput` slot at connection time.
    # slot_type: null    # Lifetime of a newly-created replication slot.
    # slot_acquire_retries: 10    # Number of times to retry acquiring the replication slot when the server reports it is still **active** (held by a not-yet-released prior connection).
    # tls: { type: disable }    # one of: disable, require, verify_ca, verify_full, from_url, prefer
      # --- Alternative variants — replace the block above with one of these ---
      # type: require
      #
      # type: verify_ca
      # ca_path: ""
      #
      # type: verify_full
      # ca_path: ""
      #
      # type: from_url
      #
      # type: prefer
    # start_lsn: null    # Optional starting LSN override (e.g.
    # proto_version: 1    # pgoutput protocol version.
    # idle_timeout: 30    # Maximum time to wait for new replication messages before returning the current batch.
    # max_messages: null    # Optional cap on the number of change events drained per fetch call.
    # max_cycle_duration: 300    # Longest a fetch cycle runs before it ends at the next transaction boundary.
    # max_staged_records: 1000000    # Maximum number of change records buffered in memory for a *single* in-progress transaction before it is aborted.
    # max_buffered_events: 100000    # Replication events read ahead while the pipeline is busy writing a page.
    # max_buffered_bytes: 268435456    # Upper bound, in bytes of decoded change data, on the read-ahead buffer described under `max_buffered_events`; reading pauses at whichever limit is reached first.
    # status_update_interval: 10    # Interval at which Standby Status Update keepalives are sent to the server.
    # tcp_keepalive: 60    # Accepted for compatibility and ignored: the replication client sets no socket keepalive (a non-default value logs a warning).
    # batch_size: 1000    # Advisory page size for `Source::stream_pages`.

Capabilities

Replay	Deterministic replay
Exactly-once replay	Yes
Incremental	Yes
Dataset discovery	No
Reports lag	Yes
Compression	No

Config fields

Field	Type	Default	Description
`connection_url` required	`string`	`—`	Connection URL pointing at the database whose WAL we want to read. The crate internally upgrades the connection to `replication=database` — callers do not need to add it themselves.
`slot_name` required	`string`	`—`	Logical replication slot name. Must match the Postgres naming rules: 1–63 chars, lowercase letters / digits / underscores only.
`publication_name` required	`string`	`—`	Publication name on the server. Must already exist (faucet does not create publications — they're a DBA-level concern that determines which tables are replicated).
`create_slot_if_missing`	`boolean`	`true`	If the slot does not exist, create it as a logical/`pgoutput` slot at connection time. Default: `true`.
`slot_type`	`object`	`permanent`	Lifetime of a newly-created slot. Only `permanent` (the default) is accepted: a `temporary` slot is dropped with the session that creates it, before replication can start on its own connection, so it is refused at config load.
`slot_acquire_retries`	`integer`	`10`	Number of times to retry acquiring the replication slot when the server reports it is still active (held by a not-yet-released prior connection). On a rapid restart — a scheduler or `serve` re-running the pipeline before the previous backend has dropped the slot — both the pre-stream `pg_replication_slot_advance` and `START_REPLICATION` fail with "replication slot … is active for PID …". Each retry waits an exponentially increasing backoff (250 ms, doubling, capped at 4 s). `0` disables retries (fail fast). Defaults to 10.
`tls`	`object`	`mode: from_url`	TLS settings for the replication connection. Default `from_url`: the `sslmode` (and `sslrootcert`) in `connection_url` apply to both the control-plane and the replication connection; without one the connection is plaintext and a warning is logged. An explicit mode that contradicts the URL's `sslmode` is refused (#789 SUPPLY-06).
`start_lsn`	`string`	`—`	Optional starting LSN override (e.g. `"0/16A4F88"`). Ignored when a state-store-managed bookmark is present (that bookmark wins). When neither is set, replication starts from the slot's `confirmed_flush_lsn`.
`proto_version`	`integer`	`1`	pgoutput protocol version. Only `1` is fully exercised in v1; `2` is accepted but streaming-transaction messages (S/E/c/A) are not yet decoded. Default: `1`.
`idle_timeout`	`integer`	`30`	Maximum time to wait for new replication messages before returning the current batch. Default: 30 s.
`max_messages`	`integer`	`—`	Optional cap on the number of change events drained per fetch call. Acts as a safety bound — `idle_timeout` is the primary terminator.
`max_cycle_duration`	`integer`	`300`	Longest a fetch cycle runs before it ends at the next transaction boundary. Default: 300 s; `0` means no bound.
`max_staged_records`	`integer`	`1000000`	Maximum number of change records buffered in memory for a single in-progress transaction before it is aborted.
`max_buffered_events`	`integer`	`100000`	Replication events read ahead while the pipeline is busy writing a page. Reading ahead keeps the replication client sending its status updates, so a long sink write does not let the server's `wal_sender_timeout` (default 60 s) end the connection. Past this many buffered events reading pauses until the pipeline catches up — keep a page write well under `wal_sender_timeout`, or raise it, if a page write can outlast that much change volume. Default 100 000.
`max_buffered_bytes`	`integer`	`268435456`	Upper bound, in bytes of decoded change data, on the read-ahead buffer described under `max_buffered_events`; reading pauses at whichever limit is reached first. This memory is held in addition to `max_staged_records`. Default 268 435 456 (256 MiB).
`status_update_interval`	`integer`	`10`	Interval at which Standby Status Update keepalives are sent to the server. Must be shorter than `idle_timeout` and well under the server's `wal_sender_timeout` (default 60 s). Default: 10 s.
`tcp_keepalive`	`integer`	`60`	Accepted for compatibility and ignored: the replication client sets no socket keepalive (a non-default value logs a warning). A dead replication connection is detected through `status_update_interval` and the server's `wal_sender_timeout`. Default: 60 s.
`batch_size`	`integer`	`1000`	Advisory page size for `Source::stream_pages`. The CDC source emits one `StreamPage` per committed transaction so the pipeline gets per-transaction durability via its per-page bookmark persist. Because transactions are atomic units they are never split across pages — a single transaction whose record count exceeds `batch_size` still emits as one page. Defaults to `DEFAULT_BATCH_SIZE`.

Get started

Your first pipeline runs in five minutes.

Install the CLI, scaffold a config, and move real data — nothing external to stand up.

curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh

brew install faucet-hq/faucet-stream/faucet-cli

Linux & macOS Homebrew

Read the guide → Star on GitHub