spate-core 0.4.0

//! Scripted fakes shared by the runtime and builder test suites: a
//! controllable Source with real `AckIssuer` plumbing and a counting
//! chain. The Checkpointer, DrainBarrier, and WatermarkController in the
//! loop are always the real implementations.

use crate::checkpoint::AckIssuer;
use crate::config::{
    AdminSection, BackpressureSection, CheckpointSection, ComponentConfig, MetricsExporter,
    MetricsSection, PinningMode, PipelineConfig, PipelineSection,
};
use crate::error::{ErrorClass, FatalError, SourceError};
use crate::feed::memory::MemoryFeed;
use crate::ops::{BlockReason, PushOutcome, RunnableChain};
use crate::pipeline::runtime::{RuntimeOptions, ShutdownHandle};
use crate::record::{PartitionId, RawPayload};
use crate::source::{
    DrainBarrier, LaneId, PayloadBatch, Source, SourceCtx, SourceEvent, SourceLane,
};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::time::{Duration, Instant};

/// Once `polled` has been polled, the first finished `finished()` read of
/// `lane` that directly follows a finished read of `after` signals `reached`
/// and blocks until `go` fires.
pub(crate) struct FinishedHook {
    pub(crate) polled: LaneId,
    pub(crate) after: LaneId,
    pub(crate) lane: LaneId,
    pub(crate) reached: mpsc::Sender<()>,
    pub(crate) go: mpsc::Receiver<()>,
}

#[derive(Default)]
pub(crate) struct SourceLog {
    pub(crate) committed: BTreeMap<PartitionId, i64>,
    pub(crate) pauses: Vec<Vec<LaneId>>,
    pub(crate) resumes: Vec<Vec<LaneId>>,
    pub(crate) flush_commits: usize,
    pub(crate) opened: bool,
    /// Whether `open` received the framework's source-stage handles. Guards
    /// the seam a source publishes consumer lag through; if the handles never
    /// arrive, the lag gauge renders a permanent zero on every pipeline.
    pub(crate) stage_metrics_attached: bool,
    /// The thread count `open` received.
    pub(crate) threads: Option<NonZeroUsize>,
    /// Lanes that report `finished` from the empty poll after their last
    /// batch, returning it without waiting, as a lane that has decided its
    /// end of input does.
    pub(crate) finishing: BTreeSet<LaneId>,
    /// Every lane poll, with the timeout the driver passed.
    pub(crate) lane_polls: Vec<(LaneId, Duration)>,
    /// How many times the driver asked each lane whether it has finished.
    pub(crate) finished_calls: BTreeMap<LaneId, usize>,
    /// When set, blocks one `finished()` read; see [`FinishedHook`].
    pub(crate) finished_hook: Option<FinishedHook>,
    /// The previous `finished()` read: the lane and its answer.
    pub(crate) last_finished_read: Option<(LaneId, bool)>,
    /// When set, `commit` and `flush_commits` fail retryably, modeling a
    /// checkpoint store outage; nothing is recorded as committed.
    pub(crate) fail_commits: bool,
    /// When set, `commit` fails with a fatal error.
    pub(crate) fatal_commits: bool,
    /// Interleaved ordering log shared with the chain fake.
    pub(crate) log: Vec<String>,
    /// Taken and triggered by the next `commit` with a position, modeling a
    /// stop that arrives while a tick commit waits on its store.
    pub(crate) trigger_in_commit: Option<ShutdownHandle>,
    /// The positions of each `commit` that saw the stop signal set and failed
    /// retryably, as a source whose store refuses writes during a stop does.
    pub(crate) stopped_commits: Vec<Vec<(PartitionId, i64)>>,
    /// When set, `configure_feed` accepts the section, as a feed-reading
    /// source does; otherwise it refuses it as the trait default does.
    pub(crate) accept_feed: bool,
    /// The origin `configure_feed` was handed.
    pub(crate) feed_origin: Option<crate::feed::FeedOrigin>,
    /// The feed `configure_feed` was handed, kept so a test reads through
    /// what the source reads through.
    pub(crate) feed: Option<Arc<dyn crate::feed::Feed>>,
    /// The answer to `reading_on_stop`, modeling a feed reader that still
    /// holds units it is reading.
    pub(crate) read_on_stop: bool,
    /// How many times the controller asked `reading_on_stop`.
    pub(crate) reading_on_stop_calls: usize,
    /// How many times the controller called `renew_held`.
    pub(crate) renew_held_calls: usize,
    /// How many times the controller called `stop_reading`.
    pub(crate) stop_reading_calls: usize,
    /// When set, a dropped lane logs `lane-dropped`, which a stop's drain
    /// does once the drivers have their `Shutdown`.
    pub(crate) log_lane_drops: bool,
    /// When set, `open` fails fatally.
    pub(crate) fail_open: bool,
    /// Moved forward by this much in `open`, modeling a source that takes
    /// that long to open.
    pub(crate) advance_in_open: Option<(Arc<crate::clock::TestClock>, Duration)>,
    /// The timeout of every `poll_events` call, in order.
    pub(crate) poll_timeouts: Vec<Duration>,
    /// The index into `poll_timeouts` of the last call that returned
    /// `CommitReady`.
    pub(crate) hinted_at: Option<usize>,
    /// The index into `poll_timeouts` of the last `ReleaseHeld` call.
    pub(crate) released_at: Option<usize>,
    /// `poll_timeouts.len()` at each successful `commit`.
    pub(crate) commit_polls: Vec<usize>,
}

