spate_core/pipeline/

builder.rs

//! The pipeline builder: the primary assembly path.
//!
//! [`Pipeline::from_config`] owns startup initialization (telemetry, the
//! metrics exporter, and the shared I/O runtime). Every metric handle built
//! afterwards (framework or custom) binds to the process's recorder, which
//! always exists under `exporter: prometheus`, and connectors get an I/O
//! handle before any thread spawns. The builder is a thin composition of the public
//! primitives it replaces; nothing here is required, and the desugaring
//! below remains a fully supported assembly path.
//!
//! The shape of an assembly (illustrative; connector construction is
//! elided, and the `spate` crate's examples carry complete, compiling
//! binaries):
//!
//! ```ignore
//! let pipeline = Pipeline::from_path(Path::new("pipeline.yaml"))?;
//! let source = MySource::from_component_config(&pipeline.config().source)?;
//! let sink = my_connector::from_component_config(pipeline.config().sink_config("default")?)?;
//! let report = pipeline
//!     .sink(sink)?
//!     .chains(move |ctx| {
//!         let chunk_cfg = ctx.chunk(); // bind before `with_metrics` moves `ctx.pipeline`
//!         chain_owned::<Row, _>(deserializer.clone())
//!             .with_metrics(ctx.pipeline, "main")
//!             .sink(encoder.clone(), KeyHashRouter, chunk_cfg,
//!                   ctx.queues, ctx.budget)
//!             .build()
//!     })
//!     .run(source)?;
//! report.log();
//! std::process::exit(report.exit_code());
//! ```
//!
//! # Desugaring
//!
//! Each builder step is a direct lift of the manual assembly it replaces
//! (all of it public API):
//!
//! | Builder | Primitives |
//! |---|---|
//! | `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) |
//! | `.sink(bundle)` | [`SinkBundle::into_parts`](crate::sink::SinkBundle::into_parts) → [`shard_queues`](crate::sink::shard_queues) → [`ShardQueues::with_frame_codec`](crate::sink::ShardQueues::with_frame_codec) when the parts carry a [`frame_codec`](crate::sink::SinkParts::frame_codec) → [`SinkShardMetrics::try_new`](crate::metrics::SinkShardMetrics::try_new) per shard → [`SinkPool::spawn`](crate::sink::SinkPool::spawn) → a boxed drain closure. It also resolves this sink's [`ChunkConfig`] from the YAML `chunk:` block / [`SinkOptions::with_chunk`]. This is the one builder step without a manual-assembly equivalent, since the config layer is what carries `chunk:`. |
//! | `.chains(f)` | the factory handed to [`PipelineRuntime::new`], with queue/budget/name plumbing pre-threaded per call |
//! | `.feed(feed)` | [`PipelineRuntime::with_feed`] |
//! | `.into_runtime(source)` / `.run(source)` | [`PipelineRuntime::new`]`(config, source, factory, `[`SinkRuntime`]`{..}, budget)` + [`PipelineRuntime::with_io_runtime`] |
//!
//! # Shutdown and drop ordering
//!
//! The sink only drains once every [`ShardQueues`] clone is gone. The
//! builder discharges this structurally by never exposing the queues
//! outside the chain factory. Each factory call receives a fresh clone in
//! its [`ChainCtx`], which the chain's terminal stage consumes and drops
//! with the driver threads, and the wrapper factory itself is dropped by
//! the runtime before the drain. Do not smuggle `ctx.queues` into
//! long-lived state outside the returned chain; a clone that outlives the
//! drivers turns a graceful drain into a deadline-bounded abandon.

use super::SinkRuntime;
use super::runtime::{
    PipelineRuntime, RuntimeOptions, StartError, install_or_reuse, metrics_settings,
};
use crate::backpressure::InflightBudget;
use crate::config::{ConfigError, PipelineConfig};
use crate::feed::Feed;
use crate::framing::FramingContract;
use crate::metrics::{
    ComponentLabels, Meter, MetricRole, MetricsHandle, SharedString, SinkShardMetrics,
};
use crate::ops::{ChunkConfig, RunnableChain, SplitTarget};
use crate::pipeline::ExitReport;
use crate::sink::{
    DrainReport, ShardQueues, ShardWriter, SinkBundle, SinkDrainFn, SinkFailures, SinkMetricsCtx,
    SinkPool, SinkProbeFn, shard_queues,
};
use crate::source::Source;
use crate::telemetry::{self, LogFormat};
use std::path::Path;
use std::sync::Arc;

/// Error assembling a pipeline (cold path, before anything runs).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum BuildError {
    /// The configuration failed to load or validate.
    #[error(transparent)]
    Config(#[from] ConfigError),
    /// The metrics exporter failed to install.
    #[error("metrics: {0}")]
    Metrics(String),
    /// The I/O runtime failed to build.
    #[error("io runtime: {0}")]
    Io(#[from] std::io::Error),
    /// The sink bundle's topology or labels are unusable.
    #[error("sink: {0}")]
    Sink(String),
    /// [`Pipeline::add_sink`] was called twice with the same name (a second
    /// bare [`Pipeline::sink`] collides on the reserved `"default"` name).
    #[error("a sink named {0:?} is already installed")]
    DuplicateSinkName(String),
    /// Another live handle set in this process already owns a metric series
    /// this sink would publish, usually a second pipeline with the same
    /// pipeline and sink name. Gauge series cannot be shared, so assembly
    /// stops here rather than letting one of the two publish readings the
    /// other overwrites. A pipeline rebuilt *sequentially* is fine: drop the
    /// old one first.
    #[error("{0}")]
    DuplicateSeries(String),
    /// [`Pipeline::into_runtime`]/[`Pipeline::run`] without a sink.
    #[error("no sink installed (call Pipeline::sink or Pipeline::add_sink first)")]
    MissingSink,
    /// [`Pipeline::into_runtime`]/[`Pipeline::run`] without a chain factory.
    #[error("no chain factory installed (call Pipeline::chains first)")]
    MissingChains,
    /// The builder was constructed inside an async runtime. It owns a
    /// blocking tokio runtime (dropping or `block_on`-ing one inside async
    /// context panics), so build pipelines from a plain thread, usually
    /// `main`.
    #[error(
        "Pipeline::from_config must be called outside any async runtime \
         (it owns a blocking tokio runtime)"
    )]
    AsyncContext,
}

