olympipe-rs 1.1.0

use crossbeam_channel::{Receiver, RecvTimeoutError, Sender, TryRecvError, bounded, unbounded};
use std::any::Any;
use std::collections::HashSet;
use std::panic::{self, AssertUnwindSafe};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use thiserror::Error;

const DEFAULT_BUFFER: usize = 4;

/// Most items placed in one channel message when the producer is not stalling.
///
/// A message is flushed earlier when the next input is not already available,
/// so a trickle source still travels one item at a time.
const MAX_CHUNK: usize = 64;

/// A source pull at least this slow ends packet coalescing.
///
/// Tight iterators (`0..n`) return in nanoseconds and stay coalesced. The
/// trickle benchmark sleeps 200µs inside `next`, which flushes the current
/// item before the following sleep. One clock read per full packet, not per item.
const SLOW_PULL: Duration = Duration::from_micros(64);

/// Items per channel message for this buffer size.
///
/// The channel holds `buffer_size` messages. Buffers of 1 or 2 stay at one
/// item per message so a stalled consumer stops the producer within a few
/// items. Larger buffers coalesce up to [`MAX_CHUNK`] items per message, and
/// therefore retain at most `buffer_size * MAX_CHUNK` items between stages.
fn chunk_limit(buffer_size: usize) -> usize {
    if buffer_size <= 2 { 1 } else { MAX_CHUNK }
}

/// Published by [`FeederRelease`] when the pipeline is first used or dropped.
///
/// The feeder thread is spawned in [`Pipeline::new`] / [`Pipeline::new_with_buffer`]
/// but waits here, so [`Pipeline::with_max_packet`] can still change its packet
/// size. Without the wait, a fast source would fill the default 64-item packets
/// before the caller could opt out.
struct FeederGate {
    limit: Mutex<Option<usize>>,
    cv: Condvar,
}

impl FeederGate {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            limit: Mutex::new(None),
            cv: Condvar::new(),
        })
    }

    fn open(&self, limit: usize) {
        let mut guard = lock(&self.limit);
        if guard.is_none() {
            *guard = Some(limit);
            self.cv.notify_one();
        }
    }

    fn wait(&self) -> usize {
        let mut guard = lock(&self.limit);
        while guard.is_none() {
            guard = self.cv.wait(guard).unwrap_or_else(|err| err.into_inner());
        }
        (*guard).expect("packet size is published before wait returns")
    }
}

/// Opens [`FeederGate`] with `limit` on drop.
struct FeederRelease {
    gate: Arc<FeederGate>,
    limit: usize,
}

impl Drop for FeederRelease {
    fn drop(&mut self) {
        self.gate.open(self.limit);
    }
}

/// Failure of a running pipeline.
///
/// The first failure is kept; later ones are ignored. Workers stop pulling new
/// items once a failure is recorded, and terminal methods surface it.
///
/// [`Pipeline::task_or`] is the per-item callback form: an `Err` from its
/// closure does not produce a [`PipelineError`]. [`Pipeline::try_task`] in
/// [`ErrorMode::FailFast`] records [`PipelineError::TaskFailed`] instead.
/// [`ErrorMode::Skip`], [`ErrorMode::SkipAndLog`], and [`ErrorMode::Replace`]
/// absorb both `Err` and, unlike `task_or`, a panic raised on that item.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
#[non_exhaustive]
pub enum PipelineError {
    /// A worker thread panicked. `stage` names the stage that failed
    /// (`"task"`, `"filter"`, `"split"`, `"reduce"`, `"checkpoint"`, `"feeder"`, …).
    /// `message` is the panic payload (`&str` or `String`).
    #[error("worker panicked in stage {stage}: {message}")]
    WorkerPanicked {
        /// Stage that panicked.
        stage: &'static str,
        /// Panic payload, or `"(unknown panic)"` when it was neither `&str` nor `String`.
        message: String,
    },
    /// [`Pipeline::try_task`] returned `Err` while in [`ErrorMode::FailFast`].
    ///
    /// `stage` is `"try_task"`. `message` is the error's [`Display`](std::fmt::Display)
    /// text. The pipeline is cancelled, the same way a panic is.
    ///
    /// ```
    /// use olympipe_rs::{ErrorMode, Pipeline, PipelineError};
    ///
    /// let err = Pipeline::new(vec![1, -1])
    ///     .try_task(
    ///         |x| if x < 0 { Err("negative") } else { Ok(x) },
    ///         1,
    ///         ErrorMode::FailFast,
    ///     )
    ///     .try_collect()
    ///     .unwrap_err();
    /// match err {
    ///     PipelineError::TaskFailed { stage, message } => {
    ///         assert_eq!(stage, "try_task");
    ///         assert_eq!(message, "negative");
    ///     }
    ///     other => panic!("unexpected {other:?}"),
    /// }
    /// ```
    #[error("task failed in stage {stage}: {message}")]
    TaskFailed {
        /// Stage that returned the error. [`Pipeline::try_task`] uses `"try_task"`.
        /// `cached_task` (feature `cache`) uses `"cached_task"` when its input
        /// cannot be serialized.
        stage: &'static str,
        /// `Display` text of the `Err` value.
        message: String,
    },
    /// [`Pipeline::timeout`] elapsed between two items.
    #[error("timeout after {0:?}")]
    Timeout(Duration),
}

/// What [`Pipeline::try_task`] and [`Pipeline::task_with_mode`] do with a
/// per-item failure.
///
/// [`ErrorMode::FailFast`] is the default. New modes may be added later.
///
/// ```
/// use olympipe_rs::ErrorMode;
///
/// let mode: ErrorMode<i32> = ErrorMode::default();
/// assert!(matches!(mode, ErrorMode::FailFast));
/// let _replaced = ErrorMode::Replace(0);
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum ErrorMode<U> {
    /// Stop the pipeline on the first failure.
    ///
    /// For [`Pipeline::try_task`], the first `Err` becomes
    /// [`PipelineError::TaskFailed`] and a panic stays
    /// [`PipelineError::WorkerPanicked`] (nothing is caught). For
    /// [`Pipeline::task_with_mode`], this is [`Pipeline::task`]: a panic fails
    /// the pipeline with stage `"task"`.
    #[default]
    FailFast,
    /// Drop the failed item and keep going.
    Skip,
    /// Drop the failed item, write one line to stderr, and keep going.
    ///
    /// The line is `olympipe-rs: [try_task] skipped item: {err}` or
    /// `olympipe-rs: [task_with_mode] skipped item: {err}`. `{err}` is the
    /// `Display` text of a returned error, or the panic payload.
    SkipAndLog,
    /// Emit a clone of this value in place of the failed item.
    ///
    /// `U: Clone` is required by the methods that accept a mode, because each
    /// failure clones `U` again.
    Replace(U),
}

#[derive(Default)]
struct FinishState {
    started: bool,
    done: bool,
    error: Option<PipelineError>,
}

/// Handles, the first recorded failure, and the cancellation flag shared by
/// every stage of a pipeline graph (both branches of a `split` share one).
struct Shared {
    handles: Mutex<Vec<JoinHandle<()>>>,
    error: Mutex<Option<PipelineError>>,
    /// Cheap per-item flag. Set after the first failure; workers load it with
    /// `Relaxed` ordering and leave their loop.
    cancelled: AtomicBool,
    /// Gathered pipelines. `Weak` so the link is not a reference cycle.
    peers: Mutex<Vec<Weak<Shared>>>,
    /// Strong refs to gathered sources, held until shutdown has read their error.
    retained: Mutex<Vec<Arc<Shared>>>,
    /// Two terminals (the two branches of a split) may finish concurrently.
    /// Exactly one joins the handles; the other waits here so it cannot miss
    /// an error that is still being recorded, and so no handle is joined twice.
    finish: Mutex<FinishState>,
    finish_cv: Condvar,
}

fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(|err| err.into_inner())
}

fn panic_message(payload: &(dyn Any + Send)) -> String {
    if let Some(msg) = payload.downcast_ref::<&str>() {
        return (*msg).to_owned();
    }
    if let Some(msg) = payload.downcast_ref::<String>() {
        return msg.clone();
    }
    "(unknown panic)".to_owned()
}

/// One channel message. `One` is the trickle / tight-backpressure path and
/// does not allocate. `Many` is a burst the producer already had in hand.
pub(crate) enum Packet<T> {
    One(T),
    Many(Vec<T>),
}

/// Outcome of pulling one burst from an input channel.
#[derive(PartialEq, Eq)]
enum Step {
    /// Input would now block. Flush partial output so latency does not wait
    /// for a full packet.
    Yield,
    /// Item callback asked to stop (limit reached, or a send failed).
    Halted,
    /// Upstream disconnected.
    Closed,
    /// Cancellation flag observed. Drop partial aggregates.
    Cancelled,
}

/// Coalesces items into [`Packet`]s. `push` sends a full packet; `flush`
/// sends a partial one (or nothing).
struct BatchTx<T> {
    tx: Sender<Packet<T>>,
    buf: Vec<T>,
    limit: usize,
}

impl<T> BatchTx<T> {
    fn new(tx: Sender<Packet<T>>, limit: usize) -> Self {
        let limit = limit.max(1);
        Self {
            tx,
            buf: Vec::with_capacity(limit.min(MAX_CHUNK)),
            limit,
        }
    }

    fn is_empty(&self) -> bool {
        self.buf.is_empty()
    }

    /// `false` when the downstream channel is disconnected.
    #[inline]
    fn push(&mut self, item: T) -> bool {
        self.buf.push(item);
        if self.buf.len() >= self.limit {
            self.flush()
        } else {
            true
        }
    }

    #[inline]
    fn flush(&mut self) -> bool {
        if self.buf.is_empty() {
            return true;
        }
        let packet = if self.buf.len() == 1 {
            Packet::One(self.buf.pop().expect("len checked"))
        } else {
            Packet::Many(std::mem::replace(
                &mut self.buf,
                Vec::with_capacity(self.limit.min(MAX_CHUNK)),
            ))
        };
        self.tx.send(packet).is_ok()
    }
}

/// Apply `on_item` to every element. `false` means the callback rejected an
/// item or the pipeline was cancelled between items.
#[inline]
fn feed_packet<T>(packet: Packet<T>, shared: &Shared, on_item: &mut impl FnMut(T) -> bool) -> bool {
    match packet {
        Packet::One(item) => !shared.is_cancelled() && on_item(item),
        Packet::Many(items) => {
            for item in items {
                if shared.is_cancelled() || !on_item(item) {
                    return false;
                }
            }
            true
        }
    }
}

