Crate spate

Expand description

High-performance, at-least-once ETL pipeline framework.

spate is the facade crate for the Spate framework and the only crate applications need to depend on. The engine (spate_core) is re-exported at the root; connectors are enabled through cargo features:

Feature	Enables
`kafka`	`kafka` — Kafka source built on `rdkafka` (single consumer, partition-queue lanes)
`kafka-tls`	TLS, mTLS and SASL for `kafka`; librdkafka is built with a vendored OpenSSL. The Kafka security section lists the SASL mechanisms the feature supports. Without it a security property in the `rdkafka` passthrough fails config validation
`s3`	`s3` — coordinated bounded object-storage (S3) backfill source: the fleet’s leader plans the prefix into splits, workers lease them through the coordinator the `coordination:` section configures (solo in-process without one; progress is then ephemeral), and the job self-terminates once the plan is final and every split completes (`refresh_listing` keeps it open)
`clickhouse`	`clickhouse` — ClickHouse sink (RowBinary, dedup tokens, replica rotation)
`clickhouse-uuid`	`uuid::Uuid` fields for `UUID` columns (`clickhouse::serde::uuid`)
`clickhouse-chrono`	`chrono` fields for `Date`/`DateTime`/`DateTime64`/`Time` columns
`clickhouse-time`	`time` crate fields for the same date/time columns
`clickhouse-rust-decimal`	`rust_decimal::Decimal` conversions for `Decimal` columns
`avro`	`avro` — Avro deserialization (Confluent wire format, schema registry)
`json`	`json` — JSON deserialization (single-document, NDJSON, top-level array)
`json-float-roundtrip`, `json-arbitrary-precision`, `json-raw-value`	Opt-in `serde_json` fidelity knobs for `json`. `arbitrary-precision` is crate-wide
`json-simd`	SIMD decode backend for `json` (`simd-json` replacing `serde_json`). The fidelity knobs above do not apply under it
`coordination`	Deprecated; removed in v0.4.0. `coordination` backend — multi-instance leader-assigned work distribution for broker-less sources: the protocol, `StoreCoordinator`, and the in-memory store (zero-infrastructure embedding). The seam types and the `CoordinationDriver` live in `spate::coordination` without any feature
`coordination-nats`	Deprecated; removed in v0.4.0. The production NATS JetStream KV store (server >= 2.11), selected by `coordination.store.nats`, on top of `coordination`; pulls the async-nats dependency tree
`datagen`	`datagen` — synthetic storefront-event source (orders, payments, refunds) for a pipeline that needs no broker, bucket or coordination store. A demo and test source: it keeps no durable progress
`datagen-avro`	Avro payloads from `datagen` instead of JSON (implies `datagen`)

§Anatomy of a pipeline

One process runs one pipeline (the user guide carries the full architecture and its rationale):

                    ┌───────────────────────────────────────────────┐
 pinned std thread  │ lane.poll → deserialize (borrowed) → operator │──try_send──▶ per-shard
 (× N)              │ chain (map/filter/flat_map, monomorphized)    │             bounded queues
                    └───────────────────────────────────────────────┘                  │
                         ▲          full? pause lanes, keep polling                    ▼
                         │                                            ┌───────────────────────────┐
                    ┌──────────┐   acks (never block)                 │ sink workers: merge chunks,│
                    │ source   │◀───────────────────────────────────── │ seal batches, rotate       │
                    │ control  │   watermarks → store/commit          │ replicas, retry; admin     │
                    └──────────┘                                      │ server (/metrics, probes)  │
                                                                      └───────────────────────────┘

Delivery is at-least-once: a batch’s offsets commit only after every record derived from it was durably written (or intentionally dropped). Duplicates are possible after a crash, so design target tables to tolerate replays.

§A minimal pipeline

Operators are stateful closures chained into a graph; the YAML carries tuning and connector configuration; pipeline::Pipeline assembles and runs the process. This compiles and runs against the spate-test mocks; swap in KafkaSource::from_component_config and a ClickHouse sink for the production version (see examples/kafka_avro_to_clickhouse.rs):

use spate::prelude::*;
use spate_test::{BytesPassthrough, TestEncoder, capture_sink, memory_source};

let config = PipelineConfig::from_str(
    "pipeline: { name: demo, threads: 1, io_threads: 1 }\n\
     admin: { listen: none }\n\
     checkpoint: { interval: 100ms }\n\
     metrics: { exporter: none }\n\
     source: { memory: {} }\n\
     sink: { capture: {} }",
)?;
let (source, handle) = memory_source();
let (sink, script) = capture_sink(1, 1);

let runtime = Pipeline::from_config(config)?
    .sink(sink)?
    .chains(|ctx| {
        // The sink's YAML `chunk:` block (or the default), bound before
        // `with_metrics` moves `ctx.pipeline`.
        let chunk_cfg = ctx.chunk();
        chain_owned::<Vec<u8>, _>(BytesPassthrough)
            .with_metrics(ctx.pipeline, "main")
            .filter(|payload: &Vec<u8>| !payload.is_empty())
            .sink(TestEncoder, KeyHashRouter, chunk_cfg,
                  ctx.queues, ctx.budget)
            .build()
    })
    .runtime_options(RuntimeOptions { handle_signals: false, ..Default::default() })
    .into_runtime(source)?;

