Struct HttpSink 

pub struct HttpSink { /* private fields */ }

Available on crate feature sink-http only.

An HTTP sink that sends records to an HTTP endpoint.

Implementations

impl HttpSink

pub fn new(config: HttpSinkConfig) -> HttpSink

Create a new HTTP sink from the given configuration.

pub fn with_auth_provider(self, provider: Arc<dyn AuthProvider>) -> HttpSink

Attach a shared AuthProvider. When set, the provider supplies the credential for every request (taking precedence over inline auth), so several sinks can share one token with single-flight refresh. Used by the CLI to resolve auth: { ref }, and by library callers who construct one provider and inject it into many sinks.

Trait Implementations

impl Sink for HttpSink

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

Non-mutating preflight probe (probe name "network").

Issues a lightweight HEAD request to the configured endpoint over the existing reqwest client. We only care that the host is reachable — that DNS, TCP, TLS and the server all work — so any HTTP response (2xx, 4xx including 405 Method Not Allowed, or 5xx) counts as a pass. Only a transport/connection error (no response at all) is a failure.

fn write_batch_partial<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<(), FaucetError>>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, HttpSink: 'async_trait,

Report per-row outcomes so the DLQ router dead-letters only the records that genuinely failed.

In Individual mode every record is an independent POST, so each record’s success/failure is attributable: we attempt all of them (unlike write_batch, whose ? short-circuits on the first failure) and return one Ok/Err per record. Without this override the default impl would surface the first error as an outer Err, and under on_batch_error: dlq_all the pipeline would route the entire batch to the DLQ — duplicating the already-delivered rows against a non-idempotent endpoint (#146 M14).

In Array mode the page is POSTed chunk-by-chunk (batch_size slices), so forward progress is not atomic across the whole page — each chunk is a separate, independently-committed array POST. The override is therefore chunk-aware rather than all-or-nothing: it iterates the chunks itself and POSTs each array; a row whose chunk was delivered is reported Ok(()), while the rows of the first failing chunk (and every not-yet-sent chunk after it) are reported Err.

This is the fix for the duplicate-data bug (F15 / audit #264): the old implementation delegated to the all-or-nothing write_batch, so a late chunk failure surfaced the whole page as an outer Err. Under on_batch_error: dlq_all the router then dead-lettered every row — including rows from earlier chunks already successfully delivered to the live endpoint — producing silent downstream duplicates. By reporting per-row outcomes, an already-delivered row is never marked failed and so can never land in the DLQ.

Within a single failed chunk a single array POST cannot attribute the failure to specific rows, so all rows of that chunk are reported Err (acceptable — none of them were delivered). When the first chunk fails (nothing has been delivered yet) the override preserves the original all-or-nothing contract and surfaces an outer Err, so the router’s on_batch_error policy (abort vs. dead-letter) still applies to a wholly-undelivered page exactly as before.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this sink accepts.

fn dataset_uri(&self) -> String

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

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, HttpSink: 'async_trait,

Write a batch of records to the destination.

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

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

Flush any buffered data to the destination.

fn supports_idempotent_writes(&self) -> bool

Whether this sink can durably commit a page’s rows and a commit token in a single atomic transaction. Default: false (at-least-once only).

fn sink_guarantee(&self) -> SinkGuarantee

The strongest delivery guarantee this sink can uphold — see SinkGuarantee.

fn dedups_by_key(&self) -> bool

Whether this sink instance is configured to dedup by key — i.e. write_mode: upsert (or delete) with a non-empty key, so re-applying a record with the same key converges instead of duplicating. Default: false.

fn supported_write_modes(&self) -> &'static [WriteMode]

Write modes this sink can apply. Default: append-only. Sinks that implement key-based merge override this to include WriteMode::Upsert / WriteMode::Delete. The CLI rejects a configured mode that is not in this set at config-load time.

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

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

The sink’s live destination schema as an infer_schema-shaped object ({"type":"object","properties":{ <col>: <type-fragment>, … }}), or None for a schemaless sink or a target that does not exist yet.

fn supports_schema_evolution(&self) -> bool

Whether this sink can apply additive/widening DDL via evolve_schema. Default: false. The CLI rejects on_drift: evolve against a sink that returns false at config-load.

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

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

Apply an additive schema evolution (new columns, lossless widenings, nullability relaxations) to the destination. MUST be idempotent (ADD COLUMN IF NOT EXISTS semantics) so concurrent runs converge.

fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, records: &'life1 [Value], scope: &'life2 str, token: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

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

Write records AND durably record token for scope, atomically.

fn last_committed_token<'life0, 'life1, 'async_trait>( &'life0 self, scope: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, FaucetError>> + Send + 'async_trait>>

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

The last token durably committed for scope, or None if this sink has never committed under that scope. Default: None.

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

Stable identifier used as the connector label on metrics and the connector attribute on spans. See Source::connector_name.