spate_core/ops/
mod.rs

1//! Operator chain: statically composed push stages behind one type-erasure
2//! boundary per batch.
3//!
4//! Stages compose via [`Collector`] (monomorphized — a whole chain compiles
5//! to one loop); the only virtual call on the data path is
6//! [`RunnableChain::push_batch`], once per poll batch. Records are born
7//! (deserialized) and die (encoded into shard frames, filtered, or skipped)
8//! inside a single `push_batch` call, so borrowed payloads never cross or
9//! outlive the boundary. See `docs/DESIGN.md` (§ Frozen v1 contracts).
10
11mod builder;
12mod chain;
13mod handoff;
14mod split;
15#[cfg(test)]
16mod tests;
17
18pub use builder::{
19    Assemble, ChainBuilder, ChainFactory, FilterPart, FlatMapPart, InspectPart, MapPart, Root,
20    RoutedSplit, SinkedChain, SplitBuilder, TryMapPart, chain, chain_owned,
21};
22pub use chain::{Emitter, Filter, FlatMap, Inspect, Map, StageLifecycle, TryMap, TypedChain};
23pub use handoff::{ChunkConfig, SinkHandoff};
24pub use split::{Sink, SinkCtx, SplitEmitter, SplitTerminal};
25
26use crate::deser::RecFamily;
27use crate::error::FatalError;
28use crate::record::{Flow, Record};
29use crate::source::PayloadBatch;
30
31/// Why a batch could not complete yet. Both cases are retried with the
32/// resume cursor, but only [`BlockReason::Capacity`] engages the driver's
33/// backpressure controller — a not-ready wait is an upstream dependency
34/// (e.g. a schema fetch), not sink pressure, and pausing the source for it
35/// would misreport the pipeline's state.
36#[derive(Clone, Copy, Debug, PartialEq, Eq)]
37#[non_exhaustive]
38pub enum BlockReason {
39    /// The terminal stage could not accept more output (a shard queue is
40    /// full): genuine backpressure.
41    Capacity,
42    /// A deserializer reported
43    /// [`DeserError::NotReady`](crate::error::DeserError::NotReady): the
44    /// payload replays once its dependency arrives. Counted on
45    /// `spate_deser_not_ready_total`.
46    NotReady,
47}
48
49/// Result of pushing one batch (or a resumed suffix of one) through a
50/// chain.
51#[derive(Debug)]
52#[non_exhaustive]
53pub enum PushOutcome {
54    /// Every payload was fully processed (records may have been filtered
55    /// or skipped by policy along the way).
56    Done,
57    /// The batch could not complete yet. Payloads with index `< resume_at`
58    /// are fully processed; the driver later re-pushes the same batch with
59    /// `from = resume_at`. Any partially-emitted payload's already-emitted
60    /// records are parked inside the terminal stage and drain first on
61    /// resume — operators never re-run for them.
62    Blocked {
63        /// Index of the first payload not yet fully processed.
64        resume_at: usize,
65        /// What the batch is waiting for.
66        reason: BlockReason,
67    },
68    /// A `Fail`-policy stage tripped or an invariant broke. The batch's
69    /// [`AckRef`](crate::checkpoint::AckRef) must be failed by the driver;
70    /// the pipeline stops.
71    Fatal(FatalError),
72}
73
74/// THE SEAM — the one erasure boundary between a pipeline thread's driver
75/// loop and a typed chain. The methods are generic over the buffer lifetime
76/// only, so `Box<dyn RunnableChain>` is legal.
77pub trait RunnableChain: Send {
78    /// Push payloads `from..` of `batch` through the chain.
79    fn push_batch<'buf>(&mut self, batch: &mut dyn PayloadBatch<'buf>, from: usize) -> PushOutcome;
80
81    /// Flush terminal-stage state (parked records, partial encoder
82    /// buffers) downstream. Called by the driver on drain, on linger
83    /// deadlines, and before commit ticks.
84    fn flush(&mut self) -> PushOutcome;
85
86    /// Discard any per-batch replay/resume state after the driver failed the
87    /// current batch's acknowledgement (a shutdown-time abandonment of a
88    /// batch blocked mid-push). Terminal parked chunks — which carry their
89    /// own acks — are unaffected; only the chain's own mid-batch cursor and
90    /// any stashed not-ready payload are cleared, so the next `push_batch` of
91    /// a fresh batch starts clean instead of tripping the resume-cursor
92    /// asserts or replaying the stale payload under the new batch's ack.
93    ///
94    /// The default is a no-op for chains that keep no cross-call batch state.
95    fn abandon_batch(&mut self) {}
96}
97
98/// Push-model stage: receives one record, forwards 0..N downstream.
99///
100/// Composed statically — `Map<F, Filter<P, Term>>` monomorphizes into a
101/// single inlined loop body.
102pub trait Collector<T> {
103    /// Push one record. [`Flow::Blocked`] propagates up to the boundary.
104    fn push(&mut self, rec: Record<T>) -> Flow;
105}
106
107/// Family-erased collector: accepts the family's record type at *any*
108/// buffer lifetime through a lifetime-generic method, which keeps it
109/// dyn-compatible. This is what lets `flat_map` closures hold a plain
110/// `&mut Emitter<'_, OutF>` without naming the downstream stack type.
111pub trait CollectorFor<F: RecFamily> {
112    /// Push one record of the family at any lifetime.
113    fn push_rec<'buf>(&mut self, rec: Record<F::Rec<'buf>>) -> Flow;
114}
115
116impl<F, C> CollectorFor<F> for C
117where
118    F: RecFamily,
119    C: for<'buf> Collector<<F as RecFamily>::Rec<'buf>>,
120{
121    #[inline(always)]
122    fn push_rec<'buf>(&mut self, rec: Record<F::Rec<'buf>>) -> Flow {
123        self.push(rec)
124    }
125}
spate_core/ops/mod.rs

spate_core/ops/
mod.rs