/// Error from [`Pipeline::run`]: assembly or startup failure.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum PipelineError {
    /// The pipeline could not be assembled.
    #[error(transparent)]
    Build(#[from] BuildError),
    /// The assembled pipeline failed to start.
    #[error(transparent)]
    Start(#[from] StartError),
}

/// Per-thread wiring handed to the chain factory, carrying everything the
/// terminal [`.sink(...)`](crate::ops::ChainBuilder) stage needs, so assemblies
/// stop threading queues, budget, and the pipeline name by hand.
///
/// Passed by value, once per pipeline thread; move the fields into the
/// chain being built. Not `Clone` (see the module docs on drop ordering).
#[derive(Debug)]
#[non_exhaustive]
pub struct ChainCtx {
    /// Zero-based pipeline thread index.
    pub thread: usize,
    /// This thread's clone of the shard-queue senders for the **first**
    /// installed sink, the back-compat handle for single-sink pipelines
    /// (`.sink(...)`). Multi-sink pipelines resolve each branch's queues by
    /// name via [`sink`](Self::sink) instead.
    pub queues: ShardQueues,
    /// The shared in-flight byte budget.
    pub budget: Arc<InflightBudget>,
    /// The pipeline name
    /// ([`ChainBuilder::with_metrics`](crate::ops::ChainBuilder::with_metrics)'s
    /// first argument).
    pub pipeline: String,
    /// How the source frames its payloads
    /// ([`Source::framing_contract`](crate::source::Source::framing_contract)).
    /// Hand it to a deserializer builder (e.g. `JsonDeserializerBuilder::
    /// for_source_framing`) so the deserializer's granularity is derived from
    /// the source and a double-framing configuration is rejected.
    pub source_framing: FramingContract,
    /// The **first** installed sink's resolved terminal-stage chunking, the
    /// back-compat handle for single-sink pipelines, mirroring [`queues`](Self::queues).
    /// Pass it to the chain's [`.sink(...)`](crate::ops::ChainBuilder::sink)
    /// terminal via [`chunk`](Self::chunk); split pipelines resolve each
    /// branch's chunk by name through [`sink`](Self::sink) instead.
    chunk: ChunkConfig,
    /// This thread's clone of every installed sink's queues and resolved
    /// chunking, keyed by the name passed to [`Pipeline::add_sink`]. Resolved
    /// through [`sink`](Self::sink). The clones die with the driver.
    named: Vec<(String, ShardQueues, ChunkConfig)>,
}

impl ChainCtx {
    /// The named sink's handles (name, shard queues, shared in-flight budget)
    /// for a split-terminal branch. Pass the result straight to
    /// [`SplitBuilder::add`](crate::ops::SplitBuilder::add). The single-sink
    /// `.sink()` sugar installs its sink under the name `"default"`.
    ///
    /// # Panics
    ///
    /// Panics if no sink was installed under `name`. The chain factory runs
    /// once per thread, on the cold path, before any data flows.
    #[must_use]
    pub fn sink(&self, name: &str) -> SplitTarget {
        let (_, queues, chunk) = self
            .named
            .iter()
            .find(|(n, _, _)| n == name)
            .unwrap_or_else(|| {
                let known: Vec<&str> = self.named.iter().map(|(n, _, _)| n.as_str()).collect();
                panic!("ChainCtx::sink: no sink named {name:?} (installed sinks: {known:?})")
            });
        SplitTarget::new(name.to_string(), queues.clone(), Arc::clone(&self.budget))
            .with_chunk(*chunk)
    }

    /// The **first** installed sink's resolved terminal-stage chunking, the
    /// single-sink counterpart to [`queues`](Self::queues), fed straight to the
    /// chain's [`.sink(...)`](crate::ops::ChainBuilder::sink) terminal:
    ///
    /// ```
    /// # use spate_core::deser::{BytesPassthrough, Owned};
    /// # use spate_core::error::SinkError;
    /// # use spate_core::ops::chain;
    /// # use spate_core::pipeline::ChainCtx;
    /// # use spate_core::record::Record;
    /// # use spate_core::sink::{KeyHashRouter, RowEncoder};
    /// # #[derive(Clone)]
    /// # struct RowBytes;
    /// # impl RowEncoder<Owned<Vec<u8>>> for RowBytes {
    /// #     fn encode<'buf>(
    /// #         &mut self,
    /// #         rec: &Record<Vec<u8>>,
    /// #         buf: &mut bytes::BytesMut,
    /// #     ) -> Result<(), SinkError> {
    /// #         buf.extend_from_slice(&rec.payload);
    /// #         Ok(())
    /// #     }
    /// # }
    /// # fn wire(ctx: ChainCtx) {
    /// # let encoder = RowBytes;
    /// let chunk_cfg = ctx.chunk(); // bind before `with_metrics` moves `ctx.pipeline`
    /// let orders = chain(BytesPassthrough)
    ///     .with_metrics(ctx.pipeline, "orders")
    ///     .sink(encoder, KeyHashRouter, chunk_cfg, ctx.queues, ctx.budget)
    ///     .build();
    /// # let _ = orders;
    /// # }
    /// ```
    ///
    /// It resolves the per-sink YAML `chunk:` block (or `SinkOptions::with_chunk`,
    /// or the 64 KiB default) at assembly time. Split pipelines take each
    /// branch's chunk from [`sink`](Self::sink) instead.
    #[must_use]
    pub fn chunk(&self) -> ChunkConfig {
        self.chunk
    }

    /// A [`Meter`] for a pipeline author's own metrics, pre-labeled with the
    /// pipeline name plus the `component` / `component_type` you name and
    /// scoped to the `spate_custom_` namespace. Resolve handles from it **once
    /// here** (the factory runs once per thread, before data flows) and move
    /// them into the operator closures that touch them:
    ///
    /// ```no_run
    /// # use spate_core::pipeline::ChainCtx;
    /// # fn wire(ctx: ChainCtx) {
    /// // Pass the LOCAL name — this registers `spate_custom_enrich_hits_total`.
    /// let hits = ctx.meter("enrich", "map").counter("enrich_hits_total", &[]);
    /// // ... move `hits` into a `.inspect(move |r| { hits.increment(1); })`
    /// # let _ = hits;
    /// # }
    /// ```
    ///
    /// The resulting series carry `pipeline`/`component`/`component_type` like
    /// every framework series and live under the `spate_` umbrella, so they join
    /// cleanly in a query. You pass local names; the `Meter` adds the
    /// `spate_custom_` prefix. [The metrics reference] carries the umbrella and
    /// the label set in full.
    ///
    /// [the metrics reference]: https://spate.kainth.dev/docs/METRICS
    #[must_use]
    pub fn meter(
        &self,
        component: impl Into<SharedString>,
        component_type: impl Into<SharedString>,
    ) -> Meter {
        Meter::new(self.pipeline.clone(), component, component_type)
    }
}

/// Sink wiring knobs that live outside connector config.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct SinkOptions {
    /// Per-shard chunk queue capacity, in chunks. The default suits most
    /// pipelines. Queued chunks are charged to the in-flight byte budget, so
    /// `shards × queue_capacity × chunk.target_bytes` has to fit under the
    /// budget's low watermark *alongside* pending writes. When it does not, a
    /// saturated pipeline sits permanently above the high watermark and the
    /// pause controller duty-cycles at its minimum pause, collapsing
    /// throughput with nothing otherwise wrong. [The backpressure page][bp]
    /// carries the full sizing rule and a worked example.
    ///
    /// [bp]: https://spate.kainth.dev/docs/user-guide/concepts/backpressure
    pub queue_capacity: usize,
    /// Programmatic override for this sink's terminal-stage chunking. `None`
    /// (the default) defers to the per-sink YAML `chunk:` block, or to
    /// [`ChunkConfig::default`] if that is absent too. Setting it **and** a
    /// YAML `chunk:` block on the same sink is a decl-once
    /// [`ConfigError`](crate::config::ConfigError) at install; the knob is
    /// declared in exactly one place. The resolved value reaches the chain
    /// terminal via [`ChainCtx::chunk`] / [`ChainCtx::sink`].
    pub chunk: Option<ChunkConfig>,
}

impl SinkOptions {
    /// Override the per-shard queue capacity. (`SinkOptions` is
    /// `#[non_exhaustive]`, so construct via `default()` + `with_*`.)
    #[must_use]
    pub fn with_queue_capacity(mut self, capacity: usize) -> Self {
        self.queue_capacity = capacity;
        self
    }

    /// Set this sink's chunking programmatically, instead of via the YAML
    /// `chunk:` block. Providing both on the same sink is a load error (see
    /// [`chunk`](Self::chunk)).
    #[must_use]
    pub fn with_chunk(mut self, chunk: ChunkConfig) -> Self {
        self.chunk = Some(chunk);
        self
    }
}

