faucet-sink-http

HTTP POST sink connector for the faucet-stream ecosystem. Sends JSON records to any HTTP(S) endpoint — a webhook, an ingest API, a serverless function, another faucet pipeline's webhook source.

Reach for it when the destination speaks HTTP but isn't one of the first-class warehouse/database/queue sinks: a SaaS ingest endpoint, an internal microservice, a Lambda/Cloud Function URL, or a fan-out webhook. It reuses a single reqwest::Client, sends records concurrently, and retries transient failures so it stays fast and resilient under load.

Feature highlights

• Two batch modes — Individual (one POST per record, sent concurrently up to a concurrency cap) and Array (records packed into a JSON-array body, re-chunked by batch_size).
• Concurrent sends — Individual mode drives a FuturesUnordered with at most concurrency requests in flight, refilling as each completes (no up-front permit acquisition, no deadlock).
• Four auth methods — none, bearer, basic, and custom headers — all on the project-wide { type, config } wire shape. Tokens/passwords are masked (***) in Debug output.
• Shared auth providers — point auth at a { ref: <name> } in the CLI's top-level auth: catalog to share one OAuth2 token (with single-flight refresh) across many sinks.
• Configurable retry — retriable failures (network errors, 5xx, 429) are retried up to max_retries; 4xx client errors fail fast.
• Per-row DLQ in Individual mode — write_batch_partial attempts every record and dead-letters only the ones that actually failed, so already-delivered rows are never duplicated against a non-idempotent endpoint.
• Client built once — the reqwest::Client is created in new() and reused for every request, with connection pooling and keep-alive.

Installation

# As a library:
cargo add faucet-sink-http

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features sink-http

Or via the umbrella crate: cargo add faucet-stream --features sink-http.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: webhook
    config:
      listen_addr: 0.0.0.0:8080
      path: /webhook
  sink:
    type: http
    config:
      url: https://downstream.example.com/ingest
      method: POST
      auth:
        type: bearer
        config:
          token: ${env:INGEST_TOKEN}
      batch_mode:
        type: Individual
      max_retries: 3
      concurrency: 16

faucet run pipeline.yaml

Configuration reference

Core

Batching & concurrency

Reliability

Authentication

auth accepts the project-wide { type, config } shape — either inline, or { ref: <name> } pointing at a shared provider in the CLI's top-level auth: catalog.

# Bearer token (typically via env / secret indirection)
auth:
  type: bearer
  config:
    token: ${env:INGEST_TOKEN}

