Struct PostgresCdcSourceConfig 

pub struct PostgresCdcSourceConfig {

    pub connection_url: String,
    pub slot_name: String,
    pub publication_name: String,
    pub create_slot_if_missing: bool,
    pub slot_type: SlotType,
    pub tls: CdcTls,
    pub start_lsn: Option<String>,
    pub proto_version: u32,
    pub idle_timeout: Duration,
    pub max_messages: Option<usize>,
    pub max_staged_records: Option<usize>,
    pub status_update_interval: Duration,
    pub tcp_keepalive: Duration,
    pub batch_size: usize,
    pub slot_acquire_retries: u32,
}

Available on crate features source-postgres-cdc and source-rest only.

Configuration for PostgresCdcSource.

Fields

connection_url: 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: String

Logical replication slot name. Must match the Postgres naming rules: 1–63 chars, lowercase letters / digits / underscores only.

publication_name: 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: bool

If the slot does not exist, create it as a logical/pgoutput slot at connection time. Default: true.

slot_type: SlotType

Whether a newly-created slot is permanent (survives disconnect) or temporary (auto-dropped when the replication connection closes).

Default permanent (back-compatible). A permanent slot pins WAL on the server until it is consumed or dropped — an abandoned permanent slot fills pg_wal and can take the whole instance down. Use temporary for ephemeral / test runs (note: a temporary slot resets on reconnect, so bookmark-based resume across runs requires a permanent slot). Drop an unused permanent slot explicitly with PostgresCdcSource::drop_slot.

tls: CdcTls

TLS settings for the replication connection. Default disable (plaintext) for back-compatibility, but credentials and all WAL data then travel unencrypted — set require/verify_ca/verify_full in production.

start_lsn: Option<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: u32

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: Duration

Maximum time to wait for new replication messages before returning the current batch. Default: 30 s.

max_messages: Option<usize>

Optional cap on the number of change events drained per fetch call. Acts as a safety bound — idle_timeout is the primary terminator.

Note: the cap is checked after each COMMIT, never mid- transaction. A single transaction larger than max_messages will still be emitted atomically (the fetch returns only after that transaction’s COMMIT and may produce more records than max_messages). To bound the memory a single in-progress transaction can consume, use max_staged_records instead.

max_staged_records: Option<usize>

Maximum number of change records buffered in memory for a single in-progress transaction before it is aborted.

Logical replication requires a transaction to be buffered until its COMMIT so it can be emitted atomically (partial transactions must never leak downstream). A single bulk UPDATE/DELETE/COPY of millions of rows therefore buffers every decoded row as a serde_json::Value in RAM, which can OOM the process. This bound is a safety valve: when an in-progress transaction’s staged record count exceeds it, the source aborts with a typed FaucetError::Source rather than being OOM-killed.

None (the default) means unbounded — atomic delivery of arbitrarily large transactions at the cost of unbounded memory. Set a value sized to your available memory if you replicate tables subject to large bulk writes.

status_update_interval: Duration

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: Duration

TCP keepalive for the replication connection. Default: 60 s.

batch_size: usize

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.

batch_size = 0 is the “no batching” sentinel: every committed transaction during the run window is accumulated into a single page that is emitted at the end with bookmark = max(commit_lsn). This negates per-transaction durability and is only useful for tests or initial-snapshot style runs.

slot_acquire_retries: u32

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.

Implementations

impl PostgresCdcSourceConfig

pub fn with_batch_size(self, batch_size: usize) -> PostgresCdcSourceConfig

Override the advisory per-page record count emitted by Source::stream_pages.

Pass 0 to disable per-transaction emission — every transaction in the run window will be accumulated into a single trailing page with bookmark = max(commit_lsn). Transactions are never split regardless of batch_size.

pub fn validate(&self) -> Result<(), FaucetError>

Validate fail-fast invariants. Called from PostgresCdcSource::new.

Trait Implementations

impl Clone for PostgresCdcSourceConfig

fn clone(&self) -> PostgresCdcSourceConfig

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for PostgresCdcSourceConfig

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for PostgresCdcSourceConfig

fn deserialize<__D>( __deserializer: __D, ) -> Result<PostgresCdcSourceConfig, <__D as Deserializer<'de>>::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl JsonSchema for PostgresCdcSourceConfig

fn schema_name() -> Cow<'static, str>

The name of the generated JSON Schema.

fn schema_id() -> Cow<'static, str>

Returns a string that uniquely identifies the schema produced by this type.

fn json_schema(generator: &mut SchemaGenerator) -> Schema

Generates a JSON Schema for this type.

fn inline_schema() -> bool

Whether JSON Schemas generated for this type should be included directly in parent schemas, rather than being re-used where possible using the $ref keyword.

impl Serialize for PostgresCdcSourceConfig

fn serialize<__S>( &self, __serializer: __S, ) -> Result<<__S as Serializer>::Ok, <__S as Serializer>::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.