pub(crate) enum Script {
    Assign(Vec<LaneSpec>),
    /// Additive gain. Lanes join the *current* epoch, as a feed reader
    /// emits for each unit part it takes up.
    Add(Vec<LaneSpec>),
    Revoke(Vec<LaneId>),
    /// Finished lanes, whose input is delivered, acknowledged and committed,
    /// as a feed reader retires a unit part. Push only after the lanes'
    /// commit is observed.
    Retire(Vec<LaneId>),
    /// Hint that the named partitions should be committed ahead of the tick.
    CommitReady(Vec<PartitionId>),
    /// Report the source permanently exhausted, modeling a bounded source
    /// (backfill) requesting the graceful completion drain.
    Drained,
    /// Panic inside `poll_events`, modeling a source bug that kills the
    /// controller thread outside its drain choreography.
    PanicPoll,
    /// Drop the stashed acks inside `poll_events`, so they resolve while the
    /// controller waits on the source, then report `Idle`.
    ReleaseHeld(Arc<Mutex<Vec<crate::checkpoint::AckRef>>>),
}

pub(crate) struct LaneSpec {
    pub(crate) id: LaneId,
    pub(crate) partition: PartitionId,
    pub(crate) batches: Vec<Vec<(i64, Vec<u8>)>>,
}

pub(crate) type SharedLog = Arc<Mutex<SourceLog>>;
pub(crate) type SharedScript = Arc<Mutex<VecDeque<Script>>>;

/// The lag `FakeSource` publishes for partition 0 at `open`. Distinctive so a
/// test can prove it came from this source and not a registration default.
pub(crate) const FAKE_SOURCE_LAG: u64 = 4242;

pub(crate) struct FakeSource {
    shared: SharedLog,
    script: SharedScript,
    issuer: Option<AckIssuer>,
    stop: crate::source::StopSignal,
}

impl FakeSource {
    pub(crate) fn new() -> (Self, SharedLog, SharedScript) {
        let shared = Arc::new(Mutex::new(SourceLog::default()));
        let script = Arc::new(Mutex::new(VecDeque::new()));
        (
            FakeSource {
                shared: Arc::clone(&shared),
                script: Arc::clone(&script),
                issuer: None,
                stop: crate::source::StopSignal::default(),
            },
            shared,
            script,
        )
    }
}

impl Source for FakeSource {
    type Lane = FakeLane;

