spate_core/pipeline/mod.rs
1//! Pipeline runtime: pinned driver threads, the source controller, and
2//! process assembly.
3//!
4//! Thread anatomy (see `docs/DESIGN.md` § Process anatomy):
5//!
6//! ```text
7//! main (run()) controller (std thread) driver 0..N (std threads)
8//! metrics/admin/io ── owns Source + Checkpointer ── own lanes + chain
9//! joins everything poll_events / commit tick poll → push_batch → route
10//! pause/resume application backpressure ticks
11//! ```
12//!
13//! Communication: the controller sends [`ThreadControl`] messages to
14//! drivers (lane assignment, drain barriers); drivers send [`DriverEvent`]
15//! requests back (pause/resume — only the controller touches the
16//! [`Source`](crate::source::Source) — and fatal reports). All channels are
17//! unbounded crossbeam channels: control traffic is rare and must never
18//! block a poll loop.
19//!
20//! Shutdown (also the full-revocation path, per DESIGN.md § Shutdown):
21//! SIGTERM → controller stops event polling and sends `Shutdown` to every
22//! driver → each driver flushes its chain, drops its lanes, and arrives at
23//! the barrier → main joins driver threads (dropping chains closes the
24//! shard queues) → the sink drains under the remaining deadline → the
25//! controller runs a final drain + commit + `flush_commits` → the process
26//! reports an [`ExitReport`]. A sink that cannot flush by the deadline is
27//! abandoned loudly; unacknowledged offsets are never committed, so the
28//! data replays after restart (at-least-once).
29
30mod builder;
31mod controller;
32mod driver;
33mod runtime;
34
35pub use builder::{BuildError, ChainCtx, Pipeline, PipelineError, SinkOptions};
36pub use runtime::{PipelineRuntime, RuntimeOptions, ShutdownHandle, StartError, metrics_settings};
37
38use crate::error::FatalError;
39use crate::record::PartitionId;
40use crate::sink::ShardQueues;
41use crate::source::{DrainBarrier, LaneId};
42use std::time::Instant;
43
44/// Control messages the controller sends to a driver thread.
45pub(crate) enum ThreadControl<L> {
46 /// Take ownership of a newly assigned lane.
47 AddLane(L),
48 /// Stop and drop the listed lanes (revocation): flush the chain, then
49 /// arrive at `barrier` once per stopped lane before `deadline`.
50 StopLanes {
51 lanes: Vec<LaneId>,
52 barrier: DrainBarrier,
53 deadline: Instant,
54 },
55 /// Flush the chain's partial terminal state now, without waiting out
56 /// the idle-flush lull
57 /// ([`SourceEvent::CommitReady`](crate::source::SourceEvent::CommitReady)).
58 ///
59 /// A lane that reached genuine end-of-input has no more data coming, so
60 /// its tail would otherwise sit in the chain until `idle_flush` elapses
61 /// — unacknowledged, and therefore blocking the completion of whatever
62 /// unit of work it belongs to. Best-effort and unsynchronised: the
63 /// controller does not wait, and a blocked chain simply retries on the
64 /// ordinary lull check.
65 FlushNow,
66 /// Drop the listed lanes without flushing or synchronising
67 /// ([`SourceEvent::LanesRetired`](crate::source::SourceEvent::LanesRetired)):
68 /// their input is fully delivered, acknowledged *and* committed, so
69 /// nothing of theirs can sit unflushed in the chain. Pure bookkeeping —
70 /// no barrier, no deadline, no `flush_until`. Forcing a flush here would
71 /// fragment sink batches and park the thread once per completed unit of
72 /// work; sources with anything in flight must use
73 /// [`ThreadControl::StopLanes`] instead.
74 DropLanes { lanes: Vec<LaneId> },
75 /// Stop everything (shutdown): flush the chain, drop all lanes, arrive
76 /// once at `barrier`, and exit the thread.
77 Shutdown {
78 barrier: DrainBarrier,
79 deadline: Instant,
80 },
81}
82
83/// Requests and reports a driver thread sends the controller.
84#[derive(Debug)]
85pub(crate) enum DriverEvent {
86 /// Backpressure tripped: pause these lanes at the source.
87 PauseLanes { lanes: Vec<LaneId> },
88 /// Backpressure cleared: resume these lanes.
89 ResumeLanes { lanes: Vec<LaneId> },
90 /// The chain failed or panicked; the pipeline must stop.
91 Fatal { thread: usize, error: FatalError },
92}
93
94/// What the sink reported when draining at shutdown — the sink layer's
95/// [`DrainReport`](crate::sink::DrainReport), re-exported so assemblies
96/// hand `SinkPool::drain`'s result straight through.
97pub use crate::sink::DrainReport;
98
99/// The sink half the runtime drives: the shared shard-queue handle plus a
100/// drain hook invoked once at shutdown with the remaining drain budget.
101///
102/// Built by the sink layer (`SinkPool`) or by tests; the runtime is
103/// deliberately ignorant of worker internals.
104pub struct SinkRuntime {
105 /// Sending side of the per-shard chunk queues, one entry per installed
106 /// sink (the runtime only uses capacity introspection for the
107 /// backpressure resume gate; the chain's terminal stage holds clones).
108 /// A single-sink pipeline has one entry.
109 pub queues: Vec<ShardQueues>,
110 /// Drain the sinks: flush what's pending within the budget, fail the
111 /// acknowledgements of anything abandoned, and report. For multi-sink
112 /// pipelines this is the composed hook that drains every sink.
113 pub drain: SinkDrainFn,
114 /// Optional connectivity probe (e.g. `SinkPool::probe_all`). The
115 /// runtime probes at startup and then periodically, driving the
116 /// sinks-connected half of `/readyz`. Without a probe the flag is set
117 /// unconditionally.
118 pub probe: Option<SinkProbeFn>,
119}
120
121impl std::fmt::Debug for SinkRuntime {
122 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123 f.debug_struct("SinkRuntime")
124 .field("queues", &self.queues)
125 .finish_non_exhaustive()
126 }
127}
128
129/// Boxed sink drain hook — defined next to the sink layer that produces
130/// it, re-exported here where the runtime consumes it.
131pub use crate::sink::{SinkDrainFn, SinkProbeFn};
132
133/// Terminal state of a pipeline run.
134#[derive(Clone, Debug, PartialEq, Eq)]
135#[non_exhaustive]
136pub enum ExitState {
137 /// Drained and committed cleanly: a requested shutdown (SIGTERM or
138 /// programmatic), or a bounded source reporting
139 /// [`SourceEvent::Drained`](crate::source::SourceEvent::Drained) after
140 /// exhausting its input. For the drained case this additionally
141 /// guarantees every batch was acknowledged and the final watermark
142 /// commit persisted (anything less exits `Failed`).
143 Completed,
144 /// A fatal error stopped the pipeline; the process should exit
145 /// non-zero.
146 Failed(FatalErrorReport),
147}
148
149/// Owned copy of the fatal error carried in the exit report.
150#[derive(Clone, Debug, PartialEq, Eq)]
151#[non_exhaustive]
152pub struct FatalErrorReport {
153 /// Component that failed.
154 pub component: String,
155 /// Human-readable cause.
156 pub reason: String,
157}
158
159impl std::fmt::Display for FatalErrorReport {
160 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
161 write!(f, "pipeline failed in {}: {}", self.component, self.reason)
162 }
163}
164
165impl std::error::Error for FatalErrorReport {}
166
167/// Outcome of [`PipelineRuntime::run`].
168#[derive(Debug)]
169#[non_exhaustive]
170pub struct ExitReport {
171 /// How the run ended.
172 pub state: ExitState,
173 /// The sink's drain report (absent when the sink drain hook could not
174 /// run, e.g. the I/O runtime was already gone).
175 pub sink_drain: Option<DrainReport>,
176 /// The last committed watermark per partition, as reported by the
177 /// final commit.
178 pub final_watermarks: Vec<(PartitionId, i64)>,
179}
180
181impl ExitReport {
182 /// Log the outcome — state, sink drain, final watermarks — at the level
183 /// matching it (`info` for a clean exit, `error` for a failure).
184 pub fn log(&self) {
185 match &self.state {
186 ExitState::Completed => tracing::info!(
187 state = ?self.state,
188 drain = ?self.sink_drain,
189 watermarks = ?self.final_watermarks,
190 "pipeline finished"
191 ),
192 ExitState::Failed(failure) => tracing::error!(
193 component = %failure.component,
194 reason = %failure.reason,
195 drain = ?self.sink_drain,
196 watermarks = ?self.final_watermarks,
197 "pipeline failed"
198 ),
199 }
200 }
201
202 /// The process exit code this outcome maps to: `0` for a clean exit,
203 /// `1` for a failure.
204 #[must_use]
205 pub fn exit_code(&self) -> i32 {
206 match self.state {
207 ExitState::Completed => 0,
208 ExitState::Failed(_) => 1,
209 }
210 }
211
212 /// The report as a `Result`, so a `main` can `?` a failed run: a clean
213 /// exit passes the report through, a failure returns the
214 /// [`FatalErrorReport`] as the error.
215 pub fn ok(self) -> Result<ExitReport, FatalErrorReport> {
216 match &self.state {
217 ExitState::Completed => Ok(self),
218 ExitState::Failed(failure) => Err(failure.clone()),
219 }
220 }
221}
222
223#[cfg(all(test, not(loom)))]
224pub(crate) mod fakes;
225#[cfg(all(test, not(loom)))]
226mod tests;