impl Default for SinkOptions {
    fn default() -> Self {
        SinkOptions {
            queue_capacity: 8,
            chunk: None,
        }
    }
}

struct SinkAssembly {
    queues: ShardQueues,
    drain: SinkDrainFn,
    probe: Option<SinkProbeFn>,
    /// This sink's resolved terminal-stage chunking, threaded into every
    /// per-thread [`ChainCtx`] (the default sink's becomes [`ChainCtx::chunk`];
    /// named sinks reach it through [`ChainCtx::sink`]).
    chunk: ChunkConfig,
}

type ChainFactoryFn = Box<dyn FnMut(ChainCtx) -> Box<dyn RunnableChain> + Send>;

/// The pipeline builder. See the [module docs](crate::pipeline) for the full
/// picture.
///
/// Non-generic, nameable, and storable; the source type enters only at the
/// terminal [`into_runtime`](Self::into_runtime)/[`run`](Self::run) call.
pub struct Pipeline {
    config: PipelineConfig,
    metrics: MetricsHandle,
    io: tokio::runtime::Runtime,
    budget: Arc<InflightBudget>,
    sink_failures: SinkFailures,
    sinks: Vec<(String, SinkAssembly)>,
    chains: Option<ChainFactoryFn>,
    options: RuntimeOptions,
    feed: Option<Arc<dyn Feed>>,
}

impl std::fmt::Debug for Pipeline {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let sink_names: Vec<&str> = self.sinks.iter().map(|(n, _)| n.as_str()).collect();
        f.debug_struct("Pipeline")
            .field("pipeline", &self.config.pipeline.name)
            .field("sinks", &sink_names)
            .field("chains", &self.chains.is_some())
            .finish_non_exhaustive()
    }
}

impl Pipeline {
    /// Load configuration from a YAML file and initialize the process; see
    /// [`from_config`](Self::from_config).
    pub fn from_path(path: &Path) -> Result<Self, BuildError> {
        Self::from_config(PipelineConfig::from_path(path)?)
    }

    /// Initialize the process from an already-loaded configuration:
    ///
    /// 1. **Telemetry** — [`telemetry::init`]`(Json, "info")`. Idempotent:
    ///    to customize the format or filter, call [`telemetry::init`]
    ///    yourself *first* (the binaries-init convention).
    /// 2. **Metrics exporter** — installed from the config's `metrics`
    ///    section before you can construct any handle, so every handle
    ///    built while holding the `Pipeline` binds to the process's
    ///    recorder, which always exists under `exporter: prometheus`.
    ///    When a foreign recorder already owns the process, the pipeline
    ///    continues against it with a warning.
    /// 3. **The I/O runtime** — `pipeline.io_threads` workers, thread name
    ///    `spate-io`. Connectors that need a handle before `run` (schema
    ///    fetchers, async pre-flight validation) use
    ///    [`io_handle`](Self::io_handle)/[`block_on`](Self::block_on).
    ///
    /// # Errors
    ///
    /// [`BuildError::AsyncContext`] when called from inside an async runtime.
    /// Build pipelines from a plain thread, usually `main`.
    pub fn from_config(config: PipelineConfig) -> Result<Self, BuildError> {
        if tokio::runtime::Handle::try_current().is_ok() {
            return Err(BuildError::AsyncContext);
        }
        if config.pipeline.io_threads == 0 {
            return Err(BuildError::Config(ConfigError::Validation(
                "pipeline.io_threads must be non-zero".into(),
            )));
        }
        // The YAML loaders run the full `PipelineConfig::validate`; a
        // programmatically built config skips it by design (minimal test
        // fixtures). But `ComponentConfig::new` peels the reserved `chunk` key
        // before the connector's `deny_unknown_fields` could reject it, so
        // without this check a stray `chunk:` on a source/deserializer body
        // would be silently swallowed here.
        config.reject_stray_chunk().map_err(BuildError::Config)?;
        telemetry::init(LogFormat::Json, "info");
        let metrics = install_or_reuse(&metrics_settings(&config)).map_err(|e| match e {
            StartError::Metrics(m) => BuildError::Metrics(m),
            other => BuildError::Metrics(other.to_string()),
        })?;
        let io = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(config.pipeline.io_threads)
            .thread_name("spate-io")
            .enable_all()
            .build()?;
        Ok(Pipeline {
            config,
            metrics,
            io,
            budget: Arc::new(InflightBudget::new()),
            sink_failures: SinkFailures::new(),
            sinks: Vec::new(),
            chains: None,
            options: RuntimeOptions::default(),
            feed: None,
        })
    }

    /// The loaded configuration. Connector sections (`config().source`,
    /// `.deserializer`, `.sink`) still belong to the caller's connector
    /// factories.
    #[must_use]
    pub fn config(&self) -> &PipelineConfig {
        &self.config
    }

    /// The installed exporter's handle (rendering, upkeep).
    #[must_use]
    pub fn metrics(&self) -> &MetricsHandle {
        &self.metrics
    }

    /// The shared in-flight byte budget.
    #[must_use]
    pub fn budget(&self) -> &Arc<InflightBudget> {
        &self.budget
    }

    /// A handle to the I/O runtime, for connector edge work that must
    /// start before the chain exists (schema-registry fetchers, ...).
    /// Valid until `run` returns; once the pipeline has started, that
    /// includes the source's `Drop`.
    #[must_use]
    pub fn io_handle(&self) -> tokio::runtime::Handle {
        self.io.handle().clone()
    }

    /// Run a future on the I/O runtime, blocking this thread. Use it for
    /// async pre-flight steps such as schema validation.
    pub fn block_on<F: Future>(&self, future: F) -> F::Output {
        self.io.block_on(future)
    }

    /// Install the single sink under the reserved name `"default"` with
    /// default [`SinkOptions`], the ergonomic path for single-sink
    /// pipelines. Sugar for [`add_sink`](Self::add_sink)`("default", bundle)`.
    pub fn sink<B: SinkBundle>(self, bundle: B) -> Result<Self, BuildError> {
        self.add_sink_with("default", bundle, SinkOptions::default())
    }

    /// [`sink`](Self::sink) with explicit [`SinkOptions`]. Sugar for
    /// [`add_sink_with`](Self::add_sink_with)`("default", bundle, options)`.
    pub fn sink_with<B: SinkBundle>(
        self,
        bundle: B,
        options: SinkOptions,
    ) -> Result<Self, BuildError> {
        self.add_sink_with("default", bundle, options)
    }

    /// Install a named sink with default [`SinkOptions`]; see
    /// [`add_sink_with`](Self::add_sink_with). Call once per destination
    /// table/stream; the chain's [`split`](crate::ops::ChainBuilder) terminal
    /// resolves each branch's queues by this name via
    /// [`ChainCtx::sink`](ChainCtx::sink).
    pub fn add_sink<B: SinkBundle>(
        self,
        name: impl Into<String>,
        bundle: B,
    ) -> Result<Self, BuildError> {
        self.add_sink_with(name, bundle, SinkOptions::default())
    }

