Module ops

Expand description

Operator chain: statically composed push stages behind one type-erasure boundary per batch.

Stages compose via Collector (monomorphized, so a whole chain compiles to one loop); the only virtual call on the data path is RunnableChain::push_batch, once per poll batch. Records are born (deserialized) and die (encoded into shard frames, filtered, or skipped) inside a single push_batch call, so borrowed payloads never cross or outlive the boundary (ADR-0013).

§Owned vs borrowed record families

For owned families (Owned<T>) the builder offers ChainBuilder::map / ChainBuilder::try_map with plain closure bounds; bare closures infer. For borrowing families a rustc limitation (E0582: a higher-ranked lifetime may not appear only in associated-type positions) rules out FnMut-with-projection-output bounds at the definition site; use ChainBuilder::map_rec / ChainBuilder::try_map_rec, whose bound goes through MapFn / TryMapFn. Pass a fn item where you can: it satisfies a higher-ranked bound by construction, where a closure satisfies it only when the compiler infers a higher-ranked signature for it.

A stage over a borrowing family, written as a fn item:

struct Compact<'buf> {
    key: &'buf str,
}
struct CompactF;
impl RecFamily for CompactF {
    type Rec<'buf> = Compact<'buf>;
}

fn shrink<'a>(e: LogEvent<'a>) -> Compact<'a> {
    Compact { key: e.key }
}
let stage = chain(log_deser).map_rec::<CompactF, _>(shrink);

ChainBuilder::filter, ChainBuilder::inspect, and ChainBuilder::flat_map have no output binding, so a single generic method serves both kinds of family.

use spate_core::backpressure::InflightBudget;
use spate_core::deser::{BytesPassthrough, Owned};
use spate_core::error::ErrorPolicy;
use spate_core::ops::{ChunkConfig, chain};
use spate_core::record::Record;
use spate_core::sink::{KeyHashRouter, RowEncoder, shard_queues};
use std::sync::Arc;

// A trivial encoder writing `<u32 len><bytes>` rows.
#[derive(Clone)]
struct LenPrefix;
impl RowEncoder<Owned<Vec<u8>>> for LenPrefix {
    fn encode<'buf>(
        &mut self,
        rec: &Record<Vec<u8>>,
        buf: &mut bytes::BytesMut,
    ) -> Result<(), spate_core::error::SinkError> {
        buf.extend_from_slice(&(rec.payload.len() as u32).to_le_bytes());
        buf.extend_from_slice(&rec.payload);
        Ok(())
    }
}

let (queues, _rx) = shard_queues(2, 64);
let budget = Arc::new(InflightBudget::new());

let mut pipeline_chain = chain(BytesPassthrough)
    .map(|mut bytes: Vec<u8>| {
        bytes.make_ascii_uppercase();
        bytes
    })
    .filter(|bytes: &Vec<u8>| !bytes.is_empty())
    .try_map(
        |bytes: Vec<u8>| String::from_utf8(bytes).map(String::into_bytes),
        ErrorPolicy::Skip,
    )
    .sink(LenPrefix, KeyHashRouter, ChunkConfig::default(), queues, budget)
    .build();

Structs§