/// Blocking receive of one packet, then (when `drain`) every packet already
/// queued. Returns once the next receive would block, so the caller can flush.
fn pull<T>(
    rx: &Receiver<Packet<T>>,
    shared: &Shared,
    drain: bool,
    mut on_item: impl FnMut(T) -> bool,
) -> Step {
    if shared.is_cancelled() {
        return Step::Cancelled;
    }
    let packet = match rx.recv() {
        Ok(packet) => packet,
        Err(_) => return Step::Closed,
    };
    if !feed_packet(packet, shared, &mut on_item) {
        return if shared.is_cancelled() {
            Step::Cancelled
        } else {
            Step::Halted
        };
    }
    if drain {
        loop {
            match rx.try_recv() {
                Ok(packet) => {
                    if !feed_packet(packet, shared, &mut on_item) {
                        return if shared.is_cancelled() {
                            Step::Cancelled
                        } else {
                            Step::Halted
                        };
                    }
                }
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => return Step::Closed,
            }
        }
    }
    if shared.is_cancelled() {
        Step::Cancelled
    } else {
        Step::Yield
    }
}

fn map_out<T, U>(
    rx: Receiver<Packet<T>>,
    tx: Sender<Packet<U>>,
    shared: &Shared,
    limit: usize,
    drain: bool,
    mut on_item: impl FnMut(T, &mut BatchTx<U>) -> bool,
) {
    let mut out = BatchTx::new(tx, limit);
    // Coalesce only after two fast callbacks with no idle gap. A slow
    // callback (the timeout bench sleeps inside `task`) is never entered
    // while the previous result is still unsent, and a trickle gap resets
    // the run so each item is published on its own.
    let mut hold = false;
    let mut prev_fast = false;
    loop {
        let step = pull(&rx, shared, drain, |item| {
            let started = if hold { None } else { Some(Instant::now()) };
            if !on_item(item, &mut out) {
                return false;
            }
            let Some(started) = started else {
                return true;
            };
            let fast = started.elapsed() < SLOW_PULL;
            if !fast {
                prev_fast = false;
                hold = false;
                return out.flush();
            }
            if prev_fast {
                hold = true;
            }
            prev_fast = true;
            hold || out.flush()
        });
        match step {
            Step::Yield => {
                hold = false;
                prev_fast = false;
                if !out.flush() {
                    break;
                }
            }
            Step::Closed | Step::Halted => {
                if !shared.is_cancelled() {
                    let _ = out.flush();
                }
                break;
            }
            Step::Cancelled => break,
        }
    }
}

/// Feeder: coalesce only after two fast pulls in a row, and only while the
/// buffer is non-empty. A slow `next` (trickle) is always shipped alone,
/// before the following pull can block.
fn feed<I, T>(iter: I, tx: Sender<Packet<T>>, shared: &Shared, limit: usize)
where
    I: IntoIterator<Item = T>,
{
    let mut iter = iter.into_iter();
    if limit <= 1 {
        for item in iter {
            if shared.is_cancelled() || tx.send(Packet::One(item)).is_err() {
                break;
            }
        }
        return;
    }
    let mut out = BatchTx::new(tx, limit);
    let mut prev_fast = false;
    let mut hold = false;
    loop {
        if shared.is_cancelled() {
            break;
        }
        let time_this = out.is_empty();
        let started = time_this.then(Instant::now);
        let Some(item) = iter.next() else {
            break;
        };
        if let Some(started) = started {
            let fast = started.elapsed() < SLOW_PULL;
            if !fast {
                prev_fast = false;
                hold = false;
                if !out.push(item) || !out.flush() {
                    break;
                }
                continue;
            }
            if prev_fast {
                hold = true;
            }
            prev_fast = true;
            if !out.push(item) {
                break;
            }
            if !hold && !out.flush() {
                break;
            }
        } else if !out.push(item) {
            break;
        }
    }
    let _ = out.flush();
}

impl Shared {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            handles: Mutex::new(Vec::new()),
            error: Mutex::new(None),
            cancelled: AtomicBool::new(false),
            peers: Mutex::new(Vec::new()),
            retained: Mutex::new(Vec::new()),
            finish: Mutex::new(FinishState::default()),
            finish_cv: Condvar::new(),
        })
    }

    #[inline]
    fn is_cancelled(&self) -> bool {
        self.cancelled.load(Ordering::Relaxed)
    }

    fn spawn(self: &Arc<Self>, stage: &'static str, body: impl FnOnce() + Send + 'static) {
        let shared = Arc::clone(self);
        let handle = thread::Builder::new()
            .name(format!("olympipe-{stage}"))
            .spawn(move || {
                if let Err(payload) = panic::catch_unwind(AssertUnwindSafe(body)) {
                    shared.fail(PipelineError::WorkerPanicked {
                        stage,
                        message: panic_message(&*payload),
                    });
                }
            })
            .unwrap_or_else(|_| panic!("olympipe-rs failed to spawn {stage} worker"));
        lock(&self.handles).push(handle);
    }

    fn fail(self: &Arc<Self>, err: PipelineError) {
        self.distribute(&err);
        self.cancel_all();
    }

    /// Record `err` on this node and on linked graphs, without overwriting an
    /// error that was stored earlier.
    fn distribute(self: &Arc<Self>, err: &PipelineError) {
        let mut seen = HashSet::<*const Self>::new();
        let mut stack = vec![Arc::clone(self)];
        while let Some(node) = stack.pop() {
            if !seen.insert(Arc::as_ptr(&node)) {
                continue;
            }
            {
                let mut slot = lock(&node.error);
                if slot.is_none() {
                    *slot = Some(err.clone());
                }
            }
            stack.extend(live_peers(&node));
        }
    }

    fn cancel_all(self: &Arc<Self>) {
        let mut seen = HashSet::<*const Self>::new();
        let mut stack = vec![Arc::clone(self)];
        while let Some(node) = stack.pop() {
            if !seen.insert(Arc::as_ptr(&node)) {
                continue;
            }
            node.cancelled.store(true, Ordering::Relaxed);
            stack.extend(live_peers(&node));
        }
    }

    /// `parent` is the gathered pipeline, `child` is a source consumed by it.
    fn link(parent: &Arc<Self>, child: &Arc<Self>) {
        if Arc::ptr_eq(parent, child) {
            return;
        }
        // Strong on the parent only: the child stays alive until shutdown, and
        // the back-edge is weak so an abandoned gather does not leak.
        lock(&parent.retained).push(Arc::clone(child));
        lock(&parent.peers).push(Arc::downgrade(child));
        lock(&child.peers).push(Arc::downgrade(parent));
        // A source may already have failed before it was gathered.
        if let Some(err) = stored_error(child) {
            parent.distribute(&err);
        }
        if let Some(err) = stored_error(parent) {
            child.distribute(&err);
        }
        if parent.is_cancelled() || child.is_cancelled() {
            parent.cancel_all();
        }
    }

    fn join_handles(&self) {
        let handles = {
            let mut guard = lock(&self.handles);
            std::mem::take(&mut *guard)
        };
        for handle in handles {
            if let Err(payload) = handle.join() {
                let mut slot = lock(&self.error);
                if slot.is_none() {
                    *slot = Some(PipelineError::WorkerPanicked {
                        stage: "worker",
                        message: panic_message(&*payload),
                    });
                }
                self.cancelled.store(true, Ordering::Relaxed);
            }
        }
    }

    /// Join every worker exactly once, then return the first error.
    ///
    /// The handle mutex is not held while joining: the other branch of a split
    /// may be blocked in its own terminal, and a joined thread must be able to
    /// finish without taking that lock. The second terminal waits until the
    /// joiner has published the error, so it cannot observe a success that has
    /// not been joined yet.
    fn shutdown(self: &Arc<Self>) -> Option<PipelineError> {
        {
            let mut guard = lock(&self.finish);
            if guard.done {
                return guard.error.clone();
            }
            if guard.started {
                while !guard.done {
                    guard = self
                        .finish_cv
                        .wait(guard)
                        .unwrap_or_else(|err| err.into_inner());
                }
                return guard.error.clone();
            }
            guard.started = true;
        }

        // Publish even if joining panics, so the waiting terminal cannot block
        // forever on `finish_cv`.
        struct Publish<'a> {
            shared: &'a Shared,
            error: Option<PipelineError>,
        }
        impl Drop for Publish<'_> {
            fn drop(&mut self) {
                let mut guard = lock(&self.shared.finish);
                if !guard.done {
                    guard.error = self.error.take();
                    guard.done = true;
                    self.shared.finish_cv.notify_all();
                }
            }
        }

        let mut publish = Publish {
            shared: self,
            error: None,
        };
        publish.error = self.join_graph();
        let error = publish.error.clone();
        drop(publish);
        error
    }

    fn join_graph(self: &Arc<Self>) -> Option<PipelineError> {
        self.join_handles();
        let peers = live_peers(self);
        for peer in &peers {
            peer.join_handles();
        }
        let error = first_error(self);
        lock(&self.retained).clear();
        lock(&self.peers).clear();
        error
    }
}

fn live_peers(shared: &Shared) -> Vec<Arc<Shared>> {
    lock(&shared.peers)
        .iter()
        .filter_map(Weak::upgrade)
        .collect()
}

fn stored_error(shared: &Shared) -> Option<PipelineError> {
    lock(&shared.error).clone()
}

fn first_error(root: &Shared) -> Option<PipelineError> {
    if let Some(err) = stored_error(root) {
        return Some(err);
    }
    let mut seen = HashSet::<*const Shared>::from([root as *const Shared]);
    let mut stack = live_peers(root);
    while let Some(node) = stack.pop() {
        if !seen.insert(Arc::as_ptr(&node)) {
            continue;
        }
        if let Some(err) = stored_error(&node) {
            return Some(err);
        }
        stack.extend(live_peers(&node));
    }
    None
}