    /// Install a named sink. This builds the per-shard chunk queues, registers
    /// the per-shard metrics (E2E basis from the config; the sink `name`
    /// becomes the `component` label, so each sink's `spate_sink_*` series is
    /// distinct), spawns the [`SinkPool`] workers on the I/O runtime, and
    /// wires the drain and readiness probe. The named sinks share the one
    /// pipeline
    /// [`InflightBudget`] and one backpressure controller (a stall on any sink
    /// pauses the shared source).
    ///
    /// # Errors
    ///
    /// [`BuildError::DuplicateSinkName`] when `name` is already installed;
    /// [`BuildError::Sink`] for an empty or reserved name (`"sink"` is the
    /// default sink's metric label), an empty or ragged topology, label
    /// shapes that do not match it, or a zero queue capacity.
    pub fn add_sink_with<B: SinkBundle>(
        mut self,
        name: impl Into<String>,
        bundle: B,
        options: SinkOptions,
    ) -> Result<Self, BuildError> {
        let sink_name = name.into();
        if self.sinks.iter().any(|(n, _)| n == &sink_name) {
            return Err(BuildError::DuplicateSinkName(sink_name));
        }
        if sink_name.is_empty() {
            return Err(BuildError::Sink("sink name must be non-empty".into()));
        }
        // "default" maps to the historical component="sink" metric label, so
        // a sink literally named "sink" would silently merge its spate_sink_*
        // series with the default's. Reject it up front.
        if sink_name == "sink" {
            return Err(BuildError::Sink(
                "the sink name \"sink\" is reserved (it is the default sink's \
                 metric label); pick another name"
                    .into(),
            ));
        }
        if options.queue_capacity == 0 {
            return Err(BuildError::Sink("queue_capacity must be non-zero".into()));
        }
        // Resolve the terminal-stage chunking BEFORE any I/O worker spawns, so
        // a config error can't leak a running SinkPool. Decl-once: the per-sink
        // YAML `chunk:` block and `SinkOptions::with_chunk` are mutually
        // exclusive.
        let yaml_chunk = match self.config.sink_config(&sink_name) {
            // Prefix the sink name onto the error so a multi-sink pipeline
            // says *which* sink is misconfigured (the dotted Component path
            // alone is `sink.<type>.…`, identical for same-typed sinks).
            Ok(cc) => cc.resolved_chunk().map_err(|e| {
                BuildError::Config(match e {
                    ConfigError::Validation(m) => {
                        ConfigError::Validation(format!("sink {sink_name:?}: {m}"))
                    }
                    ConfigError::Component { context, message } => ConfigError::Component {
                        context: format!("sink {sink_name:?}: {context}"),
                        message,
                    },
                    other => other,
                })
            })?,
            // No YAML section under this name. The sink is configured purely
            // in code (capture sinks in tests, named programmatic sinks beside
            // a placeholder `sink:`). A declared `chunk:` block that nothing
            // ever installs is warned about at `into_runtime`, where the
            // installed-name set is complete.
            Err(_) => None,
        };
        let chunk = match (yaml_chunk, options.chunk) {
            (Some(_), Some(_)) => {
                return Err(BuildError::Config(ConfigError::Validation(format!(
                    "sink {sink_name:?}: set the chunk config in exactly one place — the \
                     YAML `chunk:` block or `SinkOptions::with_chunk`, not both"
                ))));
            }
            (Some(c), None) | (None, Some(c)) => c,
            (None, None) => ChunkConfig::default(),
        };
        // The YAML path already checked `> 0` in `ChunkSection::resolve`; the
        // programmatic `with_chunk` path skips that, so enforce parity here
        // (naming the sink). There is no upper bound. The in-flight budget can
        // be smaller than one chunk on a throttled config; that is heavy
        // backpressure rather than a misconfiguration. `target_bytes` is a
        // per-shard pre-allocation; sizing it is the operator's call (see the
        // tuning docs).
        if chunk.target_bytes == 0 {
            return Err(BuildError::Config(ConfigError::Validation(format!(
                "sink {sink_name:?}: chunk.target_bytes must be greater than zero"
            ))));
        }
        let parts = bundle.into_parts();
        let num_shards = parts.shard_endpoints.len();
        if num_shards == 0 {
            return Err(BuildError::Sink("sink topology has no shards".into()));
        }
        if let Some(shard) = parts.shard_endpoints.iter().position(Vec::is_empty) {
            return Err(BuildError::Sink(format!("shard {shard} has no replicas")));
        }
        let replica_labels = parts.effective_replica_labels();
        let label_shape: Vec<usize> = replica_labels.iter().map(Vec::len).collect();
        let endpoint_shape: Vec<usize> = parts.shard_endpoints.iter().map(Vec::len).collect();
        if label_shape != endpoint_shape {
            return Err(BuildError::Sink(format!(
                "replica_labels shape {label_shape:?} does not match the \
                 endpoint topology {endpoint_shape:?}"
            )));
        }

        let pipeline_name = self.config.pipeline.name.clone();
        // The single-sink default keeps the historical `component="sink"`
        // label; named sinks use their name so their series never collide.
        let component = if sink_name == "default" {
            "sink".to_string()
        } else {
            sink_name.clone()
        };
        let (mut queues, receivers) = shard_queues(num_shards, options.queue_capacity);
        // The sink's custom-metrics scope (`spate_<component_type>_sink_*`),
        // handed to the writer before it is shared across shard workers.
        let sink_meter = Meter::for_component(
            &parts.component_type,
            MetricRole::Sink,
            pipeline_name.clone(),
            component.clone(),
        );
        let sink_labels = ComponentLabels::new(
            pipeline_name.clone(),
            component,
            parts.component_type.clone(),
        );
        // Pre-register the queue-edge handles before `queues` is cloned into
        // any terminal, so every producer shares the same `spate_queue_*` series.
        queues
            .attach_metrics(&sink_labels)
            .map_err(|e| BuildError::DuplicateSeries(e.to_string()))?;
        // Before any clone, like the metrics: every chain's terminal makes its
        // own codec instance from these queues.
        if let Some(codec) = parts.frame_codec {
            queues = queues.with_frame_codec(codec);
        }
        let e2e_basis = metrics_settings(&self.config).e2e_basis;
        // A shard's gauges are edge-triggered, so two live handle sets on one
        // series leave a stale reading standing rather than a double count. On
        // the assembly path that is a wiring mistake we can still refuse,
        // before any data flows.
        let shard_metrics: Vec<SinkShardMetrics> = replica_labels
            .iter()
            .enumerate()
            .map(|(shard, replicas)| {
                SinkShardMetrics::try_new(
                    &sink_labels,
                    u32::try_from(shard).unwrap_or(u32::MAX),
                    replicas,
                    e2e_basis,
                )
                .map(|m| m.with_queue(queues.metrics(), shard))
                .map_err(|e| BuildError::DuplicateSeries(e.to_string()))
            })
            .collect::<Result<_, _>>()?;
        let mut writer = parts.writer;
        writer.attach_metrics(SinkMetricsCtx::new(sink_meter));
        let pool = SinkPool::spawn(
            Arc::new(writer),
            parts.shard_endpoints,
            receivers,
            parts.pool,
            Arc::clone(&self.budget),
            shard_metrics,
            &pipeline_name,
            &sink_name,
            self.sink_failures.clone(),
            self.io.handle(),
        );
        self.sinks.push((
            sink_name,
            SinkAssembly {
                queues,
                drain: Box::new(move |deadline| {
                    Box::pin(async move { pool.drain(deadline).await })
                }),
                probe: parts.probe,
                chunk,
            },
        ));
        Ok(self)
    }

    /// Install the chain factory, called once per pipeline thread with
    /// that thread's [`ChainCtx`]. Composition inside the closure is fully
    /// monomorphized ([`chain_owned`](crate::ops::chain_owned) and
    /// friends); the returned `Box<dyn RunnableChain>` is the single
    /// per-batch erasure boundary.
    #[must_use]
    pub fn chains<F>(mut self, factory: F) -> Self
    where
        F: FnMut(ChainCtx) -> Box<dyn RunnableChain> + Send + 'static,
    {
        self.chains = Some(Box::new(factory));
        self
    }