# HTTP Basic
auth:
  type: basic
  config:
    username: ingest-user
    password: ${secret:vault:secret/ingest#password}

# Custom API-key header
auth:
  type: custom
  config:
    headers:
      X-Api-Key: ${env:API_KEY}

# Shared provider from the top-level `auth:` catalog (OAuth2 with single-flight refresh)
auth: { ref: my_idp }

When a shared provider is attached it takes precedence over inline auth; a bare { ref: <name> } with no provider wired up is a config error.

Examples

Individual mode — one POST per record, sent concurrently

sink:
  type: http
  config:
    url: https://webhooks.example.com/event
    auth:
      type: bearer
      config: { token: ${env:WEBHOOK_TOKEN} }
    batch_mode:
      type: Individual
    concurrency: 20
    max_retries: 2

Array mode — bulk POST as a JSON array

sink:
  type: http
  config:
    url: https://api.example.com/bulk-ingest
    batch_mode:
      type: Array
    batch_size: 500     # one POST per 500-record chunk
    max_retries: 3

Custom method + API-key header

sink:
  type: http
  config:
    url: https://api.example.com/data
    method: PUT
    auth:
      type: custom
      config:
        headers:
          X-Api-Key: ${env:API_KEY}
    batch_mode:
      type: Array
    batch_size: 0       # forward the whole upstream page as one array

Streaming & batching

The pipeline calls Sink::write_batch once per upstream StreamPage. The on-the-wire effect depends on batch_mode:

• Individual — one HTTP request per record, executed concurrently up to concurrency via a refilling FuturesUnordered. batch_size has no effect on framing here (each record is already its own request); tune throughput with concurrency. For per-record send semantics, prefer batch_mode: Individual over batch_size: 1 — it's the direct expression of intent and the only one that drives concurrent in-flight requests.
• Array — the page is re-chunked into batch_size-row slices and one POST is issued per chunk, each body a JSON array. With the default batch_size: 1000, a 2 500-record page produces 3 POSTs (1000 + 1000 + 500). With batch_size: 0 the whole page is forwarded as a single JSON array — useful when the upstream source already chunks the stream to a size the destination accepts (e.g. a Postgres source with its own batch_size).

Recommended batch_size for array-accepting endpoints: match the destination's documented batch limit (commonly 100–1000 records per request).

Dead-letter queue

The sink overrides write_batch_partial so a configured DLQ gets accurate per-row outcomes:

• Individual mode — every record is an independent POST, so each success/failure is attributable. The sink attempts all records (failures don't short-circuit siblings) and returns one outcome per record; only the genuinely-failed rows are dead-lettered. This avoids duplicating already-delivered rows against a non-idempotent endpoint under on_batch_error: dlq_all.
• Array mode — a single array POST cannot attribute a failure to specific rows, so it stays all-or-nothing: the whole array surfaces as an error and the DLQ on_batch_error policy decides whether to abort or dead-letter the batch.

This sink does not support effectively-once delivery or upsert/delete write modes — HTTP endpoints are opaque to faucet, so it can only append (POST). For idempotent delivery, make the destination endpoint idempotent (e.g. key on a record field) and use Individual mode + a DLQ.

Config loading & schema

Load config from YAML/JSON, environment variables, or a .env file. Inspect the full JSON Schema with:

faucet schema sink http

Note: headers is #[serde(skip)] (library-only, set via the headers() builder). The custom auth variant's header map round-trips through JSON/YAML.

Library usage

use faucet_core::Pipeline;
use faucet_sink_http::{HttpBatchMode, HttpSink, HttpSinkAuth, HttpSinkConfig};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = HttpSinkConfig::new("https://api.example.com/ingest")
    .auth(HttpSinkAuth::Bearer { token: "my-token".into() })
    .batch_mode(HttpBatchMode::Array)
    .with_batch_size(500)
    .max_retries(3)
    .concurrency(20);

let sink = HttpSink::new(cfg);

// let result = Pipeline::new(source, sink).run().await?;
# let _ = sink;
# Ok(())
# }

Attach a shared AuthProvider (so several sinks share one token with single-flight refresh) via HttpSink::new(cfg).with_auth_provider(provider).

How it works

• new() builds the reqwest::Client once and reuses it for every request (connection pooling + keep-alive).
• Auth is resolved once per batch — the shared provider (if attached) wins, otherwise inline auth is used.
• Individual mode drives a FuturesUnordered, seeding it with up to concurrency futures and refilling as each completes — so exactly concurrency requests are in flight without acquiring permits up-front.
• Array mode collects each batch_size chunk into a Value::Array and POSTs it as one request body.
• Every response runs through faucet_core::util::check_http_response, which maps non-2xx statuses to typed FaucetErrors; retriable ones (network, 5xx, 429) are retried up to max_retries.

Lineage dataset URI

https://<host><path> with credentials stripped — e.g. https://api.example.com/ingest (a user:pass@ prefix in the URL is redacted).

Feature flags

This crate has no optional features of its own; enable it in the CLI/umbrella via the sink-http feature.

Troubleshooting / FAQ

See also

• Sinks reference — capability matrix across all connectors.
• Authentication cookbook — shared auth: providers and the { type, config } shape.
• Dead-letter queue cookbook — on_batch_error policies and per-row dead-lettering.
• faucet-source-webhook — the natural upstream for HTTP fan-out pipelines.
• faucet-source-rest — pull from a REST API and forward it over HTTP.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.