    // A distinctive non-default value so tests can prove the source's own
    // contract (not the trait default) reaches `ChainCtx`.
    fn framing_contract(&self) -> crate::framing::FramingContract {
        crate::framing::FramingContract::PerRecord
    }

    fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError> {
        {
            let mut log = self.shared.lock().unwrap();
            if log.fail_open {
                return Err(SourceError::Client {
                    class: ErrorClass::Fatal,
                    reason: "scripted open failure".into(),
                });
            }
            log.opened = true;
            log.stage_metrics_attached = ctx.stage_metrics.is_some();
            log.threads = ctx.threads;
            if let Some((clock, by)) = &log.advance_in_open {
                clock.advance(*by);
            }
        }
        // Publish a lag figure like a real source would, so a test can tell
        // whether the handle set it was given owns the series. Only the owner
        // registers a per-partition gauge; a shadow's write is dropped and the
        // series never appears.
        if let Some(m) = &ctx.stage_metrics {
            m.set_partition_lag(PartitionId(0), FAKE_SOURCE_LAG);
        }
        self.issuer = Some(ctx.issuer);
        self.stop = ctx.stop;
        Ok(())
    }

    fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<FakeLane>, SourceError> {
        let next = self.script.lock().unwrap().pop_front();
        {
            let mut log = self.shared.lock().unwrap();
            if matches!(next, Some(Script::CommitReady(_))) {
                log.hinted_at = Some(log.poll_timeouts.len());
            }
            if matches!(next, Some(Script::ReleaseHeld(_))) {
                log.released_at = Some(log.poll_timeouts.len());
            }
            log.poll_timeouts.push(timeout);
        }
        match next {
            Some(Script::Assign(specs)) => {
                let issuer = self.issuer.as_ref().expect("open before assign");
                let lanes = specs
                    .into_iter()
                    .map(|s| FakeLane {
                        id: s.id,
                        partition: s.partition,
                        batches: s.batches.into(),
                        current: Vec::new(),
                        issuer: issuer.clone(),
                        log: Arc::clone(&self.shared),
                        finished: false,
                    })
                    .collect();
                Ok(SourceEvent::LanesAssigned(lanes))
            }
            Some(Script::Add(specs)) => {
                let issuer = self.issuer.as_ref().expect("open before add");
                let lanes = specs
                    .into_iter()
                    .map(|s| FakeLane {
                        id: s.id,
                        partition: s.partition,
                        batches: s.batches.into(),
                        current: Vec::new(),
                        issuer: issuer.clone(),
                        log: Arc::clone(&self.shared),
                        finished: false,
                    })
                    .collect();
                Ok(SourceEvent::LanesAdded(lanes))
            }
            Some(Script::Revoke(ids)) => {
                let mut log = self.shared.lock().unwrap();
                log.log.push("revoke-delivered".into());
                Ok(SourceEvent::LanesRevoked {
                    barrier: DrainBarrier::new(ids.len()),
                    lanes: ids,
                })
            }
            Some(Script::Retire(lanes)) => {
                let mut log = self.shared.lock().unwrap();
                log.log.push("retire-delivered".into());
                Ok(SourceEvent::LanesRetired { lanes })
            }
            Some(Script::CommitReady(partitions)) => Ok(SourceEvent::CommitReady { partitions }),
            Some(Script::Drained) => {
                self.shared
                    .lock()
                    .unwrap()
                    .log
                    .push("drained-delivered".into());
                Ok(SourceEvent::Drained)
            }
            Some(Script::PanicPoll) => panic!("scripted poll_events panic"),
            Some(Script::ReleaseHeld(held)) => {
                held.lock().unwrap().clear();
                Ok(SourceEvent::Idle)
            }
            None => {
                std::thread::sleep(timeout.min(Duration::from_millis(5)));
                Ok(SourceEvent::Idle)
            }
        }
    }

    fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError> {
        let mut log = self.shared.lock().unwrap();
        if let Some(trigger) = log.trigger_in_commit.take() {
            trigger.trigger();
        }
        if self.stop.is_set() {
            log.stopped_commits.push(watermarks.to_vec());
            return Err(SourceError::Client {
                class: ErrorClass::Retryable,
                reason: "the stop began during this commit".into(),
            });
        }
        if log.fatal_commits {
            return Err(SourceError::Client {
                class: ErrorClass::Fatal,
                reason: "scripted fatal commit failure".into(),
            });
        }
        if log.fail_commits {
            log.log.push("commit-failed".into());
            return Err(SourceError::Client {
                class: ErrorClass::Retryable,
                reason: "scripted commit failure".into(),
            });
        }
        for &(p, o) in watermarks {
            let slot = log.committed.entry(p).or_insert(o);
            *slot = (*slot).max(o);
        }
        log.log.push("commit".into());
        let polls = log.poll_timeouts.len();
        log.commit_polls.push(polls);
        Ok(())
    }

    fn flush_commits(&mut self) -> Result<(), SourceError> {
        let mut log = self.shared.lock().unwrap();
        log.flush_commits += 1;
        log.log.push("flush_commits".into());
        if log.fail_commits {
            return Err(SourceError::Client {
                class: ErrorClass::Retryable,
                reason: "scripted flush failure".into(),
            });
        }
        Ok(())
    }

    fn pause(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> {
        self.shared.lock().unwrap().pauses.push(lanes.to_vec());
        Ok(())
    }

    fn resume(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> {
        self.shared.lock().unwrap().resumes.push(lanes.to_vec());
        Ok(())
    }

    fn renew_held(&mut self) {
        self.shared.lock().unwrap().renew_held_calls += 1;
    }

    fn reading_on_stop(&mut self) -> bool {
        let mut log = self.shared.lock().unwrap();
        log.reading_on_stop_calls += 1;
        log.read_on_stop
    }

    fn stop_reading(&mut self) -> Result<(), SourceError> {
        let mut log = self.shared.lock().unwrap();
        log.stop_reading_calls += 1;
        log.log.push("stop_reading".into());
        Ok(())
    }

    fn configure_feed(
        &mut self,
        section: &crate::config::FeedSection,
        feed: Arc<dyn crate::feed::Feed>,
        origin: crate::feed::FeedOrigin,
    ) -> Result<(), crate::config::ConfigError> {
        let _ = section;
        self.shared.lock().unwrap().feed_origin = Some(origin);
        self.shared.lock().unwrap().feed = Some(feed);
        if self.shared.lock().unwrap().accept_feed {
            return Ok(());
        }
        Err(crate::config::ConfigError::Component {
            context: "feed".into(),
            message: format!(
                "the source (`{}`) reads no feed; remove the `feed:` section",
                self.component_type()
            ),
        })
    }
}

pub(crate) struct FakeLane {
    id: LaneId,
    partition: PartitionId,
    batches: VecDeque<Vec<(i64, Vec<u8>)>>,
    current: Vec<(i64, Vec<u8>)>,
    issuer: AckIssuer,
    log: SharedLog,
    finished: bool,
}

impl Drop for FakeLane {
    fn drop(&mut self) {
        // A lane can drop while a panicking driver holds the lock poisoned.
        if let Ok(mut log) = self.log.lock()
            && log.log_lane_drops
        {
            log.log.push("lane-dropped".into());
        }
    }
}

impl SourceLane for FakeLane {
    type Batch<'a> = FakeBatch<'a>;

    fn id(&self) -> LaneId {
        self.id
    }

    fn partition(&self) -> PartitionId {
        self.partition
    }

    fn poll(
        &mut self,
        _max_records: usize,
        timeout: Duration,
    ) -> Result<Option<FakeBatch<'_>>, SourceError> {
        let finishing = {
            let mut log = self.log.lock().unwrap();
            log.lane_polls.push((self.id, timeout));
            log.finishing.contains(&self.id)
        };
        match self.batches.pop_front() {
            Some(batch) => {
                self.current = batch;
                let last = self.current.last().expect("non-empty batch").0;
                let ack = self.issuer.issue(self.partition, last);
                Ok(Some(FakeBatch {
                    payloads: &self.current,
                    idx: 0,
                    partition: self.partition,
                    ack,
                }))
            }
            None if finishing => {
                self.finished = true;
                Ok(None)
            }
            None => {
                std::thread::sleep(timeout.min(Duration::from_millis(2)));
                Ok(None)
            }
        }
    }

    fn finished(&self) -> bool {
        let hook = {
            let mut log = self.log.lock().unwrap();
            *log.finished_calls.entry(self.id).or_default() += 1;
            let previous = log.last_finished_read.replace((self.id, self.finished));
            if self.finished
                && log.finished_hook.as_ref().is_some_and(|h| {
                    h.lane == self.id
                        && previous == Some((h.after, true))
                        && log.lane_polls.iter().any(|(l, _)| *l == h.polled)
                })
            {
                log.finished_hook.take()
            } else {
                None
            }
        };
        if let Some(hook) = hook {
            hook.reached.send(()).unwrap();
            hook.go.recv().unwrap();
        }
        self.finished
    }
}

pub(crate) struct FakeBatch<'a> {
    payloads: &'a [(i64, Vec<u8>)],
    idx: usize,
    partition: PartitionId,
    ack: crate::checkpoint::AckRef,
}