    /// Override the runtime options (signal handling, loop timings).
    #[must_use]
    pub fn runtime_options(mut self, options: RuntimeOptions) -> Self {
        self.options = options;
        self
    }

    /// Hand the source a feed built in code, for the config's `feed:`
    /// section to name. `run` passes it to [`Source::configure_feed`] with
    /// the section, wrapped so the source's calls on it are recorded under
    /// `spate_feed_*`.
    ///
    /// The feed's [`backend`](Feed::backend) must equal the key the section
    /// names, and the config must have a `feed:` section; `run` fails at
    /// startup otherwise. The section's backend keys, and `attempt_cap` as
    /// the feed's delivery cap, apply only when a feed is built from the
    /// section, so a feed passed here keeps the settings it was built with.
    /// The source still reads `units_held`, `location`, `settling_window`
    /// and, as the cap of a source that re-enqueues units, `attempt_cap`
    /// from the section. Without this call, `run`
    /// builds the in-process feed itself for `feed: { memory: {} }` and
    /// refuses any other backend. To publish to
    /// the in-process feed, keep a handle to a `MemoryFeed` and pass it here.
    /// Calls on a handle kept this way are not recorded.
    #[must_use]
    pub fn feed(mut self, feed: Arc<dyn Feed>) -> Self {
        self.feed = Some(feed);
        self
    }

    /// Finish assembly into a [`PipelineRuntime`], for callers that need
    /// [`shutdown_handle`](PipelineRuntime::shutdown_handle) before a
    /// spawned `run` (tests, embedded pipelines). The I/O runtime moves
    /// into it and is shut down when `run` returns.
    ///
    /// # Errors
    ///
    /// [`BuildError::MissingSink`] / [`BuildError::MissingChains`] when a
    /// step was skipped.
    pub fn into_runtime<S: Source + 'static>(
        mut self,
        source: S,
    ) -> Result<PipelineRuntime<S>, BuildError> {
        if self.sinks.is_empty() {
            return Err(BuildError::MissingSink);
        }
        let mut factory = self.chains.take().ok_or(BuildError::MissingChains)?;

        // Decompose the installed sinks into: the per-thread named queue set
        // (cloned into each ChainCtx), the introspection queues (the
        // backpressure resume gate spans every sink), and the drain/probe
        // hooks (composed into one of each for the runtime).
        let mut intro_queues = Vec::with_capacity(self.sinks.len());
        let mut drains = Vec::with_capacity(self.sinks.len());
        let mut probes = Vec::new();
        let mut named = Vec::with_capacity(self.sinks.len());
        for (sink_name, assembly) in std::mem::take(&mut self.sinks) {
            intro_queues.push(assembly.queues.clone());
            named.push((sink_name, assembly.queues, assembly.chunk));
            drains.push(assembly.drain);
            if let Some(probe) = assembly.probe {
                probes.push(probe);
            }
        }
        // A declared sink section whose name nothing installed can carry a
        // `chunk:` block that no install-time resolution will ever read, the
        // one mismatch invisible to `add_sink_with`. (A *malformed* block is
        // already rejected by `PipelineConfig::validate` on the YAML loaders.)
        for config_name in self.config.sink_names() {
            if !named.iter().any(|(n, _, _)| *n == config_name)
                && self
                    .config
                    .sink_config(&config_name)
                    .is_ok_and(crate::config::ComponentConfig::has_chunk)
            {
                tracing::warn!(
                    sink = %config_name,
                    "config declares a `chunk:` block for this sink, but no sink was \
                     installed under that name — the block is ignored (name mismatch \
                     between the config and add_sink)"
                );
            }
        }
        let default_queues = named[0].1.clone();
        let default_chunk = named[0].2;
        let budget = Arc::clone(&self.budget);
        let name = self.config.pipeline.name.clone();
        // Read the source's framing contract once, before it moves into the
        // runtime, and hand it to every per-thread ChainCtx.
        let source_framing = source.framing_contract();
        // This wrapper is the factory the runtime drops before the sink
        // drain; the queue clones it captures die there.
        let chains = move |thread: usize| {
            factory(ChainCtx {
                thread,
                queues: default_queues.clone(),
                budget: Arc::clone(&budget),
                pipeline: name.clone(),
                source_framing,
                chunk: default_chunk,
                named: named.clone(),
            })
        };
        let runtime = PipelineRuntime::new(
            self.config,
            source,
            chains,
            SinkRuntime {
                queues: intro_queues,
                drain: combine_drains(drains),
                probe: combine_probes(probes),
                failures: self.sink_failures,
            },
            self.budget,
        )
        .with_options(self.options)
        .with_io_runtime(self.io);
        Ok(match self.feed {
            Some(feed) => runtime.with_feed(feed),
            None => runtime,
        })
    }

    /// [`into_runtime`](Self::into_runtime) + [`PipelineRuntime::run`]:
    /// run the pipeline to completion, blocking until a shutdown signal
    /// drains it or a fatal error stops it.
    pub fn run<S: Source + 'static>(self, source: S) -> Result<ExitReport, PipelineError> {
        Ok(self.into_runtime(source)?.run()?)
    }
}

/// Compose per-sink drain hooks into one. Every sink drains concurrently under
/// the shared deadline and their reports are summed, so a multi-sink drain
/// respects one wall-clock budget instead of N sequential ones.
fn combine_drains(drains: Vec<SinkDrainFn>) -> SinkDrainFn {
    Box::new(move |deadline| {
        Box::pin(async move {
            let mut set = tokio::task::JoinSet::new();
            for drain in drains {
                #[expect(
                    clippy::disallowed_methods,
                    reason = "the set drops with the drain future, which aborts every sink's drain still running"
                )]
                set.spawn(drain(deadline));
            }
            let mut total = DrainReport::default();
            while let Some(res) = set.join_next().await {
                match res {
                    Ok(report) => {
                        total.flushed += report.flushed;
                        total.abandoned += report.abandoned;
                    }
                    // The panicked sink's counts are unknowable; its parked
                    // acks still fail on drop, so at-least-once holds. The
                    // report is incomplete and says so.
                    Err(e) => tracing::error!(
                        error = %e,
                        "a sink drain task panicked; its counts are missing \
                         from the drain report"
                    ),
                }
            }
            total
        })
    })
}

/// Compose per-sink readiness probes into one. Every sink is probed and the
/// result is connected only when all succeed (readiness is not a hot path, so
/// the sequential short-circuit is fine).
fn combine_probes(probes: Vec<SinkProbeFn>) -> Option<SinkProbeFn> {
    if probes.is_empty() {
        return None;
    }
    let probes = Arc::new(probes);
    Some(Box::new(move || {
        let probes = Arc::clone(&probes);
        Box::pin(async move {
            for probe in probes.iter() {
                probe().await?;
            }
            Ok(())
        })
    }))
}

#[expect(
    clippy::disallowed_methods,
    reason = "a test thread ends with the call it runs, or serves a stub until the test process exits"
)]
#[cfg(all(test, not(loom)))]
mod tests {
    use super::*;
    use crate::config::ComponentConfig;
    use crate::error::SinkError;
    use crate::pipeline::ExitState;
    use crate::pipeline::fakes::{
        ChainMode, ChainShared, FakeChain, FakeSource, LaneSpec, Script, SourceLog, batches,
        test_config, test_options, wait_for,
    };
    use crate::record::PartitionId;
    use crate::sink::{SealedBatch, SinkParts, SinkPoolConfig};
    use crate::source::LaneId;
    use std::sync::Mutex;
    use std::sync::atomic::Ordering;
    use std::time::Duration;