/// A parallel data pipeline backed by bounded crossbeam channels and worker threads.
///
/// Each stage method consumes the pipeline and returns a new one whose type
/// reflects the output of that stage. Worker threads share an internal state:
/// their [`JoinHandle`]s, the first [`PipelineError`] (if any), and a
/// cancellation flag. A panic in any worker — including the feeder that drains
/// the source iterator — or a fired [`Pipeline::timeout`] records that error,
/// cancels the graph, and makes every stage stop. [`Pipeline::try_collect`],
/// [`Pipeline::try_for_each`], and [`Pipeline::try_wait_for_completion`] return
/// the error; [`Pipeline::collect`], [`Pipeline::for_each`], and
/// [`Pipeline::wait_for_completion`] panic with it. The same split applies to
/// [`Pipeline::try_wait_for_results`], [`Pipeline::try_wait_for_all_completions`],
/// [`Pipeline::try_wait_and_reduce`], and [`try_collect_pair`]: those drain every
/// pipeline concurrently, because bounded [`Pipeline::split`] branches deadlock
/// if one side is read to completion before the other starts.
///
/// Per-item `Result` failures stay on [`Pipeline::task_or`] (the callback sees
/// the item) and on [`Pipeline::try_task`] / [`Pipeline::task_with_mode`]. Only
/// [`ErrorMode::FailFast`] cancels the pipeline; `try_task` then records
/// [`PipelineError::TaskFailed`]. The other modes also catch a panic raised for
/// that single item.
///
/// # Channels and back-pressure
///
/// Each inter-stage channel holds `buffer_size` messages (default 4). A
/// message is one item when `buffer_size` is 1 or 2, or when
/// [`Pipeline::with_max_packet`]`(1)` is set. Otherwise the feeder packs up to
/// 64 items that are already available into one message, and flushes a partial
/// message as soon as the next source item is not ready. Downstream stages do
/// the same: they forward a partial packet when the next input would block, so
/// trickle and startup latency stay one item deep.
///
/// At most `buffer_size` items are buffered for a one-item packet, and at most
/// `buffer_size * packet` items otherwise (`packet` defaults to 64). A slow
/// consumer still stalls the producer; the stall just happens on packet
/// boundaries.
///
/// The feeder thread is created in [`Pipeline::new`] but does not pull the
/// source until the first stage or terminal method (or until the pipeline is
/// dropped). [`Pipeline::with_max_packet`] called before that also caps the
/// feeder. After the feeder has started, it keeps the packet size it was given
/// and a later `with_max_packet` only affects stages added afterwards.
///
/// Worker threads are named `olympipe-<stage>` (`olympipe-task`,
/// `olympipe-feeder`, …). A single worker preserves input order. `task`
/// with more than one worker does not.
pub struct Pipeline<T: Send + 'static> {
    /// `Some` until the feeder is allowed to pull. Dropped first so a discarded
    /// pipeline unblocks the waiting feeder before the receiver disconnects.
    pending_release: Option<FeederRelease>,
    pub(crate) receiver: Receiver<Packet<T>>,
    shared: Arc<Shared>,
    pub(crate) buffer_size: usize,
    chunk_limit: usize,
}

fn downstream<T: Send + 'static>(
    receiver: Receiver<Packet<T>>,
    shared: Arc<Shared>,
    buffer_size: usize,
    chunk_limit: usize,
) -> Pipeline<T> {
    Pipeline {
        pending_release: None,
        receiver,
        shared,
        buffer_size,
        chunk_limit,
    }
}

// ─── constructor ──────────────────────────────────────────────────────────────

impl<T: Send + 'static> Pipeline<T> {
    /// Create a pipeline from any iterator.  A feeder thread is spawned
    /// immediately to push items into the first bounded channel.
    ///
    /// A panic while pulling the iterator fails the pipeline with stage
    /// `"feeder"`.
    pub fn new(iter: impl IntoIterator<Item = T> + Send + 'static) -> Self {
        Self::new_with_buffer(iter, DEFAULT_BUFFER)
    }

    /// Like [`Self::new`], but sets the feeder and subsequent stage channel capacity.
    ///
    /// `buffer_size` is the number of **messages** in each channel, not the
    /// number of items. For 1 or 2 that is the same thing (one item per
    /// message). Above that, a message holds up to 64 items that the producer
    /// already has, so a full channel retains at most `buffer_size * 64` items.
    /// [`Self::with_max_packet`]`(1)` opts back into one item per message.
    /// See the [channel notes](Pipeline#channels-and-back-pressure).
    ///
    /// The default packet size stays 64 whenever `buffer_size > 2`, including
    /// when a later [`Self::with_buffer`] picks a larger buffer. The feeder
    /// channel created here is not resized afterwards.
    pub fn new_with_buffer(
        iter: impl IntoIterator<Item = T> + Send + 'static,
        buffer_size: usize,
    ) -> Self {
        let buffer_size = buffer_size.max(1);
        let limit = chunk_limit(buffer_size);
        let (tx, rx) = bounded(buffer_size);
        let shared = Shared::new();
        let gate = FeederGate::new();
        let gate_feeder = Arc::clone(&gate);
        let shared_feeder = Arc::clone(&shared);
        Shared::spawn(&shared, "feeder", move || {
            let limit = gate_feeder.wait();
            feed(iter, tx, &shared_feeder, limit);
        });
        Pipeline {
            pending_release: Some(FeederRelease { gate, limit }),
            receiver: rx,
            shared,
            buffer_size,
            chunk_limit: limit,
        }
    }

    /// Override the inter-stage channel capacity for *subsequent* stages
    /// (default: 4). Prefer [`Self::new_with_buffer`] if the feeder channel
    /// must also use that capacity.
    ///
    /// The same packet rules as [`Self::new_with_buffer`] apply: capacity is a
    /// message count, and messages grow past one item only when `size > 2`.
    /// This resets the packet size of subsequent stages to that default, so
    /// call [`Self::with_max_packet`] after `with_buffer` when both are set.
    /// It does not change the feeder's channel capacity or packet size.
    pub fn with_buffer(mut self, size: usize) -> Self {
        self.buffer_size = size.max(1);
        self.chunk_limit = chunk_limit(self.buffer_size);
        self
    }

    /// Cap items per channel message for stages added after this call.
    ///
    /// `n` is clamped to `1..=64`. [`Self::with_max_packet`]`(1)` restores
    /// strict per-item messages: each channel holds at most `buffer_size`
    /// items, which is the behaviour from before packets carried 64 items.
    ///
    /// The feeder has not pulled the source yet on a pipeline fresh from
    /// [`Self::new`] or [`Self::new_with_buffer`], so this also sets the
    /// feeder's packet size.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let values = Pipeline::new(0..10)
    ///     .with_max_packet(1)
    ///     .task(|x: i32| x + 1, 1)
    ///     .collect();
    /// assert_eq!(values, (1..11).collect::<Vec<_>>());
    /// ```
    ///
    /// After the first stage or terminal method the feeder is already pulling
    /// and keeps the packet size it started with. A later `with_max_packet`
    /// only affects stages added from then on, the same way as
    /// [`Self::with_buffer`]. The feeder channel itself is fixed at
    /// construction and holds at most `buffer_size * packet` items.
    ///
    /// For large items (images, tensors), call this before the stages that
    /// move them, typically `.with_max_packet(1)` so a default channel keeps
    /// 4 items instead of 256.
    pub fn with_max_packet(mut self, n: usize) -> Self {
        let n = n.clamp(1, MAX_CHUNK);
        self.chunk_limit = n;
        if let Some(release) = self.pending_release.as_mut() {
            release.limit = n;
        }
        self
    }
}

// ─── internal helpers ─────────────────────────────────────────────────────────