impl<'a> PayloadBatch<'a> for FakeBatch<'a> {
    fn next_payload(&mut self) -> Option<RawPayload<'a>> {
        let (offset, bytes) = self.payloads.get(self.idx)?;
        self.idx += 1;
        Some(RawPayload {
            bytes,
            key: None,
            partition: self.partition,
            offset: *offset,
            timestamp_ms: *offset,
        })
    }

    fn ack(&self) -> &crate::checkpoint::AckRef {
        &self.ack
    }
}

/// A feed that hands back, over an in-process feed, under the backend key a
/// test names. It records each `close` and its bound, and settles the close
/// at once unless told to hold it.
pub(crate) struct HandingBack {
    inner: MemoryFeed,
    backend: &'static str,
    hold_close: bool,
    log: Option<SharedLog>,
    pub(crate) closes: Mutex<Vec<Duration>>,
    // A held close's confirmer, kept so the close never settles.
    held: Mutex<Vec<crate::feed::Confirmer>>,
}

impl HandingBack {
    pub(crate) fn new(backend: &'static str) -> HandingBack {
        HandingBack {
            inner: MemoryFeed::new(),
            backend,
            hold_close: false,
            log: None,
            closes: Mutex::new(Vec::new()),
            held: Mutex::new(Vec::new()),
        }
    }

    /// Never settle a close.
    pub(crate) fn holding_its_close(mut self) -> HandingBack {
        self.hold_close = true;
        self
    }

    /// Reject every close held so far with `error`.
    pub(crate) fn reject_held_closes(&self, error: &crate::feed::FeedError) {
        for confirmer in self.held.lock().unwrap().drain(..) {
            confirmer.reject(error.clone());
        }
    }

    /// Settle every close held so far.
    pub(crate) fn settle_held_closes(&self) {
        for confirmer in self.held.lock().unwrap().drain(..) {
            confirmer.confirm();
        }
    }

    /// Log `close` into `log` beside the source's entries.
    pub(crate) fn logging_into(mut self, log: SharedLog) -> HandingBack {
        self.log = Some(log);
        self
    }
}

impl crate::feed::Feed for HandingBack {
    fn backend(&self) -> &str {
        self.backend
    }

    fn set_waker(&self, waker: crate::feed::FeedWaker) {
        self.inner.set_waker(waker);
    }

    fn receive(&self, max: usize) -> Result<Vec<crate::feed::Delivery>, crate::feed::FeedError> {
        self.inner.receive(max)
    }

    fn renew(&self, receipt: crate::feed::Receipt) -> Result<(), crate::feed::FeedError> {
        self.inner.renew(receipt)
    }