// Drive it: assign a lane, produce, wait for the durable commit.
let shutdown = runtime.shutdown_handle();
let join = std::thread::spawn(move || runtime.run());
handle.assign_lanes(&[(spate::source::LaneId(0), PartitionId(0))]);
let last = handle.push(PartitionId(0), None, b"hello");
assert!(
    handle.wait_committed(PartitionId(0), last + 1, std::time::Duration::from_secs(10)),
    "the offset never committed"
);
shutdown.trigger();
let report = join.join().expect("join")?;
assert_eq!(report.exit_code(), 0);
assert!(!script.writes().is_empty());

§Where things live

ops — the chain builder and operator combinators.
source / sink — the connector traits (source::Source, source::SourceLane, sink::RowEncoder, sink::ShardWriter) and the framework-owned sink pool.
pipeline — the runtime: pinned threads, controller, shutdown.
checkpoint / backpressure — acknowledgments and flow control.
config — YAML with ${VAR:-default} interpolation and opaque per-connector sections.
metrics / admin / telemetry — observability. The metrics facade is the instrumentation API, and the metrics reference carries the taxonomy.
Testing your pipelines: the spate-test crate (in-memory sources and sinks with scripting handles).

Re-exports§

pub use spate_avro as avro;avro
pub use spate_json as json;json
pub use spate_kafka as kafka;kafka
pub use spate_clickhouse as clickhouse;clickhouse
pub use spate_s3 as s3;s3
pub use spate_datagen as datagen;datagen
pub use spate_coordination as coordination;coordination

Modules§

admin: Admin HTTP server: /metrics, /healthz, and /readyz.
backpressure: Backpressure: the global in-flight byte budget and the watermark pause/resume controller with hysteresis.
bytes: Re-export of the bytes crate: RowEncoder signatures take bytes::BytesMut, so connector authors can use this re-export instead of declaring their own bytes dependency (declaring one is also fine; versions are compatible per the workspace pin). Provides abstractions for working with bytes.
checkpoint: Checkpointing: acknowledgment tracking and watermark commits.
clock: Injectable time, so a test controls the clock that time-dependent code reads.
config: Pipeline configuration: typed framework sections plus opaque per-component passthrough, loaded from YAML with ${VAR:-default} environment interpolation.
deser: Deserialization contract: one borrowed payload in, 0..N records out, push-style.
error: Error taxonomy and per-stage error policies.
framing: Streaming record framing: the seam that cuts a decoded byte stream into records.
metrics: Metrics: exporter installation and pre-registered handle structs for every pipeline stage.
ops: Operator chain: statically composed push stages behind one type-erasure boundary per batch.
pipeline: Pipeline runtime: pinned driver threads, the source controller, and process assembly.
prelude: Curated imports for pipeline assemblies: one use spate::prelude::*; brings in the builder, chain entry points, and the types every assembly touches.
record: Core record types: payloads, metadata, and the flow-control signal.
sink: Sink abstraction: pipeline threads encode, shard workers batch and write.
source: Source abstraction: a control plane (Source) and a data plane (SourceLane).
telemetry: Structured logging: tracing initialization (JSON for Kubernetes) and rate-limited hot-path logging helpers.

Macros§

rate_limited_warn: tracing::warn! behind a RateLimit. When events were suppressed since the last allowed one, the emitted line carries a suppressed field with the count.
tls_rejection: The TLS rejection in an error’s source chain, as an Option<&rustls::Error>: a certificate the client failed to verify, an alert in TLS_REJECTION_ALERTS, or a peer that shares no protocol version, cipher suite or other handshake parameter with the client (PeerIncompatible).

Structs§

FatalError: An unrecoverable pipeline failure: an invariant was violated or a Fail-policy stage tripped. The pipeline transitions to Failed, the partition watermarks stop advancing, and the process exits non-zero.
PartitionId: Identifier of a source partition (dense, source-assigned).
RawPayload: A raw payload borrowed from the source’s buffers, valid for 'buf.
Record: A record flowing through the chain.
RecordMeta: Per-record metadata. Copy, no drop glue, so it moves through operators for free and never touches the heap.

Enums§

DeserError: A payload could not be deserialized.
ErrorClass: Broad classification used in metrics labels (error_type) and by the retry layer.
ErrorPolicy: What to do when a record fails in a stage.
Flow: Flow-control result of pushing one record downstream.
FramingContract: What one payload emitted by a source represents, so the framework can pair a source with a deserializer without the two being coordinated by hand.
SinkError: A sink failed to write a batch.
SourceError: A source failed to poll, commit, or manage its assignment.

Crate spate

Crate spate Copy item path

§Anatomy of a pipeline

§A minimal pipeline

§Where things live

Re-exports§

Modules§

Macros§

Structs§

Enums§

Crate spate