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 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 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 connector_name(&self) -> &'static str { ... }
    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 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 supports_exactly_once(&self) -> bool

Whether this source deterministically replays the same page sequence from a given bookmark — the requirement for exactly-once delivery (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 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 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,