    fn acknowledge(&self, receipt: crate::feed::Receipt) -> Result<(), crate::feed::FeedError> {
        self.inner.acknowledge(receipt)
    }

    fn fail(
        &self,
        receipt: crate::feed::Receipt,
        reason: &str,
    ) -> Result<(), crate::feed::FeedError> {
        self.inner.fail(receipt, reason)
    }

    fn publish(&self, unit: crate::feed::Unit) -> crate::feed::Confirmation {
        self.inner.publish(unit)
    }

    fn hands_back(&self) -> bool {
        true
    }

    fn close(&self, within: Duration) -> Option<crate::feed::Confirmation> {
        self.closes.lock().unwrap().push(within);
        if let Some(log) = &self.log {
            log.lock().unwrap().log.push("close".into());
        }
        let (confirmer, confirmation) = crate::feed::confirmation();
        if self.hold_close {
            self.held.lock().unwrap().push(confirmer);
        } else {
            confirmer.confirm();
        }
        Some(confirmation)
    }
}

#[derive(Default)]
pub(crate) struct ChainShared {
    pub(crate) consumed: AtomicUsize,
    pub(crate) flushes: AtomicUsize,
}

pub(crate) enum ChainMode {
    Ok,
    BlockOnce(AtomicBool),
    BlockForever,
    /// Block every batch until the flag is set, then take it.
    BlockUntil(Arc<AtomicBool>),
    FatalAtBatch(usize),
    PanicAtBatch(usize),
    /// Fail the ack of the n-th batch (1-based) after consuming it, stalling
    /// that partition's watermark permanently, as a fatal sink write does.
    FailAckAtBatch(usize),
    /// Clone-and-stash every batch's ack so watermarks never advance and the
    /// checkpointer's pending count climbs. Setting `release` stops stashing;
    /// the test then drops the stash to resolve the batches Delivered.
    HoldAcks {
        held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>>,
        release: Arc<AtomicBool>,
    },
    /// Park every batch's ack until the next `flush`, as a real terminal's
    /// chunk buffer does. Watermarks advance only after a flush resolves the
    /// stash. Pins choreography that must flush the chain before committing
    /// (revocation, shutdown).
    HoldAcksUntilFlush {
        held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>>,
    },
    /// Take every batch, and report every flush blocked, as a terminal whose
    /// sink queue stays full does. A drain's flush then runs to its deadline.
    FlushBlocked,
}

pub(crate) struct FakeChain {
    pub(crate) shared: Arc<ChainShared>,
    pub(crate) log: Arc<Mutex<SourceLog>>,
    pub(crate) mode: ChainMode,
    pub(crate) batches_seen: usize,
}

impl RunnableChain for FakeChain {
    fn push_batch<'buf>(&mut self, batch: &mut dyn PayloadBatch<'buf>, from: usize) -> PushOutcome {
        if from == 0 {
            self.batches_seen += 1;
        }
        match &self.mode {
            ChainMode::BlockOnce(done) if !done.swap(true, Ordering::Relaxed) => {
                return PushOutcome::Blocked {
                    resume_at: from,
                    reason: BlockReason::Capacity,
                };
            }
            ChainMode::BlockForever => {
                return PushOutcome::Blocked {
                    resume_at: from,
                    reason: BlockReason::Capacity,
                };
            }
            ChainMode::BlockUntil(open) if !open.load(Ordering::Relaxed) => {
                return PushOutcome::Blocked {
                    resume_at: from,
                    reason: BlockReason::Capacity,
                };
            }
            ChainMode::FatalAtBatch(n) if self.batches_seen == *n => {
                return PushOutcome::Fatal(FatalError {
                    component: "fake-chain".into(),
                    reason: "scripted fatal".into(),
                });
            }
            ChainMode::PanicAtBatch(n) if self.batches_seen == *n => {
                panic!("scripted panic in operator chain");
            }
            _ => {}
        }
        while let Some(_p) = batch.next_payload() {
            self.shared.consumed.fetch_add(1, Ordering::Relaxed);
        }
        match &self.mode {
            ChainMode::FailAckAtBatch(n) if self.batches_seen == *n => batch.ack().fail(),
            ChainMode::HoldAcks { held, release } if !release.load(Ordering::Relaxed) => {
                held.lock().unwrap().push(batch.ack().clone());
            }
            ChainMode::HoldAcksUntilFlush { held } => {
                held.lock().unwrap().push(batch.ack().clone());
            }
            _ => {}
        }
        PushOutcome::Done
    }

    fn flush(&mut self) -> PushOutcome {
        self.shared.flushes.fetch_add(1, Ordering::Relaxed);
        self.log.lock().unwrap().log.push("flush".into());
        if matches!(self.mode, ChainMode::FlushBlocked) {
            return PushOutcome::Blocked {
                resume_at: 0,
                reason: BlockReason::Capacity,
            };
        }
        if let ChainMode::HoldAcksUntilFlush { held } = &self.mode {
            // Dropping the stashed clones resolves the batches Delivered, the
            // fake equivalent of sealing a partial chunk and having the sink
            // write it.
            held.lock().unwrap().clear();
        }
        PushOutcome::Done
    }
}