    struct NullWriter;
    impl crate::sink::ShardWriter for NullWriter {
        type Endpoint = ();
        async fn write_batch(&self, (): &(), _batch: &SealedBatch) -> Result<(), SinkError> {
            Ok(())
        }
    }

    /// Records each [`SinkMetricsCtx::meter`] the builder hands it.
    struct ProbeWriter {
        seen: Arc<Mutex<Vec<Option<crate::metrics::Meter>>>>,
    }
    impl crate::sink::ShardWriter for ProbeWriter {
        type Endpoint = ();
        fn attach_metrics(&mut self, cx: SinkMetricsCtx) {
            self.seen.lock().expect("probe lock").push(cx.meter);
        }
        async fn write_batch(&self, (): &(), _batch: &SealedBatch) -> Result<(), SinkError> {
            Ok(())
        }
    }

    fn null_sink(shards: usize) -> SinkParts<NullWriter> {
        SinkParts::new(
            NullWriter,
            (0..shards).map(|_| vec![()]).collect(),
            SinkPoolConfig::default(),
        )
        .with_component_type("null")
    }

    fn fake_chain(
        shared: &Arc<ChainShared>,
        log: &Arc<Mutex<SourceLog>>,
    ) -> Box<dyn RunnableChain> {
        Box::new(FakeChain {
            shared: Arc::clone(shared),
            log: Arc::clone(log),
            mode: ChainMode::Ok,
            batches_seen: 0,
        })
    }

    #[test]
    fn missing_sink_then_missing_chains_error() {
        let (source, _shared, _script) = FakeSource::new();
        let p = Pipeline::from_config(test_config(1)).expect("builder");
        assert!(matches!(
            p.into_runtime(source).err(),
            Some(BuildError::MissingSink)
        ));

        let (source, _shared, _script) = FakeSource::new();
        let p = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink(null_sink(1))
            .expect("sink");
        assert!(matches!(
            p.into_runtime(source).err(),
            Some(BuildError::MissingChains)
        ));
    }

    /// The builder hands a sink writer one context whose `Meter` is scoped to
    /// the sink role and the declared `component_type`, and no `Meter` when
    /// the sink keeps the default `"custom"`.
    #[test]
    fn a_sink_writer_receives_one_context_scoped_to_the_sink_role() {
        let seen = Arc::new(Mutex::new(Vec::new()));
        let cfg = test_config(1);
        let name = cfg.pipeline.name.clone();
        let parts = SinkParts::new(
            ProbeWriter {
                seen: Arc::clone(&seen),
            },
            vec![vec![()]],
            SinkPoolConfig::default(),
        )
        .with_component_type("probe");
        let _pipeline = Pipeline::from_config(cfg)
            .expect("builder")
            .sink(parts)
            .expect("sink");
        let meter = {
            let mut seen = seen.lock().expect("probe lock");
            assert_eq!(seen.len(), 1, "attach_metrics runs once per writer");
            seen.pop()
                .flatten()
                .expect("a declared component_type scopes a Meter")
        };
        let rendered = crate::metrics::render_local(|| {
            meter.counter("probe_total", &[]).increment(1);
        });
        let expected = format!(
            "spate_probe_sink_probe_total{{pipeline=\"{name}\",component=\"sink\",component_type=\"probe\"}} 1"
        );
        assert!(
            rendered.lines().any(|line| line == expected),
            "missing `{expected}` in:\n{rendered}"
        );

        let seen = Arc::new(Mutex::new(Vec::new()));
        let parts = SinkParts::new(
            ProbeWriter {
                seen: Arc::clone(&seen),
            },
            vec![vec![()]],
            SinkPoolConfig::default(),
        );
        let _pipeline = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink(parts)
            .expect("sink");
        let seen = seen.lock().expect("probe lock");
        assert_eq!(seen.len(), 1, "attach_metrics runs once per writer");
        assert!(seen[0].is_none(), "the default `custom` type gets no Meter");
    }

    #[test]
    fn duplicate_sink_name_errors() {
        // A second bare `.sink()` collides on the reserved "default" name.
        let p = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink(null_sink(1))
            .expect("first sink");
        assert!(matches!(
            p.sink(null_sink(1)).err(),
            Some(BuildError::DuplicateSinkName(name)) if name == "default"
        ));

        // The same explicit name twice also collides.
        let p = Pipeline::from_config(test_config(1))
            .expect("builder")
            .add_sink("a", null_sink(1))
            .expect("first");
        assert!(matches!(
            p.add_sink("a", null_sink(1)).err(),
            Some(BuildError::DuplicateSinkName(name)) if name == "a"
        ));
    }

    #[test]
    fn distinct_named_sinks_install() {
        Pipeline::from_config(test_config(1))
            .expect("builder")
            .add_sink("a", null_sink(1))
            .expect("first")
            .add_sink("b", null_sink(2))
            .expect("second install with a distinct name");
    }

    /// Two *live* pipelines with the same pipeline and sink name would resolve
    /// the same `spate_sink_*` and `spate_queue_*` gauge series. The second's
    /// `add_sink` must refuse; those gauges cannot be shared, and letting both
    /// through leaves each overwriting the other's readings.
    ///
    /// A pipeline rebuilt *sequentially* (the supported way to replace one in
    /// a process) is fine. The first's claim frees when it is dropped, so the
    /// rebuild re-owns the series. Only overlap collides.
    #[test]
    fn two_live_pipelines_on_one_name_collide_but_sequential_reuse_is_fine() {
        let named = || {
            let mut cfg = test_config(1);
            cfg.pipeline.name = "shared-name".into();
            cfg
        };

        // First pipeline, held alive across the second's build.
        let first = Pipeline::from_config(named())
            .expect("builder")
            .sink(null_sink(2))
            .expect("first pipeline claims the sink series");

        let collision = Pipeline::from_config(named())
            .expect("builder")
            .sink(null_sink(2));
        assert!(
            matches!(collision.err(), Some(BuildError::DuplicateSeries(msg)) if msg.contains("spate_")),
            "a second live pipeline on the same name must fail on the series claim"
        );

        // Drop the first, freeing its claims, and rebuild: the series is
        // available again.
        drop(first);
        Pipeline::from_config(named())
            .expect("builder")
            .sink(null_sink(2))
            .expect("sequential rebuild re-owns the freed series");
    }

    #[test]
    fn reserved_and_empty_sink_names_error() {
        // "sink" is the default sink's metric label; installing a sink under
        // that name would silently merge the two series.
        let p = Pipeline::from_config(test_config(1)).expect("builder");
        assert!(matches!(
            p.add_sink("sink", null_sink(1)).err(),
            Some(BuildError::Sink(msg)) if msg.contains("reserved")
        ));

        let p = Pipeline::from_config(test_config(1)).expect("builder");
        assert!(matches!(
            p.add_sink("", null_sink(1)).err(),
            Some(BuildError::Sink(msg)) if msg.contains("non-empty")
        ));
    }

