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

etl_core/ops/
mod.rs
