Trait Source 

pub trait Source: Send + Sync {
    // Required method
    fn fetch_with_context<'life0, 'life1, 'async_trait>(
        &'life0 self,
        context: &'life1 HashMap<String, Value>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait;

    // Provided methods
    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 { ... }
    fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>(
        &'life0 self,
        context: &'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,
             Self: 'async_trait { ... }
    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 { ... }
    fn stream_pages<'a>(
        &'a self,
        context: &'a HashMap<String, Value>,
        batch_size: usize,
    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>> { ... }
    fn native_output_formats(&self) -> &'static [NativeFormat] { ... }
    fn stream_native<'a>(
        &'a self,
        context: &'a HashMap<String, Value>,
        format: NativeFormat,
        batch_size: usize,
    ) -> Pin<Box<dyn Stream<Item = Result<NativeBatch, FaucetError>> + Send + 'a>> { ... }
    fn config_schema(&self) -> Value { ... }
    fn state_key(&self) -> Option<String> { ... }
    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,
             Self: 'async_trait { ... }
    fn capture_resume_position<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn lag<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<SourceLag>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn state_schema(&self) -> u32 { ... }
    fn migrate_state(
        &self,
        from: u32,
        data: Value,
    ) -> Result<Value, FaucetError> { ... }
    fn supports_exactly_once(&self) -> bool { ... }
    fn replay_guarantee(&self) -> ReplayGuarantee { ... }
    fn is_shardable(&self) -> bool { ... }
    fn enumerate_shards<'life0, 'async_trait>(
        &'life0 self,
        _target: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ShardSpec>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn apply_shard<'life0, 'life1, 'async_trait>(
        &'life0 self,
        _shard: &'life1 ShardSpec,
    ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait { ... }
    fn range_digest<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        range: &'life1 KeyRange,
        key: &'life2 str,
        columns: &'life3 [String],
    ) -> Pin<Box<dyn Future<Output = Result<Option<ServerDigest>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait,
             Self: 'async_trait { ... }
    fn supports_discover(&self) -> bool { ... }
    fn discover<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<DatasetDescriptor>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn record_table(&self, _record: &Value) -> Option<String> { ... }
    fn position_le(&self, _a: &Value, _b: &Value) -> Option<bool> { ... }
    fn position_min(&self, positions: &[Value]) -> Option<Value> { ... }
    fn connector_name(&self) -> &'static str { ... }
    fn set_roundtrip_recorder(&self, _recorder: Arc<RoundtripRecorder>) { ... }
    fn set_run_clock(&self, _now: DateTime<Utc>) { ... }
    fn dataset_uri(&self) -> String { ... }
    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,
             Self: 'async_trait { ... }
}

A source fetches records from an external system.

Required Methods

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

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

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

An empty context map means this is a root source (no parent). Connectors that support being a child should use substitute_context() to resolve {placeholder} tokens in their URL path, query parameters, headers, or body. Connectors that don’t need parent context ignore the map.

Provided Methods

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_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, context: &'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, Self: 'async_trait,

Incremental fetch with parent context support.

Returns the records and an optional bookmark value for incremental replication. The default delegates to fetch_with_context and returns None for the bookmark.

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.

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

Stream records page-by-page so the pipeline can write to the sink as pages arrive instead of buffering the full result set.

batch_size is the hint the pipeline passes down; sources are free to use a larger or smaller native chunk (e.g. one page per HTTP response, one row-group per Parquet file) but should approximate it where feasible. The special value batch_size = 0 means “do not batch — emit the entire result set in a single page.” Sources that stream natively should treat 0 as “skip the chunking layer and yield one page after the underlying read completes” (useful for small lookup tables or for sinks like SQL COPY / BigQuery load jobs that prefer one large request).

The default implementation fetches the full result set via fetch_with_context_incremental and chunks it in memory by batch_size. The bookmark (when present) is attached to the final page so the pipeline only persists after the entire fetch has been written. Sources that can stream natively override this method and may emit per-page bookmarks (e.g. CDC).

An empty result with a Some(bookmark) still yields one empty page carrying the bookmark, so incremental runs that produce no records still advance their checkpoint.

fn native_output_formats(&self) -> &'static [NativeFormat]