ChainBuilder: Fluent builder for one pipeline’s operator chain. DF is the deserializer’s record family; CurF the family at the current end of the chain (changed by map_rec and flat_map, and by map for owned payloads).
ChainFactory: Stamps out identical chains, one per pipeline thread. Send + Sync when the deserializer, stage closures, encoder, and router are.
ChunkConfig: Tuning for the terminal stage’s per-shard chunking.
Emitter: Stack-borrowed emitter handed to flat_map closures. Parameterized by the output family (a 'static tag), so user closures never name the concrete downstream stack type or the buffer lifetime. Each emit is one virtual call, confined to flat_map stages.
Filter: filter: drop records failing the predicate. A drop releases the record’s ack share, so it counts as success for the batch.
FilterPart: Recorded filter stage.
FlatMap: flat_map: one record in, 0..N out via a stack-borrowed Emitter. Carries the output family as a type parameter so the impl is fully constrained (closure argument types are not associated bindings).
FlatMapPart: Recorded flat_map stage.
Inspect: inspect: observe without transforming (no metrics of its own).
InspectPart: Recorded inspect stage.
Map: map: transform the payload.
MapPart: Recorded map/map_rec stage.
Root: The empty stage list.
RoutedSplit: A split-terminated chain, ready to build. Not Clone; the branches are built once. Stamp identical per-thread chains by re-running the chains factory closure (as the pipeline builder does).
Sink: A typed, Copy handle to one split branch, carrying a branch index plus the destination family, so SplitEmitter::emit both type-checks the row and recovers the branch with zero per-call lookup. Minted by SplitBuilder::add.
SinkCtx: The per-sink handles a split branch needs, resolved by name from the chain factory’s ChainCtx via ChainCtx::sink. The bundle carries the sink’s name, so SplitBuilder::add does not repeat it.
SinkHandoff: The chain’s terminal stage. Owns one accumulation buffer per shard, seals EncodedChunks at ChunkConfig::target_bytes, and hands them to the sink workers through the bounded ShardQueues, with a try_send that never blocks the pipeline thread.
SinkedChain: A fully specified chain, ready to build, or to stamp out one instance per pipeline thread via SinkedChain::build_factory.
SplitBuilder: Accumulates split-sink branches before the routing closure is supplied. Built by ChainBuilder::split; each add declares one destination and hands back a typed handle, then route takes the closure that dispatches to them.
SplitEmitter: Stack-borrowed emitter handed to a route closure. emit routes one derived record to the branch named by a Sink<F> handle; a record that emits to no branch triggers the split’s unmatched policy. One Any downcast per emit, no per-call name lookup.
SplitTerminal: The chain’s split terminal. Runs the route closure over each record and aggregates the branches’ lifecycle (relieve/flush/fatal). See the module docs.
TryMap: try_map: fallible transform with a per-stage ErrorPolicy. Skip drops the record (releasing its ack share) and counts it; Fail records a fatal error; the batch aborts and the pipeline stops.
TryMapPart: Recorded try_map/try_map_rec stage.
TypedChain: A concrete chain: deserializer + statically composed operator stack, erased behind RunnableChain. Ops must accept the family’s record type at every buffer lifetime; the HRTB is what makes borrowed records legal behind the erased boundary.

Enums§

BlockReason: Why a batch could not complete yet. Both cases are retried with the resume cursor, but only BlockReason::Capacity engages the driver’s backpressure controller. A not-ready wait is an upstream dependency (e.g. a schema fetch), not sink pressure, and pausing the source for it would misreport the pipeline’s state.
PushOutcome: Result of pushing one batch (or a resumed suffix of one) through a chain.

Traits§

Assemble: Assembles recorded parts into the concrete collector stack, given the terminal stage. Takes &self so one set of parts can assemble many identical chains. Stage closures must therefore be Clone (plain closures and closures over Clone/Arc state are).
Collector: Push-model stage: receives one record, forwards 0..N downstream.
CollectorFor: Family-erased collector: accepts the family’s record type at any buffer lifetime through a lifetime-generic method, which keeps it dyn-compatible. This is what lets flat_map closures hold a plain &mut Emitter<'_, OutF> without naming the downstream stack type.
MapFn: A record-to-record transform between families. Implemented for every FnMut(In) -> Out; expressed as an independent two-parameter trait so higher-ranked builder bounds stay legal for borrowing families (see the module docs on E0582). fn items satisfy it at every lifetime.
RunnableChain: The one erasure boundary between a pipeline thread’s driver loop and a typed chain. The methods are generic over the buffer lifetime only, so Box<dyn RunnableChain> is legal.
StageLifecycle: Per-batch lifecycle cascade implemented by every stage. Combinators handle their own concern and delegate downstream; the terminal stage anchors the recursion.
TryMapFn: Fallible variant of MapFn, with the error type as a third parameter.

Functions§

chain: Start a chain from a deserializer producing family F.
chain_owned: Start a chain from a deserializer producing owned records T.

Module ops

Module ops Copy item path

§Owned vs borrowed record families

Structs§

Enums§

Traits§

Functions§

Module ops