etl_core/pipeline/
builder.rs

1//! The pipeline builder: the primary assembly path.
2//!
3//! [`Pipeline::from_config`] owns startup initialization — telemetry, the
4//! metrics exporter, and the shared I/O runtime — so holding a `Pipeline`
5//! *guarantees* a live recorder: every metric handle built afterwards
6//! (framework or custom) is live, and connectors get an I/O handle before
7//! any thread spawns. The builder is a thin composition of the public
8//! primitives it replaces; nothing here is required — the desugaring below
9//! remains a fully supported assembly path.
10//!
11//! The shape of an assembly (illustrative — connector construction elided;
12//! see the `etl` crate's examples for complete, compiling binaries):
13//!
14//! ```ignore
15//! let pipeline = Pipeline::from_path(Path::new("pipeline.yaml"))?;
16//! let source = MySource::from_component_config(&pipeline.config().source)?;
17//! let sink = my_connector::from_component_config(&pipeline.config().sink)?;
18//! let report = pipeline
19//!     .sink(sink)?
20//!     .chains(move |ctx| {
21//!         chain_owned::<Row, _>(deserializer.clone())
22//!             .with_metrics(ctx.pipeline, "main")
23//!             .sink(encoder.clone(), KeyHashRouter, ChunkConfig::default(),
24//!                   ctx.queues, ctx.budget)
25//!             .build()
26//!     })
27//!     .run(source)?;
28//! report.log();
29//! std::process::exit(report.exit_code());
30//! ```
31//!
32//! # Desugaring
33//!
34//! Each builder step is a direct lift of the manual assembly it replaces
35//! (all of it public API):
36//!
37//! | Builder | Primitives |
38//! |---|---|
39//! | `from_config(config)` | [`telemetry::init`](crate::telemetry::init) → [`metrics::install`](crate::metrics::install)`(&`[`metrics_settings`](crate::pipeline::metrics_settings)`(&config))` → `tokio::runtime::Builder` (`io_threads` workers) → [`InflightBudget::new`](crate::backpressure::InflightBudget::new) |
40//! | `.sink(bundle)` | [`SinkBundle::into_parts`](crate::sink::SinkBundle::into_parts) → [`shard_queues`](crate::sink::shard_queues) → [`SinkShardMetrics::new`](crate::metrics::SinkShardMetrics::new) per shard → [`SinkPool::spawn`](crate::sink::SinkPool::spawn) → a boxed drain closure |
41//! | `.chains(f)` | the factory handed to [`PipelineRuntime::new`], with queue/budget/name plumbing pre-threaded per call |
42//! | `.into_runtime(source)` / `.run(source)` | [`PipelineRuntime::new`]`(config, source, factory, `[`SinkRuntime`]`{..}, budget)` + [`PipelineRuntime::with_io_runtime`] |
43//!
44//! # Shutdown and drop ordering
45//!
46//! The sink only drains once every [`ShardQueues`] clone is gone. The
47//! builder discharges this structurally: it never exposes the queues
48//! outside the chain factory — each factory call receives a fresh clone in
49//! its [`ChainCtx`], which the chain's terminal stage consumes and drops
50//! with the driver threads, and the wrapper factory itself is dropped by
51//! the runtime before the drain. Do not smuggle `ctx.queues` into
52//! long-lived state outside the returned chain; a clone that outlives the
53//! drivers turns a graceful drain into a deadline-bounded abandon.
54
55use super::SinkRuntime;
56use super::runtime::{
57    PipelineRuntime, RuntimeOptions, StartError, install_or_reuse, metrics_settings,
58};
59use crate::backpressure::InflightBudget;
60use crate::config::{ConfigError, PipelineConfig};
61use crate::metrics::{ComponentLabels, MetricsHandle, SinkShardMetrics};
62use crate::ops::RunnableChain;
63use crate::pipeline::ExitReport;
64use crate::sink::{ShardQueues, SinkBundle, SinkDrainFn, SinkPool, SinkProbeFn, shard_queues};
65use crate::source::Source;
66use crate::telemetry::{self, LogFormat};
67use std::path::Path;
68use std::sync::Arc;
69
70/// Error assembling a pipeline (cold path, before anything runs).
71#[derive(Debug, thiserror::Error)]
72#[non_exhaustive]
73pub enum BuildError {
74    /// The configuration failed to load or validate.
75    #[error(transparent)]
76    Config(#[from] ConfigError),
77    /// The metrics exporter failed to install.
78    #[error("metrics: {0}")]
79    Metrics(String),
80    /// The I/O runtime failed to build.
81    #[error("io runtime: {0}")]
82    Io(#[from] std::io::Error),
83    /// The sink bundle's topology or labels are unusable.
84    #[error("sink: {0}")]
85    Sink(String),
86    /// [`Pipeline::sink`] was called twice.
87    #[error("a sink is already installed")]
88    SinkAlreadySet,
89    /// [`Pipeline::into_runtime`]/[`Pipeline::run`] without a sink.
90    #[error("no sink installed (call Pipeline::sink first)")]
91    MissingSink,
92    /// [`Pipeline::into_runtime`]/[`Pipeline::run`] without a chain factory.
93    #[error("no chain factory installed (call Pipeline::chains first)")]
94    MissingChains,
95    /// The builder was constructed inside an async runtime. It owns a
96    /// blocking tokio runtime (dropping or `block_on`-ing one inside async
97    /// context panics), so build pipelines from a plain thread — usually
98    /// `main`.
99    #[error(
100        "Pipeline::from_config must be called outside any async runtime \
101         (it owns a blocking tokio runtime)"
102    )]
103    AsyncContext,
104}
105
106/// Error from [`Pipeline::run`]: assembly or startup failure.
107#[derive(Debug, thiserror::Error)]
108#[non_exhaustive]
109pub enum PipelineError {
110    /// The pipeline could not be assembled.
111    #[error(transparent)]
112    Build(#[from] BuildError),
113    /// The assembled pipeline failed to start.
114    #[error(transparent)]
115    Start(#[from] StartError),
116}
117
118/// Per-thread wiring handed to the chain factory — everything the terminal
119/// [`.sink(...)`](crate::ops::ChainBuilder) stage needs, so assemblies stop
120/// threading queues, budget, and the pipeline name by hand.
121///
122/// Passed by value, once per pipeline thread; move the fields into the
123/// chain being built. Deliberately not `Clone` — see the module docs on
124/// drop ordering.
125#[derive(Debug)]
126#[non_exhaustive]
127pub struct ChainCtx {
128    /// Zero-based pipeline thread index.
129    pub thread: usize,
130    /// This thread's clone of the shard-queue senders.
131    pub queues: ShardQueues,
132    /// The shared in-flight byte budget.
133    pub budget: Arc<InflightBudget>,
134    /// The pipeline name — [`ChainBuilder::with_metrics`](crate::ops::ChainBuilder::with_metrics)'s
135    /// first argument.
136    pub pipeline: String,
137}
138
139/// Sink wiring knobs that live outside connector config.
140#[derive(Clone, Debug)]
141#[non_exhaustive]
142pub struct SinkOptions {
143    /// Per-shard chunk queue capacity, in chunks. The default suits most
144    /// pipelines; see `docs/DESIGN.md` § Backpressure for the sizing rule.
145    pub queue_capacity: usize,
146}
147
148impl SinkOptions {
149    /// Override the per-shard queue capacity. (`SinkOptions` is
150    /// `#[non_exhaustive]`, so construct via `default()` + `with_*`.)
151    #[must_use]
152    pub fn with_queue_capacity(mut self, capacity: usize) -> Self {
153        self.queue_capacity = capacity;
154        self
155    }
156}
157
158impl Default for SinkOptions {
159    fn default() -> Self {
160        SinkOptions { queue_capacity: 8 }
161    }
162}
163
164struct SinkAssembly {
165    queues: ShardQueues,
166    drain: SinkDrainFn,
167    probe: Option<SinkProbeFn>,
168}
169
170type ChainFactoryFn = Box<dyn FnMut(ChainCtx) -> Box<dyn RunnableChain> + Send>;
171
172/// The pipeline builder — see the [module docs](self) for the full picture.
173///
174/// Non-generic, nameable, and storable: the source type enters only at the
175/// terminal [`into_runtime`](Self::into_runtime)/[`run`](Self::run) call.
176pub struct Pipeline {
177    config: PipelineConfig,
178    metrics: MetricsHandle,
179    io: tokio::runtime::Runtime,
180    budget: Arc<InflightBudget>,
181    sink: Option<SinkAssembly>,
182    chains: Option<ChainFactoryFn>,
183    options: RuntimeOptions,
184}
185
186impl std::fmt::Debug for Pipeline {
187    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
188        f.debug_struct("Pipeline")
189            .field("pipeline", &self.config.pipeline.name)
190            .field("sink", &self.sink.is_some())
191            .field("chains", &self.chains.is_some())
192            .finish_non_exhaustive()
193    }
194}
195
196impl Pipeline {
197    /// Load configuration from a YAML file and initialize the process; see
198    /// [`from_config`](Self::from_config).
199    pub fn from_path(path: &Path) -> Result<Self, BuildError> {
200        Self::from_config(PipelineConfig::from_path(path)?)
201    }
202
203    /// Initialize the process from an already-loaded configuration:
204    ///
205    /// 1. **Telemetry** — [`telemetry::init`]`(Json, "info")`. Idempotent:
206    ///    to customize the format or filter, call [`telemetry::init`]
207    ///    yourself *first* (the binaries-init convention).
208    /// 2. **Metrics exporter** — installed from the config's `metrics`
209    ///    section before you can construct any handle, so every handle
210    ///    built while holding the `Pipeline` is live. When a foreign
211    ///    recorder already owns the process, the pipeline continues
212    ///    against it with a warning.
213    /// 3. **The I/O runtime** — `pipeline.io_threads` workers, thread name
214    ///    `etl-io`. Connectors that need a handle before `run` (schema
215    ///    fetchers, async pre-flight validation) use
216    ///    [`io_handle`](Self::io_handle)/[`block_on`](Self::block_on).
217    ///
218    /// # Errors
219    ///
220    /// [`BuildError::AsyncContext`] when called from inside an async
221    /// runtime — build pipelines from a plain thread, usually `main`.
222    pub fn from_config(config: PipelineConfig) -> Result<Self, BuildError> {
223        if tokio::runtime::Handle::try_current().is_ok() {
224            return Err(BuildError::AsyncContext);
225        }
226        if config.pipeline.io_threads == 0 {
227            return Err(BuildError::Config(ConfigError::Validation(
228                "pipeline.io_threads must be non-zero".into(),
229            )));
230        }
231        telemetry::init(LogFormat::Json, "info");
232        let metrics = install_or_reuse(&metrics_settings(&config)).map_err(|e| match e {
233            StartError::Metrics(m) => BuildError::Metrics(m),
234            other => BuildError::Metrics(other.to_string()),
235        })?;
236        let io = tokio::runtime::Builder::new_multi_thread()
237            .worker_threads(config.pipeline.io_threads)
238            .thread_name("etl-io")
239            .enable_all()
240            .build()?;
241        Ok(Pipeline {
242            config,
243            metrics,
244            io,
245            budget: Arc::new(InflightBudget::new()),
246            sink: None,
247            chains: None,
248            options: RuntimeOptions::default(),
249        })
250    }
251
252    /// The loaded configuration — connector sections (`config().source`,
253    /// `.deserializer`, `.sink`) still belong to the caller's connector
254    /// factories.
255    #[must_use]
256    pub fn config(&self) -> &PipelineConfig {
257        &self.config
258    }
259
260    /// The installed exporter's handle (rendering, upkeep).
261    #[must_use]
262    pub fn metrics(&self) -> &MetricsHandle {
263        &self.metrics
264    }
265
266    /// The shared in-flight byte budget.
267    #[must_use]
268    pub fn budget(&self) -> &Arc<InflightBudget> {
269        &self.budget
270    }
271
272    /// A handle to the I/O runtime, for connector edge work that must
273    /// start before the chain exists (schema-registry fetchers, ...).
274    /// Valid until `run` returns.
275    #[must_use]
276    pub fn io_handle(&self) -> tokio::runtime::Handle {
277        self.io.handle().clone()
278    }
279
280    /// Run a future on the I/O runtime, blocking this thread — for async
281    /// pre-flight steps such as schema validation.
282    pub fn block_on<F: Future>(&self, future: F) -> F::Output {
283        self.io.block_on(future)
284    }
285
286    /// Install the sink with default [`SinkOptions`]; see
287    /// [`sink_with`](Self::sink_with).
288    pub fn sink<B: SinkBundle>(self, bundle: B) -> Result<Self, BuildError> {
289        self.sink_with(bundle, SinkOptions::default())
290    }
291
292    /// Install the sink: builds the per-shard chunk queues, registers the
293    /// per-shard metrics (E2E basis from the config), spawns the
294    /// [`SinkPool`] workers on the I/O runtime, and wires the drain and
295    /// readiness probe.
296    ///
297    /// # Errors
298    ///
299    /// [`BuildError::SinkAlreadySet`] on a second call;
300    /// [`BuildError::Sink`] for an empty or ragged topology, label shapes
301    /// that do not match it, or a zero queue capacity.
302    pub fn sink_with<B: SinkBundle>(
303        mut self,
304        bundle: B,
305        options: SinkOptions,
306    ) -> Result<Self, BuildError> {
307        if self.sink.is_some() {
308            return Err(BuildError::SinkAlreadySet);
309        }
310        if options.queue_capacity == 0 {
311            return Err(BuildError::Sink("queue_capacity must be non-zero".into()));
312        }
313        let parts = bundle.into_parts();
314        let num_shards = parts.shard_endpoints.len();
315        if num_shards == 0 {
316            return Err(BuildError::Sink("sink topology has no shards".into()));
317        }
318        if let Some(shard) = parts.shard_endpoints.iter().position(Vec::is_empty) {
319            return Err(BuildError::Sink(format!("shard {shard} has no replicas")));
320        }
321        let replica_labels = parts.effective_replica_labels();
322        let label_shape: Vec<usize> = replica_labels.iter().map(Vec::len).collect();
323        let endpoint_shape: Vec<usize> = parts.shard_endpoints.iter().map(Vec::len).collect();
324        if label_shape != endpoint_shape {
325            return Err(BuildError::Sink(format!(
326                "replica_labels shape {label_shape:?} does not match the \
327                 endpoint topology {endpoint_shape:?}"
328            )));
329        }
330
331        let name = self.config.pipeline.name.clone();
332        let (queues, receivers) = shard_queues(num_shards, options.queue_capacity);
333        let sink_labels = ComponentLabels::new(name.clone(), "sink", parts.component_type.clone());
334        let e2e_basis = metrics_settings(&self.config).e2e_basis;
335        let shard_metrics: Vec<SinkShardMetrics> = replica_labels
336            .iter()
337            .enumerate()
338            .map(|(shard, replicas)| {
339                SinkShardMetrics::new(
340                    &sink_labels,
341                    u32::try_from(shard).unwrap_or(u32::MAX),
342                    replicas,
343                    e2e_basis,
344                )
345            })
346            .collect();
347        let pool = SinkPool::spawn(
348            Arc::new(parts.writer),
349            parts.shard_endpoints,
350            receivers,
351            parts.pool,
352            Arc::clone(&self.budget),
353            shard_metrics,
354            &name,
355            self.io.handle(),
356        );
357        self.sink = Some(SinkAssembly {
358            queues,
359            drain: Box::new(move |deadline| Box::pin(async move { pool.drain(deadline).await })),
360            probe: parts.probe,
361        });
362        Ok(self)
363    }
364
365    /// Install the chain factory, called once per pipeline thread with
366    /// that thread's [`ChainCtx`]. Composition inside the closure is fully
367    /// monomorphized ([`chain_owned`](crate::ops::chain_owned) and
368    /// friends); the returned `Box<dyn RunnableChain>` is the same single
369    /// per-batch erasure boundary as always.
370    #[must_use]
371    pub fn chains<F>(mut self, factory: F) -> Self
372    where
373        F: FnMut(ChainCtx) -> Box<dyn RunnableChain> + Send + 'static,
374    {
375        self.chains = Some(Box::new(factory));
376        self
377    }
378
379    /// Override the runtime options (signal handling, loop timings).
380    #[must_use]
381    pub fn runtime_options(mut self, options: RuntimeOptions) -> Self {
382        self.options = options;
383        self
384    }
385
386    /// Finish assembly into a [`PipelineRuntime`] — for callers that need
387    /// [`shutdown_handle`](PipelineRuntime::shutdown_handle) before a
388    /// spawned `run` (tests, embedded pipelines). The I/O runtime moves
389    /// into it and is shut down when `run` returns.
390    ///
391    /// # Errors
392    ///
393    /// [`BuildError::MissingSink`] / [`BuildError::MissingChains`] when a
394    /// step was skipped.
395    pub fn into_runtime<S: Source + 'static>(
396        mut self,
397        source: S,
398    ) -> Result<PipelineRuntime<S>, BuildError> {
399        let assembly = self.sink.take().ok_or(BuildError::MissingSink)?;
400        let mut factory = self.chains.take().ok_or(BuildError::MissingChains)?;
401        let queues = assembly.queues.clone();
402        let budget = Arc::clone(&self.budget);
403        let name = self.config.pipeline.name.clone();
404        // This wrapper is the factory the runtime drops before the sink
405        // drain — the queue clone it captures dies exactly there.
406        let chains = move |thread: usize| {
407            factory(ChainCtx {
408                thread,
409                queues: queues.clone(),
410                budget: Arc::clone(&budget),
411                pipeline: name.clone(),
412            })
413        };
414        Ok(PipelineRuntime::new(
415            self.config,
416            source,
417            chains,
418            SinkRuntime {
419                queues: assembly.queues,
420                drain: assembly.drain,
421                probe: assembly.probe,
422            },
423            self.budget,
424        )
425        .with_options(self.options)
426        .with_io_runtime(self.io))
427    }
428
429    /// [`into_runtime`](Self::into_runtime) + [`PipelineRuntime::run`]:
430    /// run the pipeline to completion, blocking until a shutdown signal
431    /// drains it or a fatal error stops it.
432    pub fn run<S: Source + 'static>(self, source: S) -> Result<ExitReport, PipelineError> {
433        Ok(self.into_runtime(source)?.run()?)
434    }
435}
436
437#[cfg(all(test, not(loom)))]
438mod tests {
439    use super::*;
440    use crate::error::SinkError;
441    use crate::pipeline::ExitState;
442    use crate::pipeline::fakes::{
443        ChainMode, ChainShared, FakeChain, FakeSource, LaneSpec, Script, SourceLog, batches,
444        test_config, test_options, wait_for,
445    };
446    use crate::record::PartitionId;
447    use crate::sink::{SealedBatch, SinkParts, SinkPoolConfig};
448    use crate::source::LaneId;
449    use std::sync::Mutex;
450    use std::time::Duration;
451
452    struct NullWriter;
453    impl crate::sink::ShardWriter for NullWriter {
454        type Endpoint = ();
455        async fn write_batch(&self, (): &(), _batch: &SealedBatch) -> Result<(), SinkError> {
456            Ok(())
457        }
458    }
459
460    fn null_sink(shards: usize) -> SinkParts<NullWriter> {
461        SinkParts::new(
462            NullWriter,
463            (0..shards).map(|_| vec![()]).collect(),
464            SinkPoolConfig::default(),
465        )
466        .with_component_type("null")
467    }
468
469    fn fake_chain(
470        shared: &Arc<ChainShared>,
471        log: &Arc<Mutex<SourceLog>>,
472    ) -> Box<dyn RunnableChain> {
473        Box::new(FakeChain {
474            shared: Arc::clone(shared),
475            log: Arc::clone(log),
476            mode: ChainMode::Ok,
477            batches_seen: 0,
478        })
479    }
480
481    #[test]
482    fn missing_sink_then_missing_chains_error() {
483        let (source, _shared, _script) = FakeSource::new();
484        let p = Pipeline::from_config(test_config(1)).expect("builder");
485        assert!(matches!(
486            p.into_runtime(source).err(),
487            Some(BuildError::MissingSink)
488        ));
489
490        let (source, _shared, _script) = FakeSource::new();
491        let p = Pipeline::from_config(test_config(1))
492            .expect("builder")
493            .sink(null_sink(1))
494            .expect("sink");
495        assert!(matches!(
496            p.into_runtime(source).err(),
497            Some(BuildError::MissingChains)
498        ));
499    }
500
501    #[test]
502    fn second_sink_errors() {
503        let p = Pipeline::from_config(test_config(1))
504            .expect("builder")
505            .sink(null_sink(1))
506            .expect("first sink");
507        assert!(matches!(
508            p.sink(null_sink(1)).err(),
509            Some(BuildError::SinkAlreadySet)
510        ));
511    }
512
513    #[test]
514    fn bad_topologies_error_instead_of_panicking() {
515        let p = Pipeline::from_config(test_config(1)).expect("builder");
516        let empty = SinkParts::new(NullWriter, Vec::new(), SinkPoolConfig::default());
517        assert!(matches!(p.sink(empty).err(), Some(BuildError::Sink(_))));
518
519        let p = Pipeline::from_config(test_config(1)).expect("builder");
520        let ragged = SinkParts::new(
521            NullWriter,
522            vec![vec![()], vec![]],
523            SinkPoolConfig::default(),
524        );
525        assert!(matches!(p.sink(ragged).err(), Some(BuildError::Sink(_))));
526
527        let p = Pipeline::from_config(test_config(1)).expect("builder");
528        let bad_labels = SinkParts::new(NullWriter, vec![vec![()]], SinkPoolConfig::default())
529            .with_replica_labels(vec![vec!["a".into(), "b".into()]]);
530        assert!(matches!(
531            p.sink(bad_labels).err(),
532            Some(BuildError::Sink(_))
533        ));
534
535        let p = Pipeline::from_config(test_config(1)).expect("builder");
536        assert!(matches!(
537            p.sink_with(null_sink(1), SinkOptions { queue_capacity: 0 })
538                .err(),
539            Some(BuildError::Sink(_))
540        ));
541    }
542
543    #[tokio::test]
544    async fn from_config_inside_async_context_errors() {
545        assert!(matches!(
546            Pipeline::from_config(test_config(1)).err(),
547            Some(BuildError::AsyncContext)
548        ));
549    }
550
551    /// The chain factory sees every thread index exactly once, with the
552    /// pipeline name from the config, and the assembled pipeline runs to a
553    /// clean `Completed` through the real `SinkPool`. This guards `ChainCtx`
554    /// coverage and end-to-end assembly — not drop ordering: the drain
555    /// containment fix (`sink/worker.rs`) deliberately converts a leaked
556    /// `ShardQueues` clone from an unbounded hang into a bounded, loud
557    /// abandon, so completion here no longer implies clean drop ordering.
558    /// The drop-ordering + at-least-once contract is covered where it *can*
559    /// still fail observably: the whole-assembly test in `etl-test`'s
560    /// `tests/bundle.rs`, which routes real data through `ctx.queues` and
561    /// asserts the watermark only advances past the last record after a
562    /// durable write.
563    #[test]
564    fn chain_ctx_covers_every_thread_and_run_completes() {
565        let (source, shared, script) = FakeSource::new();
566        script
567            .lock()
568            .unwrap()
569            .push_back(Script::Assign(vec![LaneSpec {
570                id: LaneId(0),
571                partition: PartitionId(0),
572                batches: batches(&[0..10, 10..20]),
573            }]));
574        let chain_shared = Arc::new(ChainShared::default());
575        let seen_threads: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
576
577        let cs = Arc::clone(&chain_shared);
578        let log = Arc::clone(&shared);
579        let seen = Arc::clone(&seen_threads);
580        let runtime = Pipeline::from_config(test_config(2))
581            .expect("builder")
582            .sink(null_sink(1))
583            .expect("sink")
584            .chains(move |ctx| {
585                assert_eq!(ctx.pipeline, "test");
586                seen.lock().unwrap().push(ctx.thread);
587                fake_chain(&cs, &log)
588            })
589            .runtime_options(test_options())
590            .into_runtime(source)
591            .expect("into_runtime");
592
593        let shutdown = runtime.shutdown_handle();
594        let join = std::thread::spawn(move || runtime.run());
595        wait_for("payloads consumed", Duration::from_secs(5), || {
596            chain_shared
597                .consumed
598                .load(std::sync::atomic::Ordering::Relaxed)
599                == 20
600        });
601        shutdown.trigger();
602        let report = join.join().unwrap().unwrap();
603        assert_eq!(report.state, ExitState::Completed);
604
605        let mut threads = seen_threads.lock().unwrap().clone();
606        threads.sort_unstable();
607        assert_eq!(threads, vec![0, 1], "one ChainCtx per pipeline thread");
608    }
609
610    /// The whole-builder happy path through `run()` (not `into_runtime`),
611    /// exercised over the real SinkPool: completes and commits.
612    #[test]
613    fn run_completes_via_builder_terminal() {
614        let (source, shared, script) = FakeSource::new();
615        script
616            .lock()
617            .unwrap()
618            .push_back(Script::Assign(vec![LaneSpec {
619                id: LaneId(0),
620                partition: PartitionId(0),
621                batches: batches(std::slice::from_ref(&(0..5))),
622            }]));
623        let chain_shared = Arc::new(ChainShared::default());
624        let cs = Arc::clone(&chain_shared);
625        let log = Arc::clone(&shared);
626
627        let pipeline = Pipeline::from_config(test_config(1))
628            .expect("builder")
629            .sink(null_sink(2))
630            .expect("sink")
631            .chains(move |_ctx| fake_chain(&cs, &log))
632            .runtime_options(test_options());
633
634        // Drive shutdown from a watcher thread once the payloads land.
635        let consumed = Arc::clone(&chain_shared);
636        let runtime = pipeline.into_runtime(source).expect("into_runtime");
637        let shutdown = runtime.shutdown_handle();
638        std::thread::spawn(move || {
639            wait_for("payloads consumed", Duration::from_secs(5), || {
640                consumed.consumed.load(std::sync::atomic::Ordering::Relaxed) == 5
641            });
642            shutdown.trigger();
643        });
644        let report = runtime.run().unwrap();
645        assert_eq!(report.state, ExitState::Completed);
646        assert_eq!(report.final_watermarks, vec![(PartitionId(0), 5)]);
647    }
648}
etl_core/pipeline/builder.rs

etl_core/pipeline/
builder.rs