Wire formats this source can emit as raw bytes for the native byte-passthrough fast path (#633), in preference order (first = best). Default: &[] (no native fast path).

The pipeline uses the path only when this returns a non-empty slice, a sink advertises a matching NativeLoadCapability, and every prerequisite holds (see plan_native_transfer). A TransformingSource deliberately does not override this, so any attached transform disables the fast path.

fn stream_native<'a>( &'a self, context: &'a HashMap<String, Value>, format: NativeFormat, batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<NativeBatch, FaucetError>> + Send + 'a>>

Stream the source natively as byte batches in format (#633).

Only invoked after the pipeline negotiated format — one this source advertised via native_output_formats. The default yields a single typed “unsupported” error so a source that advertises support but forgets to override this fails loudly. Each batch’s bookmark carries the same checkpoint semantics as StreamPage.

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.

Returning Some(key) opts this source into resumable runs: when the pipeline is configured with a state store via Pipeline::with_state_store, it reads the bookmark at key before fetching and writes the new bookmark back only after the sink confirms the batch was written.

The default returns None, meaning the source is not persisted. Keys must satisfy validate_state_key.

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, Self: 'async_trait,

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

The default implementation ignores the value, which keeps existing sources backwards-compatible. Sources that support incremental replication override this — typically by storing the value behind interior mutability and consulting it inside fetch_with_context_incremental.

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

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

Capture the source’s current replication position without consuming any changes, ensuring any server-side resource (e.g. a logical replication slot) needed to later resume from that position exists.

Returns the position as a bookmark Value — the same shape apply_start_bookmark accepts — or None if this source does not support position capture.

Used by the snapshot→CDC replication orchestrator (faucet replicate) to anchor the CDC stream at-or-before the bulk snapshot’s read point so the handoff has no gap. The default returns None.

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

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

How far this source is behind its head (#733) — unread WAL bytes, binlog distance, unconsumed messages, or the age of the oldest unread change.

The pipeline polls it at page boundaries and exports faucet_source_lag_{bytes,events,seconds}; faucet status --probe asks for it on demand after applying the stored bookmark. It must be a cheap, read-only query. An Err is logged once and reported as no lag; it never fails a run. Default: Ok(None) (no notion of a head). Decorators must forward this.

fn state_schema(&self) -> u32

The version of this source’s bookmark shape (#736). Stored state carries it; bump it whenever the shape changes and teach migrate_state the step. Default 0. Decorators must forward this.

fn migrate_state(&self, from: u32, data: Value) -> Result<Value, FaucetError>

Bring a bookmark stored at shape version from up to state_schema (#736). Must be pure — the migrated value is only persisted by the next bookmark write, so a crash in between re-runs it. The default knows only its own version. Decorators must forward this.

fn supports_exactly_once(&self) -> bool

Whether this source deterministically replays the same page sequence from a given bookmark — the requirement for the atomic-watermark effectively-once path (a non-deterministic replay could cause the pipeline to skip a page whose contents differ from the one already committed). Default: false.

Sources with a durable monotonic position and per-page bookmarks (CDC) override this to return true. The pipeline rejects DeliveryMode::ExactlyOnce against a source that returns false.

fn replay_guarantee(&self) -> ReplayGuarantee

The typed replay capability this source advertises — see ReplayGuarantee.

The default derives from supports_exactly_once (the boolean stays the back-compat primitive: existing connectors that override only the boolean automatically advertise Deterministic here). Override this directly only to diverge from the boolean — there is currently no reason to.

fn is_shardable(&self) -> bool

Whether this source can split its work into independent shards for clustered (Mode B) execution. Default: false (single whole-dataset shard). Sources with a natural partition (object-store prefixes, table primary-key ranges) override this to true and implement enumerate_shards + apply_shard.

fn enumerate_shards<'life0, 'async_trait>( &'life0 self, _target: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ShardSpec>, FaucetError>> + Send + 'async_trait>>

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

Enumerate the shards this source splits into, aiming for roughly target of them (a hint — the source may return fewer, e.g. when the dataset is small, or one per natural partition regardless of target).

Called once per run by the cluster coordinator; enumeration must be deterministic enough that re-enumeration yields a compatible set (stable shard ids), since it may run on more than one instance and is reconciled by idempotent insert. May perform read-only I/O (a LIST, a MIN/MAX query). The default returns a single whole-dataset shard (ShardSpec::whole), preserving today’s single-worker behavior.

fn apply_shard<'life0, 'life1, 'async_trait>( &'life0 self, _shard: &'life1 ShardSpec, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

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

Narrow this source instance to a single shard before streaming.

Called on the worker that claims shard, after construction and before any stream_pages call. Like apply_start_bookmark this takes &self and is expected to record the shard behind interior mutability (the source consults it when building its query / listing). The default ignores the shard — a non-shardable source only ever receives ShardSpec::whole, so ignoring it streams the whole dataset. Implementations should accept the whole shard as a no-op.

fn range_digest<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, range: &'life1 KeyRange, key: &'life2 str, columns: &'life3 [String], ) -> Pin<Box<dyn Future<Output = Result<Option<ServerDigest>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: 'async_trait,

Compute a ServerDigest of the rows in range inside the backend, so a matching range ships no rows (#701). columns are the compared columns; key the integer key the range is over. Default Ok(None): not supported, the verifier streams the range instead. Two digests compare only when both sides report the same algorithm.

fn supports_discover(&self) -> bool

Whether this source can enumerate the datasets behind its connection via discover. Default: false. Sources backed by an introspectable catalog (database information_schema, MongoDB collections, Elasticsearch indices, object-store prefixes) override this to true.

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

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

Enumerate the datasets living behind this source’s connection — one DatasetDescriptor per table / collection / index / prefix, each carrying a partial config override that selects it (used by faucet discover to scaffold one matrix row per dataset).

Must be read-only and cheap: catalog metadata queries and listings only, never a data scan. Descriptors must never embed credentials. The default returns a typed “unsupported” error; override it (and return true from supports_discover) only for sources with a real catalog to introspect.

fn record_table(&self, _record: &Value) -> Option<String>

The dataset (table / collection) a change record belongs to, for a change stream that carries several tables (#731). A multi-table faucet mirror runs one stream and routes each record to its table’s pipeline by this name, which must match the name the paired bulk source’s discover reports (e.g. public.orders). None means the record belongs to no table (a DDL / control event) — and is the default, so a source that does not override this cannot be used for a multi-table mirror.

fn position_le(&self, _a: &Value, _b: &Value) -> Option<bool>

Order two of this source’s bookmarks (#731): Some(true) when stream position a is at or before b (every change up to a is also covered by b), Some(false) when it is not, and None when the source cannot tell (the default, or two positions it cannot relate). A multi-table mirror uses it to resume one shared stream from the earliest table and skip, per table, pages that table has already committed.

fn position_min(&self, positions: &[Value]) -> Option<Value>

The earliest stream position every one of positions can resume from (#731): the shared change stream restarts there and each table skips what it has already applied. The default picks the position that position_le orders at or before all the others, and None when there is none (an empty slice, or positions this source cannot order). Sources whose positions are only partially ordered (one cursor per capture instance, say) override it with a component-wise minimum.

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 set_roundtrip_recorder(&self, _recorder: Arc<RoundtripRecorder>)

Receive a pre-labelled handle for counting upstream round trips (#638) — calls this connector makes to its own backend.

The pipeline calls this once before streaming, because it is the only place that knows the pipeline / row / connector labels every other metric carries. A connector opts in by storing the handle behind interior mutability (the pattern the REST source already uses for runtime_start) and calling recorder.record("<op>") at each real backend call — including retries, which are real round trips.

Defaulted to a no-op: an uninstrumented connector emits nothing, so instrumentation lands connector by connector with no behaviour change in between, and a third-party connector is unaffected.

fn set_run_clock(&self, _now: DateTime<Utc>)

Receive the run clock (#769): the instant ${now.*} renders from — faucet run --clock, a schedule tick, a backfill unit’s start — so a source that bounds reads by “now” (REST window slicing) reproduces the run as of that instant rather than the wall clock.

Called once before streaming. Defaulted to a no-op; a connector opts in by storing the instant behind interior mutability.

fn dataset_uri(&self) -> String

Logical dataset identity for lineage emission, following OpenLineage naming conventions (https://openlineage.io/docs/spec/naming).

The default returns "<connector_name>://unknown". Built-in connectors override with a credential-free URI derived from their config. Strip any credentials with redact_uri_credentials. Informational metadata only — never used for I/O.

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, Self: 'async_trait,

Run a fast, non-mutating preflight probe (used by faucet doctor).

The default pulls a single page via stream_pages and reports success/failure — it exercises the real read path (DNS, TLS, auth, the first request, the first-record decode) but never paginates the full dataset and never repeats. The page stream is dropped immediately after the first page.

Sources whose first page blocks waiting for inbound data (webhook, websocket) or has side effects (CDC consuming WAL) override this with a cheaper, side-effect-free probe. Probe-level failures are returned as a ProbeStatus::Fail inside Ok(report).

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl Source for PostgresCdcSource

impl Source for TransformingSource

impl<'a, S> Source for InstrumentedSource<'a, S>

where S: Source + ?Sized,