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)
`s3`	`s3` — coordinated bounded object-storage (S3) backfill source: the fleet’s leader plans the prefix into splits, workers lease them through an injected coordinator (solo in-process by default; progress is ephemeral without a durable store), 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`. Not part of `full` — `arbitrary-precision` is crate-wide
`coordination`	`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`	The production NATS JetStream KV store (server >= 2.11) on top of `coordination`; pulls the async-nats dependency tree
`full`	All connectors (`avro`, `json`, `kafka`, `clickhouse`, `coordination-nats`, `s3`)

§Anatomy of a pipeline

One process runs one pipeline (see docs/DESIGN.md in the repository for 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 — 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\
     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");
while handle.last_committed(PartitionId(0)) != Some(last + 1) {
    std::thread::sleep(std::time::Duration::from_millis(5));
}
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 — acknowledgements 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; see docs/METRICS.md for 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;
pub use spate_json as json;
pub use spate_kafka as kafka;
pub use spate_clickhouse as clickhouse;
pub use spate_s3 as s3;
pub use spate_coordination as 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: Provides abstractions for working with bytes.
checkpoint: Checkpointing: acknowledgement tracking and watermark commits.
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 initialisation (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.

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 — 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