impl<T: Send + 'static> Pipeline<T> {
    /// Let the feeder pull. No-op once a stage or terminal has already done so.
    fn activate(&mut self) {
        // Drop publishes `FeederRelease::limit` and wakes the feeder.
        self.pending_release.take();
    }

    /// Spawn `count` worker threads that each read from `self.receiver` and
    /// write to a fresh bounded channel.  `worker` must be `Clone` so it can
    /// be duplicated per thread.  Use `spawn_single` for stages that run
    /// exactly one thread and whose closure cannot be cloned.
    fn spawn_stage<U, W>(mut self, stage: &'static str, count: usize, worker: W) -> Pipeline<U>
    where
        U: Send + 'static,
        W: Fn(Receiver<Packet<T>>, Sender<Packet<U>>, Arc<Shared>) + Send + Clone + 'static,
    {
        self.activate();
        let (tx, rx) = bounded::<Packet<U>>(self.buffer_size);
        let shared = Arc::clone(&self.shared);
        for _ in 0..count {
            let worker = worker.clone();
            let in_rx = self.receiver.clone();
            let out_tx = tx.clone();
            let shared_worker = Arc::clone(&shared);
            Shared::spawn(&shared, stage, move || {
                worker(in_rx, out_tx, shared_worker);
            });
        }
        drop(tx);
        downstream(rx, shared, self.buffer_size, self.chunk_limit)
    }

    /// Like `spawn_stage` but for a single worker thread whose closure does
    /// not need to implement `Clone`.
    fn spawn_single<U, W>(mut self, stage: &'static str, worker: W) -> Pipeline<U>
    where
        U: Send + 'static,
        W: FnOnce(Receiver<Packet<T>>, Sender<Packet<U>>, Arc<Shared>) + Send + 'static,
    {
        self.activate();
        let (tx, rx) = bounded::<Packet<U>>(self.buffer_size);
        let shared = Arc::clone(&self.shared);
        let in_rx = self.receiver;
        let shared_worker = Arc::clone(&shared);
        let buffer_size = self.buffer_size;
        let chunk_limit = self.chunk_limit;
        Shared::spawn(&shared, stage, move || worker(in_rx, tx, shared_worker));
        downstream(rx, shared, buffer_size, chunk_limit)
    }

    /// Drain `self`, drop the receiver so upstream senders unblock, join every
    /// worker, and return the first recorded error.
    fn finish<R>(
        mut self,
        drain: impl FnOnce(&Receiver<Packet<T>>, &Shared) -> R,
    ) -> Result<R, PipelineError> {
        self.activate();
        let output = drain(&self.receiver, &self.shared);
        let shared = Arc::clone(&self.shared);
        drop(self);
        match shared.shutdown() {
            Some(err) => Err(err),
            None => Ok(output),
        }
    }
}

// ─── stage methods ────────────────────────────────────────────────────────────

impl<T: Send + 'static> Pipeline<T> {
    /// Map each item through `f`.  `count` worker threads run concurrently,
    /// sharing the input channel (MPMC) and writing to a shared output channel.
    ///
    /// A `count` of zero runs one worker. One worker preserves input order;
    /// several workers do not (each takes a whole packet).
    ///
    /// A panic in `f` fails the pipeline (stage `"task"`).
    pub fn task<U, F>(self, f: F, count: usize) -> Pipeline<U>
    where
        U: Send + 'static,
        F: Fn(T) -> U + Send + Clone + 'static,
    {
        let count = count.max(1);
        // One worker may drain every packet already queued. Several workers
        // take one packet each so a large packet is not stolen from the others.
        let drain = count == 1;
        let limit = self.chunk_limit;
        self.spawn_stage("task", count, move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, drain, |item, out| {
                if shared.is_cancelled() {
                    return false;
                }
                let mapped = f(item);
                !shared.is_cancelled() && out.push(mapped)
            });
        })
    }

    /// Map each item through `method` on a worker-local value built by `init`.
    ///
    /// `init` runs once on each worker thread, after the thread has started, so
    /// `S` does not have to be [`Send`]. `method` borrows that state mutably
    /// for every item the worker receives. `count == 0` runs one worker. One
    /// worker preserves input order; several workers do not (each takes a whole
    /// packet), same as [`Pipeline::task`].
    ///
    /// The state is dropped at the end of the worker. Use
    /// [`Pipeline::class_task_with_close`] when that drop is not enough.
    /// A panic in `init` or `method` fails the pipeline with stage `"class_task"`.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let values = Pipeline::new(1..=3)
    ///     .class_task(
    ///         || 0,
    ///         |state, x: i32| {
    ///             *state += x;
    ///             *state
    ///         },
    ///         1,
    ///     )
    ///     .collect();
    /// assert_eq!(values, vec![1, 3, 6]);
    /// ```
    pub fn class_task<S, U, I, M>(self, init: I, method: M, count: usize) -> Pipeline<U>
    where
        S: 'static,
        U: Send + 'static,
        I: Fn() -> S + Send + Clone + 'static,
        M: Fn(&mut S, T) -> U + Send + Clone + 'static,
    {
        self.class_task_with_close(init, method, |_| {}, count)
    }

    /// [`Pipeline::class_task`] plus a hook invoked with the state when the
    /// worker stops.
    ///
    /// `close` is [`Fn`], not [`FnOnce`], so it can be cloned onto each worker
    /// (the stage body passed to the thread pool is itself [`Fn`]). It is
    /// called once per worker, on that worker's thread, after the worker has
    /// stopped pulling, in every case where the worker returns normally:
    ///
    /// * the input was exhausted;
    /// * the output channel disconnected (the downstream stage stopped reading,
    ///   for example [`Pipeline::limit`], or the branch was dropped);
    /// * the pipeline was cancelled (a failure in this stage's `method` is not
    ///   this case — that panics — but a failure elsewhere in the graph is).
    ///
    /// `close` is not called if `init` or `method` panics: the panic unwinds
    /// out of the worker first, and [`Drop`] of `S` still runs during that
    /// unwind when `init` had already returned. A panic inside `close` fails
    /// the pipeline with stage `"class_task"` after `close` has started.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    /// use std::sync::atomic::{AtomicUsize, Ordering};
    /// use std::sync::Arc;
    ///
    /// let closed = Arc::new(AtomicUsize::new(0));
    /// let flag = Arc::clone(&closed);
    /// let values = Pipeline::new(vec![1, 2, 3])
    ///     .class_task_with_close(
    ///         || 0,
    ///         |state, x: i32| {
    ///             *state += x;
    ///             *state
    ///         },
    ///         move |state| {
    ///             flag.store(state as usize, Ordering::Relaxed);
    ///         },
    ///         1,
    ///     )
    ///     .collect();
    /// assert_eq!(values, vec![1, 3, 6]);
    /// assert_eq!(closed.load(Ordering::Relaxed), 6);
    /// ```
    pub fn class_task_with_close<S, U, I, M, C>(
        self,
        init: I,
        method: M,
        close: C,
        count: usize,
    ) -> Pipeline<U>
    where
        S: 'static,
        U: Send + 'static,
        I: Fn() -> S + Send + Clone + 'static,
        M: Fn(&mut S, T) -> U + Send + Clone + 'static,
        C: Fn(S) + Send + Clone + 'static,
    {
        let count = count.max(1);
        // One worker may drain every packet already queued. Several workers
        // take one packet each, matching `task`, so a large packet is not
        // stolen from the others and a single worker keeps input order.
        let drain = count == 1;
        let limit = self.chunk_limit;
        self.spawn_stage("class_task", count, move |rx, tx, shared| {
            // `S` is built here, so it never has to cross a thread boundary.
            let mut state = init();
            map_out(rx, tx, &shared, limit, drain, |item, out| {
                if shared.is_cancelled() {
                    return false;
                }
                let mapped = method(&mut state, item);
                !shared.is_cancelled() && out.push(mapped)
            });
            // Unwind from `init` or `method` skips this call. `Shared::spawn`
            // records that panic; a panic from `close` itself is recorded too.
            close(state);
        })
    }

    /// Map each item through `f`, which may fail.
    ///
    /// This is the item-aware callback form of per-item error handling: `Err`
    /// from `f` is passed to `on_error` together with the original item and
    /// does **not** fail the pipeline. `Some(v)` forwards `v`; `None` skips the
    /// item. [`Pipeline::try_task`] is the mode-based form ([`ErrorMode`]) and
    /// does not hand the item to a callback.
    ///
    /// A panic inside `f` or `on_error` is not recoverable: it fails the
    /// pipeline with stage `"task_or"` (see [`PipelineError`]).
    pub fn task_or<U, E, F, H>(self, f: F, on_error: H) -> Pipeline<U>
    where
        T: Clone,
        U: Send + 'static,
        E: 'static,
        F: Fn(T) -> Result<U, E> + Send + 'static,
        H: Fn(T, E) -> Option<U> + Send + 'static,
    {
        let limit = self.chunk_limit;
        self.spawn_single("task_or", move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, true, |item, out| {
                let cloned = item.clone();
                match f(item) {
                    Ok(value) => out.push(value),
                    Err(err) => match on_error(cloned, err) {
                        Some(value) => out.push(value),
                        None => true,
                    },
                }
            });
        })
    }

    /// Map each item through `f`, applying `mode` when `f` returns [`Err`].
    ///
    /// `count == 0` runs one worker. One worker preserves input order; several
    /// workers do not, same as [`Pipeline::task`].
    ///
    /// * [`ErrorMode::FailFast`] — the first `Err` cancels the pipeline with
    ///   [`PipelineError::TaskFailed`] `{ stage: "try_task", message }` where
    ///   `message` is `err.to_string()`. Panics are not caught, so they still
    ///   fail the stage as [`PipelineError::WorkerPanicked`] with no extra
    ///   `catch_unwind` on the success path.
    /// * [`ErrorMode::Skip`] — the item is dropped.
    /// * [`ErrorMode::SkipAndLog`] — the item is dropped and
    ///   `olympipe-rs: [try_task] skipped item: {err}` is written to stderr.
    /// * [`ErrorMode::Replace`] — a clone of the replacement is emitted instead.
    ///
    /// In every mode except `FailFast`, a panic inside `f` is a per-item
    /// failure too (the panic payload is the message, or `"(unknown panic)"`).
    /// That matches a policy which catches every failure of one item and lets
    /// the rest of the stream continue. `FailFast` does not catch it.
    ///
    /// [`Pipeline::task_or`] remains the callback form that also receives the
    /// original item.
    ///
    /// ```
    /// use olympipe_rs::{ErrorMode, Pipeline};
    ///
    /// let values = Pipeline::new(vec!["1", "x", "3"])
    ///     .try_task(|s: &str| s.parse::<i32>(), 1, ErrorMode::Skip)
    ///     .collect();
    /// assert_eq!(values, vec![1, 3]);
    /// ```
    pub fn try_task<U, E, F>(self, f: F, count: usize, mode: ErrorMode<U>) -> Pipeline<U>
    where
        U: Clone + Send + 'static,
        E: std::fmt::Display + 'static,
        F: Fn(T) -> Result<U, E> + Send + Clone + 'static,
    {
        let count = count.max(1);
        let drain = count == 1;
        let limit = self.chunk_limit;
        // Fail-fast must not pay for `catch_unwind`. A panic still reaches
        // `Shared::spawn`, which records stage "try_task".
        if matches!(mode, ErrorMode::FailFast) {
            return self.spawn_stage("try_task", count, move |rx, tx, shared| {
                map_out(rx, tx, &shared, limit, drain, |item, out| {
                    if shared.is_cancelled() {
                        return false;
                    }
                    match f(item) {
                        Ok(value) => !shared.is_cancelled() && out.push(value),
                        Err(err) => {
                            shared.fail(PipelineError::TaskFailed {
                                stage: "try_task",
                                message: err.to_string(),
                            });
                            false
                        }
                    }
                });
            });
        }
        self.spawn_stage("try_task", count, move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, drain, |item, out| {
                if shared.is_cancelled() {
                    return false;
                }
                match panic::catch_unwind(AssertUnwindSafe(|| f(item))) {
                    Ok(Ok(value)) => !shared.is_cancelled() && out.push(value),
                    Ok(Err(err)) => {
                        !shared.is_cancelled() && note_skipped(&mode, "try_task", &err, out)
                    }
                    Err(payload) => {
                        !shared.is_cancelled()
                            && note_skipped(&mode, "try_task", &panic_message(&payload), out)
                    }
                }
            });
        })
    }

    /// [`Pipeline::task`] with an [`ErrorMode`] applied to panics inside `f`.
    ///
    /// `f` itself cannot return `Err`. [`ErrorMode::FailFast`] is `task`: the
    /// panic is not caught and the stage name stays `"task"`. The other modes
    /// catch the panic per item (`Skip` drops it, `SkipAndLog` also writes
    /// `olympipe-rs: [task_with_mode] skipped item: {message}` to stderr,
    /// `Replace` emits a clone of the replacement). `count` follows `task`.
    ///
    /// ```
    /// use olympipe_rs::{ErrorMode, Pipeline};
    ///
    /// let values = Pipeline::new(vec![1, 0, 4])
    ///     .task_with_mode(
    ///         |x: i32| {
    ///             if x == 0 {
    ///                 panic!("zero");
    ///             }
    ///             10 / x
    ///         },
    ///         1,
    ///         ErrorMode::Replace(-1),
    ///     )
    ///     .collect();
    /// assert_eq!(values, vec![10, -1, 2]);
    /// ```
    pub fn task_with_mode<U, F>(self, f: F, count: usize, mode: ErrorMode<U>) -> Pipeline<U>
    where
        U: Clone + Send + 'static,
        F: Fn(T) -> U + Send + Clone + 'static,
    {
        if matches!(mode, ErrorMode::FailFast) {
            return self.task(f, count);
        }
        let count = count.max(1);
        let drain = count == 1;
        let limit = self.chunk_limit;
        self.spawn_stage("task_with_mode", count, move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, drain, |item, out| {
                if shared.is_cancelled() {
                    return false;
                }
                match panic::catch_unwind(AssertUnwindSafe(|| f(item))) {
                    Ok(value) => !shared.is_cancelled() && out.push(value),
                    Err(payload) => {
                        !shared.is_cancelled()
                            && note_skipped(&mode, "task_with_mode", &panic_message(&payload), out)
                    }
                }
            });
        })
    }

    /// Map each item through `f`, reusing a JSON file from `cache_dir` when the
    /// same `step` has already seen that input.
    ///
    /// Rust cannot hash a closure, so `step` is the cache identity. Changing it
    /// (or the JSON produced for an input) invalidates previous files. Put a
    /// version in the name when the computation changes, for example
    /// `"resize-v2"`.
    ///
    /// The key is the lowercase hex SHA-256 of the raw 32-byte SHA-256 of
    /// `step` (UTF-8) concatenated with the `serde_json` bytes of the input.
    /// The file is `{cache_dir}/{first 16 hex chars of sha256(step)}/{key}.json`
    /// ([`cache_file`](crate::cache_file)). A hit is read and deserialized and `f` is not called.
    /// A miss calls `f` and writes the JSON atomically: a unique temporary file
    /// in that directory, then `rename`, so concurrent workers never observe a
    /// partial file. Directories are created as needed. A missing, corrupt, or
    /// undeserializable file is a miss and is overwritten.
    ///
    /// Keys follow `serde_json`. A [`std::collections::HashMap`] does not have
    /// a stable key order, so equal maps can miss. Use
    /// [`std::collections::BTreeMap`] (or another stably encoded type) when the
    /// input has a map.
    ///
    /// `count == 0` runs one worker, same as [`Pipeline::task`]. One worker
    /// preserves input order; several workers do not. The output is `U` itself.
    /// Python olympipe wraps the value in a `CacheEnvelope` and needs
    /// `uncache`; this method does not.
    ///
    /// Serializing the input is part of the key. If that fails the pipeline
    /// stops with [`PipelineError::TaskFailed`] `{ stage: "cached_task", .. }`
    /// and `f` is not called. A panic in `f` fails the pipeline with stage
    /// `"cached_task"`. An error while writing the cache (I/O, or serializing
    /// `U`) does not drop the value: it is still sent downstream, and the
    /// worker writes `olympipe-rs: [cached_task] cache write failed: {err}` to
    /// stderr once.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    /// use std::sync::atomic::{AtomicUsize, Ordering};
    /// use std::sync::Arc;
    ///
    /// let dir = std::env::temp_dir().join(format!(
    ///     "olympipe-rs-doc-cache-{}",
    ///     std::process::id()
    /// ));
    /// let _ = std::fs::remove_dir_all(&dir);
    /// let calls = Arc::new(AtomicUsize::new(0));
    /// let counted = Arc::clone(&calls);
    /// let values = Pipeline::new(vec![1, 2, 1])
    ///     .cached_task("double-v1", &dir, move |x: i32| {
    ///         counted.fetch_add(1, Ordering::SeqCst);
    ///         x * 2
    ///     }, 1)
    ///     .collect();
    /// assert_eq!(values, vec![2, 4, 2]);
    /// assert_eq!(calls.load(Ordering::SeqCst), 2);
    /// let _ = std::fs::remove_dir_all(&dir);
    /// ```
    #[cfg(feature = "cache")]
    #[cfg_attr(docsrs, doc(cfg(feature = "cache")))]
    pub fn cached_task<U, F>(
        self,
        step: &str,
        cache_dir: impl Into<std::path::PathBuf>,
        f: F,
        count: usize,
    ) -> Pipeline<U>
    where
        T: serde::Serialize,
        U: serde::Serialize + serde::de::DeserializeOwned + Send + 'static,
        F: Fn(T) -> U + Send + Clone + 'static,
    {
        let count = count.max(1);
        let drain = count == 1;
        let limit = self.chunk_limit;
        let step = step.to_owned();
        let cache_dir = cache_dir.into();
        self.spawn_stage("cached_task", count, move |rx, tx, shared| {
            let mut logged = false;
            map_out(rx, tx, &shared, limit, drain, |item, out| {
                if shared.is_cancelled() {
                    return false;
                }
                match crate::cache::resolve(&step, &cache_dir, item, &f, &mut logged) {
                    Ok(value) => !shared.is_cancelled() && out.push(value),
                    Err(err) => {
                        shared.fail(err);
                        false
                    }
                }
            });
        })
    }

    /// Keep only items for which `f` returns `true`.
    ///
    /// A panic in `f` fails the pipeline (stage `"filter"`).
    pub fn filter<F>(self, f: F) -> Pipeline<T>
    where
        F: Fn(&T) -> bool + Send + 'static,
    {
        let limit = self.chunk_limit;
        self.spawn_single("filter", move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, true, |item, out| {
                if f(&item) { out.push(item) } else { true }
            });
        })
    }

    /// Group items into `Vec<T>` of at most `size` elements.
    /// The last (potentially incomplete) batch is always emitted.
    pub fn batch(self, size: usize) -> Pipeline<Vec<T>> {
        assert!(size >= 1, "batch size must be >= 1");
        let limit = self.chunk_limit;
        self.spawn_single("batch", move |rx, tx, shared| {
            let mut buf: Vec<T> = Vec::with_capacity(size);
            let mut out = BatchTx::new(tx, limit);
            loop {
                let step = pull(&rx, &shared, true, |item| {
                    buf.push(item);
                    if buf.len() >= size {
                        let full = std::mem::replace(&mut buf, Vec::with_capacity(size));
                        return out.push(full);
                    }
                    true
                });
                match step {
                    Step::Yield => {
                        if !out.flush() {
                            break;
                        }
                    }
                    Step::Closed => {
                        if !shared.is_cancelled() {
                            if !buf.is_empty() {
                                let _ = out.push(std::mem::take(&mut buf));
                            }
                            let _ = out.flush();
                        }
                        break;
                    }
                    Step::Halted | Step::Cancelled => break,
                }
            }
        })
    }

    /// Collect items arriving within `window` of each other into a `Vec<T>`.
    /// A new batch begins whenever the inter-item gap exceeds `window`, or when
    /// the upstream channel disconnects.
    pub fn temporal_batch(self, window: Duration) -> Pipeline<Vec<T>> {
        self.spawn_single("temporal_batch", move |rx, tx, shared| {
            loop {
                if shared.is_cancelled() {
                    return;
                }
                let first = match rx.recv() {
                    Ok(packet) => packet,
                    Err(_) => return,
                };
                if shared.is_cancelled() {
                    return;
                }
                let mut batch = Vec::new();
                if !extend_batch(&mut batch, first, &shared) {
                    return;
                }
                loop {
                    if shared.is_cancelled() {
                        return;
                    }
                    match rx.recv_timeout(window) {
                        Ok(packet) => {
                            if !extend_batch(&mut batch, packet, &shared) {
                                return;
                            }
                        }
                        Err(RecvTimeoutError::Timeout) => break,
                        Err(RecvTimeoutError::Disconnected) => {
                            if !shared.is_cancelled() {
                                let _ = tx.send(Packet::One(batch));
                            }
                            return;
                        }
                    }
                }
                if shared.is_cancelled() || tx.send(Packet::One(batch)).is_err() {
                    return;
                }
            }
        })
    }

    /// Apply `f` to each item and forward every element yielded by the
    /// returned iterator (flatMap).
    ///
    /// A panic in `f` fails the pipeline (stage `"explode"`).
    pub fn explode<U, I, F>(self, f: F) -> Pipeline<U>
    where
        U: Send + 'static,
        I: IntoIterator<Item = U>,
        F: Fn(T) -> I + Send + 'static,
    {
        let limit = self.chunk_limit;
        self.spawn_single("explode", move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, true, |item, out| {
                for produced in f(item) {
                    if shared.is_cancelled() || !out.push(produced) {
                        return false;
                    }
                }
                true
            });
        })
    }

    /// Route each item to one or both output pipelines according to `f`.
    ///
    /// Both output channels are **bounded** (`self.buffer_size`). A branch that
    /// is not consumed applies back-pressure to the router, so the two branches
    /// must be consumed **concurrently** — via [`Pipeline::gather`], or with one
    /// branch drained on another thread. Draining them one after the other can
    /// block the router once a buffer fills and stall the pipeline.
    ///
    /// Dropping a branch disconnects it. The router keeps feeding the other
    /// branch and does not block on the disconnected one. If both branches are
    /// gone, the router stops.
    ///
    /// A panic in `f` fails the pipeline (stage `"split"`).
    ///
    /// For more than two branches, use [`Pipeline::split_by`] or
    /// [`Pipeline::split_array`]. For sequential consumption, use
    /// [`Pipeline::split_unbounded`].
    ///
    /// # Example
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    /// use std::thread;
    ///
    /// let (evens, odds) = Pipeline::new(0..10i32).split(|x| {
    ///     if x % 2 == 0 {
    ///         (Some(x), None)
    ///     } else {
    ///         (None, Some(x))
    ///     }
    /// });
    ///
    /// thread::scope(|scope| {
    ///     let odds = scope.spawn(|| odds.collect());
    ///     let mut evens = evens.collect();
    ///     let mut odds = odds.join().unwrap();
    ///     evens.sort();
    ///     odds.sort();
    ///     assert_eq!(evens, vec![0, 2, 4, 6, 8]);
    ///     assert_eq!(odds, vec![1, 3, 5, 7, 9]);
    /// });
    /// ```
    pub fn split<A, B, F>(self, f: F) -> (Pipeline<A>, Pipeline<B>)
    where
        A: Send + 'static,
        B: Send + 'static,
        F: Fn(T) -> (Option<A>, Option<B>) + Send + 'static,
    {
        self.split_inner(false, f)
    }

    /// Like [`Pipeline::split`], but both output channels are unbounded.
    ///
    /// There is no back-pressure between the router and its branches: a branch
    /// that is not yet consumed can retain the whole input. That makes it safe
    /// to drain one branch completely before the other. Prefer [`Pipeline::split`]
    /// when the branches are consumed concurrently and memory should stay bounded.
    ///
    /// # Example
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let (evens, odds) = Pipeline::new(0..10i32).split_unbounded(|x| {
    ///     if x % 2 == 0 {
    ///         (Some(x), None)
    ///     } else {
    ///         (None, Some(x))
    ///     }
    /// });
    /// // Sequential consumption is safe because the router never blocks on send.
    /// let evens = evens.collect();
    /// let odds = odds.collect();
    /// assert_eq!(evens, vec![0, 2, 4, 6, 8]);
    /// assert_eq!(odds, vec![1, 3, 5, 7, 9]);
    /// ```
    pub fn split_unbounded<A, B, F>(self, f: F) -> (Pipeline<A>, Pipeline<B>)
    where
        A: Send + 'static,
        B: Send + 'static,
        F: Fn(T) -> (Option<A>, Option<B>) + Send + 'static,
    {
        self.split_inner(true, f)
    }

    fn split_inner<A, B, F>(mut self, unbounded_out: bool, f: F) -> (Pipeline<A>, Pipeline<B>)
    where
        A: Send + 'static,
        B: Send + 'static,
        F: Fn(T) -> (Option<A>, Option<B>) + Send + 'static,
    {
        self.activate();
        let buffer_size = self.buffer_size;
        let chunk_limit = self.chunk_limit;
        let shared = Arc::clone(&self.shared);
        let (tx_a, rx_a) = if unbounded_out {
            unbounded()
        } else {
            bounded(buffer_size)
        };
        let (tx_b, rx_b) = if unbounded_out {
            unbounded()
        } else {
            bounded(buffer_size)
        };
        let input = self.receiver;
        let shared_router = Arc::clone(&shared);
        Shared::spawn(&shared, "split", move || {
            route(input, tx_a, tx_b, chunk_limit, f, &shared_router);
        });
        (
            downstream(rx_a, Arc::clone(&shared), buffer_size, chunk_limit),
            downstream(rx_b, shared, buffer_size, chunk_limit),
        )
    }

    /// Route each item to branch `f(&item)` of `n`.
    ///
    /// `n` must be at least 1. Every branch is a bounded channel of
    /// `buffer_size` messages, packetised on its own with the pipeline's chunk
    /// limit, and all branches share one cancellation state. An index `>= n`
    /// panics the router; the pipeline fails with
    /// [`PipelineError::WorkerPanicked`] `{ stage: "split", .. }`.
    ///
    /// Dropping a branch disconnects it. The router keeps feeding the others
    /// and stops when every branch is gone. A branch nobody reads applies
    /// back-pressure, so the branches must be consumed concurrently
    /// ([`Pipeline::gather_all`], [`Pipeline::wait_for_results`], or one thread
    /// per branch). [`Pipeline::split_by_unbounded`] is the sequential opt-in.
    ///
    /// # Panics
    ///
    /// Panics if `n` is 0.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut values = Pipeline::gather_all(
    ///     Pipeline::new(0..6i32).split_by(2, |x| if x % 2 == 0 { 0 } else { 1 }),
    /// )
    /// .collect();
    /// values.sort();
    /// assert_eq!(values, vec![0, 1, 2, 3, 4, 5]);
    /// ```
    pub fn split_by<F>(self, n: usize, f: F) -> Vec<Pipeline<T>>
    where
        F: Fn(&T) -> usize + Send + 'static,
    {
        self.split_by_inner(n, false, f)
    }

    /// Like [`Pipeline::split_by`], but every branch channel is unbounded.
    ///
    /// A branch that is not yet consumed can retain the whole input, which
    /// makes it safe to drain the [`Vec`] from front to back. Prefer
    /// [`Pipeline::split_by`] when the branches are read concurrently.
    ///
    /// # Panics
    ///
    /// Panics if `n` is 0.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut branches = Pipeline::new(0..6i32).split_by_unbounded(2, |x| usize::from(x % 2 == 1));
    /// let odds = branches.pop().expect("odds");
    /// let evens = branches.pop().expect("evens");
    /// assert_eq!(evens.collect(), vec![0, 2, 4]);
    /// assert_eq!(odds.collect(), vec![1, 3, 5]);
    /// ```
    pub fn split_by_unbounded<F>(self, n: usize, f: F) -> Vec<Pipeline<T>>
    where
        F: Fn(&T) -> usize + Send + 'static,
    {
        self.split_by_inner(n, true, f)
    }

    fn split_by_inner<F>(mut self, n: usize, unbounded_out: bool, f: F) -> Vec<Pipeline<T>>
    where
        F: Fn(&T) -> usize + Send + 'static,
    {
        assert!(n >= 1, "split_by requires at least one branch");
        self.activate();
        let buffer_size = self.buffer_size;
        let chunk_limit = self.chunk_limit;
        let shared = Arc::clone(&self.shared);
        let cap = if unbounded_out {
            None
        } else {
            Some(buffer_size)
        };
        let ends = open_channels::<T>(n, cap);
        let txs = ends.txs;
        let rxs = ends.rxs;
        let input = self.receiver;
        let shared_router = Arc::clone(&shared);
        Shared::spawn(&shared, "split", move || {
            route_by(input, txs, chunk_limit, f, &shared_router);
        });
        rxs.into_iter()
            .map(|rx| downstream(rx, Arc::clone(&shared), buffer_size, chunk_limit))
            .collect()
    }

    /// Route each item through `f` into `N` branches.
    ///
    /// `f` returns one [`Option`] per branch. `Some` is sent on that branch,
    /// `None` skips it, so one item can land on several branches or on none.
    /// `N` must be at least 1. Channels are bounded (`buffer_size` messages),
    /// each branch packetises on its own, and every branch shares one
    /// cancellation state. Dropping a branch disconnects it; the router keeps
    /// feeding the rest and stops when all of them are gone.
    ///
    /// Consume the branches concurrently. [`Pipeline::split_array_unbounded`]
    /// allows draining them one after another.
    ///
    /// A panic in `f` fails the pipeline (stage `"split"`).
    ///
    /// # Panics
    ///
    /// Panics if `N` is 0.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let [evens, odds] = Pipeline::new(0..6i32).split_array(|x| {
    ///     if x % 2 == 0 {
    ///         [Some(x), None]
    ///     } else {
    ///         [None, Some(x)]
    ///     }
    /// });
    /// let mut values = evens.gather(vec![odds]).collect();
    /// values.sort();
    /// assert_eq!(values, (0..6).collect::<Vec<_>>());
    /// ```
    pub fn split_array<const N: usize, U, F>(self, f: F) -> [Pipeline<U>; N]
    where
        U: Send + 'static,
        F: Fn(T) -> [Option<U>; N] + Send + 'static,
    {
        self.split_array_inner(false, f)
    }

    /// Like [`Pipeline::split_array`], but every branch channel is unbounded.
    ///
    /// Sequential consumption is safe. The router can retain every item that a
    /// branch has not read yet.
    ///
    /// # Panics
    ///
    /// Panics if `N` is 0.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let [evens, odds] = Pipeline::new(0..6i32).split_array_unbounded(|x| {
    ///     if x % 2 == 0 {
    ///         [Some(x), None]
    ///     } else {
    ///         [None, Some(x)]
    ///     }
    /// });
    /// assert_eq!(evens.collect(), vec![0, 2, 4]);
    /// assert_eq!(odds.collect(), vec![1, 3, 5]);
    /// ```
    pub fn split_array_unbounded<const N: usize, U, F>(self, f: F) -> [Pipeline<U>; N]
    where
        U: Send + 'static,
        F: Fn(T) -> [Option<U>; N] + Send + 'static,
    {
        self.split_array_inner(true, f)
    }

    fn split_array_inner<const N: usize, U, F>(
        mut self,
        unbounded_out: bool,
        f: F,
    ) -> [Pipeline<U>; N]
    where
        U: Send + 'static,
        F: Fn(T) -> [Option<U>; N] + Send + 'static,
    {
        assert!(N >= 1, "split_array requires at least one branch");
        self.activate();
        let buffer_size = self.buffer_size;
        let chunk_limit = self.chunk_limit;
        let shared = Arc::clone(&self.shared);
        let cap = if unbounded_out {
            None
        } else {
            Some(buffer_size)
        };
        let ends = open_channels::<U>(N, cap);
        let txs = ends.txs;
        let rxs = ends.rxs;
        let input = self.receiver;
        let shared_router = Arc::clone(&shared);
        Shared::spawn(&shared, "split", move || {
            route_fanout(input, txs, chunk_limit, f, &shared_router);
        });
        let mut rxs = rxs.into_iter();
        std::array::from_fn(|_| {
            downstream(
                rxs.next().expect("one receiver per branch"),
                Arc::clone(&shared),
                buffer_size,
                chunk_limit,
            )
        })
    }

    /// Merge `self` and all pipelines in `others` into a single output stream.
    /// One forwarder thread is spawned per source; all handles are merged into
    /// the returned pipeline's handle registry.
    ///
    /// A failure in any source (including a source iterator panic) fails the
    /// gathered pipeline.
    ///
    /// [`Pipeline::gather_all`] is the same merge when every branch, including
    /// this one, is already in a [`Vec`] (the shape [`Pipeline::split_by`]
    /// returns).
    pub fn gather(mut self, others: Vec<Pipeline<T>>) -> Pipeline<T> {
        self.activate();
        let others: Vec<Pipeline<T>> = others
            .into_iter()
            .map(|mut other| {
                other.activate();
                other
            })
            .collect();
        let buffer_size = self.buffer_size;
        let chunk_limit = self.chunk_limit;
        let (tx, rx) = bounded::<Packet<T>>(buffer_size);
        let shared = Arc::clone(&self.shared);

        let tx0 = tx.clone();
        let self_rx = self.receiver;
        let shared_fwd = Arc::clone(&shared);
        Shared::spawn(&shared, "gather", move || forward(self_rx, tx0, shared_fwd));

        for other in others {
            Shared::link(&shared, &other.shared);
            let drained = {
                let mut guard = lock(&other.shared.handles);
                std::mem::take(&mut *guard)
            };
            lock(&shared.handles).extend(drained);

            let tx_n = tx.clone();
            let other_rx = other.receiver;
            let shared_fwd = Arc::clone(&shared);
            Shared::spawn(&shared, "gather", move || {
                forward(other_rx, tx_n, shared_fwd)
            });
        }
        drop(tx);

        downstream(rx, shared, buffer_size, chunk_limit)
    }

    /// Merge every pipeline in `pipes` into one stream.
    ///
    /// The output channel uses the first pipeline's buffer size and packet
    /// limit, because this calls [`Pipeline::gather`] with that pipeline as
    /// `self`. A failure in any source fails the gathered pipeline.
    ///
    /// # Panics
    ///
    /// Panics if `pipes` is empty.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut values = Pipeline::gather_all(vec![
    ///     Pipeline::new(vec![1, 2]),
    ///     Pipeline::new(vec![3, 4]),
    /// ])
    /// .collect();
    /// values.sort();
    /// assert_eq!(values, vec![1, 2, 3, 4]);
    /// ```
    pub fn gather_all(pipes: Vec<Pipeline<T>>) -> Pipeline<T> {
        assert!(
            !pipes.is_empty(),
            "gather_all requires at least one pipeline"
        );
        let mut pipes = pipes.into_iter();
        let first = pipes
            .next()
            .expect("gather_all requires at least one pipeline");
        first.gather(pipes.collect())
    }

    /// Fold all items into a single accumulated value and emit it as the sole
    /// output item.
    ///
    /// A panic in `f` fails the pipeline (stage `"reduce"`). The accumulator is
    /// not emitted when the pipeline is cancelled.
    pub fn reduce<U, F>(self, init: U, f: F) -> Pipeline<U>
    where
        U: Send + 'static,
        F: Fn(U, T) -> U + Send + 'static,
    {
        self.spawn_single("reduce", move |rx, tx, shared| {
            let mut acc = Some(init);
            loop {
                match pull(&rx, &shared, true, |item| {
                    let current = acc.take().expect("accumulator");
                    acc = Some(f(current, item));
                    true
                }) {
                    Step::Yield => {}
                    Step::Closed => {
                        if !shared.is_cancelled() {
                            if let Some(value) = acc.take() {
                                let _ = tx.send(Packet::One(value));
                            }
                        }
                        break;
                    }
                    Step::Halted | Step::Cancelled => break,
                }
            }
        })
    }

    /// Fold every input into `initial`, then stream what `emitter` produces
    /// from that state.
    ///
    /// A single worker runs the fold, so `reducer` sees items in input order.
    /// `reducer` is `(state, item) -> state`, the same order as
    /// [`Pipeline::reduce`] and [`Iterator::fold`]. Python olympipe passes
    /// `(item, state)`.
    ///
    /// `emitter` runs only after the input is exhausted, and only if this
    /// pipeline was not cancelled or failed. Each produced item is sent
    /// downstream with the same packetised channel as other stages, so a full
    /// buffer stalls this worker and a disconnected or cancelled downstream
    /// stops the rest of the iterator. Nothing is sent when the fold stops
    /// because the pipeline failed or was cancelled, including when the input
    /// ends after a failure.
    ///
    /// A panic in `reducer` or `emitter` fails the pipeline with stage
    /// `"checkpoint"`.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let sorted = Pipeline::new(vec![3, 1, 2]).checkpoint(
    ///     Vec::new(),
    ///     |mut state, item| {
    ///         state.push(item);
    ///         state
    ///     },
    ///     |mut state| {
    ///         state.sort();
    ///         state
    ///     },
    /// );
    /// assert_eq!(sorted.collect(), vec![1, 2, 3]);
    /// ```
    pub fn checkpoint<S, U, I, R, E>(self, initial: S, reducer: R, emitter: E) -> Pipeline<U>
    where
        S: Send + 'static,
        U: Send + 'static,
        R: Fn(S, T) -> S + Send + 'static,
        E: FnOnce(S) -> I + Send + 'static,
        I: IntoIterator<Item = U>,
    {
        let limit = self.chunk_limit;
        self.spawn_single("checkpoint", move |rx, tx, shared| {
            let mut state = Some(initial);
            loop {
                match pull(&rx, &shared, true, |item| {
                    let current = state.take().expect("checkpoint state");
                    state = Some(reducer(current, item));
                    true
                }) {
                    Step::Yield => {}
                    Step::Closed => {
                        if !shared.is_cancelled() {
                            if let Some(finished) = state.take() {
                                let mut out = BatchTx::new(tx, limit);
                                for produced in emitter(finished) {
                                    if shared.is_cancelled() || !out.push(produced) {
                                        return;
                                    }
                                }
                                if !shared.is_cancelled() {
                                    let _ = out.flush();
                                }
                            }
                        }
                        break;
                    }
                    Step::Halted | Step::Cancelled => break,
                }
            }
        })
    }

    /// Pass at most `n` items downstream, then stop.
    pub fn limit(self, n: usize) -> Pipeline<T> {
        let limit = self.chunk_limit;
        self.spawn_single("limit", move |rx, tx, shared| {
            let mut sent = 0usize;
            map_out(rx, tx, &shared, limit, true, |item, out| {
                if !out.push(item) {
                    return false;
                }
                sent += 1;
                sent < n
            });
        })
    }

    /// Fail the pipeline if no item arrives within `duration` between two
    /// consecutive items.
    ///
    /// The first item is awaited without a deadline. On expiry the stage
    /// records [`PipelineError::Timeout`], cancels the graph, and drops its
    /// output sender. [`Pipeline::try_collect`] (and the other `try_*`
    /// terminals) return that error; [`Pipeline::collect`] panics.
    pub fn timeout(self, duration: Duration) -> Pipeline<T> {
        let limit = self.chunk_limit;
        self.spawn_single("timeout", move |rx, tx, shared| {
            let mut out = BatchTx::new(tx, limit);
            let first = match rx.recv() {
                Ok(packet) => packet,
                Err(_) => return,
            };
            if !forward_packet(&mut out, first, &shared) {
                return;
            }
            loop {
                if shared.is_cancelled() {
                    return;
                }
                match rx.recv_timeout(duration) {
                    Ok(packet) => {
                        if !forward_packet(&mut out, packet, &shared) {
                            break;
                        }
                    }
                    Err(RecvTimeoutError::Timeout) => {
                        shared.fail(PipelineError::Timeout(duration));
                        break;
                    }
                    Err(RecvTimeoutError::Disconnected) => break,
                }
            }
        })
    }

    /// Print each item to stdout and forward it unchanged.
    pub fn debug(self) -> Pipeline<T>
    where
        T: std::fmt::Debug,
    {
        let limit = self.chunk_limit;
        self.spawn_single("debug", move |rx, tx, shared| {
            map_out(rx, tx, &shared, limit, true, |item, out| {
                println!("[olympipe-rs] {item:?}");
                out.push(item)
            });
        })
    }
}

/// Push every item of `packet` into `batch`. `false` if the pipeline is cancelled.
fn extend_batch<T>(batch: &mut Vec<T>, packet: Packet<T>, shared: &Shared) -> bool {
    feed_packet(packet, shared, &mut |item| {
        batch.push(item);
        true
    })
}

fn forward_packet<T>(out: &mut BatchTx<T>, packet: Packet<T>, shared: &Shared) -> bool {
    let ok = feed_packet(packet, shared, &mut |item| out.push(item));
    ok && !shared.is_cancelled() && out.flush()
}

/// Forward packets unchanged. Gather does not repack: the source stage already
/// coalesced, and one send per packet keeps per-source order.
fn forward<T>(rx: Receiver<Packet<T>>, tx: Sender<Packet<T>>, shared: Arc<Shared>) {
    loop {
        if shared.is_cancelled() {
            break;
        }
        match rx.recv() {
            Ok(packet) => {
                if shared.is_cancelled() || tx.send(packet).is_err() {
                    break;
                }
            }
            Err(_) => break,
        }
    }
}

fn route<T, A, B, F>(
    rx: Receiver<Packet<T>>,
    tx_a: Sender<Packet<A>>,
    tx_b: Sender<Packet<B>>,
    limit: usize,
    f: F,
    shared: &Shared,
) where
    F: Fn(T) -> (Option<A>, Option<B>),
{
    let mut out_a = BatchTx::new(tx_a, limit);
    let mut out_b = BatchTx::new(tx_b, limit);
    let mut a_open = true;
    let mut b_open = true;
    loop {
        if shared.is_cancelled() || !(a_open || b_open) {
            break;
        }
        let step = pull(&rx, shared, true, |item| {
            if shared.is_cancelled() {
                return false;
            }
            let (a, b) = f(item);
            if a_open {
                if let Some(value) = a {
                    if !out_a.push(value) {
                        a_open = false;
                    }
                }
            }
            if b_open {
                if let Some(value) = b {
                    if !out_b.push(value) {
                        b_open = false;
                    }
                }
            }
            a_open || b_open
        });
        if a_open && !out_a.flush() {
            a_open = false;
        }
        if b_open && !out_b.flush() {
            b_open = false;
        }
        if step != Step::Yield || !(a_open || b_open) {
            break;
        }
    }
}

/// N-way channels. `None` is unbounded; `Some(capacity)` is bounded. Kept off
/// `route` so the two-branch split path does not grow a `Vec` of senders.
struct ChannelEnds<T> {
    txs: Vec<Sender<Packet<T>>>,
    rxs: Vec<Receiver<Packet<T>>>,
}

fn open_channels<T>(n: usize, capacity: Option<usize>) -> ChannelEnds<T> {
    let mut txs = Vec::with_capacity(n);
    let mut rxs = Vec::with_capacity(n);
    for _ in 0..n {
        let (tx, rx) = match capacity {
            Some(capacity) => bounded(capacity),
            None => unbounded(),
        };
        txs.push(tx);
        rxs.push(rx);
    }
    ChannelEnds { txs, rxs }
}

fn flush_open<T>(outs: &mut [Option<BatchTx<T>>]) {
    for slot in outs.iter_mut() {
        let failed = match slot.as_mut() {
            Some(out) => !out.flush(),
            None => false,
        };
        if failed {
            *slot = None;
        }
    }
}

/// Index routing. An out-of-range index panics so [`Shared::spawn`] records
/// stage `"split"`, matching a panic inside the two-way split closure.
fn route_by<T, F>(
    rx: Receiver<Packet<T>>,
    txs: Vec<Sender<Packet<T>>>,
    limit: usize,
    f: F,
    shared: &Shared,
) where
    F: Fn(&T) -> usize,
{
    let n = txs.len();
    let mut outs: Vec<Option<BatchTx<T>>> = txs
        .into_iter()
        .map(|tx| Some(BatchTx::new(tx, limit)))
        .collect();
    loop {
        if shared.is_cancelled() || outs.iter().all(Option::is_none) {
            break;
        }
        let step = pull(&rx, shared, true, |item| {
            if shared.is_cancelled() {
                return false;
            }
            let index = f(&item);
            if index >= n {
                panic!("split index {index} is outside 0..{n}");
            }
            let disconnected = match outs[index].as_mut() {
                Some(out) => !out.push(item),
                None => false,
            };
            if disconnected {
                outs[index] = None;
            }
            outs.iter().any(Option::is_some)
        });
        flush_open(&mut outs);
        if step != Step::Yield || outs.iter().all(Option::is_none) {
            break;
        }
    }
}

/// One item may be copied onto several branches, or onto none. Separate from
/// `route` so the two-output hot path stays a pair of senders.
fn route_fanout<T, U, const N: usize, F>(
    rx: Receiver<Packet<T>>,
    txs: Vec<Sender<Packet<U>>>,
    limit: usize,
    f: F,
    shared: &Shared,
) where
    F: Fn(T) -> [Option<U>; N],
{
    let mut outs: Vec<Option<BatchTx<U>>> = txs
        .into_iter()
        .map(|tx| Some(BatchTx::new(tx, limit)))
        .collect();
    loop {
        if shared.is_cancelled() || outs.iter().all(Option::is_none) {
            break;
        }
        let step = pull(&rx, shared, true, |item| {
            if shared.is_cancelled() {
                return false;
            }
            let parts = f(item);
            for (index, part) in parts.into_iter().enumerate() {
                let Some(value) = part else {
                    continue;
                };
                let disconnected = match outs[index].as_mut() {
                    Some(out) => !out.push(value),
                    None => false,
                };
                if disconnected {
                    outs[index] = None;
                }
            }
            outs.iter().any(Option::is_some)
        });
        flush_open(&mut outs);
        if step != Step::Yield || outs.iter().all(Option::is_none) {
            break;
        }
    }
}

/// `Skip` / `SkipAndLog` / `Replace` for one item. `FailFast` never reaches here.
fn note_skipped<U: Clone>(
    mode: &ErrorMode<U>,
    stage: &str,
    err: &dyn std::fmt::Display,
    out: &mut BatchTx<U>,
) -> bool {
    match mode {
        ErrorMode::Skip => true,
        ErrorMode::SkipAndLog => {
            eprintln!("olympipe-rs: [{stage}] skipped item: {err}");
            true
        }
        ErrorMode::Replace(value) => out.push(value.clone()),
        ErrorMode::FailFast => unreachable!("FailFast does not recover per item"),
    }
}

// ─── terminal methods ─────────────────────────────────────────────────────────

impl<T: Send + 'static> Pipeline<T> {
    /// Collect all output items into a `Vec`, then join every worker thread.
    ///
    /// # Panics
    ///
    /// Panics if a worker panics or [`Self::timeout`] fires:
    /// `panic!("olympipe-rs pipeline failed: {err}")`.
    /// The message includes the original worker panic text.
    /// Use [`Self::try_collect`] to receive the [`PipelineError`] instead.
    /// Recoverable per-item errors belong in [`Self::task_or`] and do not panic.
    pub fn collect(self) -> Vec<T> {
        match self.try_collect() {
            Ok(items) => items,
            Err(err) => panic!("olympipe-rs pipeline failed: {err}"),
        }
    }

    /// Apply `f` to each output item, then join every worker thread.
    ///
    /// # Panics
    ///
    /// Panics if a worker panics or [`Self::timeout`] fires:
    /// `panic!("olympipe-rs pipeline failed: {err}")`.
    /// The message includes the original worker panic text.
    /// Use [`Self::try_for_each`] to receive the [`PipelineError`] instead.
    /// Recoverable per-item errors belong in [`Self::task_or`] and do not panic.
    pub fn for_each<F>(self, f: F)
    where
        F: FnMut(T),
    {
        if let Err(err) = self.try_for_each(f) {
            panic!("olympipe-rs pipeline failed: {err}");
        }
    }

    /// Drain and discard all output, then join every worker thread.
    ///
    /// # Panics
    ///
    /// Panics if a worker panics or [`Self::timeout`] fires:
    /// `panic!("olympipe-rs pipeline failed: {err}")`.
    /// The message includes the original worker panic text.
    /// Use [`Self::try_wait_for_completion`] to receive the [`PipelineError`]
    /// instead. Recoverable per-item errors belong in [`Self::task_or`] and do
    /// not panic.
    pub fn wait_for_completion(self) {
        if let Err(err) = self.try_wait_for_completion() {
            panic!("olympipe-rs pipeline failed: {err}");
        }
    }

    /// Collect all output items, join every worker, and return the first
    /// pipeline failure.
    ///
    /// `Ok` means every worker finished without a panic or a timeout. On `Err`,
    /// items already queued are discarded: the pipeline's output is not a
    /// partial success.
    pub fn try_collect(self) -> Result<Vec<T>, PipelineError> {
        self.finish(|rx, shared| {
            let mut items = Vec::new();
            while let Some(packet) = recv_user(rx, shared) {
                match packet {
                    Packet::One(item) => items.push(item),
                    Packet::Many(more) => items.extend(more),
                }
            }
            items
        })
    }

    /// Apply `f` to each output item, join every worker, and return the first
    /// pipeline failure.
    ///
    /// A panic in `f` is a panic of the calling thread, not a [`PipelineError`].
    pub fn try_for_each<F>(self, mut f: F) -> Result<(), PipelineError>
    where
        F: FnMut(T),
    {
        self.finish(|rx, shared| {
            while let Some(packet) = recv_user(rx, shared) {
                match packet {
                    Packet::One(item) => f(item),
                    Packet::Many(more) => {
                        for item in more {
                            f(item);
                        }
                    }
                }
            }
        })
    }

    /// Drain and discard all output, join every worker, and return the first
    /// pipeline failure.
    pub fn try_wait_for_completion(self) -> Result<(), PipelineError> {
        self.finish(|rx, shared| while recv_user(rx, shared).is_some() {})
    }

    /// Collect this pipeline and `others`, with this pipeline's items first.
    ///
    /// Every pipeline is drained at the same time. Bounded branches of one
    /// [`Pipeline::split`] / [`Pipeline::split_by`] graph deadlock if they are
    /// read one after another; concurrent drains keep the router moving.
    ///
    /// # Panics
    ///
    /// Panics if any pipeline fails, with `olympipe-rs pipeline failed: {err}`,
    /// the same message as [`Pipeline::collect`].
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut branches = Pipeline::new(0..6i32).split_by(2, |x| usize::from(x % 2 == 1));
    /// let odds = branches.pop().expect("odds");
    /// let evens = branches.pop().expect("evens");
    /// let mut groups = evens.wait_for_results(vec![odds]);
    /// for group in &mut groups {
    ///     group.sort();
    /// }
    /// assert_eq!(groups[0], vec![0, 2, 4]);
    /// assert_eq!(groups[1], vec![1, 3, 5]);
    /// ```
    pub fn wait_for_results(self, others: Vec<Pipeline<T>>) -> Vec<Vec<T>> {
        unwrap_pipeline(self.try_wait_for_results(others))
    }

    /// [`Pipeline::wait_for_results`] as a [`Result`].
    ///
    /// On failure the successful pipelines' items are discarded. When several
    /// pipelines fail, the error is the one from the earliest pipeline in the
    /// list (`self`, then `others` in order) — each terminal still runs to
    /// completion so a sibling blocked on a shared router can observe the
    /// cancellation and return.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut branches = Pipeline::new(vec![1, 2, 3]).split_by(1, |_| 0);
    /// let only = branches.pop().expect("branch");
    /// let groups = only.try_wait_for_results(vec![]).expect("results");
    /// assert_eq!(groups, vec![vec![1, 2, 3]]);
    /// ```
    pub fn try_wait_for_results(
        self,
        others: Vec<Pipeline<T>>,
    ) -> Result<Vec<Vec<T>>, PipelineError> {
        let mut pipes = Vec::with_capacity(1 + others.len());
        pipes.push(self);
        pipes.extend(others);
        // Reading one bounded branch to completion before the next fills the
        // router and deadlocks. Every terminal has to be in `recv` together.
        collect_concurrent(pipes, Pipeline::try_collect)
    }

    /// Drain and discard this pipeline and `others`, concurrently.
    ///
    /// # Panics
    ///
    /// Panics if any pipeline fails, with `olympipe-rs pipeline failed: {err}`.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let mut branches = Pipeline::new(0..4i32).split_by(2, |x| usize::from(x % 2 == 1));
    /// let odds = branches.pop().expect("odds");
    /// let evens = branches.pop().expect("evens");
    /// evens.wait_for_all_completions(vec![odds]);
    /// ```
    pub fn wait_for_all_completions(self, others: Vec<Pipeline<T>>) {
        unwrap_pipeline(self.try_wait_for_all_completions(others));
    }

    /// [`Pipeline::wait_for_all_completions`] as a [`Result`].
    ///
    /// Output is discarded. Failure handling matches
    /// [`Pipeline::try_wait_for_results`]: siblings are still joined, so a
    /// shared cancellation flag can unblock them.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// Pipeline::new(0..3i32)
    ///     .try_wait_for_all_completions(vec![])
    ///     .expect("completed");
    /// ```
    pub fn try_wait_for_all_completions(
        self,
        others: Vec<Pipeline<T>>,
    ) -> Result<(), PipelineError> {
        let mut pipes = Vec::with_capacity(1 + others.len());
        pipes.push(self);
        pipes.extend(others);
        collect_concurrent(pipes, Pipeline::try_wait_for_completion).map(|_| ())
    }

    /// Fold every item with `f` and return the accumulator.
    ///
    /// `f` takes `(accumulator, item)`, the same order as [`Pipeline::reduce`].
    /// An empty input returns `init`. One worker performs the fold, so input
    /// order is preserved.
    ///
    /// # Panics
    ///
    /// Panics if the pipeline fails, with `olympipe-rs pipeline failed: {err}`.
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let sum = Pipeline::new(1..=4).wait_and_reduce(0, |acc, x| acc + x);
    /// assert_eq!(sum, 10);
    /// ```
    pub fn wait_and_reduce<U, F>(self, init: U, f: F) -> U
    where
        U: Send + 'static,
        F: Fn(U, T) -> U + Send + 'static,
    {
        unwrap_pipeline(self.try_wait_and_reduce(init, f))
    }

    /// [`Pipeline::wait_and_reduce`] as a [`Result`].
    ///
    /// A panic in `f` is [`PipelineError::WorkerPanicked`] with stage
    /// `"reduce"`, because the fold is [`Pipeline::reduce`].
    ///
    /// ```
    /// use olympipe_rs::Pipeline;
    ///
    /// let sum = Pipeline::new(1..=3)
    ///     .try_wait_and_reduce(0, |acc, x| acc + x)
    ///     .expect("sum");
    /// assert_eq!(sum, 6);
    /// ```
    pub fn try_wait_and_reduce<U, F>(self, init: U, f: F) -> Result<U, PipelineError>
    where
        U: Send + 'static,
        F: Fn(U, T) -> U + Send + 'static,
    {
        let mut reduced = self.reduce(init, f).try_collect()?;
        Ok(reduced.pop().expect("reduce emits the accumulator"))
    }
}