    #[test]
    fn bad_topologies_error_instead_of_panicking() {
        let p = Pipeline::from_config(test_config(1)).expect("builder");
        let empty = SinkParts::new(NullWriter, Vec::new(), SinkPoolConfig::default());
        assert!(matches!(p.sink(empty).err(), Some(BuildError::Sink(_))));

        let p = Pipeline::from_config(test_config(1)).expect("builder");
        let ragged = SinkParts::new(
            NullWriter,
            vec![vec![()], vec![]],
            SinkPoolConfig::default(),
        );
        assert!(matches!(p.sink(ragged).err(), Some(BuildError::Sink(_))));

        let p = Pipeline::from_config(test_config(1)).expect("builder");
        let bad_labels = SinkParts::new(NullWriter, vec![vec![()]], SinkPoolConfig::default())
            .with_replica_labels(vec![vec!["a".into(), "b".into()]]);
        assert!(matches!(
            p.sink(bad_labels).err(),
            Some(BuildError::Sink(_))
        ));

        let p = Pipeline::from_config(test_config(1)).expect("builder");
        assert!(matches!(
            p.sink_with(null_sink(1), SinkOptions::default().with_queue_capacity(0))
                .err(),
            Some(BuildError::Sink(_))
        ));
    }

    #[tokio::test]
    async fn from_config_inside_async_context_errors() {
        assert!(matches!(
            Pipeline::from_config(test_config(1)).err(),
            Some(BuildError::AsyncContext)
        ));
    }