/// A distinct pipeline name per config.
///
/// Metric gauge series are owned by exactly one live handle set per process
/// (`metrics::ownership`), and the pipeline name is part of every key. Under
/// nextest each test is its own process, but `cargo test` runs a binary's
/// tests concurrently in one process, where two pipelines sharing a name
/// collide and the second fails to start. Naming them apart keeps the two
/// runners equivalent.
pub(crate) fn test_pipeline_name() -> String {
    static NEXT: AtomicU64 = AtomicU64::new(0);
    format!("test-{}", NEXT.fetch_add(1, Ordering::Relaxed))
}

pub(crate) fn test_config(threads: usize) -> PipelineConfig {
    PipelineConfig {
        pipeline: PipelineSection {
            name: test_pipeline_name(),
            threads: Some(threads),
            io_threads: 1,
            pinning: PinningMode::Off,
        },
        checkpoint: CheckpointSection {
            interval: Duration::from_millis(20),
            max_pending_batches: 1024,
            drain_timeout: Duration::from_secs(3),
            stalled_fail_after: Duration::from_secs(120),
        },
        backpressure: BackpressureSection {
            min_pause: Duration::from_millis(10),
            ..Default::default()
        },
        metrics: MetricsSection {
            exporter: MetricsExporter::None,
            ..Default::default()
        },
        admin: AdminSection {
            listen: Some("127.0.0.1:0".parse().expect("addr")),
        },
        source: ComponentConfig::new("fake", serde_yaml::Value::Null),
        deserializer: None,
        sink: Some(ComponentConfig::new("fake", serde_yaml::Value::Null)),
        sinks: None,
        feed: None,
    }
}

pub(crate) fn test_options() -> RuntimeOptions {
    RuntimeOptions {
        handle_signals: false,
        max_records: 512,
        poll_timeout: Duration::from_millis(2),
        idle_flush: Duration::from_millis(20),
        blocked_retry: Duration::from_millis(1),
        event_poll_timeout: Duration::from_millis(5),
        version: "test".into(),
        dead_letter_read_interval: Some(crate::feed::DEAD_LETTER_READ_INTERVAL),
    }
}

pub(crate) fn batches(ranges: &[std::ops::Range<i64>]) -> Vec<Vec<(i64, Vec<u8>)>> {
    ranges
        .iter()
        .map(|r| r.clone().map(|o| (o, vec![0u8; 8])).collect())
        .collect()
}

/// Spin until `cond` holds or the timeout elapses; panic on timeout.
pub(crate) fn wait_for(what: &str, timeout: Duration, cond: impl Fn() -> bool) {
    let deadline = Instant::now() + timeout;
    while !cond() {
        assert!(Instant::now() < deadline, "timed out waiting for {what}");
        std::thread::sleep(Duration::from_millis(2));
    }
}