Struct PostgresCdcSource 

pub struct PostgresCdcSource { /* private fields */ }

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

Implementations

impl PostgresCdcSource

pub async fn new( config: PostgresCdcSourceConfig, ) -> Result<PostgresCdcSource, FaucetError>

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

Drop this source’s replication slot on the server, freeing the WAL it pins. A no-op if the slot doesn’t exist; errors if the slot is still active (in use by a live replication connection). Call this when decommissioning a permanent slot so it doesn’t leak WAL (#78/#12).

Trait Implementations

impl Source for PostgresCdcSource

fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresCdcSource: 'async_trait,

Drain the replication stream into a single Vec<Value> plus the bookmark of the most recent COMMIT.

Implemented by collecting Source::stream_pages with the batch_size = 0 sentinel — that sentinel coalesces every transaction in the run window into a single trailing page, which exactly matches the historical fetch_with_context_incremental contract (one aggregated buffer, one max-LSN bookmark). The streaming pipeline (Pipeline::run / run_stream) drives stream_pages directly with the per-source batch_size config field instead so it gets per-transaction durability.

fn stream_pages<'a>( &'a self, ctx: &'a HashMap<String, Value>, _batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>>

Per-transaction streaming.

Each committed transaction is emitted as its own StreamPage with bookmark = Some(commit_lsn). Because the pipeline flushes the sink and persists the bookmark on every page that carries one, a mid-stream crash recovers from the last fully- committed transaction with no partial-transaction leakage.

Atomic transactions. A transaction is never split across pages. If a single transaction’s record count exceeds PostgresCdcSourceConfig::batch_size it is still emitted as one page; batch_size is advisory.

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

The trait-level batch_size argument is intentionally ignored in favour of the config field (matches the convention used by the query-mode postgres source and the rest source).

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresCdcSource: 'async_trait,

Preflight probe that does not start replication.

The default Source::check would call stream_pages, which opens the replication stream and consumes/holds WAL (a side effect that pins server resources) — unacceptable as a preflight. Instead we open a normal (non-replication) SQL connection — the same connection style ensure_slot uses — and inspect the slot catalog without touching the replication protocol:

• connection fails → auth probe Fail (could not connect / authenticate),
• connected but the slot row is absent → slot probe Skip (faucet can create it on the first run),
• slot present → slot probe Pass.

The whole call is bounded by ctx.timeout.

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresCdcSource: 'async_trait,

Primary fetch method. Receives context from a parent source’s records.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

fn state_key(&self) -> Option<String>

Stable key under which this source’s incremental-replication bookmark should be persisted in a StateStore.

fn apply_start_bookmark<'life0, 'async_trait>( &'life0 self, bookmark: Value, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresCdcSource: 'async_trait,

Apply a bookmark loaded from a StateStore as this run’s starting point.

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. Defaults to the final segment of std::any::type_name::<Self>(), e.g. "RestSource". Built-in connectors override with a short, friendly snake_case name (e.g. "rest"). Must return a non-empty string; observability decorators fall back to "unknown" in release builds if it is empty (and debug_assert! in debug builds).

fn fetch_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: fetch with no parent context.

fn fetch_all_incremental<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: incremental fetch with no parent context.