/// Drain `a` and `b` at the same time and return both vectors.
///
/// The two pipelines may have different item types. They are read concurrently
/// so a bounded split (see [`Pipeline::split_array`]) cannot deadlock on a
/// full branch. If both fail, `a`'s error is returned. The other terminal is
/// still joined: a shared cancellation flag must be able to wake it.
///
/// ```
/// use olympipe_rs::{try_collect_pair, Pipeline};
///
/// let numbers = Pipeline::new(vec![1, 2]);
/// let labels = Pipeline::new(vec!["a", "b"]);
/// let (numbers, labels) = try_collect_pair(numbers, labels).expect("pair");
/// assert_eq!(numbers, vec![1, 2]);
/// assert_eq!(labels, vec!["a", "b"]);
/// ```
pub fn try_collect_pair<A, B>(
    a: Pipeline<A>,
    b: Pipeline<B>,
) -> Result<(Vec<A>, Vec<B>), PipelineError>
where
    A: Send + 'static,
    B: Send + 'static,
{
    thread::scope(|scope| {
        let right = scope.spawn(|| b.try_collect());
        let left = a.try_collect();
        let right = match right.join() {
            Ok(result) => result,
            Err(payload) => std::panic::resume_unwind(payload),
        };
        match (left, right) {
            (Ok(left), Ok(right)) => Ok((left, right)),
            (Err(err), _) => Err(err),
            (Ok(_), Err(err)) => Err(err),
        }
    })
}

