# faucet-sink-http
[](https://crates.io/crates/faucet-sink-http)
[](https://docs.rs/faucet-sink-http)
[](https://github.com/PawanSikawat/faucet-stream/blob/main/rust-toolchain.toml)
[](https://github.com/PawanSikawat/faucet-stream#license)
HTTP POST sink connector for the [faucet-stream](https://github.com/PawanSikawat/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
```bash
# 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
```yaml
# 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
```
```bash
faucet run pipeline.yaml
```
## Configuration reference
### Core
| `url` | string | — *(required)* | Target endpoint URL. |
| `method` | string | `POST` | HTTP method (`POST`, `PUT`, `PATCH`, …). |
| `headers` | header map | empty | Additional request headers. **Library-only** — `#[serde(skip)]`, set via the `headers()` builder; not settable from YAML/JSON. |
| `auth` | `HttpSinkAuth` | `{ type: none }` | Authentication — inline `{ type, config }` or `{ ref: <name> }`. See [Authentication](#authentication). |
### Batching & concurrency
| `batch_mode` | `HttpBatchMode` | `Individual` | `Individual` = one POST per record; `Array` = records sent as one JSON-array body. See [Batch modes](#batch-modes). |
| `concurrency` | int | `10` | Max concurrent in-flight requests in `Individual` mode. No effect in `Array` mode (one POST per chunk, issued sequentially). Clamped to a floor of `1`. |
| `batch_size` | int | `1000` | Max records per outbound HTTP request in `Array` mode (re-chunks the upstream page). **`0` = "no batching" sentinel** — forwards the whole page as one JSON array. **No effect in `Individual` mode** (each record is already its own request); kept for config-shape parity and validated via `faucet_core::validate_batch_size`. |
### Reliability
| `max_retries` | int | `0` | Number of retries on retriable failures (network errors, 5xx, 429). Each retry is immediate (no backoff). 4xx client errors fail fast. After exhausting retries the last error is returned. |
## 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.
| `none` | *(none)* | No authentication. |
| `bearer` | `{ token }` | Sent as `Authorization: Bearer <token>`. |
| `basic` | `{ username, password }` | HTTP Basic authentication. |
| `custom` | `{ headers: { <name>: <value>, … } }` | Arbitrary header map applied to every request (e.g. an API key header). |
```yaml
# Bearer token (typically via env / secret indirection)
auth:
type: bearer
config:
token: ${env:INGEST_TOKEN}
```
```yaml
# HTTP Basic
auth:
type: basic
config:
username: ingest-user
password: ${secret:vault:secret/ingest#password}
```
```yaml
# Custom API-key header
auth:
type: custom
config:
headers:
X-Api-Key: ${env:API_KEY}
```
```yaml
# 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
```yaml
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
```yaml
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
```yaml
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](https://pawansikawat.github.io/faucet-stream/cookbook/dlq.html) 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:
```bash
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
```rust
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 `FaucetError`s; 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
| `Auth` error: "references provider … but no provider was supplied" | `auth: { ref: <name> }` points at a provider not in the top-level `auth:` catalog (or no provider was injected via `with_auth_provider`). Add the named provider, or use inline `auth`. |
| `401` / `403` from the endpoint | Wrong or expired credentials. Verify the `bearer` token / `basic` credentials, or that the shared provider's scopes are correct. |
| `Auth` error: "invalid custom header name/value" | A `custom` header name or value isn't a valid HTTP header. Remove non-ASCII / control characters. |
| Requests not parallel / slow in `Individual` mode | `concurrency` defaults to `10`. Raise it for high-fan-out endpoints; it's clamped to a floor of `1`. |
| `batch_size` seems ignored | You're in `Individual` mode, where each record is its own request — `batch_size` only affects `Array` mode. Switch to `batch_mode: Array` to pack records, or use `concurrency` to tune `Individual`. |
| Endpoint rejects large array bodies (`413` / timeout) | The `Array` body exceeds the destination's limit. Lower `batch_size` to its documented per-request cap. |
| Duplicate rows after a partial failure | The destination isn't idempotent. Use `batch_mode: Individual` + a DLQ so only failed rows are retried/dead-lettered, and make the endpoint idempotent (key on a record field). |
| `4xx` errors not retried | Intentional — only retriable failures (network, 5xx, 429) are retried. Fix the request shape (`url`, `method`, body) for 4xx. |
## See also
- [Sinks reference](https://pawansikawat.github.io/faucet-stream/reference/connectors.html) — capability matrix across all connectors.
- [Authentication cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/auth.html) — shared `auth:` providers and the `{ type, config }` shape.
- [Dead-letter queue cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/dlq.html) — `on_batch_error` policies and per-row dead-lettering.
- [`faucet-source-webhook`](https://crates.io/crates/faucet-source-webhook) — the natural upstream for HTTP fan-out pipelines.
- [`faucet-source-rest`](https://crates.io/crates/faucet-source-rest) — pull from a REST API and forward it over HTTP.
## License
Licensed under either of [Apache License, Version 2.0](https://www.apache.org/licenses/LICENSE-2.0) or [MIT license](https://opensource.org/licenses/MIT) at your option.