    /// The chain factory sees every thread index exactly once, with the
    /// pipeline name from the config, and the assembled pipeline runs to a
    /// clean `Completed` through the real `SinkPool`. This guards `ChainCtx`
    /// coverage and end-to-end assembly, not drop ordering. Drain containment
    /// (`sink/worker.rs`) turns a leaked `ShardQueues` clone into a bounded
    /// abandon rather than an unbounded hang, so completion here does not
    /// imply clean drop ordering. The drop-ordering and at-least-once contract
    /// is covered where it can still fail observably: the whole-assembly test
    /// in `spate-test`'s `tests/bundle.rs`, which routes real data through
    /// `ctx.queues` and asserts the watermark only advances past the last
    /// record after a durable write.
    #[test]
    fn chain_ctx_covers_every_thread_and_run_completes() {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(&[0..10, 10..20]),
            }]));
        let chain_shared = Arc::new(ChainShared::default());
        let seen_threads: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));

        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);
        let seen = Arc::clone(&seen_threads);
        let config = test_config(2);
        let pipeline_name = config.pipeline.name.clone();
        let runtime = Pipeline::from_config(config)
            .expect("builder")
            .sink(null_sink(1))
            .expect("sink")
            .chains(move |ctx| {
                assert_eq!(ctx.pipeline, pipeline_name);
                // The source's framing contract threads into every ChainCtx;
                // FakeSource overrides it to the non-default PerRecord.
                assert_eq!(ctx.source_framing, FramingContract::PerRecord);
                seen.lock().unwrap().push(ctx.thread);
                fake_chain(&cs, &log)
            })
            .runtime_options(test_options())
            .into_runtime(source)
            .expect("into_runtime");

        let shutdown = runtime.shutdown_handle();
        let join = std::thread::spawn(move || runtime.run());
        wait_for("payloads consumed", Duration::from_secs(5), || {
            chain_shared.consumed.load(Ordering::Relaxed) == 20
        });
        shutdown.trigger();
        let report = join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed);

        let mut threads = seen_threads.lock().unwrap().clone();
        threads.sort_unstable();
        assert_eq!(threads, vec![0, 1], "one ChainCtx per pipeline thread");
    }

    /// A `chunk:` body peeled onto a `ComponentConfig`, the shape the framework
    /// resolves at install without the connector ever seeing the key.
    fn chunk_body(yaml: &str) -> ComponentConfig {
        ComponentConfig::new("fake", serde_yaml::from_str(yaml).expect("chunk yaml"))
    }

    #[test]
    fn yaml_and_programmatic_chunk_on_one_sink_collide() {
        // Decl-once: setting the YAML `chunk:` block AND `SinkOptions::with_chunk`
        // on the same sink is a config error, not a silent override.
        let mut config = test_config(1);
        config.sink = Some(chunk_body("chunk: { target_bytes: 128KiB }"));
        let err = Pipeline::from_config(config)
            .expect("builder")
            .sink_with(
                null_sink(1),
                SinkOptions::default().with_chunk(ChunkConfig::default()),
            )
            .err();
        assert!(
            matches!(&err, Some(BuildError::Config(ConfigError::Validation(m))) if m.contains("exactly one place")),
            "{err:?}"
        );
    }

    #[test]
    fn programmatic_zero_target_bytes_is_rejected_at_install() {
        let err = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink_with(
                null_sink(1),
                SinkOptions::default().with_chunk(ChunkConfig {
                    target_bytes: 0,
                    encode_policy: crate::error::ErrorPolicy::Skip,
                }),
            )
            .err();
        assert!(
            matches!(&err, Some(BuildError::Config(ConfigError::Validation(m))) if m.contains("chunk.target_bytes")),
            "{err:?}"
        );
    }

    #[test]
    fn each_split_branch_resolves_its_own_chunk() {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(&[0..1, 1..2]),
            }]));

        let mut config = test_config(1);
        config.sink = None;
        let mut sinks = std::collections::BTreeMap::new();
        sinks.insert(
            "a".to_string(),
            chunk_body("chunk: { target_bytes: 128KiB }"),
        );
        sinks.insert(
            "b".to_string(),
            chunk_body("chunk: { target_bytes: 512KiB }"),
        );
        config.sinks = Some(sinks);

        let captured: Arc<Mutex<Vec<(usize, usize)>>> = Arc::new(Mutex::new(Vec::new()));
        let cap = Arc::clone(&captured);
        let chain_shared = Arc::new(ChainShared::default());
        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);
        let runtime = Pipeline::from_config(config)
            .expect("builder")
            .add_sink("a", null_sink(1))
            .expect("sink a")
            .add_sink("b", null_sink(1))
            .expect("sink b")
            .chains(move |ctx| {
                cap.lock().unwrap().push((
                    ctx.sink("a").chunk.target_bytes,
                    ctx.sink("b").chunk.target_bytes,
                ));
                fake_chain(&cs, &log)
            })
            .runtime_options(test_options())
            .into_runtime(source)
            .expect("into_runtime");

        let shutdown = runtime.shutdown_handle();
        let join = std::thread::spawn(move || runtime.run());
        wait_for("chain factory ran", Duration::from_secs(5), || {
            !captured.lock().unwrap().is_empty()
        });
        shutdown.trigger();
        let _ = join.join().unwrap();

        assert_eq!(
            captured.lock().unwrap()[0],
            (128 * 1024, 512 * 1024),
            "each branch resolves its own per-sink chunk"
        );
    }

    /// A [`FrameCodec`](crate::sink::FrameCodec) that writes a fixed prefix
    /// before the chunk.
    struct PrefixCodec {
        prefix: &'static [u8],
        out: Vec<u8>,
    }

    impl crate::sink::FrameCodec for PrefixCodec {
        fn encode_frame(&mut self, chunk: &[u8]) -> Result<&[u8], SinkError> {
            self.out.clear();
            self.out.extend_from_slice(self.prefix);
            self.out.extend_from_slice(chunk);
            Ok(&self.out)
        }
    }

    fn prefix_codec(prefix: &'static [u8]) -> crate::sink::FrameCodecFactory {
        crate::sink::FrameCodecFactory::new(move || {
            Box::new(PrefixCodec {
                prefix,
                out: Vec::new(),
            })
        })
    }

    /// The codec a sink's parts carry reaches that sink's queues in every
    /// chain, and no other sink's.
    #[test]
    fn each_sinks_frame_codec_reaches_every_chain() {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(&[0..1, 1..2]),
            }]));

        let config = test_config(2);
        type Coded = Option<Vec<u8>>;
        let captured: Arc<Mutex<Vec<(Coded, Coded)>>> = Arc::new(Mutex::new(Vec::new()));
        let cap = Arc::clone(&captured);
        let chain_shared = Arc::new(ChainShared::default());
        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);
        let code = |ctx: &ChainCtx, sink: &str| -> Coded {
            let mut codec = ctx.sink(sink).queues.new_frame_codec()?;
            Some(codec.encode_frame(b"probe").expect("prefix codec").to_vec())
        };
        let runtime = Pipeline::from_config(config)
            .expect("builder")
            .add_sink("a", null_sink(1).with_frame_codec(prefix_codec(b"A:")))
            .expect("sink a")
            .add_sink("b", null_sink(1).with_frame_codec(prefix_codec(b"B:")))
            .expect("sink b")
            .chains(move |ctx| {
                cap.lock().unwrap().push((code(&ctx, "a"), code(&ctx, "b")));
                fake_chain(&cs, &log)
            })
            .runtime_options(test_options())
            .into_runtime(source)
            .expect("into_runtime");

        let shutdown = runtime.shutdown_handle();
        let join = std::thread::spawn(move || runtime.run());
        wait_for("every chain factory ran", Duration::from_secs(5), || {
            captured.lock().unwrap().len() == 2
        });
        shutdown.trigger();
        let _ = join.join().unwrap();

        let captured = captured.lock().unwrap();
        for (thread, coded) in captured.iter().enumerate() {
            assert_eq!(
                coded,
                &(Some(b"A:probe".to_vec()), Some(b"B:probe".to_vec())),
                "chain {thread} codes each sink with that sink's codec"
            );
        }
    }

    /// Drive a capture of `ctx.chunk()` through a full assembly, shared by
    /// the YAML-block and `with_chunk` propagation tests below.
    fn captured_default_chunk(
        build: impl FnOnce(Pipeline) -> Result<Pipeline, BuildError>,
        config: PipelineConfig,
    ) -> usize {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(&[0..1, 1..2]),
            }]));
        let captured: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
        let cap = Arc::clone(&captured);
        let chain_shared = Arc::new(ChainShared::default());
        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);
        let runtime = build(Pipeline::from_config(config).expect("builder"))
            .expect("sink install")
            .chains(move |ctx| {
                cap.lock().unwrap().push(ctx.chunk().target_bytes);
                fake_chain(&cs, &log)
            })
            .runtime_options(test_options())
            .into_runtime(source)
            .expect("into_runtime");

        let shutdown = runtime.shutdown_handle();
        let join = std::thread::spawn(move || runtime.run());
        wait_for("chain factory ran", Duration::from_secs(5), || {
            !captured.lock().unwrap().is_empty()
        });
        shutdown.trigger();
        let _ = join.join().unwrap();
        let captured = captured.lock().unwrap();
        captured[0]
    }

    #[test]
    fn default_sink_yaml_chunk_reaches_ctx_chunk() {
        // The headline single-sink path: `sink.<type>.chunk` must arrive at
        // the chain factory via `ctx.chunk()`, not silently stay the default.
        let mut config = test_config(1);
        config.sink = Some(chunk_body("chunk: { target_bytes: 128KiB }"));
        let bytes = captured_default_chunk(|p| p.sink(null_sink(1)), config);
        assert_eq!(bytes, 128 * 1024, "YAML chunk block reaches ctx.chunk()");
    }

    #[test]
    fn with_chunk_reaches_ctx_chunk() {
        let bytes = captured_default_chunk(
            |p| {
                p.sink_with(
                    null_sink(1),
                    SinkOptions::default().with_chunk(ChunkConfig {
                        target_bytes: 96 * 1024,
                        encode_policy: crate::error::ErrorPolicy::Skip,
                    }),
                )
            },
            test_config(1),
        );
        assert_eq!(bytes, 96 * 1024, "with_chunk reaches ctx.chunk()");
    }

    #[test]
    fn stray_chunk_on_a_source_body_is_rejected_by_from_config() {
        // `ComponentConfig::new` peels the reserved key before the connector's
        // `deny_unknown_fields` could reject it, so `from_config` must reject
        // it itself. A stray `chunk:` on a source is an error, not a silent
        // no-op, even on the programmatic path that skips `validate`.
        let mut config = test_config(1);
        config.source = ComponentConfig::new(
            "fake",
            serde_yaml::from_str("chunk: { target_bytes: 64KiB }").expect("yaml"),
        );
        let err = Pipeline::from_config(config).err();
        assert!(
            matches!(&err, Some(BuildError::Config(ConfigError::Validation(m))) if m.contains("source")),
            "{err:?}"
        );
    }

    /// Regression: the runtime must hand the source the framework's
    /// source-stage handles at `open`.
    ///
    /// `SourceMetrics::set_partition_lag` has exactly one possible caller, the
    /// source, because only the client can see the log end. When those handles
    /// are reachable only through a connector-side builder that nothing calls,
    /// `spate_source_lag_records` renders a permanent `0` on every Kafka
    /// pipeline. A maximally backlogged consumer reports no lag, and any alert
    /// or autoscaler keyed on it reads as caught up. Nothing fails; the series
    /// is always zero. This asserts the seam is connected. The Kafka crate's
    /// `a_backlogged_consumer_publishes_its_lag` covers the value arriving,
    /// which this test does not.
    #[test]
    fn source_open_receives_the_stage_metrics() {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(&[0..2, 2..4]),
            }]));
        let chain_shared = Arc::new(ChainShared::default());
        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);
        let runtime = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink(null_sink(1))
            .expect("sink")
            .chains(move |_| fake_chain(&cs, &log))
            .runtime_options(test_options())
            .into_runtime(source)
            .expect("into_runtime");

        let shutdown = runtime.shutdown_handle();
        let join = std::thread::spawn(move || runtime.run());
        wait_for("source opened", Duration::from_secs(5), || {
            shared.lock().unwrap().opened
        });
        shutdown.trigger();
        join.join().unwrap().unwrap();

        assert!(
            shared.lock().unwrap().stage_metrics_attached,
            "the runtime must share SourceMetrics with the source at open, \
             or consumer lag can never be published"
        );
    }

    /// The whole-builder happy path through `run()` (not `into_runtime`),
    /// exercised over the real SinkPool. Completes and commits.
    #[test]
    fn run_completes_via_builder_terminal() {
        let (source, shared, script) = FakeSource::new();
        script
            .lock()
            .unwrap()
            .push_back(Script::Assign(vec![LaneSpec {
                id: LaneId(0),
                partition: PartitionId(0),
                batches: batches(std::slice::from_ref(&(0..5))),
            }]));
        let chain_shared = Arc::new(ChainShared::default());
        let cs = Arc::clone(&chain_shared);
        let log = Arc::clone(&shared);

        let pipeline = Pipeline::from_config(test_config(1))
            .expect("builder")
            .sink(null_sink(2))
            .expect("sink")
            .chains(move |_ctx| fake_chain(&cs, &log))
            .runtime_options(test_options());

        // Drive shutdown from a watcher thread once the payloads land.
        let consumed = Arc::clone(&chain_shared);
        let runtime = pipeline.into_runtime(source).expect("into_runtime");
        let shutdown = runtime.shutdown_handle();
        std::thread::spawn(move || {
            wait_for("payloads consumed", Duration::from_secs(5), || {
                consumed.consumed.load(Ordering::Relaxed) == 5
            });
            shutdown.trigger();
        });
        let report = runtime.run().unwrap();
        assert_eq!(report.state, ExitState::Completed);
        assert_eq!(report.final_watermarks, vec![(PartitionId(0), 5)]);
    }
}