fn unwrap_pipeline<T>(result: Result<T, PipelineError>) -> T {
    match result {
        Ok(value) => value,
        Err(err) => panic!("olympipe-rs pipeline failed: {err}"),
    }
}

/// Run `drive` on every pipeline at once and keep the first error in list order.
fn collect_concurrent<T, R, Drive>(
    pipes: Vec<Pipeline<T>>,
    drive: Drive,
) -> Result<Vec<R>, PipelineError>
where
    T: Send + 'static,
    R: Send,
    Drive: Fn(Pipeline<T>) -> Result<R, PipelineError> + Sync,
{
    thread::scope(|scope| {
        let handles: Vec<_> = pipes
            .into_iter()
            .map(|pipe| scope.spawn(|| drive(pipe)))
            .collect();
        let mut values = Vec::with_capacity(handles.len());
        let mut failure = None;
        for handle in handles {
            match handle.join() {
                Ok(Ok(value)) => {
                    if failure.is_none() {
                        values.push(value);
                    }
                }
                Ok(Err(err)) => {
                    if failure.is_none() {
                        failure = Some(err);
                    }
                }
                Err(payload) => std::panic::resume_unwind(payload),
            }
        }
        match failure {
            Some(err) => Err(err),
            None => Ok(values),
        }
    })
}

/// One packet for a terminal, or `None` on cancel / disconnect.
/// The cancellation load is relaxed; it is not taken per item inside a packet.
fn recv_user<T>(rx: &Receiver<Packet<T>>, shared: &Shared) -> Option<Packet<T>> {
    if shared.is_cancelled() {
        return None;
    }
    match rx.recv() {
        Ok(packet) if !shared.is_cancelled() => Some(packet),
        _ => None,
    }
}