spate-core 0.4.0

//! Runtime integration tests over scripted fakes: a controllable Source
//! with real `AckIssuer` plumbing, a counting chain, and a mock sink
//! bundle. The Checkpointer, DrainBarrier, and WatermarkController in the
//! loop are the real implementations.

#![expect(
    clippy::disallowed_methods,
    reason = "each test joins its pipeline thread, and its io tasks end when the io runtime drops"
)]

use super::fakes::*;
use super::*;
use crate::config::PipelineConfig;
use crate::error::{ErrorClass, SourceError};
use crate::feed::memory::MemoryFeed;
use crate::metrics::names;
use crate::ops::RunnableChain;
use crate::pipeline::runtime::{PipelineRuntime, default_thread_count};
use crate::record::PartitionId;
use crate::sink::shard_queues;
use crate::source::{LaneId, Source, SourceCtx, SourceEvent};
use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
#[cfg(unix)]
use tokio::signal::unix::{SignalKind, signal};

fn test_sink() -> (SinkRuntime, Arc<AtomicBool>) {
    test_sink_recording(Arc::default())
}

/// [`test_sink`], recording the budget each drain is given into `budgets`.
fn test_sink_recording(budgets: Arc<Mutex<Vec<Duration>>>) -> (SinkRuntime, Arc<AtomicBool>) {
    let (queues, receivers) = shard_queues(1, 8);
    let drained = Arc::new(AtomicBool::new(false));
    let flag = Arc::clone(&drained);
    let drain: super::SinkDrainFn = Box::new(move |budget| {
        budgets.lock().unwrap().push(budget);
        Box::pin(async move {
            let _receivers = receivers;
            flag.store(true, Ordering::Relaxed);
            DrainReport::default()
        })
    });
    (
        SinkRuntime {
            queues: vec![queues],
            drain,
            probe: None,
            failures: SinkFailures::new(),
        },
        drained,
    )
}

struct Harness {
    /// The pipeline's in-flight budget, cloned before the runtime takes it.
    /// A test charges it directly, with no sink involved.
    budget: Arc<crate::backpressure::InflightBudget>,
    shared: Arc<Mutex<SourceLog>>,
    script: Arc<Mutex<VecDeque<Script>>>,
    chain: Arc<ChainShared>,
    drained: Arc<AtomicBool>,
    /// The register the runtime reads; a test records into it in place of a
    /// sink worker.
    sink_failures: SinkFailures,
    shutdown: super::runtime::ShutdownHandle,
    /// Every WARN `run` logged on its own thread.
    warnings: Warnings,
    /// The budget each sink drain was given.
    sink_budgets: Arc<Mutex<Vec<Duration>>>,
    /// Where the admin server bound, once `run` logs it.
    admin_addr: AdminAddr,
    join: std::thread::JoinHandle<Result<ExitReport, super::runtime::StartError>>,
}

fn start(
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    start_with_config(test_config(1), mode_factory)
}

fn start_with_config(
    config: PipelineConfig,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    start_with_options(config, test_options(), mode_factory)
}

fn start_with_options(
    config: PipelineConfig,
    options: RuntimeOptions,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    start_with(config, options, |runtime| runtime, mode_factory)
}

/// [`start_with_options`], with `customize` applied to the assembled runtime
/// before it runs.
fn start_with(
    config: PipelineConfig,
    options: RuntimeOptions,
    customize: impl FnOnce(PipelineRuntime<FakeSource>) -> PipelineRuntime<FakeSource>,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    start_prepared(
        config,
        options,
        |_| {},
        |runtime, _| customize(runtime),
        mode_factory,
    )
}

/// [`start_with`], with `prepare` run on the source's log before `run`
/// starts, for settings `run` reads at startup.
fn start_prepared(
    config: PipelineConfig,
    options: RuntimeOptions,
    prepare: impl FnOnce(&mut SourceLog),
    customize: impl FnOnce(PipelineRuntime<FakeSource>, &SharedLog) -> PipelineRuntime<FakeSource>,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    let (source, shared, script) = FakeSource::new();
    prepare(&mut shared.lock().unwrap());
    let chain_shared = Arc::new(ChainShared::default());
    let sink_budgets = Arc::new(Mutex::new(Vec::new()));
    let (sink, drained) = test_sink_recording(Arc::clone(&sink_budgets));
    let sink_failures = sink.failures.clone();
    let budget = Arc::new(crate::backpressure::InflightBudget::new());
    let budget_for_test = Arc::clone(&budget);
    let cs = Arc::clone(&chain_shared);
    let log = Arc::clone(&shared);
    let runtime = PipelineRuntime::new(
        config,
        source,
        move |_thread| {
            Box::new(mode_factory(Arc::clone(&cs), Arc::clone(&log))) as Box<dyn RunnableChain>
        },
        sink,
        budget,
    )
    .with_options(options);
    let runtime = customize(runtime, &shared);
    let shutdown = runtime.shutdown_handle();
    let warnings = Warnings::default();
    let layer = warnings.clone();
    let admin_addr = AdminAddr::default();
    let admin_layer = admin_addr.clone();
    std::sync::LazyLock::force(&BYSTANDER);
    let join = std::thread::spawn(move || {
        use tracing_subscriber::layer::SubscriberExt;
        let subscriber = tracing_subscriber::registry().with(layer).with(admin_layer);
        tracing::subscriber::with_default(subscriber, || runtime.run())
    });
    Harness {
        budget: budget_for_test,
        shared,
        script,
        chain: chain_shared,
        drained,
        sink_failures,
        shutdown,
        warnings,
        sink_budgets,
        admin_addr,
        join,
    }
}

fn assign_one_lane(h: &Harness, ranges: &[std::ops::Range<i64>]) {
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Assign(vec![LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(ranges),
        }]));
}

/// [`start_prepared`] with `clock` as both the pipeline clock and the stop
/// clock, and `customize` applied after.
fn start_on_clock(
    config: PipelineConfig,
    clock: &Arc<crate::clock::TestClock>,
    customize: impl FnOnce(PipelineRuntime<FakeSource>) -> PipelineRuntime<FakeSource>,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    start_on_clock_with(config, test_options(), clock, customize, mode_factory)
}

/// [`start_on_clock`] with `options`.
fn start_on_clock_with(
    config: PipelineConfig,
    options: RuntimeOptions,
    clock: &Arc<crate::clock::TestClock>,
    customize: impl FnOnce(PipelineRuntime<FakeSource>) -> PipelineRuntime<FakeSource>,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    let clock = Arc::clone(clock);
    start_prepared(
        config,
        options,
        |_| {},
        move |runtime, _| {
            customize(
                runtime
                    .with_clock(Arc::clone(&clock) as _)
                    .with_stop_clock(clock),
            )
        },
        mode_factory,
    )
}

/// The controller's passes so far, one `poll_events` call each.
fn controller_passes(h: &Harness) -> usize {
    h.shared.lock().unwrap().poll_timeouts.len()
}

/// Wait for `n` more controller passes.
fn passes(h: &Harness, n: usize) {
    let target = controller_passes(h) + n;
    wait_for("controller passes", Duration::from_secs(5), || {
        controller_passes(h) >= target
    });
}

/// Advance `clock` by `by` and wait for the commit tick it fires: the tick's
/// `FlushNow` reaching the chain, and the controller finishing that pass.
fn tick(h: &Harness, clock: &crate::clock::TestClock, by: Duration) {
    let flushed = flushes(h);
    let target = controller_passes(h) + 2;
    clock.advance(by);
    wait_for("the commit tick", Duration::from_secs(5), || {
        flushes(h) > flushed && controller_passes(h) >= target
    });
}

// ---------------------------------------------------------------- tests --

/// The controller broadcasts `FlushNow` on every commit tick, so a partial
/// terminal buffer can never hold acknowledgments (and with them the
/// partition watermark) longer than the checkpoint interval just because
/// polls keep returning data. The idle flush is pushed out of reach (60s
/// against a 20ms commit interval), so every flush observed here can only
/// have come from the tick broadcast.
#[test]
fn commit_tick_flushes_chains_without_an_idle_lull() {
    let options = RuntimeOptions {
        idle_flush: Duration::from_secs(60),
        ..test_options()
    };
    let h = start_with_options(test_config(1), options, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("batches consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) >= 20
    });
    let flushes_before = h.chain.flushes.load(Ordering::Relaxed);
    wait_for("commit-tick flushes", Duration::from_secs(5), || {
        h.chain.flushes.load(Ordering::Relaxed) >= flushes_before + 3
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

#[test]
fn happy_path_consumes_and_commits() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    wait_for("watermark committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&30)
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 30)]);
    assert!(h.drained.load(Ordering::Relaxed), "sink drain must run");
    let log = h.shared.lock().unwrap();
    assert!(log.opened);
    assert!(log.flush_commits >= 1, "shutdown must flush commits");
}

/// A tick commit that fails because the stop began during it leaves its
/// positions pending, and the final commit, run with the stop signal clear,
/// stores them.
#[test]
fn a_tick_commit_cut_by_the_stop_goes_out_in_the_final_commit() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().trigger_in_commit = Some(h.shutdown.clone());
    // One batch, so the first position committed is the last one.
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    let report = h.join.join().unwrap().unwrap();

    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
    let log = h.shared.lock().unwrap();
    assert_eq!(log.stopped_commits, vec![vec![(PartitionId(0), 10)]]);
    assert_eq!(
        log.log.iter().filter(|entry| *entry == "commit").count(),
        1,
        "{:?}",
        log.log
    );
    assert_eq!(log.committed.get(&PartitionId(0)), Some(&10));
}

/// Revocation stops the lane's consumption and commits every consumed offset
/// from the revocation drain. Regression for #664.
#[test]
fn revocation_drains_flushes_and_commits_in_order() {
    let mut config = test_config(1);
    // No commit tick fires during the test, so only the revocation drain can
    // commit.
    config.checkpoint.interval = Duration::from_secs(60);
    let h = start_prepared(
        config,
        test_options(),
        |log| log.log_lane_drops = true,
        |runtime, _| runtime,
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::Ok,
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    wait_for("first batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) >= 10
    });
    assert!(
        h.shared.lock().unwrap().committed.is_empty(),
        "nothing may commit before the revocation drain"
    );
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    wait_for("revocation processed", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        log.log.iter().any(|e| e == "revoke-delivered") && log.flush_commits >= 1
    });
    let consumed_at_revoke = h.chain.consumed.load(Ordering::Relaxed);
    wait_for("the revoked lane dropped", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        let revoke_at = log.log.iter().position(|e| e == "revoke-delivered");
        revoke_at.is_some_and(|at| log.log[at..].iter().any(|e| e == "lane-dropped"))
    });
    assert_eq!(
        h.chain.consumed.load(Ordering::Relaxed),
        consumed_at_revoke,
        "no consumption after lanes were revoked"
    );
    {
        let log = h.shared.lock().unwrap();
        let revoke_at = log
            .log
            .iter()
            .position(|e| e == "revoke-delivered")
            .unwrap();
        let flush_after = log.log[revoke_at..].iter().position(|e| e == "flush");
        let commit_after = log.log[revoke_at..].iter().position(|e| e == "commit");
        assert!(
            flush_after.is_some(),
            "chain must flush during revocation drain"
        );
        assert!(
            commit_after.is_some(),
            "revocation must commit acknowledged offsets"
        );
        // This chain acks on consume, so flush-before-commit ordering is
        // pinned by `revocation_flushes_parked_acks_before_committing`.
        assert_eq!(
            log.committed.get(&PartitionId(0)),
            Some(&(consumed_at_revoke as i64))
        );
    }
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// Revocation must flush the chain *before* committing. The drain turns
/// parked acknowledgments into committable watermarks. The chain here
/// parks every batch's ack until it is flushed, as a real terminal's partial
/// chunk buffer does, and both periodic flush paths are out of reach (60s
/// commit interval, 60s idle flush, sub-second test), so the only thing that
/// can resolve the acks before the revocation commit is the `StopLanes`
/// drain itself. Deleting `flush_until` from the `StopLanes` arm fails this
/// test; `revocation_drains_flushes_and_commits_in_order` cannot see that.
#[test]
fn revocation_flushes_parked_acks_before_committing() {
    let mut config = test_config(1);
    config.checkpoint.interval = Duration::from_secs(60);
    let options = RuntimeOptions {
        idle_flush: Duration::from_secs(60),
        ..test_options()
    };
    let held = Arc::new(Mutex::new(Vec::new()));
    let h = start_with_options(config, options, move |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::HoldAcksUntilFlush {
            held: Arc::clone(&held),
        },
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("batches consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) >= 20
    });
    assert!(
        h.shared.lock().unwrap().committed.is_empty(),
        "acks are parked and no commit tick is due; nothing may commit \
         before the revocation drain"
    );
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    // StopLanes → flush_until resolves the parked acks; the commit that
    // follows inside `revoke_lanes` must then cover everything consumed.
    wait_for(
        "revocation committed the drained acks",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&20),
    );
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

#[test]
fn blocked_chain_pauses_then_resumes() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::BlockOnce(AtomicBool::new(false)),
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    wait_for("pause and resume observed", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        !log.pauses.is_empty() && !log.resumes.is_empty()
    });
    {
        let log = h.shared.lock().unwrap();
        assert_eq!(log.pauses[0], vec![LaneId(0)]);
        assert_eq!(log.resumes[0], vec![LaneId(0)]);
    }
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// A batch that can never unblock must not hold shutdown hostage until the
/// barrier deadline. The driver's retry loop observes the shutdown flag,
/// abandons the batch (failing its acknowledgment so the data replays),
/// and exits promptly.
#[test]
fn shutdown_during_permanently_blocked_batch_exits_promptly_and_fails_the_batch() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::BlockForever,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    // A block pauses the lane, so the driver is inside its retry loop.
    wait_for(
        "the blocked batch pauses the lane",
        Duration::from_secs(5),
        || !h.shared.lock().unwrap().pauses.is_empty(),
    );
    let begun = std::time::Instant::now();
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert!(
        begun.elapsed() < Duration::from_secs(5),
        "shutdown must not wait for the barrier deadline"
    );
    // The abandoned batch's offsets never became committable. Nothing was
    // consumed, so no watermark advanced and no commit happened.
    assert_eq!(h.chain.consumed.load(Ordering::Relaxed), 0);
    let log = h.shared.lock().unwrap();
    assert!(
        log.committed.is_empty(),
        "a blocked, abandoned batch must not commit: {:?}",
        log.committed
    );
}

#[test]
fn fatal_chain_fails_pipeline_and_stalls_watermark() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::FatalAtBatch(2),
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("pipeline must fail");
    };
    assert_eq!(failure.component, "fake-chain");
    let log = h.shared.lock().unwrap();
    // Batch 1 delivered → committable to 10. The failed batch 2 stalls the
    // watermark. 20 and 30 must never be committed.
    let committed = log.committed.get(&PartitionId(0)).copied();
    assert!(
        committed == Some(10) || committed.is_none(),
        "watermark must not pass the failed batch (got {committed:?})"
    );
    assert!(
        h.drained.load(Ordering::Relaxed),
        "failure path drains sink"
    );
}

#[test]
fn panicking_chain_fails_pipeline() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::PanicAtBatch(2),
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20]);
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("pipeline must fail");
    };
    assert!(failure.reason.contains("panicked"), "{}", failure.reason);
    let log = h.shared.lock().unwrap();
    let committed = log.committed.get(&PartitionId(0)).copied();
    assert!(committed == Some(10) || committed.is_none());
}

/// A watermark permanently stalled behind a failed batch (as a fatal sink
/// write produces) fails the pipeline as `checkpoint` once it has been stalled
/// longer than `stalled_fail_after` on the pipeline's clock, and not at the
/// limit itself. The tracker stamps the stall and the controller ages it on
/// the clock the runtime hands both.
#[test]
fn a_stall_fails_only_past_its_limit_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    // An hour ahead of real time, so a stall stamped on any other clock
    // reads an hour old.
    clock.advance(Duration::from_secs(3600));
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    cfg.checkpoint.stalled_fail_after = Duration::from_secs(600);
    let h = start_on_clock(
        cfg,
        &clock,
        |runtime| runtime,
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::FailAckAtBatch(1),
            batches_seen: 0,
        },
    );
    // The first batch's ack fails, stalling the partition; later batches
    // deliver but their watermark can never pass the failure.
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    wait_for("every batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    passes(&h, 3);

    tick(&h, &clock, Duration::from_secs(600));
    assert!(!h.join.is_finished(), "a stall at its limit must not fail");

    clock.advance(Duration::from_secs(60));
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("a permanent stall must fail the pipeline");
    };
    assert_eq!(failure.component, "checkpoint");
    assert!(failure.reason.contains("stalled"), "{}", failure.reason);
}

/// `max_pending_batches` is a hard per-partition bound, enforced at the
/// driver's poll boundary (the regression #196 pins). With every ack
/// withheld, exactly `limit` batches enter the chain and no more, however
/// long the pipeline runs. The bound holds across commit ticks rather than
/// being re-measured by them.
#[test]
fn pending_batch_ceiling_is_a_hard_bound() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = test_config(1);
    cfg.checkpoint.max_pending_batches = 3;
    let interval = cfg.checkpoint.interval;
    let h = start_on_clock(
        cfg,
        &clock,
        |runtime| runtime,
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release_c),
            },
            batches_seen: 0,
        },
    );
    // Six batches whose acks are all withheld. The gate closes after the
    // third and the remaining three never leave the lane.
    assign_one_lane(&h, &[0..10, 10..20, 20..30, 30..40, 40..50, 50..60]);
    wait_for("the ceiling fills", Duration::from_secs(5), || {
        held.lock().unwrap().len() == 3
    });
    // Ten commit ticks later the count has not moved. The bound holds
    // instead of overshooting on the next tick's slack.
    for _ in 0..10 {
        tick(&h, &clock, interval);
    }
    assert_eq!(
        held.lock().unwrap().len(),
        3,
        "pending exceeded max_pending_batches"
    );
    // The bound is the driver's, not a controller pause. The source was
    // never asked to pause for checkpoint pressure.
    assert!(
        h.shared.lock().unwrap().pauses.is_empty(),
        "the ceiling must not be enforced via source pauses"
    );
    // Releasing the acks retires the held batches; the gate reopens and
    // the remainder flows to completion.
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    // Commits ride the tick, so each check moves the clock one interval.
    wait_for("everything commits", Duration::from_secs(5), || {
        clock.advance(interval);
        h.shared
            .lock()
            .unwrap()
            .committed
            .get(&PartitionId(0))
            .is_some_and(|&w| w >= 60)
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// A permanently failed batch stalls its partition's watermark, so the
/// gate pins that partition at the ceiling. The replay a restart faces is
/// bounded by `max_pending_batches`, not by fetch rate x
/// `stalled_fail_after`.
#[test]
fn a_stalled_partition_plateaus_at_the_ceiling() {
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = test_config(1);
    cfg.checkpoint.max_pending_batches = 3;
    let interval = cfg.checkpoint.interval;
    let h = start_on_clock(
        cfg,
        &clock,
        |runtime| runtime,
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::FailAckAtBatch(1),
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, &[0..10, 10..20, 20..30, 30..40, 40..50, 50..60]);
    // The failed head blocks advancement forever; the gate lets exactly
    // the ceiling through and the other three batches stay in the lane.
    wait_for("the ceiling fills", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    for _ in 0..10 {
        tick(&h, &clock, interval);
    }
    assert_eq!(
        h.chain.consumed.load(Ordering::Relaxed),
        30,
        "a stalled partition must plateau at the ceiling"
    );
    h.shutdown.trigger();
    let _ = h.join.join().unwrap();
}

/// The ceiling is per partition. A lane added while a sibling partition
/// sits at its ceiling runs immediately, gated only by its own count, as
/// `checkpoint.max_pending_batches` documents.
#[test]
fn a_partition_at_the_ceiling_gates_only_its_own_lanes() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.max_pending_batches = 3;
    let h = start_with_config(cfg, move |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::HoldAcks {
            held: Arc::clone(&held_c),
            release: Arc::clone(&release_c),
        },
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20, 20..30, 30..40, 40..50, 50..60]);
    wait_for(
        "partition 0 reaches its ceiling",
        Duration::from_secs(5),
        || held.lock().unwrap().len() == 3,
    );

    // A fresh lane on a fresh partition joins while partition 0 is gated.
    // Its batch flows immediately; the newcomer's gate counts only its own
    // partition's pending.
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Add(vec![LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(100..110))),
        }]));
    wait_for(
        "the added partition's batch is delivered beside the gated sibling",
        Duration::from_secs(5),
        || held.lock().unwrap().len() == 4,
    );
    assert!(
        h.shared.lock().unwrap().pauses.is_empty(),
        "no source pause is involved in gating"
    );

    // Releasing drains both partitions to completion.
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wait_for("both partitions commit", Duration::from_secs(5), || {
        let shared = h.shared.lock().unwrap();
        shared
            .committed
            .get(&PartitionId(0))
            .is_some_and(|&w| w >= 60)
            && shared
                .committed
                .get(&PartitionId(1))
                .is_some_and(|&w| w >= 110)
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// Lane 0 on partition 0 and lane 1 on partition 1, one batch each, both
/// committed. Returns once the source has both commits.
fn two_committed_partitions(h: &Harness) {
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..5))),
        },
    ]));
    wait_for("both partitions commit", Duration::from_secs(5), || {
        let shared = h.shared.lock().unwrap();
        shared.committed.get(&PartitionId(0)) == Some(&10)
            && shared.committed.get(&PartitionId(1)) == Some(&5)
    });
}

fn logged(h: &Harness, entry: &str) -> bool {
    h.shared.lock().unwrap().log.iter().any(|e| e == entry)
}

/// A partition whose only lane retires leaves `final_watermarks`, so the
/// report lists only what the source still holds.
#[test]
fn a_retired_partition_leaves_the_final_watermarks() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    two_committed_partitions(&h);
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Retire(vec![LaneId(1)]));
    wait_for("retirement delivered", Duration::from_secs(5), || {
        logged(&h, "retire-delivered")
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

/// A partition whose only lane is revoked leaves `final_watermarks`.
#[test]
fn a_revoked_partition_leaves_the_final_watermarks() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    two_committed_partitions(&h);
    let flushes = h.shared.lock().unwrap().flush_commits;
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(1)]));
    wait_for("revocation processed", Duration::from_secs(5), || {
        logged(&h, "revoke-delivered") && h.shared.lock().unwrap().flush_commits > flushes
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

/// Lane 0 on partition 0 with one committed batch, and lane 1 on the same
/// partition with no input, so it holds the partition and issues nothing.
fn two_lanes_on_one_committed_partition(h: &Harness) {
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(0),
            batches: Vec::new(),
        },
    ]));
    wait_for("partition 0 commits", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
}

/// Retiring one of a partition's lanes keeps its `final_watermarks` entry
/// while another lane still holds it.
#[test]
fn a_partition_keeps_its_final_watermark_while_a_lane_of_it_remains() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    two_lanes_on_one_committed_partition(&h);
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Retire(vec![LaneId(1)]));
    wait_for("retirement delivered", Duration::from_secs(5), || {
        logged(&h, "retire-delivered")
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

/// Revoking one of a partition's lanes keeps its `final_watermarks` entry
/// while another lane still holds it.
#[test]
fn a_partition_keeps_its_final_watermark_after_a_revocation() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    two_lanes_on_one_committed_partition(&h);
    let flushes = h.shared.lock().unwrap().flush_commits;
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(1)]));
    wait_for("revocation processed", Duration::from_secs(5), || {
        logged(&h, "revoke-delivered") && h.shared.lock().unwrap().flush_commits > flushes
    });
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

/// A commit that fails during a revocation and succeeds on a later tick
/// reaches the source, and its partition stays out of `final_watermarks`.
#[test]
fn a_commit_retried_after_revocation_is_not_listed() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    // Set before any lane runs, so partition 1's position is still pending
    // when its lane is revoked.
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..5))),
        },
    ]));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 15
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(1)]));
    wait_for(
        "the revocation's commit fails",
        Duration::from_secs(5),
        || {
            let log = h.shared.lock().unwrap();
            log.log
                .iter()
                .position(|e| e == "revoke-delivered")
                .is_some_and(|at| log.log[at..].iter().any(|e| e == "commit-failed"))
        },
    );
    h.shared.lock().unwrap().fail_commits = false;
    wait_for(
        "the retry commits partition 1",
        Duration::from_secs(5),
        || {
            h.shared
                .lock()
                .unwrap()
                .committed
                .contains_key(&PartitionId(1))
        },
    );
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

fn ok_harness() -> Harness {
    start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    })
}

fn revoke_and_wait(h: &Harness, lanes: Vec<LaneId>) {
    let flushes = h.shared.lock().unwrap().flush_commits;
    h.script.lock().unwrap().push_back(Script::Revoke(lanes));
    wait_for("revocation processed", Duration::from_secs(5), || {
        logged(h, "revoke-delivered") && h.shared.lock().unwrap().flush_commits > flushes
    });
}

fn lane(id: u32, partition: u32, ranges: &[std::ops::Range<i64>]) -> LaneSpec {
    LaneSpec {
        id: LaneId(id),
        partition: PartitionId(partition),
        batches: batches(ranges),
    }
}

fn wait_committed(h: &Harness, partition: u32, at: i64) {
    wait_for("the partition commits", Duration::from_secs(5), || {
        h.shared
            .lock()
            .unwrap()
            .committed
            .get(&PartitionId(partition))
            == Some(&at)
    });
}

/// A partition an eager rebalance revokes and hands straight back keeps its
/// `final_watermarks` entry though nothing commits on it afterwards.
#[test]
fn a_partition_assigned_back_keeps_its_final_watermark() {
    let h = ok_harness();
    two_committed_partitions(&h);
    revoke_and_wait(&h, vec![LaneId(0), LaneId(1)]);
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        lane(2, 0, &[]),
        lane(3, 2, std::slice::from_ref(&(0..3))),
    ]));
    wait_committed(&h, 2, 3);
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(
        report.final_watermarks,
        vec![(PartitionId(0), 10), (PartitionId(2), 3)],
        "partition 0 is held again; partition 1 went elsewhere"
    );
}

/// A lane gain drops the entries of partitions revoked earlier in the
/// epoch, so a later assignment that hands one back lists it only once it
/// commits again.
#[test]
fn a_lane_gain_drops_revoked_partitions_from_the_final_watermarks() {
    let h = ok_harness();
    two_committed_partitions(&h);
    revoke_and_wait(&h, vec![LaneId(1)]);
    h.script.lock().unwrap().push_back(Script::Add(vec![lane(
        2,
        2,
        std::slice::from_ref(&(0..3)),
    )]));
    wait_committed(&h, 2, 3);
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        lane(3, 1, &[]),
        lane(4, 3, std::slice::from_ref(&(0..2))),
    ]));
    wait_committed(&h, 3, 2);
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(3), 2)]);
}

/// A commit retried after its partition's revocation is listed once an
/// assignment hands the partition back.
#[test]
fn a_commit_retried_after_revocation_is_listed_once_assigned_back() {
    let h = ok_harness();
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        lane(0, 0, std::slice::from_ref(&(0..10))),
        lane(1, 1, std::slice::from_ref(&(0..5))),
    ]));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 15
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(1)]));
    wait_for(
        "the revocation's commit fails",
        Duration::from_secs(5),
        || {
            let log = h.shared.lock().unwrap();
            log.log
                .iter()
                .position(|e| e == "revoke-delivered")
                .is_some_and(|at| log.log[at..].iter().any(|e| e == "commit-failed"))
        },
    );
    h.shared.lock().unwrap().fail_commits = false;
    wait_committed(&h, 1, 5);
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        lane(2, 1, &[]),
        lane(3, 2, std::slice::from_ref(&(0..3))),
    ]));
    wait_committed(&h, 2, 3);
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(
        report.final_watermarks,
        vec![(PartitionId(1), 5), (PartitionId(2), 3)]
    );
}

/// An added lane on a partition that is still live fails the run from the
/// checkpointer before the lane reaches a thread.
#[test]
fn an_added_live_partition_fails_the_run() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..5)));
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Add(vec![LaneSpec {
            id: LaneId(1),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(5..10))),
        }]));
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("re-adding a live partition must fail the run");
    };
    assert_eq!(failure.component, "checkpoint");
    assert!(
        failure.reason.contains("already tracked"),
        "{}",
        failure.reason
    );
}

#[test]
fn shutdown_flushes_chain_before_exit() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..5)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 5
    });
    let flushes_before = h.chain.flushes.load(Ordering::Relaxed);
    h.shutdown.trigger();
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert!(
        h.chain.flushes.load(Ordering::Relaxed) > flushes_before,
        "shutdown must flush the chain"
    );
    assert!(h.drained.load(Ordering::Relaxed));
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 5)]);
}

/// A bounded source reporting `Drained` completes the pipeline on its own,
/// with no external shutdown trigger, through the ordinary drain
/// choreography: chain flush, sink drain, final synchronous commit.
#[test]
fn drained_source_completes_pipeline_without_external_trigger() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    // Model the bounded-source contract: Drained only after every lane's
    // data has been polled and pushed.
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 30)]);
    assert!(h.drained.load(Ordering::Relaxed), "sink drain must run");
    let log = h.shared.lock().unwrap();
    assert!(
        log.flush_commits >= 1,
        "drained completion must flush commits synchronously"
    );
}

/// An empty bounded source (nothing to assign, e.g. an empty backfill
/// listing) still completes cleanly via `Drained`.
#[test]
fn drained_before_any_data_completes_with_empty_watermarks() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert!(report.final_watermarks.is_empty());
    assert!(h.drained.load(Ordering::Relaxed), "sink drain must run");
}

/// `Completed` after `Drained` must mean every batch was acknowledged and
/// committed. If acknowledgments never resolve (a wedged sink abandoning
/// batches at the drain deadline), the backstop converts the exit into a
/// failure instead of reporting a silently incomplete backfill as done.
#[test]
fn drained_with_unacknowledged_batches_fails_instead_of_completing() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let h = start(move |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::HoldAcks {
            held: Arc::clone(&held_c),
            release: Arc::clone(&release_c),
        },
        batches_seen: 0,
    });
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    // Acks for both batches are held (never delivered, never failed) for the
    // life of the test, so the final commit cycle still sees them pending.
    h.sink_failures.record("orders", "write timed out".into());
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("drained exit with pending acks must fail, not complete");
    };
    assert_eq!(failure.component, "source");
    assert!(
        failure.reason.contains("unacknowledged")
            && failure
                .reason
                .contains("sink `orders` abandoned a batch: write timed out"),
        "{}",
        failure.reason
    );
    assert!(report.final_watermarks.is_empty(), "nothing was committed");
    drop(held);
}

/// `Completed` after `Drained` must also mean the final watermark commit
/// persisted. When the checkpoint store is down for the whole run (every
/// commit and the final flush fail retryably), all batches acknowledge
/// fine, but nothing was durably committed, and there is no next tick to
/// retry into. The backstop must fail the run instead of reporting a
/// completion whose checkpoint holds stale offsets.
#[test]
fn drained_with_failing_commit_fails_instead_of_completing() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().fail_commits = true;
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("drained exit with an unpersisted commit must fail, not complete");
    };
    assert_eq!(failure.component, "source");
    assert!(
        failure.reason.contains("did not persist"),
        "{}",
        failure.reason
    );
    assert!(
        report.final_watermarks.is_empty(),
        "nothing was durably committed"
    );
}

/// The last commit of a stop goes through `Source::commit`, once, after the
/// drain.
#[test]
fn the_final_commit_runs_after_the_drain_through_commit() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().fail_commits = true;
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    wait_for("a tick commit", Duration::from_secs(5), || {
        h.shared
            .lock()
            .unwrap()
            .log
            .iter()
            .any(|e| e == "commit-failed")
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();

    let log = h.shared.lock().unwrap().log.clone();
    let drained = log
        .iter()
        .position(|e| e == "drained-delivered")
        .expect("the drain was delivered");
    let finals = log[drained..]
        .iter()
        .filter(|e| *e == "commit-failed")
        .count();
    assert_eq!(finals, 1, "one final commit after the drain: {log:?}");
    assert!(report.final_watermarks.is_empty(), "nothing was committed");
    let ExitState::Failed(failure) = report.state else {
        panic!("a drained exit with nothing committed must fail");
    };
    assert!(
        failure.reason.contains("did not persist"),
        "{}",
        failure.reason
    );
}

/// A failed final commit records one error outcome, the same as a failed
/// tick commit, and leaves every position to the drained-exit backstop.
#[test]
fn a_failed_final_commit_counts_one_error_outcome() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(&[0..10, 10..20]),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(&[0..10, 10..20]),
        },
    ]));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 40
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();

    let log = h.shared.lock().unwrap();
    assert!(report.final_watermarks.is_empty());
    let ExitState::Failed(failure) = report.state else {
        panic!("a drained exit with nothing committed must fail");
    };
    assert!(
        failure.reason.contains("(2 partition(s) uncommitted"),
        "{}",
        failure.reason
    );

    let rendered = handle.render();
    let failed = log.log.iter().filter(|e| *e == "commit-failed").count();
    assert_eq!(
        commits_counter(&rendered, &pipeline_name, "error"),
        Some(failed as f64),
        "{rendered}"
    );
    assert!(
        commits_counter(&rendered, &pipeline_name, "ok").is_none_or(|n| n == 0.0),
        "{rendered}"
    );
}

/// A stored commit records an `ok` outcome and no `error` outcome.
#[test]
fn a_stored_commit_counts_an_ok_outcome() {
    let mut cfg = test_config(1);
    cfg.pipeline.name = "ok-commit-outcome".into();
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();

    let rendered = handle.render();
    assert!(
        commits_counter(&rendered, &pipeline_name, "ok").is_some_and(|n| n >= 1.0),
        "{rendered}"
    );
    assert!(
        commits_counter(&rendered, &pipeline_name, "error").is_none_or(|n| n == 0.0),
        "{rendered}"
    );
}

/// A fatal commit error in the final pass fails the run with the commit's own
/// reason.
#[test]
fn a_fatal_final_commit_fails_the_run_with_its_reason() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().fail_commits = true;
    assign_one_lane(&h, &[0..10, 10..20]);
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    wait_for("a tick commit", Duration::from_secs(5), || {
        h.shared
            .lock()
            .unwrap()
            .log
            .iter()
            .any(|e| e == "commit-failed")
    });
    h.shared.lock().unwrap().fatal_commits = true;
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("a fatal final commit must fail the run");
    };
    assert!(
        failure.reason.contains("commit failed fatally"),
        "{}",
        failure.reason
    );
}

/// A commit-ready hint commits the named partitions ahead of the tick and
/// leaves the others to it.
#[test]
fn a_commit_ready_hint_commits_only_the_chased_partitions() {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    {
        let mut script = h.script.lock().unwrap();
        script.push_back(Script::CommitReady(vec![PartitionId(9)]));
        script.push_back(Script::CommitReady(vec![PartitionId(0)]));
    }
    wait_for(
        "the chased partition committed",
        Duration::from_secs(5),
        || {
            h.shared
                .lock()
                .unwrap()
                .committed
                .contains_key(&PartitionId(0))
        },
    );
    {
        let log = h.shared.lock().unwrap();
        assert!(
            !log.committed.contains_key(&PartitionId(1)),
            "{:?}",
            log.log
        );
        assert_eq!(
            log.log.iter().filter(|e| *e == "commit").count(),
            1,
            "{:?}",
            log.log
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A hinted partition whose ack resolves after its tightened window is
/// committed without waiting for the periodic tick, and the commit carries
/// only the hinted partition.
#[test]
fn a_hinted_partition_acked_after_its_window_commits_before_the_tick() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_chase_window(Duration::ZERO),
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release_c),
            },
            batches_seen: 0,
        },
    );
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("the hint delivered", Duration::from_secs(5), || {
        h.shared.lock().unwrap().hinted_at.is_some()
    });
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wait_for(
        "the hinted partition committed",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10),
    );
    {
        let log = h.shared.lock().unwrap();
        assert!(
            !log.committed.contains_key(&PartitionId(1)),
            "{:?}",
            log.committed
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A hinted partition inside its window is polled at the tightened cadence
/// while its acks are outstanding.
#[test]
fn a_hinted_partition_in_its_window_is_polled_at_the_tightened_cadence() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with_config(cfg, move |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::HoldAcks {
            held: Arc::clone(&held_c),
            release: Arc::clone(&release_c),
        },
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("a poll after the hint", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        log.hinted_at
            .is_some_and(|at| log.poll_timeouts.len() > at + 1)
    });
    {
        let log = h.shared.lock().unwrap();
        let at = log.hinted_at.unwrap();
        assert_eq!(
            log.poll_timeouts[at + 1],
            super::controller::FAST_COMMIT_POLL,
            "{:?}",
            &log.poll_timeouts[at..]
        );
    }
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wait_for(
        "the hinted partition committed",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10),
    );
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A post-window commit that fails retryably is not sent again on every pass:
/// with nothing new harvested, the next attempt waits for the periodic tick.
#[test]
fn a_failed_post_window_commit_is_not_retried_every_pass() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_chase_window(Duration::ZERO),
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release_c),
            },
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.shared.lock().unwrap().fail_commits = true;
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("the hint delivered", Duration::from_secs(5), || {
        h.shared.lock().unwrap().hinted_at.is_some()
    });
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    let failed = |h: &Harness| {
        h.shared
            .lock()
            .unwrap()
            .log
            .iter()
            .filter(|e| *e == "commit-failed")
            .count()
    };
    wait_for("the first commit attempt", Duration::from_secs(5), || {
        failed(&h) >= 1
    });
    let polls = h.shared.lock().unwrap().poll_timeouts.len();
    wait_for("forty more passes", Duration::from_secs(5), || {
        h.shared.lock().unwrap().poll_timeouts.len() >= polls + 40
    });
    let attempts = failed(&h);
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
    assert_eq!(attempts, 1, "commit attempts over forty passes");
}

/// An ack of a post-window partition that resolves while the controller waits
/// on the source is committed on that same pass.
#[test]
fn a_post_window_ack_resolved_during_the_poll_commits_on_that_pass() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_chase_window(Duration::ZERO),
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release_c),
            },
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    release.store(true, Ordering::Relaxed);
    {
        let mut script = h.script.lock().unwrap();
        script.push_back(Script::CommitReady(vec![PartitionId(0)]));
        script.push_back(Script::ReleaseHeld(Arc::clone(&held)));
    }
    wait_for(
        "the hinted partition committed",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10),
    );
    {
        let log = h.shared.lock().unwrap();
        let released_at = log.released_at.unwrap();
        assert_eq!(
            log.commit_polls.first().copied(),
            Some(released_at + 1),
            "acks resolved in poll {released_at}; commits after polls {:?}",
            log.commit_polls
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A hinted partition whose lane retires inside its window stands down, so
/// the controller returns to the event poll timeout.
#[test]
fn a_retired_hinted_partition_leaves_the_tightened_cadence() {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("partition 0 committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Retire(vec![LaneId(0)]));
    wait_for("retirement delivered", Duration::from_secs(5), || {
        logged(&h, "retire-delivered")
    });
    let after = h.shared.lock().unwrap().poll_timeouts.len();
    wait_for(
        "two polls after the retirement",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().poll_timeouts.len() > after + 2,
    );
    {
        let log = h.shared.lock().unwrap();
        assert!(
            !log.poll_timeouts[after..].contains(&super::controller::FAST_COMMIT_POLL),
            "{:?}",
            &log.poll_timeouts[after..]
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// An assignment that hands a hinted partition back inside its window ends
/// that partition's chase as well.
#[test]
fn a_new_assignment_inside_the_window_ends_the_chase() {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the first epoch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("partition 0 committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(10..20))),
        },
        LaneSpec {
            id: LaneId(2),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));
    wait_for("the second epoch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(1)]));
    wait_for("partition 1 committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(1)) == Some(&10)
    });
    {
        let log = h.shared.lock().unwrap();
        assert_eq!(
            log.committed.get(&PartitionId(0)),
            Some(&10),
            "{:?}",
            log.committed
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A hinted partition past its window leaves the tightened poll cadence while
/// its acks are still outstanding.
#[test]
fn a_chase_past_its_window_returns_to_the_event_poll_timeout() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_chase_window(Duration::ZERO),
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release_c),
            },
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("all payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("two polls after the hint", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        log.hinted_at
            .is_some_and(|at| log.poll_timeouts.len() > at + 2)
    });
    {
        let log = h.shared.lock().unwrap();
        let at = log.hinted_at.unwrap();
        assert!(
            !log.poll_timeouts[at + 1..].contains(&super::controller::FAST_COMMIT_POLL),
            "{:?}",
            &log.poll_timeouts[at..]
        );
    }
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wait_for(
        "the hinted partition committed",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10),
    );
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// An assignment that hands a hinted partition back in a new epoch ends that
/// partition's chase, so the new epoch's progress waits for its own hint or
/// the periodic tick.
#[test]
fn a_new_assignment_ends_the_chase_of_a_reused_partition() {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_chase_window(Duration::ZERO),
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::Ok,
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the first epoch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(0)]));
    wait_for("partition 0 committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(10..20))),
        },
        LaneSpec {
            id: LaneId(2),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));
    wait_for("the second epoch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![PartitionId(1)]));
    wait_for("partition 1 committed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(1)) == Some(&10)
    });
    {
        let log = h.shared.lock().unwrap();
        assert_eq!(
            log.committed.get(&PartitionId(0)),
            Some(&10),
            "{:?}",
            log.committed
        );
    }
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A fatal commit error fails the run.
#[test]
fn a_fatal_commit_fails_the_run() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    h.shared.lock().unwrap().fatal_commits = true;
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("a fatal commit must fail the run");
    };
    assert!(
        failure.reason.contains("commit failed fatally"),
        "{}",
        failure.reason
    );
}

/// The `spate_checkpoint_commits_total` value for one pipeline and outcome.
fn commits_counter(rendered: &str, pipeline_name: &str, outcome: &str) -> Option<f64> {
    rendered
        .lines()
        .filter(|l| l.starts_with(&format!("{}{{", names::CHECKPOINT_COMMITS_TOTAL)))
        .filter(|l| l.contains(&format!(r#"pipeline="{pipeline_name}""#)))
        .find(|l| l.contains(&format!(r#"outcome="{outcome}""#)))
        .and_then(|l| l.rsplit(' ').next())
        .map(|v| v.parse().expect("a numeric sample"))
}

#[test]
fn source_error_classification() {
    let retryable = SourceError::Client {
        class: ErrorClass::Retryable,
        reason: "hiccup".into(),
    };
    let fatal = SourceError::Client {
        class: ErrorClass::Fatal,
        reason: "broken".into(),
    };
    // Compile-time association test of the taxonomy used by the driver.
    assert!(matches!(
        retryable,
        SourceError::Client {
            class: ErrorClass::Retryable,
            ..
        }
    ));
    assert!(matches!(
        fatal,
        SourceError::Client {
            class: ErrorClass::Fatal,
            ..
        }
    ));
}

/// If the controller thread panics (here, a source whose `poll_events`
/// panics), it never runs the drain choreography, never tells the drivers to
/// stop, and never sets the shutdown flag. `run()` must detect the dead
/// controller, stop the drivers itself, and return `Failed` rather than
/// wedging on an untimed driver join.
#[test]
fn controller_panic_stops_drivers_and_fails_instead_of_hanging() {
    let h = start(|shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::BlockForever,
        batches_seen: 0,
    });
    {
        let mut script = h.script.lock().unwrap();
        // Assign a lane so the driver wedges on the never-unblocking batch,
        // then panic the next poll_events.
        script.push_back(Script::Assign(vec![LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        }]));
        script.push_back(Script::PanicPoll);
    }
    // Bounded wait: without the fix, main's untimed driver join hangs here.
    let deadline = Instant::now() + Duration::from_secs(30);
    while !h.join.is_finished() {
        assert!(
            Instant::now() < deadline,
            "run() hung after the controller thread panicked"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    let report = h.join.join().unwrap().unwrap();
    let ExitState::Failed(failure) = report.state else {
        panic!("a controller panic must fail the run");
    };
    assert_eq!(failure.component, "controller");
}

/// A caller-owned I/O runtime (`with_io_runtime`) must be the runtime `run()`
/// uses (tasks land on its worker threads) and must be shut down by `run()`
/// on exit like an internally built one.
#[test]
fn caller_owned_io_runtime_is_used_and_shut_down_by_run() {
    let io = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .thread_name("custom-io")
        .enable_all()
        .build()
        .expect("runtime");

    // A parked task holding a drop guard: only a runtime shutdown drops it.
    struct SetOnDrop(Arc<AtomicBool>);
    impl Drop for SetOnDrop {
        fn drop(&mut self) {
            self.0.store(true, Ordering::Relaxed);
        }
    }
    let runtime_shut_down = Arc::new(AtomicBool::new(false));
    let guard = SetOnDrop(Arc::clone(&runtime_shut_down));
    io.spawn(async move {
        let _guard = guard;
        std::future::pending::<()>().await;
    });

    // A sink drain that records which runtime executes spawned work.
    let (queues, receivers) = shard_queues(1, 8);
    let drain_worker: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
    let seen = Arc::clone(&drain_worker);
    let drain: super::SinkDrainFn = Box::new(move |_budget| {
        Box::pin(async move {
            let _receivers = receivers;
            let name = tokio::spawn(async { std::thread::current().name().map(String::from) })
                .await
                .expect("probe task");
            *seen.lock().unwrap() = name;
            DrainReport::default()
        })
    });
    let sink = SinkRuntime {
        queues: vec![queues],
        drain,
        probe: None,
        failures: SinkFailures::new(),
    };

    let (source, _shared, _script) = FakeSource::new();
    let chain_shared = Arc::new(ChainShared::default());
    let log = Arc::new(Mutex::new(SourceLog::default()));
    let cs = Arc::clone(&chain_shared);
    let runtime = PipelineRuntime::new(
        test_config(1),
        source,
        move |_thread| {
            Box::new(FakeChain {
                shared: Arc::clone(&cs),
                log: Arc::clone(&log),
                mode: ChainMode::Ok,
                batches_seen: 0,
            }) as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options())
    .with_io_runtime(io);
    let shutdown = runtime.shutdown_handle();
    let join = std::thread::spawn(move || runtime.run());
    shutdown.trigger();
    let report = join.join().unwrap().unwrap();

    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(
        drain_worker.lock().unwrap().as_deref(),
        Some("custom-io"),
        "drain-spawned work must run on the caller's runtime"
    );
    assert!(
        runtime_shut_down.load(Ordering::Relaxed),
        "run() must shut the caller-owned runtime down on exit"
    );
}

/// A caller-owned current-thread I/O runtime fails `run` with
/// `StartError::Config` before any pipeline thread starts. Regression for
/// #1155.
#[test]
fn current_thread_io_runtime_fails_startup() {
    let io = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime");
    let (queues, receivers) = shard_queues(1, 8);
    let drain: super::SinkDrainFn = Box::new(move |_budget| {
        Box::pin(async move {
            let _receivers = receivers;
            DrainReport::default()
        })
    });
    let sink = SinkRuntime {
        queues: vec![queues],
        drain,
        probe: None,
        failures: SinkFailures::new(),
    };

    let (source, shared, _script) = FakeSource::new();
    let chain_shared = Arc::new(ChainShared::default());
    let log = Arc::new(Mutex::new(SourceLog::default()));
    let chains_built = Arc::new(AtomicUsize::new(0));
    let chains_built_in = Arc::clone(&chains_built);
    let runtime = PipelineRuntime::new(
        test_config(1),
        source,
        move |_thread| {
            chains_built_in.fetch_add(1, Ordering::Relaxed);
            Box::new(FakeChain {
                shared: Arc::clone(&chain_shared),
                log: Arc::clone(&log),
                mode: ChainMode::Ok,
                batches_seen: 0,
            }) as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options())
    .with_io_runtime(io);
    // Without the check, a pre-triggered shutdown lets `run` drain and
    // return `Ok`, so the assertion below fails instead of hanging.
    runtime.shutdown_handle().trigger();
    let (result, rendered) = std::thread::spawn(move || {
        let mut result = None;
        let rendered = crate::metrics::render_local(|| result = Some(runtime.run()));
        (result.expect("run returned"), rendered)
    })
    .join()
    .unwrap();

    let Err(err @ StartError::Config(_)) = result else {
        panic!("expected StartError::Config, got {result:?}");
    };
    assert!(err.to_string().contains("multi-thread"), "{err}");
    assert!(!shared.lock().unwrap().opened, "the source must not open");
    assert_eq!(
        chains_built.load(Ordering::Relaxed),
        0,
        "no pipeline thread may start"
    );
    assert!(
        !rendered.contains(names::PIPELINE_STATE),
        "a refused runtime must register no series:\n{rendered}"
    );
}

/// The source drops while the io runtime still runs, so work its `Drop`
/// spawns there completes. Regression for #854.
#[test]
fn the_source_drops_before_the_io_runtime_stops() {
    struct NeedsIoOnDrop {
        inner: FakeSource,
        io: tokio::runtime::Handle,
        ran: Arc<AtomicBool>,
    }
    impl Source for NeedsIoOnDrop {
        type Lane = FakeLane;
        fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError> {
            self.inner.open(ctx)
        }
        fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<FakeLane>, SourceError> {
            self.inner.poll_events(timeout)
        }
        fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError> {
            self.inner.commit(watermarks)
        }
    }
    impl Drop for NeedsIoOnDrop {
        fn drop(&mut self) {
            let (tx, rx) = mpsc::channel();
            // Outlives the moment main would stop the runtime if it did not
            // wait for this thread. A stopped runtime drops the task, and
            // with it the sender.
            self.io.spawn(async move {
                tokio::time::sleep(Duration::from_millis(300)).await;
                let _ = tx.send(());
            });
            let ran = rx.recv_timeout(Duration::from_secs(10)).is_ok();
            self.ran.store(ran, Ordering::Relaxed);
        }
    }

    let io = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .expect("runtime");
    let ran = Arc::new(AtomicBool::new(false));
    let (inner, _shared, _script) = FakeSource::new();
    let source = NeedsIoOnDrop {
        inner,
        io: io.handle().clone(),
        ran: Arc::clone(&ran),
    };
    let (queues, receivers) = shard_queues(1, 8);
    let drain: super::SinkDrainFn = Box::new(move |_budget| {
        Box::pin(async move {
            let _receivers = receivers;
            DrainReport::default()
        })
    });
    let sink = SinkRuntime {
        queues: vec![queues],
        drain,
        probe: None,
        failures: SinkFailures::new(),
    };
    let chain_shared = Arc::new(ChainShared::default());
    let log = Arc::new(Mutex::new(SourceLog::default()));
    let runtime = PipelineRuntime::new(
        test_config(1),
        source,
        move |_thread| {
            Box::new(FakeChain {
                shared: Arc::clone(&chain_shared),
                log: Arc::clone(&log),
                mode: ChainMode::Ok,
                batches_seen: 0,
            }) as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options())
    .with_io_runtime(io);
    let shutdown = runtime.shutdown_handle();
    let join = std::thread::spawn(move || runtime.run());
    shutdown.trigger();
    let report = join.join().unwrap().unwrap();

    assert_eq!(report.state, ExitState::Completed);
    assert!(
        ran.load(Ordering::Relaxed),
        "work the source's Drop spawns on the io runtime must run"
    );
}

/// A fallible startup step after the driver threads are spawned (here, the
/// admin bind, forced to fail by occupying its port) must stop the drivers
/// and return the error promptly, never leaking the running pinned threads.
#[test]
fn startup_error_after_driver_spawn_stops_drivers_and_returns_err() {
    let occupied = std::net::TcpListener::bind("127.0.0.1:0").expect("bind probe port");
    let addr = occupied.local_addr().unwrap();
    let mut cfg = test_config(1);
    cfg.admin.listen = Some(addr);
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    // Prompt return proves the driver join completed (threads did not leak).
    let deadline = Instant::now() + Duration::from_secs(10);
    while !h.join.is_finished() {
        assert!(
            Instant::now() < deadline,
            "run() did not return promptly on a bind failure"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    let result = h.join.join().unwrap();
    let Err(err @ StartError::AdminBind { .. }) = result else {
        panic!("expected StartError::AdminBind, got {result:?}");
    };
    // A bare I/O error leaves the reader guessing which of a config's
    // addresses the kernel refused.
    assert!(
        err.to_string().contains(&addr.to_string()),
        "the error must name the address it could not take: {err}"
    );
    drop(occupied);
}

/// The text of the `StartError::Config` a run returns before its source
/// opens. Waits for the run under a deadline, so a run that opens the source
/// and keeps going fails the test there.
fn config_error_before_open(h: Harness) -> String {
    wait_for("run() to return", Duration::from_secs(10), || {
        h.join.is_finished()
    });
    let result = h.join.join().unwrap();
    let Err(err @ StartError::Config(_)) = result else {
        panic!("expected StartError::Config, got {result:?}");
    };
    assert!(!h.shared.lock().unwrap().opened, "the source must not open");
    err.to_string()
}

fn ok_chain(shared: Arc<ChainShared>, log: Arc<Mutex<SourceLog>>) -> FakeChain {
    FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    }
}

fn feed_section(yaml: &str) -> Option<crate::config::FeedSection> {
    Some(serde_yaml::from_str(yaml).expect("section"))
}

/// Pins that the runtime hands a `feed:` section to the source, whose default
/// `configure_feed` refuses it before the source opens.
#[test]
fn a_source_without_a_feed_rejects_the_feed_section() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(err.contains("reads no feed"), "{err}");
}

/// Pins that `run` refuses a `location` on the in-process feed it builds,
/// before the source is offered the section.
#[test]
fn a_memory_feed_with_a_location_is_refused() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}\nlocation: s3://ops/jobs/backfill-1/");
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(
        err.contains("`feed.location`") && err.contains("`memory`"),
        "{err}"
    );
    assert!(
        !err.contains("s3://ops"),
        "the location must not print: {err}"
    );
}

/// Pins the defaults `spate-core` owns for the `feed:` section: no key set
/// leaves the source's own default in force, a bounded job settles for five
/// minutes, the in-process feed dead-letters a unit on its tenth failed
/// delivery, and a stop reserves five seconds of the 25 s drain. The SQS,
/// Kafka and S3 defaults are pinned in their crates.
#[test]
fn the_feed_section_defaults() {
    use crate::feed::Feed;

    for yaml in ["memory: {}", "sqs: {}"] {
        let section = feed_section(yaml).expect("section");
        assert_eq!(section.units_held, None, "{yaml}");
        assert_eq!(section.attempt_cap, None, "{yaml}");
        assert_eq!(section.location, None, "{yaml}");
        assert_eq!(section.settling_window, None, "{yaml}");
        assert_eq!(
            section.settling_window_or_default(),
            Duration::from_secs(300),
            "{yaml}"
        );
    }

    let section = feed_section("memory: {}").expect("section");
    let feed = MemoryFeed::from_section(&section).expect("feed");
    drop(feed.publish(crate::feed::Unit::new("poison", b"poison".to_vec())));
    // The cap is written out: a loop bound read from the constant would
    // follow a changed cap.
    for attempt in 1..=10 {
        let delivery = feed.receive(usize::MAX).unwrap().remove(0);
        assert_eq!(delivery.deliveries, attempt);
        feed.fail(delivery.receipt, "unreadable").unwrap();
        let counts = feed.counts();
        let dead = u64::from(attempt == 10);
        assert_eq!(counts.dead_lettered, dead, "after attempt {attempt}");
        assert_eq!(counts.visible, 1 - dead, "after attempt {attempt}");
    }

    let cfg = PipelineConfig::from_str(
        "pipeline: { name: demo }\nsource: { memory: {} }\nsink: { memory: {} }\n",
    )
    .expect("config");
    assert_eq!(cfg.checkpoint.drain_timeout, Duration::from_secs(25));
    assert_eq!(crate::feed::STOP_RESERVE, Duration::from_secs(5));
}

/// Pins that `run` refuses a `location` on an in-process feed passed in
/// through `Pipeline::feed`, which skips building the feed from the section.
#[test]
fn a_memory_feed_passed_in_with_a_location_is_refused() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}\nlocation: s3://ops/jobs/backfill-1/");
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_feed(Arc::new(MemoryFeed::new())),
        ok_chain,
    );
    let err = config_error_before_open(h);
    assert!(
        err.contains("`feed.location`") && err.contains("`memory`"),
        "{err}"
    );
}

/// Pins that `run` refuses a `settling_window` on an in-process feed passed
/// in through `Pipeline::feed`, which skips building the feed from the
/// section.
#[test]
fn a_memory_feed_passed_in_with_a_settling_window_is_refused() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}\nsettling_window: 1m");
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_feed(Arc::new(MemoryFeed::new())),
        ok_chain,
    );
    let err = config_error_before_open(h);
    assert!(
        err.contains("`feed.settling_window`") && err.contains("`memory`"),
        "{err}"
    );
}

/// Pins that `run` refuses a settling window below the floor set on a
/// section built in code, which skips the load-time check.
#[test]
fn a_zero_settling_window_passed_in_is_refused() {
    let mut cfg = test_config(1);
    let mut section = feed_section("sqs: {}\nlocation: s3://ops/jobs/backfill-1/").unwrap();
    section.settling_window = Some(Duration::ZERO);
    cfg.feed = Some(section);
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(err.contains("`feed.settling_window`"), "{err}");
}

/// Pins that `Source::configure_feed`'s default refuses the section, naming
/// the source, so a source that reads no feed fails at startup.
#[test]
fn the_default_configure_feed_refuses_the_section() {
    struct NoFeed(FakeSource);
    impl Source for NoFeed {
        type Lane = FakeLane;
        fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError> {
            self.0.open(ctx)
        }
        fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<FakeLane>, SourceError> {
            self.0.poll_events(timeout)
        }
        fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError> {
            self.0.commit(watermarks)
        }
    }
    let (inner, _shared, _script) = FakeSource::new();
    let mut source = NoFeed(inner);
    let section = feed_section("memory: {}").unwrap();
    let err = source
        .configure_feed(
            &section,
            Arc::new(MemoryFeed::new()),
            crate::feed::FeedOrigin::Section,
        )
        .unwrap_err()
        .to_string();
    assert!(err.contains("reads no feed"), "{err}");
}

/// Pins the origin `run` hands `configure_feed`: `Section` for the in-process
/// feed it builds from the section, `Given` for a feed passed to
/// `Pipeline::feed`.
#[test]
fn the_source_learns_where_its_feed_came_from() {
    use crate::feed::FeedOrigin;
    for (given, origin) in [(false, FeedOrigin::Section), (true, FeedOrigin::Given)] {
        let mut cfg = test_config(1);
        cfg.feed = feed_section("memory: {}");
        let (source, shared, script) = FakeSource::new();
        shared.lock().unwrap().accept_feed = true;
        script.lock().unwrap().push_back(Script::Drained);
        let chain_shared = Arc::new(ChainShared::default());
        let (sink, _drained) = test_sink();
        let log = Arc::clone(&shared);
        let mut runtime = PipelineRuntime::new(
            cfg,
            source,
            move |_thread| {
                Box::new(ok_chain(Arc::clone(&chain_shared), Arc::clone(&log)))
                    as Box<dyn RunnableChain>
            },
            sink,
            Arc::new(crate::backpressure::InflightBudget::new()),
        )
        .with_options(test_options());
        if given {
            runtime = runtime.with_feed(Arc::new(MemoryFeed::new()));
        }
        let report = runtime.run().expect("the source accepts the section");
        assert_eq!(report.state, ExitState::Completed);
        assert_eq!(
            shared.lock().unwrap().feed_origin,
            Some(origin),
            "given: {given}"
        );
    }
}

/// Pins that the source reads through a feed the runtime counts, under the
/// `feed` component and the backend's key, and times on the injected clock.
#[test]
fn the_runtime_counts_the_calls_its_source_makes_through_the_feed() {
    use crate::feed::{Feed, Unit};
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    cfg.feed = feed_section("memory: {}");
    let pipeline_name = cfg.pipeline.name.clone();
    let settings = crate::pipeline::metrics_settings(&cfg);

    // One unit dead-lettered and one waiting, through calls the runtime does
    // not see.
    let backend = Arc::new(MemoryFeed::with_attempt_cap(std::num::NonZeroU32::MIN));
    drop(backend.publish(Unit::new("poison", Vec::new())));
    let poison = backend.receive(1).unwrap().remove(0);
    backend.fail(poison.receipt, "unreadable").unwrap();
    drop(backend.publish(Unit::new("work", Vec::new())));

    let given = Arc::clone(&backend);
    let clock = crate::clock::TestClock::frozen();
    let injected = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        test_options(),
        |log| log.accept_feed = true,
        move |runtime, _| runtime.with_feed(given).with_clock(injected),
        ok_chain,
    );
    wait_for("the source's feed", Duration::from_secs(5), || {
        h.shared.lock().unwrap().feed.is_some()
    });
    // The runtime's own install, taken after it: in a process of its own, as
    // nextest runs it, a feed wrapped before the recorder existed would render
    // nothing here.
    let handle = crate::metrics::install(&settings).expect("the runtime's exporter");
    let feed = h
        .shared
        .lock()
        .unwrap()
        .feed
        .clone()
        .expect("the source's feed");
    let work = feed.receive(8).unwrap().remove(0);
    feed.acknowledge(work.receipt).unwrap();
    let mut published = feed.publish(Unit::new("more", Vec::new()));
    clock.advance(Duration::from_secs(3));
    published.try_outcome().expect("settled").unwrap();

    let series = |name: &str| {
        format!(r#"{name}{{pipeline="{pipeline_name}",component="feed",component_type="memory"}} "#)
    };
    let reads = |name: &str, value: &str| {
        let line = format!("{}{value}", series(name));
        handle.render().lines().any(|l| l == line)
    };
    wait_for("the dead-letter depth", Duration::from_secs(5), || {
        reads("spate_feed_dead_letter_units", "1")
    });
    assert!(reads("spate_feed_units_received_total", "1"));
    assert!(reads("spate_feed_units_acknowledged_total", "1"));
    assert!(reads("spate_feed_held_units", "0"));
    assert!(reads("spate_feed_publish_duration_seconds_sum", "3"));

    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// The in-process feed, counting the dead-letter reads made on it and failing
/// each one when `fails` is set.
struct DeadLetterReads {
    inner: MemoryFeed,
    reads: Arc<AtomicUsize>,
    fails: bool,
}

impl crate::feed::Feed for DeadLetterReads {
    fn backend(&self) -> &str {
        self.inner.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 read_depth(&self, of: crate::feed::DepthOf) -> Option<crate::feed::DepthReading> {
        if of == crate::feed::DepthOf::DeadLetters {
            self.reads.fetch_add(1, Ordering::Relaxed);
            if self.fails {
                let (reporter, reading) = crate::feed::depth_reading();
                reporter.report(Err(crate::feed::FeedError::new(
                    crate::feed::FeedErrorKind::Retryable,
                    "throttled",
                )));
                return Some(reading);
            }
        }
        self.inner.read_depth(of)
    }
}

/// Pins that the runtime reads the feed's dead-letter depth once at startup
/// by default, and makes no such read when `dead_letter_read_interval` is
/// `None`.
#[test]
fn the_runtime_reads_dead_letters_only_on_its_interval() {
    for (interval, expected) in [(None, 0), (Some(Duration::from_secs(60)), 1)] {
        let mut cfg = test_config(1);
        cfg.feed = feed_section("memory: {}");
        let reads = Arc::new(AtomicUsize::new(0));
        let feed = DeadLetterReads {
            inner: MemoryFeed::new(),
            reads: Arc::clone(&reads),
            fails: false,
        };
        let options = RuntimeOptions {
            dead_letter_read_interval: interval,
            ..test_options()
        };
        let h = start_prepared(
            cfg,
            options,
            |log| log.accept_feed = true,
            move |runtime, _| runtime.with_feed(Arc::new(feed)),
            ok_chain,
        );
        h.script.lock().unwrap().push_back(Script::Drained);
        let report = h.join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed);
        assert_eq!(reads.load(Ordering::Relaxed), expected, "{interval:?}");
    }
}

/// Pins that the runtime times its dead-letter reads on the injected clock:
/// one at startup, and the next once that clock has moved an interval.
#[test]
fn the_runtime_reads_dead_letters_on_its_clock() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let reads = Arc::new(AtomicUsize::new(0));
    let feed = DeadLetterReads {
        inner: MemoryFeed::new(),
        reads: Arc::clone(&reads),
        fails: false,
    };
    let clock = crate::clock::TestClock::frozen();
    let injected = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        test_options(),
        |log| log.accept_feed = true,
        move |runtime, _| runtime.with_feed(Arc::new(feed)).with_clock(injected),
        ok_chain,
    );
    wait_for("the startup read", Duration::from_secs(5), || {
        reads.load(Ordering::Relaxed) == 1
    });
    clock.advance(crate::feed::DEAD_LETTER_READ_INTERVAL);
    wait_for("the second read", Duration::from_secs(5), || {
        reads.load(Ordering::Relaxed) == 2
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
}

/// Pins that a dead-letter read that fails is logged as a warning naming the
/// feed.
#[test]
fn a_failed_dead_letter_read_warns() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let feed = DeadLetterReads {
        inner: MemoryFeed::new(),
        reads: Arc::default(),
        fails: true,
    };
    let h = start_prepared(
        cfg,
        test_options(),
        |log| log.accept_feed = true,
        move |runtime, _| runtime.with_feed(Arc::new(feed)),
        ok_chain,
    );
    let warned = || {
        h.warnings.0.lock().unwrap().iter().any(|fields| {
            fields
                .iter()
                .any(|(name, value)| name == "message" && value.contains("dead-letter destination"))
                && fields
                    .iter()
                    .any(|(name, value)| name == "feed" && value.contains("memory"))
        })
    };
    wait_for("the warning", Duration::from_secs(5), warned);
    h.script.lock().unwrap().push_back(Script::Drained);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
}

/// The in-process feed under `key`, passing the metrics context it is handed to
/// `attach`.
// Redundant for the compiler. cargo-mutants skips an item under exactly
// `cfg(test)`, and these tests never call most of the forwarding methods.
#[cfg(test)]
fn keyed_feed(
    key: &'static str,
    attach: impl Fn(crate::feed::FeedMetricsCtx) + Send + Sync + 'static,
) -> Arc<dyn crate::feed::Feed> {
    use crate::feed::{
        Confirmation, Delivery, Feed, FeedError, FeedMetricsCtx, FeedWaker, Receipt, Unit,
    };

    struct Keyed<A> {
        inner: MemoryFeed,
        key: &'static str,
        attach: A,
    }

    impl<A: Fn(FeedMetricsCtx) + Send + Sync> Feed for Keyed<A> {
        fn backend(&self) -> &str {
            self.key
        }
        fn set_waker(&self, waker: FeedWaker) {
            self.inner.set_waker(waker);
        }
        fn attach_metrics(&self, cx: FeedMetricsCtx) {
            (self.attach)(cx);
        }
        fn receive(&self, max: usize) -> Result<Vec<Delivery>, FeedError> {
            self.inner.receive(max)
        }
        fn renew(&self, receipt: Receipt) -> Result<(), FeedError> {
            self.inner.renew(receipt)
        }
        fn acknowledge(&self, receipt: Receipt) -> Result<(), FeedError> {
            self.inner.acknowledge(receipt)
        }
        fn fail(&self, receipt: Receipt, reason: &str) -> Result<(), FeedError> {
            self.inner.fail(receipt, reason)
        }
        fn publish(&self, unit: Unit) -> Confirmation {
            self.inner.publish(unit)
        }
    }

    Arc::new(Keyed {
        inner: MemoryFeed::new(),
        key,
        attach,
    })
}

/// Pins that the runtime hands a feed backend one `Meter` scoped
/// `spate_<backend>_feed_*`, carrying the feed's labels, once the recorder is
/// installed and before the source is handed the feed.
#[test]
fn a_feed_backend_receives_a_meter_scoped_to_the_feed_role() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    cfg.feed = feed_section("probe: {}");
    let pipeline_name = cfg.pipeline.name.clone();
    let settings = crate::pipeline::metrics_settings(&cfg);

    let source_had_feed = Arc::new(Mutex::new(None));
    let noted = Arc::clone(&source_had_feed);
    let h = start_prepared(
        cfg,
        test_options(),
        |log| log.accept_feed = true,
        move |runtime, source| {
            let source = Arc::clone(source);
            runtime.with_feed(keyed_feed("probe", move |cx| {
                *noted.lock().unwrap() = Some(source.lock().unwrap().feed.is_some());
                cx.meter
                    .expect("`probe` can scope a family")
                    .counter("attached_total", &[])
                    .increment(1);
            }))
        },
        ok_chain,
    );
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(
        *source_had_feed.lock().unwrap(),
        Some(false),
        "the backend's `Meter` arrives before the source is handed the feed"
    );

    // The runtime's own install, taken after it: in a process of its own, as
    // nextest runs it, a `Meter` handed over before the recorder existed would
    // render nothing here.
    let handle = crate::metrics::install(&settings).expect("the runtime's exporter");
    let series = format!(
        r#"spate_probe_feed_attached_total{{pipeline="{pipeline_name}",component="feed",component_type="probe"}} 1"#
    );
    let rendered = handle.render();
    assert!(rendered.lines().any(|l| l == series), "{rendered}");
}

/// Pins that the runtime hands a feed backend its pipeline's
/// `metrics.per_partition_detail`, once per pipeline that runs it.
#[test]
fn a_feed_backend_receives_the_pipelines_per_partition_detail() {
    for detail in [false, true] {
        let mut cfg = test_config(1);
        cfg.metrics.per_partition_detail = detail;
        cfg.feed = feed_section("probe: {}");
        let seen = Arc::new(Mutex::new(Vec::new()));
        let noted = Arc::clone(&seen);
        let h = start_prepared(
            cfg,
            test_options(),
            |log| log.accept_feed = true,
            move |runtime, _| {
                runtime.with_feed(keyed_feed("probe", move |cx| {
                    noted.lock().unwrap().push(cx.per_partition_detail);
                }))
            },
            ok_chain,
        );
        h.script.lock().unwrap().push_back(Script::Drained);
        let report = h.join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed, "{detail}");
        assert_eq!(*seen.lock().unwrap(), [detail], "{detail}");
    }
}

/// Pins that a backend whose key cannot scope a family receives no `Meter` and
/// the job still runs: a reserved root and `custom` silently, a malformed key
/// with a warning naming it. The per-partition setting arrives all the same.
#[test]
fn a_feed_backend_whose_key_cannot_scope_a_family_receives_no_meter() {
    for (key, warns) in [("feed", false), ("custom", false), ("not-a-segment", true)] {
        let mut cfg = test_config(1);
        cfg.metrics.per_partition_detail = true;
        cfg.feed = feed_section(&format!("{key}: {{}}"));
        let attached = Arc::new(Mutex::new(Vec::new()));
        let noted = Arc::clone(&attached);
        let h = start_prepared(
            cfg,
            test_options(),
            |log| log.accept_feed = true,
            move |runtime, _| {
                runtime.with_feed(keyed_feed(key, move |cx| {
                    noted
                        .lock()
                        .unwrap()
                        .push((cx.meter.is_some(), cx.per_partition_detail));
                }))
            },
            ok_chain,
        );
        h.script.lock().unwrap().push_back(Script::Drained);
        let report = h.join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed, "{key}");
        assert_eq!(*attached.lock().unwrap(), [(false, true)], "{key}");
        let warned = h.warnings.0.lock().unwrap().iter().any(|fields| {
            fields
                .iter()
                .any(|(name, value)| name == "role" && value.contains("feed"))
                && fields
                    .iter()
                    .any(|(name, value)| name == "component_type" && value.contains(key))
        });
        assert_eq!(warned, warns, "{key}");
    }
}

/// Pins that `run` passes a `location` beside a backend other than the
/// in-process feed on to the feed checks that follow.
#[test]
fn a_location_beside_a_durable_backend_is_not_refused() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("sqs: {}\nlocation: s3://ops/jobs/backfill-1/");
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(err.contains("pass it to `Pipeline::feed`"), "{err}");
}

/// Pins that a section naming a backend other than `memory`, with no feed
/// passed in, fails startup with the instruction to build it in code.
#[test]
fn a_backend_built_in_code_without_a_feed_fails_startup() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("sqs: {}");
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(err.contains("pass it to `Pipeline::feed`"), "{err}");
}

/// Pins that a feed passed in with no `feed:` section fails startup.
#[test]
fn a_feed_given_without_a_feed_section_fails_startup() {
    let h = start_with(
        test_config(1),
        test_options(),
        |runtime| runtime.with_feed(Arc::new(MemoryFeed::new())),
        ok_chain,
    );
    let err = config_error_before_open(h);
    assert!(err.contains("no `feed:` section"), "{err}");
}

/// Pins that a feed passed in whose backend differs from the section's key
/// fails startup, naming both.
#[test]
fn a_feed_whose_backend_differs_from_the_section_fails_startup() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("sqs: {}");
    let h = start_with(
        cfg,
        test_options(),
        |runtime| runtime.with_feed(Arc::new(MemoryFeed::new())),
        ok_chain,
    );
    let err = config_error_before_open(h);
    assert!(err.contains("`memory`") && err.contains("`sqs`"), "{err}");
}

/// Pins that the runtime builds the in-process feed from the section, so a
/// key in the `memory` body fails startup naming its path.
#[test]
fn a_key_in_the_memory_body_fails_startup() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: { depth: 4 }");
    let h = start_with_config(cfg, ok_chain);
    let err = config_error_before_open(h);
    assert!(
        err.contains("feed.memory") && err.contains("depth"),
        "{err}"
    );
}

/// Pins that the runtime offers the `feed:` section before any pipeline
/// thread is built, so a refused section starts no driver.
#[test]
fn a_refused_feed_section_builds_no_pipeline_thread() {
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let built = Arc::new(AtomicUsize::new(0));
    let counter = Arc::clone(&built);
    let h = start_with_config(cfg, move |shared, log| {
        counter.fetch_add(1, Ordering::SeqCst);
        ok_chain(shared, log)
    });
    let err = config_error_before_open(h);
    assert!(err.contains("reads no feed"), "{err}");
    assert_eq!(
        built.load(Ordering::SeqCst),
        0,
        "no chain may be built before the section is accepted"
    );
}

/// A second registered dispatcher. With one live dispatcher, tracing-core
/// caches a callsite's interest from whichever thread reaches it first, and
/// other tests reach the feed arm on threads with no subscriber.
static BYSTANDER: std::sync::LazyLock<tracing::Dispatch> = std::sync::LazyLock::new(|| {
    use tracing_subscriber::layer::SubscriberExt;
    tracing::Dispatch::new(
        tracing_subscriber::registry().with(tracing_subscriber::filter::LevelFilter::OFF),
    )
});

/// Pins that a `feed: { memory: {} }` section the source refuses fails
/// startup with the error alone and logs no WARN.
#[test]
fn a_refused_feed_section_fails_startup_without_a_warning() {
    use tracing_subscriber::layer::SubscriberExt;
    std::sync::LazyLock::force(&BYSTANDER);
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let (source, shared, _script) = FakeSource::new();
    let chain_shared = Arc::new(ChainShared::default());
    let (sink, _drained) = test_sink();
    let log = Arc::clone(&shared);
    let runtime = PipelineRuntime::new(
        cfg,
        source,
        move |_thread| {
            Box::new(ok_chain(Arc::clone(&chain_shared), Arc::clone(&log)))
                as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options());
    let warnings = Warnings::default();
    let subscriber = tracing_subscriber::registry().with(warnings.clone());
    let result = tracing::subscriber::with_default(subscriber, || runtime.run());
    let Err(err @ StartError::Config(_)) = result else {
        panic!("expected StartError::Config, got {result:?}");
    };
    assert!(err.to_string().contains("reads no feed"), "{err}");
    assert!(!shared.lock().unwrap().opened, "the source must not open");
    let warnings = warnings.0.lock().unwrap().clone();
    assert!(warnings.is_empty(), "{warnings:#?}");
}

/// Pins that a `feed: { memory: {} }` section the source accepts starts and
/// completes the run with no WARN from the runtime.
#[test]
fn an_accepted_in_process_feed_section_starts_without_a_warning() {
    use tracing_subscriber::layer::SubscriberExt;
    std::sync::LazyLock::force(&BYSTANDER);
    let mut cfg = test_config(1);
    cfg.feed = feed_section("memory: {}");
    let (source, shared, script) = FakeSource::new();
    shared.lock().unwrap().accept_feed = true;
    script.lock().unwrap().push_back(Script::Drained);
    let chain_shared = Arc::new(ChainShared::default());
    let (sink, _drained) = test_sink();
    let log = Arc::clone(&shared);
    let runtime = PipelineRuntime::new(
        cfg,
        source,
        move |_thread| {
            Box::new(ok_chain(Arc::clone(&chain_shared), Arc::clone(&log)))
                as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options());
    let warnings = Warnings::default();
    let subscriber = tracing_subscriber::registry().with(warnings.clone());
    let report = tracing::subscriber::with_default(subscriber, || runtime.run())
        .expect("the source accepts the section");
    assert_eq!(report.state, ExitState::Completed);
    assert!(shared.lock().unwrap().opened, "the source opens");
    let warnings = warnings.0.lock().unwrap().clone();
    assert!(warnings.is_empty(), "{warnings:#?}");
}

/// `admin.listen: none` runs and drains with no admin server. With no
/// listener there is no stop channel to signal, and the run has to reach its
/// drain and its controller join anyway.
#[test]
fn no_admin_listener_still_drains_and_completes() {
    let mut cfg = test_config(1);
    cfg.admin.listen = None;
    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// The controller's `SourceMetrics`, not one of the per-thread instances,
/// must own the source series.
///
/// A pipeline builds one handle set per driver thread plus the controller's,
/// all on identical labels, because every thread counts the records it polled.
/// Only the controller's can publish the gauges. It holds the assignment, and
/// its clone reaches the source, which is the only thing that can measure lag.
/// If a driver's instance claimed the series first, the controller's would be
/// a shadow and `spate_source_lag_records` would go silent again, as it did
/// for fourteen days.
///
/// `FakeSource` publishes `FAKE_SOURCE_LAG` for partition 0 at `open`, through
/// whatever handle set it was given. A per-partition gauge registers on its
/// first published value, so a shadowed controller leaves the series *absent*.
/// Flip `SourceMetrics::shadow` to `try_new` in `runtime.rs` and this test
/// fails on a missing series, not on a wrong number.
#[test]
fn the_controller_owns_the_source_series_not_the_per_thread_shadows() {
    let mut cfg = test_config(4); // four drivers → four shadows, one owner
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    // Installed before the pipeline builds. Handles bind to the recorder
    // present at construction. Idempotent, so the runtime's own install
    // reuses this one and renders through this handle.
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);

    let rendered = handle.render();
    let needle = format!(
        r#"spate_source_lag_records{{pipeline="{pipeline_name}",component="source",component_type="source",partition="0"}} {FAKE_SOURCE_LAG}"#
    );
    assert!(
        rendered.contains(&needle),
        "the source's lag must reach the exposition — the handle set the \
         source was given owns the series:\n{rendered}"
    );
}

/// The in-flight budget's usage reaches the exposition as exactly one series.
///
/// Regression for #332.
///
/// This asserts the number of series as well as the value. The budget is one
/// counter per pipeline, so publishing it from each driver would give one
/// series per thread, all carrying the same number, and `sum()` over the
/// family would report the thread count times the usage.
#[test]
fn the_inflight_budget_reaches_the_exposition_as_one_series() {
    let mut cfg = test_config(4); // four drivers, so a per-driver regression shows up in the count
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    // Installed before the pipeline builds. Handles bind to the recorder
    // present at construction, and `install` is idempotent, so the runtime
    // reuses this one and renders through this handle.
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });

    // The controller samples the budget once per loop pass.
    const CHARGED: usize = 7 << 20;
    h.budget.add(CHARGED);

    let series = |rendered: &str| -> Vec<String> {
        rendered
            .lines()
            .filter(|l| l.starts_with(&format!("{}{{", names::BACKPRESSURE_INFLIGHT_BYTES)))
            // Under `cargo test` other tests' pipelines record into this process-wide recorder.
            .filter(|l| l.contains(&format!(r#"pipeline="{pipeline_name}""#)))
            .map(str::to_owned)
            .collect()
    };

    wait_for(
        "the budget gauge is published",
        Duration::from_secs(5),
        || {
            series(&handle.render())
                .iter()
                .any(|l| l.ends_with(&format!(" {CHARGED}")))
        },
    );

    let rendered = handle.render();
    let published = series(&rendered);
    assert_eq!(
        published,
        vec![format!(
            r#"{}{{pipeline="{pipeline_name}",component="runtime",component_type="pipeline"}} {CHARGED}"#,
            names::BACKPRESSURE_INFLIGHT_BYTES
        )],
        "the budget publishes exactly one series, under the runtime's labels:\n{rendered}"
    );

    // A drain is visible too, so the series tracks the budget.
    h.budget.sub(CHARGED);
    wait_for("the budget gauge drains", Duration::from_secs(5), || {
        series(&handle.render()).iter().any(|l| l.ends_with(" 0"))
    });

    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// The value of one pipeline's `spate_pipeline_io_busy_ratio` line, if the
/// exposition carries it.
fn io_busy_ratio(rendered: &str, pipeline_name: &str) -> Option<f64> {
    let prefix = format!(
        r#"{}{{pipeline="{pipeline_name}",component="runtime",component_type="pipeline"}} "#,
        names::PIPELINE_IO_BUSY_RATIO
    );
    rendered
        .lines()
        .find_map(|l| l.strip_prefix(&prefix))
        .and_then(|v| v.parse().ok())
}

fn install_prometheus(cfg: &PipelineConfig) -> crate::metrics::MetricsHandle {
    // Installed before the pipeline builds. Handles bind to the recorder
    // present at construction, and `install` is idempotent, so the runtime
    // reuses this one and renders through this handle.
    crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter")
}

/// The I/O busy gauge is in the exposition, at 0, before its first sample.
#[test]
fn the_io_busy_ratio_is_published_from_start() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = install_prometheus(&cfg);

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    wait_for(
        "the busy gauge is published",
        Duration::from_secs(5),
        || io_busy_ratio(&handle.render(), &pipeline_name) == Some(0.0),
    );

    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// A caller-owned I/O runtime kept busy by a task of its own reads as busy
/// in the exposition, sampled by the running pipeline.
#[test]
fn a_busy_io_runtime_reaches_the_exposition() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = install_prometheus(&cfg);

    let io = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .expect("runtime");
    let stop = Arc::new(AtomicBool::new(false));
    let spin = Arc::clone(&stop);
    io.spawn(async move {
        // CPU work inside a poll is what tokio counts as busy.
        while !spin.load(Ordering::Relaxed) {
            let polled = Instant::now();
            while polled.elapsed() < Duration::from_millis(10) {
                std::hint::spin_loop();
            }
            tokio::task::yield_now().await;
        }
    });

    let h = start_with(
        cfg,
        test_options(),
        |runtime| {
            runtime
                .with_io_runtime(io)
                .with_io_busy_sampling(Duration::from_millis(50), Duration::from_secs(1))
        },
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::Ok,
            batches_seen: 0,
        },
    );
    wait_for(
        "a busy sample is published",
        Duration::from_secs(10),
        || io_busy_ratio(&handle.render(), &pipeline_name).is_some_and(|r| r > 0.5),
    );

    stop.store(true, Ordering::Relaxed);
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// A running pipeline divides by its I/O runtime's worker count: one task
/// holding one of two workers reads about half in the exposition.
#[test]
fn one_busy_worker_of_two_reads_half_in_the_exposition() {
    const WINDOW: Duration = Duration::from_secs(2);

    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = install_prometheus(&cfg);

    let io = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .expect("runtime");
    let stop = Arc::new(AtomicBool::new(false));
    let spin = Arc::clone(&stop);
    io.spawn(async move {
        while !spin.load(Ordering::Relaxed) {
            let polled = Instant::now();
            while polled.elapsed() < Duration::from_millis(10) {
                std::hint::spin_loop();
            }
            tokio::task::yield_now().await;
        }
    });

    let h = start_with(
        cfg,
        test_options(),
        |runtime| {
            runtime
                .with_io_runtime(io)
                .with_io_busy_sampling(Duration::from_millis(50), WINDOW)
        },
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::Ok,
            batches_seen: 0,
        },
    );
    // A reading counted against one worker sits near 1 once the window
    // fills; against two it stays near half for a whole window. The hold is
    // timed because one render of a shared process's exposition can take
    // tens of milliseconds, so a count of polls has no fixed duration.
    let in_band_since = std::cell::Cell::new(None::<Instant>);
    wait_for(
        "the gauge to hold near half",
        Duration::from_secs(15),
        || {
            let half = io_busy_ratio(&handle.render(), &pipeline_name)
                .is_some_and(|r| (0.3..0.7).contains(&r));
            in_band_since.set(half.then(|| in_band_since.get().unwrap_or_else(Instant::now)));
            in_band_since
                .get()
                .is_some_and(|since| since.elapsed() >= WINDOW)
        },
    );

    stop.store(true, Ordering::Relaxed);
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// One event's fields, as `(name, debug value)` pairs.
type Fields = Vec<(String, String)>;

/// The fields of every warning event.
#[derive(Clone, Default)]
struct Warnings(Arc<Mutex<Vec<Fields>>>);

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Warnings {
    fn on_event(
        &self,
        event: &tracing::Event<'_>,
        _ctx: tracing_subscriber::layer::Context<'_, S>,
    ) {
        struct Record(Fields);
        impl tracing::field::Visit for Record {
            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
                self.0.push((field.name().to_owned(), format!("{value:?}")));
            }
        }
        if *event.metadata().level() == tracing::Level::WARN {
            let mut fields = Record(Vec::new());
            event.record(&mut fields);
            self.0.lock().unwrap().push(fields.0);
        }
    }
}

/// The address the admin server logged when it bound.
#[derive(Clone, Default)]
struct AdminAddr(Arc<Mutex<Option<std::net::SocketAddr>>>);

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for AdminAddr {
    fn on_event(
        &self,
        event: &tracing::Event<'_>,
        _ctx: tracing_subscriber::layer::Context<'_, S>,
    ) {
        struct Addr(Option<std::net::SocketAddr>);
        impl tracing::field::Visit for Addr {
            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
                if field.name() == "addr" {
                    self.0 = format!("{value:?}").parse().ok();
                }
            }
        }
        let mut addr = Addr(None);
        event.record(&mut addr);
        if let Some(addr) = addr.0 {
            *self.0.lock().unwrap() = Some(addr);
        }
    }
}

/// Feeds `sample_io_busy` a sample every 5 s for ten minutes from a
/// three-worker runtime busy `share` of the time, under a captured subscriber,
/// and returns the fields of every warning it logged.
fn io_busy_warnings(pipeline: &'static str, share: f64) -> Vec<Fields> {
    use crate::metrics::{ComponentLabels, IoBusySampler, IoRuntimeMetrics};
    use crate::pipeline::runtime::sample_io_busy;
    use tracing_subscriber::layer::SubscriberExt;

    const WORKERS: usize = 3;
    let labels = ComponentLabels::new(pipeline, "runtime", "pipeline");
    let gauge = IoRuntimeMetrics::try_new(&labels).expect("free series");
    let t0 = Instant::now();
    let mut sampler = IoBusySampler::new(
        Duration::from_secs(5),
        IoBusySampler::WINDOW,
        t0,
        Duration::ZERO,
    );
    let warnings = Warnings::default();
    let subscriber = tracing_subscriber::registry().with(warnings.clone());
    tracing::subscriber::with_default(subscriber, || {
        for secs in (5..=600).step_by(5) {
            let busy = Duration::from_secs_f64(share * (WORKERS as u64 * secs) as f64);
            sample_io_busy(
                &mut sampler,
                &gauge,
                t0 + Duration::from_secs(secs),
                busy,
                WORKERS,
                pipeline,
            );
        }
    });
    warnings.0.lock().unwrap().clone()
}

/// Ten minutes of a saturated three-worker I/O runtime log one warning,
/// carrying the pipeline and the worker count. Half busy on three workers
/// logs nothing.
#[test]
fn a_saturated_io_runtime_logs_one_warning() {
    let logged = io_busy_warnings("io-busy-warning", 0.95);
    assert_eq!(logged.len(), 1, "{logged:?}");
    let field = |name: &str| {
        logged[0]
            .iter()
            .find(|(n, _)| n == name)
            .map(|(_, v)| v.as_str())
    };
    assert_eq!(field("pipeline"), Some("\"io-busy-warning\""));
    assert_eq!(field("io_workers"), Some("3"));

    let half = io_busy_warnings("io-busy-half", 0.5);
    assert!(half.is_empty(), "half busy warned: {half:?}");
}

/// One pipeline's `spate_checkpoint_pending_batches` lines, sorted. The
/// pipeline label keeps the assertion independent of any other pipeline a
/// sibling test registered in this process.
fn pending_series(rendered: &str, pipeline_name: &str) -> Vec<String> {
    let mut out: Vec<String> = rendered
        .lines()
        .filter(|l| l.starts_with(&format!("{}{{", names::CHECKPOINT_PENDING_BATCHES)))
        .filter(|l| l.contains(&format!(r#"pipeline="{pipeline_name}""#)))
        .map(str::to_owned)
        .collect();
    out.sort();
    out
}

/// The checkpoint labels a pipeline of this name publishes under.
fn checkpoint_labels(pipeline_name: &str) -> String {
    format!(r#"pipeline="{pipeline_name}",component="checkpoint",component_type="checkpoint""#)
}

/// `metrics.per_partition_detail: true` publishes the checkpointer's
/// per-partition pending counts.
///
/// Regression for #334. `CheckpointMetrics::set_partition_pending` and
/// `retain_partitions` had no caller outside the metrics module's own tests,
/// so the flag bought a family that never gained a `partition`-labeled
/// series.
///
/// A labeled gauge registers on its first published value, so a missing call
/// leaves the series *absent* rather than `0`. Presence is what this asserts;
/// both lines render `0` because the run completed with every batch
/// acknowledged.
#[test]
fn per_partition_pending_reaches_the_exposition_when_detail_is_on() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    cfg.metrics.per_partition_detail = true;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));

    // Published on the commit tick, so it is there while the run is live and
    // not only on the final commit at shutdown.
    wait_for(
        "the per-partition series is published",
        Duration::from_secs(5),
        || {
            pending_series(&handle.render(), &pipeline_name)
                .iter()
                .any(|l| l.contains(r#"partition="0""#))
        },
    );

    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);

    let rendered = handle.render();
    let labels = checkpoint_labels(&pipeline_name);
    let mut expected = vec![
        format!("{}{{{labels}}} 0", names::CHECKPOINT_PENDING_BATCHES),
        format!(
            r#"{}{{{labels},partition="0"}} 0"#,
            names::CHECKPOINT_PENDING_BATCHES
        ),
    ];
    expected.sort();
    assert_eq!(
        pending_series(&rendered, &pipeline_name),
        expected,
        "the aggregate and the one assigned partition:\n{rendered}"
    );
}

/// Each partition's series carries that partition's own count, and a
/// partition that leaves the assignment is released.
///
/// These are the two assertions presence cannot make. A publisher writing a
/// constant satisfies presence, and so does a `commit_cycle` that never calls
/// `retain_partitions`. Both are caught here: the counts differ across the two
/// partitions, and partition 1 leaves while its batch is still unacknowledged,
/// so without the release its series freezes at `1` for the life of the
/// process rather than reading `0`.
#[test]
fn per_partition_pending_counts_each_partition_and_zeroes_a_revoked_one() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let release_c = Arc::clone(&release);

    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    cfg.metrics.per_partition_detail = true;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, move |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::HoldAcks {
            held: Arc::clone(&held_c),
            release: Arc::clone(&release_c),
        },
        batches_seen: 0,
    });

    // Three withheld batches on partition 0 against one on partition 1, so
    // the two series must differ and the aggregate must follow the larger.
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(&[0..10, 10..20, 20..30]),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));

    let labels = checkpoint_labels(&pipeline_name);
    let name = names::CHECKPOINT_PENDING_BATCHES;
    let p0 = format!(r#"{name}{{{labels},partition="0"}} 3"#);
    let p1 = format!(r#"{name}{{{labels},partition="1"}} 1"#);
    let aggregate = format!("{name}{{{labels}}} 3");
    wait_for(
        "each partition publishes its own count",
        Duration::from_secs(5),
        || {
            let rendered = handle.render();
            rendered.contains(&p0) && rendered.contains(&p1) && rendered.contains(&aggregate)
        },
    );

    // Partition 1 leaves with its batch still held, so the checkpointer drops
    // a tracker whose count is non-zero.
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(1)]));
    let p1_released = format!(r#"{name}{{{labels},partition="1"}} 0"#);
    wait_for(
        "the revoked partition is released and the remaining one is not",
        Duration::from_secs(10),
        || {
            let rendered = handle.render();
            rendered.contains(&p1_released) && rendered.contains(&p0)
        },
    );

    // Releasing retires partition 0's batches so the drain can complete.
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wait_for("partition 0 commits", Duration::from_secs(5), || {
        h.shared
            .lock()
            .unwrap()
            .committed
            .get(&PartitionId(0))
            .is_some_and(|&w| w >= 30)
    });
    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// With `per_partition_detail` off the family is the aggregate alone.
///
/// Asserting the whole list rather than the absence of a `partition=` line
/// also catches a second publisher of the aggregate.
#[test]
fn per_partition_pending_is_absent_when_detail_is_off() {
    let mut cfg = test_config(1);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    assert!(!cfg.metrics.per_partition_detail, "the default is off");
    let handle = crate::metrics::install(&crate::metrics::MetricsSettings {
        exporter: crate::metrics::Exporter::Prometheus,
        per_partition_detail: cfg.metrics.per_partition_detail,
        e2e_basis: crate::metrics::E2eBasis::Ingest,
    })
    .expect("install the exporter");

    let h = start_with_config(cfg, |shared, log| FakeChain {
        shared,
        log,
        mode: ChainMode::Ok,
        batches_seen: 0,
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });

    h.script.lock().unwrap().push_back(Script::Drained);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);

    let rendered = handle.render();
    assert_eq!(
        pending_series(&rendered, &pipeline_name),
        vec![format!(
            "{}{{{}}} 0",
            names::CHECKPOINT_PENDING_BATCHES,
            checkpoint_labels(&pipeline_name)
        )],
        "the aggregate alone:\n{rendered}"
    );
}

/// With `pipeline.threads` unset, the pipeline keeps one core per two I/O
/// workers, rounded up, plus one for the controller, and always runs at least
/// one thread.
#[test]
fn the_default_thread_count_reserves_half_a_core_per_io_worker() {
    for (cores, io_threads, threads) in [
        (1, 2, 1),
        (2, 2, 1),
        (4, 1, 2),
        (4, 2, 2),
        (8, 8, 3),
        (32, 2, 30),
        (32, 8, 27),
    ] {
        assert_eq!(
            default_thread_count(cores, io_threads),
            threads,
            "{cores} cores, io_threads: {io_threads}"
        );
    }
}

/// With `pipeline.threads` unset, the runtime starts as many drivers as
/// `default_thread_count` derives from `available_parallelism` and the
/// configured `io_threads`.
#[test]
fn an_unset_thread_count_starts_the_derived_number_of_drivers() {
    // From 4 cores up, `io_threads: 2` tells a reserve of one core per worker
    // apart. `io_threads: 4` tells apart a reserve that ignores the configured
    // count, since 1 and 2 workers both reserve one core.
    for io_threads in [2, 4] {
        let mut cfg = test_config(1);
        cfg.pipeline.threads = None;
        cfg.pipeline.io_threads = io_threads;
        let built = Arc::new(AtomicUsize::new(0));
        let counter = Arc::clone(&built);
        let h = start_with_config(cfg, move |shared, log| {
            counter.fetch_add(1, Ordering::Relaxed);
            FakeChain {
                shared,
                log,
                mode: ChainMode::Ok,
                batches_seen: 0,
            }
        });
        h.script.lock().unwrap().push_back(Script::Drained);
        let report = h.join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed);

        let cores = std::thread::available_parallelism().map_or(2, usize::from);
        assert_eq!(
            built.load(Ordering::Relaxed),
            default_thread_count(cores, io_threads),
            "{cores} cores, io_threads: {io_threads}"
        );
    }
}

/// Pins that the source is opened with the thread count the runtime
/// resolved, whether `pipeline.threads` sets it or the host derives it.
#[test]
fn the_source_is_opened_with_the_pipeline_thread_count() {
    // 3 differs from the `io_threads` of 2 set below, so a count taken
    // from the wrong setting fails.
    let cores = std::thread::available_parallelism().map_or(2, usize::from);
    for (threads, expected) in [(Some(3), 3), (None, default_thread_count(cores, 2))] {
        let mut cfg = test_config(1);
        cfg.pipeline.threads = threads;
        cfg.pipeline.io_threads = 2;
        let h = start_with_config(cfg, |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::Ok,
            batches_seen: 0,
        });
        h.script.lock().unwrap().push_back(Script::Drained);
        let report = h.join.join().unwrap().unwrap();
        assert_eq!(report.state, ExitState::Completed);
        assert_eq!(
            h.shared.lock().unwrap().threads,
            std::num::NonZeroUsize::new(expected),
            "pipeline.threads: {threads:?}"
        );
    }
}

// ------------------------------------------------------ graceful stop --

/// A config whose stop has `drain_timeout` to run in.
fn stop_config(drain_timeout: Duration) -> PipelineConfig {
    let mut cfg = test_config(1);
    cfg.checkpoint.drain_timeout = drain_timeout;
    cfg
}

/// Start a run whose source answers `reading_on_stop` with `read_on_stop`
/// and logs its lanes' drops, on the stop clock `clock`.
fn start_stopping(
    cfg: PipelineConfig,
    read_on_stop: bool,
    clock: &Arc<crate::clock::TestClock>,
    feed: impl FnOnce(&SharedLog) -> Option<Arc<dyn crate::feed::Feed>>,
    mode_factory: impl Fn(Arc<ChainShared>, Arc<Mutex<SourceLog>>) -> FakeChain + Send + 'static,
) -> Harness {
    let clock = Arc::clone(clock);
    start_prepared(
        cfg,
        test_options(),
        |log| {
            log.read_on_stop = read_on_stop;
            log.log_lane_drops = true;
            log.accept_feed = true;
        },
        move |runtime, log| {
            let runtime = runtime.with_stop_clock(clock);
            match feed(log) {
                Some(feed) => runtime.with_feed(feed),
                None => runtime,
            }
        },
        mode_factory,
    )
}

fn joined(h: &Harness) {
    wait_for("run() to return", Duration::from_secs(10), || {
        h.join.is_finished()
    });
}

/// Pins that startup refuses a drain timeout no longer than the stop reserve
/// for a feed that hands back, under any backend key, before the source
/// opens, and accepts it for one that does not.
#[test]
fn a_feed_refuses_a_drain_timeout_within_the_read_reserve() {
    for backend in ["sqs", "kafka"] {
        let mut cfg = stop_config(crate::feed::STOP_RESERVE);
        cfg.feed = feed_section(&format!("{backend}: {{}}"));
        let h = start_with(
            cfg,
            test_options(),
            |runtime| runtime.with_feed(Arc::new(HandingBack::new(backend))),
            ok_chain,
        );
        let err = config_error_before_open(h);
        assert!(
            err.contains("checkpoint.drain_timeout")
                && err.contains("5s")
                && err.contains(&format!("`{backend}`")),
            "{err}"
        );
    }

    let clock = crate::clock::TestClock::frozen();
    let completes = |cfg: PipelineConfig, feed: Option<Arc<dyn crate::feed::Feed>>| {
        let h = start_stopping(cfg, false, &clock, |_| feed, ok_chain);
        h.script.lock().unwrap().push_back(Script::Drained);
        joined(&h);
        let report = h.join.join().unwrap().expect("the run starts");
        assert_eq!(report.state, ExitState::Completed);
    };
    let mut cfg = stop_config(crate::feed::STOP_RESERVE + Duration::from_millis(1));
    cfg.feed = feed_section("sqs: {}");
    completes(cfg, Some(Arc::new(HandingBack::new("sqs"))));
    for drain_timeout in [crate::feed::STOP_RESERVE, Duration::from_secs(3)] {
        let mut cfg = stop_config(drain_timeout);
        cfg.feed = feed_section("memory: {}");
        completes(cfg, None);
    }
    let mut cfg = stop_config(crate::feed::STOP_RESERVE);
    cfg.feed = feed_section("memory: {}");
    completes(cfg, Some(Arc::new(MemoryFeed::new())));
}

/// Pins that a stop reads on while the source answers `reading_on_stop`
/// until the reserve before the deadline, in the draining state, and calls
/// `stop_reading` at the reserve, before the drivers drop their lanes.
#[test]
fn a_stop_reads_on_until_the_reserve_then_hands_back() {
    let drain_timeout = Duration::from_secs(600);
    let mut cfg = stop_config(drain_timeout);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = install_prometheus(&cfg);
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(cfg, true, &clock, |_| None, ok_chain);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });

    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    clock.advance(drain_timeout - crate::feed::STOP_RESERVE - Duration::from_millis(1));
    let asked = h.shared.lock().unwrap().reading_on_stop_calls;
    wait_for("three more passes", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls >= asked + 3
    });
    assert_eq!(h.shared.lock().unwrap().stop_reading_calls, 0);
    let draining = format!(r#"spate_pipeline_state{{pipeline="{pipeline_name}","#);
    let rendered = handle.render();
    assert!(
        rendered.lines().any(|l| l.starts_with(&draining)
            && l.contains(r#"state="draining""#)
            && l.ends_with(" 1")),
        "{rendered}"
    );

    clock.advance(Duration::from_millis(1));
    joined(&h);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    let log = h.shared.lock().unwrap();
    assert_eq!(log.stop_reading_calls, 1);
    let at = |entry: &str| log.log.iter().position(|e| e == entry);
    assert!(
        at("stop_reading") < at("lane-dropped") && at("lane-dropped").is_some(),
        "{:?}",
        log.log
    );
    // With no feed that hands back, the sink drains to the deadline itself.
    assert_eq!(
        *h.sink_budgets.lock().unwrap(),
        vec![crate::feed::STOP_RESERVE]
    );
}

/// Pins that a source with nothing left to read ends the read phase on the
/// first pass, so the stop drains without waiting for the reserve.
#[test]
fn a_stop_with_nothing_left_to_read_drains_at_once() {
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(Duration::from_secs(600)),
        false,
        &clock,
        |_| None,
        ok_chain,
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    h.shutdown.trigger();
    joined(&h);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    let log = h.shared.lock().unwrap();
    assert_eq!((log.reading_on_stop_calls, log.stop_reading_calls), (1, 1));
}

/// Pins that a batch blocked when the stop begins keeps retrying through the
/// read phase and is delivered, where the drain would abandon it for replay.
#[test]
fn a_batch_blocked_during_the_read_phase_is_not_abandoned() {
    let open = Arc::new(AtomicBool::new(false));
    let unblock = Arc::clone(&open);
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(Duration::from_secs(600)),
        true,
        &clock,
        |_| None,
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::BlockUntil(Arc::clone(&unblock)),
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for(
        "the blocked batch pauses the lane",
        Duration::from_secs(5),
        || !h.shared.lock().unwrap().pauses.is_empty(),
    );
    h.shutdown.trigger();
    let asked = h.shared.lock().unwrap().reading_on_stop_calls;
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls >= asked + 3
    });
    open.store(true, Ordering::Relaxed);
    wait_for("the batch is taken", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.shared.lock().unwrap().read_on_stop = false;
    joined(&h);
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

/// Pins that a failed run reads nothing more and hands back at once.
#[test]
fn a_failure_drain_hands_back_without_reading_on() {
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(Duration::from_secs(600)),
        true,
        &clock,
        |_| None,
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::FatalAtBatch(1),
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    joined(&h);
    let report = h.join.join().unwrap().unwrap();
    assert!(matches!(report.state, ExitState::Failed(_)), "{report:?}");
    let log = h.shared.lock().unwrap();
    assert_eq!((log.reading_on_stop_calls, log.stop_reading_calls), (0, 1));
}

/// Pins that a source whose run ends drained hands nothing back, and that a
/// drain no stop began runs to `drain_timeout` from its own start: the sink
/// to a second before it, the feed's close to the end.
#[test]
fn a_drained_run_does_not_stop_reading() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(drain_timeout);
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let h = start_stopping(
        cfg,
        true,
        &clock,
        move |_| {
            let feed = Arc::new(HandingBack::new("sqs"));
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            Some(feed)
        },
        ok_chain,
    );
    h.script.lock().unwrap().push_back(Script::Drained);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_eq!(h.shared.lock().unwrap().stop_reading_calls, 0);
    let feed = kept.lock().unwrap().take().unwrap();
    assert_eq!(*feed.closes.lock().unwrap(), vec![drain_timeout]);
    assert_eq!(
        *h.sink_budgets.lock().unwrap(),
        vec![drain_timeout - Duration::from_secs(1)]
    );
}

/// Pins that the runtime closes a feed it was given once, after the final
/// commit's flush.
#[test]
fn the_runtime_closes_its_feed_after_the_final_commit() {
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(Duration::from_secs(600));
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let h = start_stopping(
        cfg,
        false,
        &clock,
        move |log| {
            let feed = Arc::new(HandingBack::new("sqs").logging_into(Arc::clone(log)));
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            Some(feed)
        },
        ok_chain,
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.shutdown.trigger();
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let feed = kept.lock().unwrap().take().unwrap();
    assert_eq!(feed.closes.lock().unwrap().len(), 1);
    let log = h.shared.lock().unwrap();
    let close = log.log.iter().position(|e| e == "close");
    let last_flush = log.log.iter().rposition(|e| e == "flush_commits");
    assert!(last_flush.is_some() && close > last_flush, "{:?}", log.log);
    assert_eq!(
        log.log.last().map(String::as_str),
        Some("close"),
        "{:?}",
        log.log
    );
}

/// Pins that the close's bound is what is left of the stop's deadline, read
/// on the stop clock: the whole reserve when the read phase ran to it.
#[test]
fn the_close_gets_the_rest_of_the_deadline() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(drain_timeout);
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let h = start_stopping(
        cfg,
        true,
        &clock,
        move |_| {
            let feed = Arc::new(HandingBack::new("sqs"));
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            Some(feed)
        },
        ok_chain,
    );
    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    clock.advance(drain_timeout - crate::feed::STOP_RESERVE);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let feed = kept.lock().unwrap().take().unwrap();
    assert_eq!(
        *feed.closes.lock().unwrap(),
        vec![crate::feed::STOP_RESERVE]
    );
    // The sink drain ends a second early, which leaves that second to the
    // close when the sink drains slower than this one.
    assert_eq!(
        *h.sink_budgets.lock().unwrap(),
        vec![crate::feed::STOP_RESERVE - Duration::from_secs(1)]
    );
}

/// Pins that a close that never settles ends the run at the deadline, with
/// a warning, and leaves the exit state alone.
#[test]
fn a_close_that_never_settles_ends_at_the_deadline() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(drain_timeout);
    cfg.feed = feed_section("sqs: {}");
    let h = start_stopping(
        cfg,
        true,
        &clock,
        |_| Some(Arc::new(HandingBack::new("sqs").holding_its_close())),
        ok_chain,
    );
    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    clock.advance(drain_timeout + Duration::from_secs(1));
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let warnings = h.warnings.0.lock().unwrap().clone();
    assert!(
        warnings
            .iter()
            .any(|fields| fields.iter().any(|(name, value)| {
                name == "message" && value.contains("did not close before the drain deadline")
            })),
        "{warnings:#?}"
    );
}

/// Pins that a source keeping `Source::reading_on_stop`'s default reads
/// nothing on after the stop, so it drains without waiting for the reserve.
#[test]
fn a_source_that_does_not_read_on_drains_at_once() {
    struct Plain(FakeSource);
    impl Source for Plain {
        type Lane = FakeLane;
        fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError> {
            self.0.open(ctx)
        }
        fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<FakeLane>, SourceError> {
            self.0.poll_events(timeout)
        }
        fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError> {
            self.0.commit(watermarks)
        }
    }
    let (inner, shared, _script) = FakeSource::new();
    let chain_shared = Arc::new(ChainShared::default());
    let (sink, _drained) = test_sink();
    let log = Arc::clone(&shared);
    let runtime = PipelineRuntime::new(
        stop_config(Duration::from_secs(600)),
        Plain(inner),
        move |_thread| {
            Box::new(ok_chain(Arc::clone(&chain_shared), Arc::clone(&log)))
                as Box<dyn RunnableChain>
        },
        sink,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_options(test_options())
    .with_stop_clock(crate::clock::TestClock::frozen());
    runtime.shutdown_handle().trigger();
    let join = std::thread::spawn(move || runtime.run());
    wait_for("run() to return", Duration::from_secs(10), || {
        join.is_finished()
    });
    assert_eq!(join.join().unwrap().unwrap().state, ExitState::Completed);
}

/// Pins that a source whose `open` failed is not asked to hand back.
#[test]
fn a_source_that_failed_to_open_is_not_asked_to_stop_reading() {
    let clock = crate::clock::TestClock::frozen();
    let h = start_prepared(
        stop_config(Duration::from_secs(600)),
        test_options(),
        |log| log.fail_open = true,
        move |runtime, _| runtime.with_stop_clock(clock),
        ok_chain,
    );
    joined(&h);
    let report = h.join.join().unwrap().unwrap();
    assert!(matches!(report.state, ExitState::Failed(_)), "{report:?}");
    assert_eq!(h.shared.lock().unwrap().stop_reading_calls, 0);
}

/// A lane whose batch blocks, assigned and revoked, with the revocation
/// waiting on the blocked driver when this returns.
fn revoke_a_blocked_lane(h: &Harness) {
    assign_one_lane(h, std::slice::from_ref(&(0..10)));
    wait_for(
        "the blocked batch pauses the lane",
        Duration::from_secs(5),
        || !h.shared.lock().unwrap().pauses.is_empty(),
    );
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    wait_for(
        "the revocation is delivered",
        Duration::from_secs(5),
        || {
            h.shared
                .lock()
                .unwrap()
                .log
                .iter()
                .any(|e| e == "revoke-delivered")
        },
    );
}

fn blocked_chain(shared: Arc<ChainShared>, log: Arc<Mutex<SourceLog>>) -> FakeChain {
    FakeChain {
        shared,
        log,
        mode: ChainMode::BlockForever,
        batches_seen: 0,
    }
}

/// Pins that a stop which arrives while a revocation waits on a driver held
/// by a blocked batch reaches that driver when the source reads nothing on,
/// so the run drains without waiting out the revocation's budget.
#[test]
fn a_stop_during_a_blocked_revocation_drains_at_once() {
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(Duration::from_secs(600)),
        false,
        &clock,
        |_| None,
        blocked_chain,
    );
    revoke_a_blocked_lane(&h);
    h.shutdown.trigger();
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_eq!(h.chain.consumed.load(Ordering::Relaxed), 0);
    assert_revoked_while_blocked(&h.shared);
}

/// Pins that while a revocation waits on a driver held by a blocked batch,
/// the controller lets the source renew what it holds on more than one pass.
/// Regression for #1278.
#[test]
fn a_revocation_waiting_on_its_drain_lets_the_source_renew() {
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(Duration::from_secs(600)),
        false,
        &clock,
        |_| None,
        blocked_chain,
    );
    revoke_a_blocked_lane(&h);
    wait_for(
        "the source renews during the wait",
        Duration::from_secs(5),
        || h.shared.lock().unwrap().renew_held_calls >= 2,
    );
    h.shutdown.trigger();
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_revoked_while_blocked(&h.shared);
}

/// Pins that a revocation waiting on its drain during a stop's read phase
/// lets the source renew on each pass, the passes counted by the
/// `reading_on_stop` the same wait asks on each one.
#[test]
fn a_revocation_during_the_read_phase_renews_on_each_pass() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(drain_timeout),
        true,
        &clock,
        |_| None,
        blocked_chain,
    );
    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    revoke_a_blocked_lane(&h);
    let (asked, renewed) = {
        let log = h.shared.lock().unwrap();
        (log.reading_on_stop_calls, log.renew_held_calls)
    };
    wait_for("five passes of the wait", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls >= asked + 5
    });
    let (passes, renews) = {
        let log = h.shared.lock().unwrap();
        (
            log.reading_on_stop_calls - asked,
            log.renew_held_calls - renewed,
        )
    };
    // One `reading_on_stop` may precede the wait's first pass.
    assert!(
        renews + 1 >= passes,
        "{renews} renewals over {passes} passes"
    );
    clock.advance(drain_timeout - crate::feed::STOP_RESERVE);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_revoked_while_blocked(&h.shared);
}

/// The lane's batch blocked and its revocation reached the controller.
fn assert_revoked_while_blocked(shared: &SharedLog) {
    let log = shared.lock().unwrap();
    assert!(!log.pauses.is_empty(), "the batch never blocked");
    let revoked = log.log.iter().position(|e| e == "revoke-delivered");
    let stopped = log.log.iter().position(|e| e == "stop_reading");
    assert!(revoked.is_some() && revoked < stopped, "{:?}", log.log);
}

/// Pins that a stop request names whether it is set and when it was first
/// asked for.
#[test]
fn a_stop_request_reports_its_trigger() {
    let clock = crate::clock::TestClock::frozen();
    let runtime = PipelineRuntime::new(
        test_config(1),
        FakeSource::new().0,
        |_thread| unreachable!("the runtime does not run"),
        test_sink().0,
        Arc::new(crate::backpressure::InflightBudget::new()),
    )
    .with_stop_clock(Arc::clone(&clock) as _);
    let handle = runtime.shutdown_handle();
    assert!(
        format!("{handle:?}").contains("set: false, at: None"),
        "{handle:?}"
    );
    handle.trigger();
    clock.advance(Duration::from_secs(1));
    handle.trigger();
    let printed = format!("{handle:?}");
    assert!(printed.contains("set: true, at: Some("), "{printed}");
}

/// Pins that a revocation during the stop's reading, waiting on a blocked
/// batch, ends at the reserve, so the hand-back and the drain still run.
#[test]
fn a_revocation_during_the_read_phase_ends_at_the_reserve() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(
        stop_config(drain_timeout),
        true,
        &clock,
        |_| None,
        blocked_chain,
    );
    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    revoke_a_blocked_lane(&h);
    clock.advance(drain_timeout - crate::feed::STOP_RESERVE);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_eq!(h.shared.lock().unwrap().stop_reading_calls, 1);
    assert_revoked_while_blocked(&h.shared);
}

/// Pins that the drain deadline counts from the trigger: a stop asked for
/// before the source finished opening leaves the close what is left after
/// the open.
#[test]
fn the_drain_deadline_counts_from_the_trigger() {
    let drain_timeout = Duration::from_secs(600);
    let opening = Duration::from_secs(10);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(drain_timeout);
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let in_open = Arc::clone(&clock);
    let stop_clock = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        test_options(),
        |log| {
            log.accept_feed = true;
            log.advance_in_open = Some((in_open, opening));
        },
        move |runtime, _| {
            let feed = Arc::new(HandingBack::new("sqs"));
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            let runtime = runtime.with_stop_clock(stop_clock).with_feed(feed);
            runtime.shutdown_handle().trigger();
            runtime
        },
        ok_chain,
    );
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let feed = kept.lock().unwrap().take().unwrap();
    assert_eq!(*feed.closes.lock().unwrap(), vec![drain_timeout - opening]);
}

/// Pins that a close the feed rejects ends the run completed, with a
/// warning that carries the feed's reason.
#[test]
fn a_rejected_close_warns_with_its_reason() {
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(Duration::from_secs(600));
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let h = start_stopping(
        cfg,
        false,
        &clock,
        move |_| {
            let feed = Arc::new(HandingBack::new("sqs").holding_its_close());
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            Some(feed)
        },
        ok_chain,
    );
    let feed = kept.lock().unwrap().clone().unwrap();
    h.shutdown.trigger();
    wait_for("the close", Duration::from_secs(5), || {
        !feed.closes.lock().unwrap().is_empty()
    });
    feed.reject_held_closes(&crate::feed::FeedError::new(
        crate::feed::FeedErrorKind::Retryable,
        "2 units not handed back",
    ));
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let warnings = h.warnings.0.lock().unwrap().clone();
    assert!(
        warnings.iter().any(|fields| {
            fields
                .iter()
                .any(|(name, value)| name == "message" && value.contains("did not close cleanly"))
                && fields.iter().any(|(name, value)| {
                    name == "error" && value.contains("2 units not handed back")
                })
        }),
        "{warnings:#?}"
    );
}

/// Pins that the runtime waits for a close that settles after the wait
/// begins, with no warning.
#[test]
fn a_close_that_settles_later_is_waited_for() {
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(Duration::from_secs(600));
    cfg.feed = feed_section("sqs: {}");
    let kept = Arc::new(Mutex::new(None::<Arc<HandingBack>>));
    let keep = Arc::clone(&kept);
    let h = start_stopping(
        cfg,
        false,
        &clock,
        move |_| {
            let feed = Arc::new(HandingBack::new("sqs").holding_its_close());
            *keep.lock().unwrap() = Some(Arc::clone(&feed));
            Some(feed)
        },
        ok_chain,
    );
    let feed = kept.lock().unwrap().clone().unwrap();
    h.shutdown.trigger();
    wait_for("the close", Duration::from_secs(5), || {
        !feed.closes.lock().unwrap().is_empty()
    });
    feed.settle_held_closes();
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    let warnings = h.warnings.0.lock().unwrap().clone();
    assert!(
        !warnings.iter().any(|fields| fields
            .iter()
            .any(|(name, value)| name == "message" && value.contains("did not close"))),
        "{warnings:#?}"
    );
}

/// Pins that a feed that does not hand back leaves the sink drain the whole
/// deadline.
#[test]
fn a_feed_that_does_not_hand_back_leaves_the_sink_the_whole_deadline() {
    let drain_timeout = Duration::from_secs(600);
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = stop_config(drain_timeout);
    cfg.feed = feed_section("memory: {}");
    let h = start_stopping(cfg, false, &clock, |_| None, ok_chain);
    h.script.lock().unwrap().push_back(Script::Drained);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
    assert_eq!(*h.sink_budgets.lock().unwrap(), vec![drain_timeout]);
}

/// Pins that an assignment during the stop's reading leaves the pipeline in
/// the draining state.
#[test]
fn an_assignment_during_the_read_phase_keeps_the_pipeline_draining() {
    let drain_timeout = Duration::from_secs(600);
    let mut cfg = stop_config(drain_timeout);
    cfg.metrics.exporter = crate::config::MetricsExporter::Prometheus;
    let pipeline_name = cfg.pipeline.name.clone();
    let handle = install_prometheus(&cfg);
    let clock = crate::clock::TestClock::frozen();
    let h = start_stopping(cfg, true, &clock, |_| None, ok_chain);
    h.shutdown.trigger();
    wait_for("the read phase", Duration::from_secs(5), || {
        h.shared.lock().unwrap().reading_on_stop_calls > 0
    });
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    let state = |name: &str| {
        let line = format!(r#"pipeline="{pipeline_name}","#);
        handle
            .render()
            .lines()
            .find(|l| {
                l.starts_with("spate_pipeline_state{")
                    && l.contains(&line)
                    && l.contains(&format!(r#"state="{name}""#))
            })
            .map(str::to_owned)
    };
    assert!(
        state("draining").is_some_and(|l| l.ends_with(" 1")),
        "{:?}",
        state("draining")
    );
    assert!(
        state("running").is_some_and(|l| l.ends_with(" 0")),
        "{:?}",
        state("running")
    );
    clock.advance(drain_timeout);
    joined(&h);
    assert_eq!(h.join.join().unwrap().unwrap().state, ExitState::Completed);
}

/// Pins that SIGTERM begins the graceful stop, so the run drains and
/// returns.
#[cfg(unix)]
#[test]
fn a_sigterm_stops_the_run() {
    // A listener of the test's own replaces the default action first, so a
    // SIGTERM sent before the runtime listens does not end the test process.
    let signals = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let _listener = signals.block_on(async { signal(SignalKind::terminate()).unwrap() });
    let mut options = test_options();
    options.handle_signals = true;
    let h = start_with(test_config(1), options, |runtime| runtime, ok_chain);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("payloads consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    let pid = std::process::id().to_string();
    // The runtime's listener may not be installed yet, so the signal is
    // sent until the run ends.
    let deadline = Instant::now() + Duration::from_secs(10);
    while !h.join.is_finished() {
        assert!(Instant::now() < deadline, "SIGTERM did not stop the run");
        let sent = std::process::Command::new("kill")
            .args(["-TERM", &pid])
            .status()
            .unwrap();
        assert!(sent.success());
        std::thread::sleep(Duration::from_millis(50));
    }
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
    assert_eq!(report.final_watermarks, vec![(PartitionId(0), 10)]);
}

// ------------------------------------------------------- finished lanes --

const FINISH_DEADLINE: Duration = Duration::from_secs(5);

/// Options whose lane-poll and idle-flush waits outlast any test, so a
/// parked thread turns only on the control messages a test sends.
fn parked_options() -> RuntimeOptions {
    RuntimeOptions {
        poll_timeout: Duration::from_secs(3600),
        idle_flush: Duration::from_secs(3600),
        ..test_options()
    }
}

/// One pipeline thread whose `finishing` lanes report `finished` from the
/// empty poll after their last batch, with no commit tick inside the test.
fn start_finishing(
    options: RuntimeOptions,
    max_pending_batches: Option<usize>,
    finishing: &[u32],
    mode: impl Fn() -> ChainMode + Send + 'static,
) -> Harness {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(60);
    if let Some(limit) = max_pending_batches {
        cfg.checkpoint.max_pending_batches = limit;
    }
    let finishing: Vec<LaneId> = finishing.iter().map(|&id| LaneId(id)).collect();
    start_prepared(
        cfg,
        options,
        move |log| log.finishing.extend(finishing),
        |runtime, _| runtime,
        move |shared, log| FakeChain {
            shared,
            log,
            mode: mode(),
            batches_seen: 0,
        },
    )
}

/// `count` batches of ten records each, from offset 0.
fn tens(count: i64) -> Vec<Vec<(i64, Vec<u8>)>> {
    let ranges: Vec<std::ops::Range<i64>> = (0..count).map(|i| i * 10..i * 10 + 10).collect();
    batches(&ranges)
}

/// Assign one lane per `(id, batches)` pair, each on partition `id`, with
/// `batches` batches of ten records.
fn assign(h: &Harness, lanes: &[(u32, i64)]) {
    let specs = lanes
        .iter()
        .map(|&(id, count)| LaneSpec {
            id: LaneId(id),
            partition: PartitionId(id),
            batches: tens(count),
        })
        .collect();
    h.script.lock().unwrap().push_back(Script::Assign(specs));
}

fn polls_of(h: &Harness, id: u32) -> usize {
    let log = h.shared.lock().unwrap();
    log.lane_polls
        .iter()
        .filter(|(l, _)| *l == LaneId(id))
        .count()
}

fn finished_calls(h: &Harness) -> usize {
    h.shared.lock().unwrap().finished_calls.values().sum()
}

fn flushes(h: &Harness) -> usize {
    h.chain.flushes.load(Ordering::Relaxed)
}

/// Wake the parked thread `times` times with a `CommitReady` for partition
/// 0, each answered by the `FlushNow` it sends, and return the flushes the
/// wakes added.
fn wake(h: &Harness, times: usize) -> usize {
    let before = flushes(h);
    for i in 1..=times {
        h.script
            .lock()
            .unwrap()
            .push_back(Script::CommitReady(vec![PartitionId(0)]));
        wait_for("the thread serves the wake", FINISH_DEADLINE, || {
            flushes(h) >= before + i
        });
    }
    flushes(h) - before
}

fn shut_down(h: Harness) {
    h.shutdown.trigger();
    wait_for("the driver exits", FINISH_DEADLINE, || h.join.is_finished());
    let report = h.join.join().unwrap().unwrap();
    assert_eq!(report.state, ExitState::Completed);
}

/// Pins that a thread whose lanes have all finished stops polling them and
/// parks on its control channel, serving each `FlushNow` from the park.
/// Regression for #1478.
#[test]
fn a_thread_whose_lanes_are_all_finished_parks() {
    let h = start_finishing(parked_options(), None, &[0, 1], || ChainMode::Ok);
    assign(&h, &[(0, 1), (1, 1)]);
    wait_for("both lanes finish", FINISH_DEADLINE, || {
        polls_of(&h, 0) == 2 && polls_of(&h, 1) == 2
    });
    let calls = finished_calls(&h);
    let wakes = wake(&h, 5);
    let asked = finished_calls(&h) - calls;
    assert!(wakes >= 5, "{wakes} wakes");
    // Each wake turns the loop at least once and parks again within a pass.
    assert!(
        (wakes..=2 * wakes + 2).contains(&asked),
        "{asked} reads over {wakes} wakes"
    );
    assert_eq!((polls_of(&h, 0), polls_of(&h, 1)), (2, 2));
    shut_down(h);
}

/// Pins that a finished lane is not polled again while a sibling on its
/// thread still produces, and that the sibling is never handed a blocking
/// timeout while it has data.
#[test]
fn a_finished_lane_is_skipped_while_its_sibling_flows() {
    let h = start_finishing(parked_options(), None, &[0], || ChainMode::Ok);
    assign(&h, &[(0, 1), (1, 50)]);
    wait_for("every batch is consumed", FINISH_DEADLINE, || {
        h.chain.consumed.load(Ordering::Relaxed) == 510
    });
    assert_eq!(polls_of(&h, 0), 2);
    let log = h.shared.lock().unwrap();
    let flowing_polls: Vec<Duration> = log
        .lane_polls
        .iter()
        .filter(|(l, _)| *l == LaneId(1))
        .map(|(_, t)| *t)
        .take(50)
        .collect();
    drop(log);
    assert!(
        flowing_polls.iter().all(Duration::is_zero),
        "{flowing_polls:?}"
    );
    shut_down(h);
}

/// Pins that a finished lane does not stop an empty sibling's pass from
/// reaching the blocking poll timeout.
#[test]
fn a_finished_lane_beside_an_empty_one_lets_the_pass_block() {
    let h = start_finishing(test_options(), None, &[0], || ChainMode::Ok);
    assign(&h, &[(0, 0), (1, 0)]);
    wait_for("the empty lane is polled 20 times", FINISH_DEADLINE, || {
        polls_of(&h, 1) >= 20
    });
    assert_eq!(polls_of(&h, 0), 1);
    let log = h.shared.lock().unwrap();
    let timeouts: Vec<Duration> = log
        .lane_polls
        .iter()
        .filter(|(l, _)| *l == LaneId(1))
        .skip(1)
        .map(|(_, t)| *t)
        .collect();
    drop(log);
    assert!(
        timeouts.iter().all(|t| *t == test_options().poll_timeout),
        "{timeouts:?}"
    );
    shut_down(h);
}

/// Pins that a thread whose lanes are each gated or finished parks, and that
/// the gated lane flows again once its acknowledgments retire.
#[test]
fn a_finished_lane_beside_a_gated_one_parks() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let (held_c, release_c) = (Arc::clone(&held), Arc::clone(&release));
    let h = start_finishing(parked_options(), Some(3), &[1], move || {
        ChainMode::HoldAcks {
            held: Arc::clone(&held_c),
            release: Arc::clone(&release_c),
        }
    });
    assign(&h, &[(0, 6), (1, 0)]);
    wait_for(
        "the gate closes and the empty lane finishes",
        FINISH_DEADLINE,
        || held.lock().unwrap().len() == 3 && polls_of(&h, 1) == 1,
    );
    let calls = finished_calls(&h);
    let wakes = wake(&h, 5);
    let asked = finished_calls(&h) - calls;
    assert!(wakes >= 5, "{wakes} wakes");
    // Each wake turns the loop at least once and parks again within a pass.
    assert!(
        (wakes..=2 * wakes + 2).contains(&asked),
        "{asked} reads over {wakes} wakes"
    );
    assert_eq!(polls_of(&h, 1), 1);
    release.store(true, Ordering::Relaxed);
    held.lock().unwrap().clear();
    wake(&h, 1);
    wait_for("the gated lane drains", FINISH_DEADLINE, || {
        h.chain.consumed.load(Ordering::Relaxed) == 60
    });
    shut_down(h);
}

/// Pins that a thread parked on its finished lanes still flushes its chain
/// once its lull outlasts the idle flush, so the tail's acknowledgments
/// resolve.
#[test]
fn a_finished_thread_still_flushes_its_tail_when_idle() {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let held_c = Arc::clone(&held);
    let h = start_finishing(test_options(), None, &[0], move || {
        ChainMode::HoldAcksUntilFlush {
            held: Arc::clone(&held_c),
        }
    });
    assign(&h, &[(0, 1)]);
    wait_for("the tail is flushed", FINISH_DEADLINE, || {
        polls_of(&h, 0) == 2 && held.lock().unwrap().is_empty() && flushes(&h) >= 1
    });
    assert_eq!(polls_of(&h, 0), 2);
    shut_down(h);
}

/// Pins that a thread parked on a finished lane takes a lane handed to it.
#[test]
fn a_finished_thread_takes_a_new_lane() {
    let h = start_finishing(parked_options(), None, &[0], || ChainMode::Ok);
    assign(&h, &[(0, 1)]);
    wait_for("the lane finishes", FINISH_DEADLINE, || {
        polls_of(&h, 0) >= 2
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Add(vec![LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: tens(1),
        }]));
    wait_for("the new lane is read", FINISH_DEADLINE, || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    shut_down(h);
}

/// Pins that a thread parked on a finished lane stops it on a revocation and
/// arrives at the revocation's barrier.
#[test]
fn a_finished_thread_stops_a_revoked_lane() {
    let h = start_finishing(parked_options(), None, &[0], || ChainMode::Ok);
    assign(&h, &[(0, 1)]);
    wait_for("the lane finishes", FINISH_DEADLINE, || {
        polls_of(&h, 0) >= 2
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    wait_for("the revocation commits", FINISH_DEADLINE, || {
        logged(&h, "revoke-delivered")
            && h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    shut_down(h);
}

/// Pins that a thread parked on a finished lane drops it when it retires.
#[test]
fn a_finished_thread_drops_a_retired_lane() {
    let h = start_finishing(parked_options(), None, &[0], || ChainMode::Ok);
    h.shared.lock().unwrap().log_lane_drops = true;
    assign(&h, &[(0, 1)]);
    wait_for("the lane finishes", FINISH_DEADLINE, || {
        polls_of(&h, 0) >= 2
    });
    wake(&h, 1);
    wait_for("partition 0 commits", FINISH_DEADLINE, || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Retire(vec![LaneId(0)]));
    wait_for("the lane is dropped", FINISH_DEADLINE, || {
        logged(&h, "lane-dropped")
    });
    shut_down(h);
}

/// Pins that a lane-set change between skipped lanes does not carry the skip
/// streak into the smaller rotation and park the thread while a sibling lane
/// still holds batches.
#[test]
fn a_retire_between_skips_does_not_park_a_flowing_thread() {
    let h = start_finishing(parked_options(), None, &[0, 1], || ChainMode::Ok);
    let (reached_tx, reached_rx) = mpsc::channel();
    let (go_tx, go_rx) = mpsc::channel();
    h.shared.lock().unwrap().finished_hook = Some(FinishedHook {
        polled: LaneId(2),
        after: LaneId(0),
        lane: LaneId(1),
        reached: reached_tx,
        go: go_rx,
    });
    // One lane per message, each read before the next is sent, so the
    // thread holds the lanes in the order 0, 1, 2.
    assign(&h, &[(0, 1)]);
    for (id, count) in [(1, 1), (2, 200)] {
        wait_for("the previous lane is read", FINISH_DEADLINE, || {
            polls_of(&h, id - 1) >= 1
        });
        h.script
            .lock()
            .unwrap()
            .push_back(Script::Add(vec![LaneSpec {
                id: LaneId(id),
                partition: PartitionId(id),
                batches: tens(count),
            }]));
    }
    // The driver is inside lane 1's read straight after skipping finished
    // lane 0, so the streak counts lane 0.
    reached_rx.recv_timeout(FINISH_DEADLINE).unwrap();
    assert!(h.chain.consumed.load(Ordering::Relaxed) < 2020);
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Retire(vec![LaneId(0)]));
    wait_for("the retire is delivered", FINISH_DEADLINE, || {
        logged(&h, "retire-delivered")
    });
    // A `poll_events` call after the delivering one means the retire's
    // `DropLanes` is on the control channel.
    let calls = h.shared.lock().unwrap().poll_timeouts.len();
    wait_for("the retire is sent to the thread", FINISH_DEADLINE, || {
        h.shared.lock().unwrap().poll_timeouts.len() > calls
    });
    go_tx.send(()).unwrap();
    wait_for("lane 2 is read to its end", FINISH_DEADLINE, || {
        h.chain.consumed.load(Ordering::Relaxed) == 2020
    });
    shut_down(h);
}

/// Pins that a thread parked on its finished lanes reads a lane added after
/// a wake, so the pass after each park starts its skip count afresh.
#[test]
fn a_lane_added_after_a_wake_is_read() {
    let h = start_finishing(parked_options(), None, &[0, 1], || ChainMode::Ok);
    assign(&h, &[(0, 1), (1, 1)]);
    wait_for("both lanes finish", FINISH_DEADLINE, || {
        polls_of(&h, 0) == 2 && polls_of(&h, 1) == 2
    });
    wake(&h, 1);
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Add(vec![LaneSpec {
            id: LaneId(2),
            partition: PartitionId(2),
            batches: tens(1),
        }]));
    wait_for("the new lane is read", FINISH_DEADLINE, || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    shut_down(h);
}

// ------------------------------------------------------- injected clock --

fn blocked_flush_chain(shared: Arc<ChainShared>, log: Arc<Mutex<SourceLog>>) -> FakeChain {
    FakeChain {
        shared,
        log,
        mode: ChainMode::FlushBlocked,
        batches_seen: 0,
    }
}

/// Options whose idle flush never fires within a test's advances.
fn no_idle_flush() -> RuntimeOptions {
    RuntimeOptions {
        idle_flush: Duration::from_secs(3600),
        ..test_options()
    }
}

/// Whether the log holds a `commit` after the revocation was delivered.
fn committed_after_revoke(h: &Harness) -> bool {
    let log = h.shared.lock().unwrap();
    log.log
        .iter()
        .position(|e| e == "revoke-delivered")
        .is_some_and(|at| log.log[at..].iter().any(|e| e == "commit"))
}

/// The commit tick fires once `checkpoint.interval` has passed on the
/// pipeline's clock, and not a millisecond before.
#[test]
fn the_commit_tick_fires_at_the_interval_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(600);
    let h = start_on_clock_with(cfg, no_idle_flush(), &clock, |runtime| runtime, ok_chain);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    let flushed = flushes(&h);

    clock.advance(Duration::from_secs(600) - Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(flushes(&h), flushed, "the tick fired early");
    assert!(h.shared.lock().unwrap().committed.is_empty());

    clock.advance(Duration::from_millis(1));
    wait_for("the tick commits", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });
    wait_for("the tick's FlushNow", Duration::from_secs(5), || {
        flushes(&h) > flushed
    });
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// Starts a runtime on `clock` that holds every ack for the whole run, so a
/// chase window closes with its acks outstanding.
fn start_holding_acks(
    cfg: PipelineConfig,
    clock: &Arc<crate::clock::TestClock>,
    customize: impl FnOnce(PipelineRuntime<FakeSource>) -> PipelineRuntime<FakeSource>,
) -> Harness {
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    start_on_clock_with(
        cfg,
        no_idle_flush(),
        clock,
        customize,
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held),
                release: Arc::clone(&release),
            },
            batches_seen: 0,
        },
    )
}

/// Hints `partition` and waits for the next pass, so a test may move the clock.
// The fake logs the hint as it hands it over, before the controller reads the
// clock for the window's end. Advancing earlier would move that end.
//
// `cfg(test)` is redundant for the compiler. cargo-mutants skips an item under
// exactly `cfg(test)`, and a mutant here only reopens that race.
#[cfg(test)]
fn hint(h: &Harness, partition: PartitionId) {
    let before = h.shared.lock().unwrap().hinted_at;
    h.script
        .lock()
        .unwrap()
        .push_back(Script::CommitReady(vec![partition]));
    wait_for("a poll after the hint", Duration::from_secs(5), || {
        let log = h.shared.lock().unwrap();
        log.hinted_at != before
            && log
                .hinted_at
                .is_some_and(|at| log.poll_timeouts.len() > at + 1)
    });
}

/// The timeout of the controller's latest poll.
fn last_poll(h: &Harness) -> Duration {
    *h.shared.lock().unwrap().poll_timeouts.last().unwrap()
}

/// Asserts that a hinted partition is polled at the tightened cadence until
/// `window` has passed on the pipeline's clock, and at the ordinary cadence
/// after.
// Redundant for the compiler. cargo-mutants skips an item under exactly
// `cfg(test)`, and a mutant here only empties the tests that call it.
#[cfg(test)]
fn assert_chase_closes_after(
    window: Duration,
    cfg: PipelineConfig,
    customize: impl FnOnce(PipelineRuntime<FakeSource>) -> PipelineRuntime<FakeSource>,
) {
    let clock = crate::clock::TestClock::frozen();
    let h = start_holding_acks(cfg, &clock, customize);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    hint(&h, PartitionId(0));

    clock.advance(window - Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(last_poll(&h), super::controller::FAST_COMMIT_POLL);

    clock.advance(Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(last_poll(&h), test_options().event_poll_timeout);

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A hinted partition is chased at the tightened cadence until its window
/// ends on the pipeline's clock, and at the ordinary cadence after.
#[test]
fn a_chase_window_closes_at_its_end_on_the_injected_clock() {
    let window = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    assert_chase_closes_after(window, cfg, |runtime| runtime.with_chase_window(window));
}

/// Pins the one commit interval `SourceEvent::CommitReady` promises a source.
#[test]
fn a_chase_window_lasts_one_commit_interval_by_default() {
    let interval = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = interval;
    assert_chase_closes_after(interval, cfg, |runtime| runtime);
}

/// Hinting a partition already being chased leaves the end of its window where
/// it was.
#[test]
fn a_repeated_hint_does_not_extend_a_chase() {
    let clock = crate::clock::TestClock::frozen();
    let window = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let h = start_holding_acks(cfg, &clock, |runtime| runtime.with_chase_window(window));
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    hint(&h, PartitionId(0));
    clock.advance(window - Duration::from_millis(1));
    hint(&h, PartitionId(0));

    clock.advance(Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(last_poll(&h), test_options().event_poll_timeout);

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A partition hinted during another's chase gets a whole window of its own.
#[test]
fn a_later_hint_gets_a_window_of_its_own() {
    let clock = crate::clock::TestClock::frozen();
    let window = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let h = start_holding_acks(cfg, &clock, |runtime| runtime.with_chase_window(window));
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
    ]));
    wait_for("the batches consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 20
    });
    hint(&h, PartitionId(0));
    clock.advance(window - Duration::from_millis(1));
    hint(&h, PartitionId(1));

    clock.advance(window - Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(last_poll(&h), super::controller::FAST_COMMIT_POLL);

    clock.advance(Duration::from_millis(1));
    passes(&h, 3);
    assert_eq!(last_poll(&h), test_options().event_poll_timeout);

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A commit that failed past a partition's window is retried by the periodic
/// tick, and not before it.
#[test]
fn a_failed_post_window_commit_is_retried_on_the_tick() {
    let clock = crate::clock::TestClock::frozen();
    let interval = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = interval;
    let h = start_on_clock_with(
        cfg,
        no_idle_flush(),
        &clock,
        |runtime| runtime.with_chase_window(Duration::ZERO),
        ok_chain,
    );
    h.shared.lock().unwrap().fail_commits = true;
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    hint(&h, PartitionId(0));
    wait_for("a failed commit", Duration::from_secs(5), || {
        logged(&h, "commit-failed")
    });

    h.shared.lock().unwrap().fail_commits = false;
    passes(&h, 3);
    assert!(h.shared.lock().unwrap().committed.is_empty());

    clock.advance(interval);
    wait_for("the tick retries", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// Past its window, a partition whose commit failed is not committed again
/// while another partition is chased inside its window, and the tick retries it.
/// Regression for #1595.
#[test]
fn a_failed_post_window_commit_waits_while_another_partition_is_chased() {
    let clock = crate::clock::TestClock::frozen();
    let window = Duration::from_secs(600);
    let interval = Duration::from_secs(3600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = interval;
    let h = start_on_clock_with(
        cfg,
        no_idle_flush(),
        &clock,
        |runtime| runtime.with_chase_window(window),
        ok_chain,
    );
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(&[]),
        },
    ]));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    hint(&h, PartitionId(0));
    clock.advance(window);
    wait_for("a failed commit", Duration::from_secs(5), || {
        logged(&h, "commit-failed")
    });
    passes(&h, 3);
    let failed = |h: &Harness| {
        h.shared
            .lock()
            .unwrap()
            .log
            .iter()
            .filter(|e| *e == "commit-failed")
            .count()
    };
    let before = failed(&h);

    hint(&h, PartitionId(1));
    passes(&h, 40);
    assert_eq!(
        failed(&h) - before,
        0,
        "commit attempts over forty passes with no advance"
    );

    h.shared.lock().unwrap().fail_commits = false;
    clock.advance(interval - window);
    wait_for("the tick retries", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&10)
    });

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// Past its window, a partition whose commit failed is committed ahead of the
/// tick when its position advances while another partition is chased inside its
/// window.
#[test]
fn a_post_window_advance_commits_while_another_partition_is_chased() {
    let clock = crate::clock::TestClock::frozen();
    let window = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let held: Arc<Mutex<Vec<crate::checkpoint::AckRef>>> = Arc::new(Mutex::new(Vec::new()));
    let release = Arc::new(AtomicBool::new(false));
    let held_c = Arc::clone(&held);
    let h = start_on_clock_with(
        cfg,
        no_idle_flush(),
        &clock,
        |runtime| runtime.with_chase_window(window),
        move |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::HoldAcks {
                held: Arc::clone(&held_c),
                release: Arc::clone(&release),
            },
            batches_seen: 0,
        },
    );
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(&[0..10, 10..20]),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(&[]),
        },
    ]));
    wait_for("the batches held", Duration::from_secs(5), || {
        held.lock().unwrap().len() == 2
    });
    hint(&h, PartitionId(0));
    clock.advance(window);
    passes(&h, 3);
    hint(&h, PartitionId(1));

    held.lock().unwrap().remove(0);
    wait_for("a failed commit", Duration::from_secs(5), || {
        logged(&h, "commit-failed")
    });

    h.shared.lock().unwrap().fail_commits = false;
    held.lock().unwrap().clear();
    wait_for("the advance commits", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(0)) == Some(&20)
    });

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// After the pass handling a revocation, a position the revoked partition failed
/// to commit waits for the tick while another partition's chase commits.
#[test]
fn a_revoked_partitions_failed_commit_waits_while_another_is_chased() {
    let clock = crate::clock::TestClock::frozen();
    let window = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let h = start_on_clock_with(
        cfg,
        no_idle_flush(),
        &clock,
        |runtime| runtime.with_chase_window(window),
        ok_chain,
    );
    h.shared.lock().unwrap().fail_commits = true;
    h.script.lock().unwrap().push_back(Script::Assign(vec![
        LaneSpec {
            id: LaneId(0),
            partition: PartitionId(0),
            batches: batches(std::slice::from_ref(&(0..10))),
        },
        LaneSpec {
            id: LaneId(1),
            partition: PartitionId(1),
            batches: batches(std::slice::from_ref(&(0..5))),
        },
    ]));
    wait_for("the batches consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 15
    });
    hint(&h, PartitionId(0));
    clock.advance(window);
    passes(&h, 3);
    hint(&h, PartitionId(1));
    wait_for("a failed commit", Duration::from_secs(5), || {
        logged(&h, "commit-failed")
    });

    let flushed = h.shared.lock().unwrap().flush_commits;
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    wait_for("the revocation", Duration::from_secs(5), || {
        h.shared.lock().unwrap().flush_commits > flushed
    });
    // The pass handling the revocation finishes before the next poll, and it is
    // the last that may send the revoked position.
    passes(&h, 2);
    h.shared.lock().unwrap().fail_commits = false;

    wait_for("the chase commits", Duration::from_secs(5), || {
        h.shared.lock().unwrap().committed.get(&PartitionId(1)) == Some(&5)
    });
    assert!(
        !h.shared
            .lock()
            .unwrap()
            .committed
            .contains_key(&PartitionId(0)),
        "a revoked partition's failed position was sent again"
    );

    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A watermark stuck past `watermark_stuck` on the pipeline's clock turns
/// `/healthz` to 503 while data is pending, and not at the threshold itself.
#[test]
fn a_stuck_watermark_turns_liveness_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    let interval = Duration::from_secs(60);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = interval;
    cfg.checkpoint.stalled_fail_after = Duration::from_secs(3600);
    let h = start_on_clock_with(
        cfg,
        no_idle_flush(),
        &clock,
        |runtime| runtime,
        |shared, log| FakeChain {
            shared,
            log,
            mode: ChainMode::FailAckAtBatch(1),
            batches_seen: 0,
        },
    );
    assign_one_lane(&h, &[0..10, 10..20, 20..30]);
    wait_for("every batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 30
    });
    passes(&h, 3);
    let addr = h.admin_addr.0.lock().unwrap().expect("admin server bound");
    let healthz = || {
        // A lane poll after the advance beats the driver's heart on the
        // moved clock, so only the watermark decides the answer.
        let polled = polls_of(&h, 0);
        wait_for("a lane poll", Duration::from_secs(5), || {
            polls_of(&h, 0) > polled
        });
        http_status(addr, "/healthz")
    };

    // The default `watermark_stuck` is 300 s.
    for _ in 0..5 {
        tick(&h, &clock, interval);
    }
    assert_eq!(healthz(), 200, "a watermark at the threshold is not stuck");
    tick(&h, &clock, interval);
    assert_eq!(healthz(), 503);
    h.shutdown.trigger();
    let _ = h.join.join().unwrap();
}

/// The status code of a `GET path` to the admin server at `addr`.
fn http_status(addr: std::net::SocketAddr, path: &str) -> u16 {
    use std::io::{Read, Write};
    let mut stream = std::net::TcpStream::connect(addr).expect("connect to admin");
    write!(
        stream,
        "GET {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
    )
    .expect("send request");
    let mut response = String::new();
    stream.read_to_string(&mut response).expect("read response");
    response
        .split_whitespace()
        .nth(1)
        .and_then(|code| code.parse().ok())
        .unwrap_or_else(|| panic!("no status line in {response:?}"))
}

/// A revocation whose flush stays blocked is waited on until its drain
/// deadline on the stop clock, then completes with the run still running,
/// and the driver abandons the flush there so the run can stop.
#[test]
fn a_blocked_revocation_flush_ends_at_its_deadline_on_the_stop_clock() {
    let clock = crate::clock::TestClock::frozen();
    let drain = Duration::from_secs(600);
    let mut cfg = stop_config(drain);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let stop_clock = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        no_idle_flush(),
        |_| {},
        move |runtime, _| runtime.with_stop_clock(stop_clock),
        blocked_flush_chain,
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    wait_for(
        "the revocation flush running",
        Duration::from_secs(5),
        || {
            h.shared
                .lock()
                .unwrap()
                .log
                .iter()
                .any(|e| e == "revoke-delivered")
                && flushes(&h) >= 3
        },
    );

    clock.advance(drain - Duration::from_millis(1));
    let flushed = flushes(&h);
    wait_for("the flush still retrying", Duration::from_secs(5), || {
        flushes(&h) >= flushed + 3
    });
    assert!(!committed_after_revoke(&h), "the revocation ended early");

    clock.advance(Duration::from_millis(1));
    wait_for("the revocation completes", Duration::from_secs(5), || {
        committed_after_revoke(&h)
    });
    assert!(
        !h.join.is_finished(),
        "a lapsed revocation must not end the run"
    );
    h.shutdown.trigger();
    // The stop's flush blocks too, until its own deadline on the same clock.
    wait_for("run() to return", Duration::from_secs(10), || {
        clock.advance(drain);
        h.join.is_finished()
    });
}

/// A stop's blocked flush ends at the lanes' deadline on the stop clock, so
/// the run returns once that clock reaches it.
#[test]
fn a_blocked_flush_ends_at_the_lanes_deadline_on_the_stop_clock() {
    let clock = crate::clock::TestClock::frozen();
    let drain = Duration::from_secs(600);
    let mut cfg = stop_config(drain);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let stop_clock = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        no_idle_flush(),
        |_| {},
        move |runtime, _| runtime.with_stop_clock(stop_clock),
        blocked_flush_chain,
    );
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    let flushed = flushes(&h);
    h.shutdown.trigger();
    wait_for("the drain's flush retrying", Duration::from_secs(5), || {
        flushes(&h) >= flushed + 3
    });
    clock.advance(drain);
    joined(&h);
}

/// The idle flush fires once a lull reaches `idle_flush` on the pipeline's
/// clock, and once per lull.
#[test]
fn idle_flush_fires_once_at_its_lull_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    let lull = Duration::from_secs(600);
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let options = RuntimeOptions {
        idle_flush: lull,
        ..test_options()
    };
    let h = start_on_clock_with(cfg, options, &clock, |runtime| runtime, ok_chain);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    let lane_polls = |n: usize| {
        let target = polls_of(&h, 0) + n;
        wait_for("lane polls", Duration::from_secs(5), || {
            polls_of(&h, 0) >= target
        });
    };

    clock.advance(lull - Duration::from_millis(1));
    lane_polls(3);
    assert_eq!(flushes(&h), 0, "the idle flush fired early");

    clock.advance(Duration::from_millis(1));
    wait_for("the idle flush", Duration::from_secs(5), || {
        flushes(&h) == 1
    });

    clock.advance(lull);
    lane_polls(3);
    assert_eq!(flushes(&h), 1, "one idle flush per lull");
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A startup error after the drivers spawn stops them with a flush deadline
/// on the stop clock, so a blocked flush ends once that clock reaches the
/// drain grace.
#[test]
fn a_startup_error_ends_a_blocked_flush_at_the_grace_on_the_stop_clock() {
    let occupied = std::net::TcpListener::bind("127.0.0.1:0").expect("bind probe port");
    let clock = crate::clock::TestClock::frozen();
    let drain = Duration::from_secs(600);
    let mut cfg = stop_config(drain);
    cfg.admin.listen = Some(occupied.local_addr().unwrap());
    let stop_clock = Arc::clone(&clock);
    let h = start_prepared(
        cfg,
        no_idle_flush(),
        |_| {},
        move |runtime, _| runtime.with_stop_clock(stop_clock),
        blocked_flush_chain,
    );
    wait_for("the stop's flush retrying", Duration::from_secs(5), || {
        flushes(&h) >= 3
    });
    assert!(!h.join.is_finished(), "the flush ended before the grace");
    clock.advance(drain);
    joined(&h);
    let result = h.join.join().unwrap();
    assert!(
        matches!(result, Err(super::runtime::StartError::AdminBind { .. })),
        "{result:?}"
    );
    drop(occupied);
}

/// Start one pipeline thread on the pipeline clock `clock`, with an idle
/// flush after `lull`, no commit tick within the test's advances, and the
/// lanes in `finishing` reporting finished once they run dry.
fn start_idle_on_clock(
    clock: &Arc<crate::clock::TestClock>,
    lull: Duration,
    finishing: &[u32],
) -> Harness {
    let mut cfg = test_config(1);
    cfg.checkpoint.interval = Duration::from_secs(3600);
    let options = RuntimeOptions {
        idle_flush: lull,
        ..test_options()
    };
    let finishing: Vec<LaneId> = finishing.iter().map(|&id| LaneId(id)).collect();
    let clock = Arc::clone(clock);
    start_prepared(
        cfg,
        options,
        move |log| log.finishing.extend(finishing),
        move |runtime, _| runtime.with_clock(clock),
        ok_chain,
    )
}

/// A thread whose every lane is finished flushes once its lull reaches
/// `idle_flush` on the pipeline's clock.
#[test]
fn a_finished_thread_flushes_at_its_lull_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    let lull = Duration::from_secs(600);
    let h = start_idle_on_clock(&clock, lull, &[0]);
    assign(&h, &[(0, 1)]);
    wait_for("the lane finishes", Duration::from_secs(5), || {
        polls_of(&h, 0) == 2
    });
    let flushed = flushes(&h);
    clock.advance(lull);
    wait_for("the idle flush", Duration::from_secs(5), || {
        flushes(&h) > flushed
    });
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// A thread left with no lanes flushes once its lull reaches `idle_flush` on
/// the pipeline's clock.
#[test]
fn a_thread_with_no_lanes_flushes_at_its_lull_on_the_injected_clock() {
    let clock = crate::clock::TestClock::frozen();
    let lull = Duration::from_secs(600);
    let h = start_idle_on_clock(&clock, lull, &[]);
    assign_one_lane(&h, std::slice::from_ref(&(0..10)));
    wait_for("the batch consumed", Duration::from_secs(5), || {
        h.chain.consumed.load(Ordering::Relaxed) == 10
    });
    h.script
        .lock()
        .unwrap()
        .push_back(Script::Revoke(vec![LaneId(0)]));
    // The commit after the revocation follows the drain's own flush.
    wait_for("the revocation commits", Duration::from_secs(5), || {
        committed_after_revoke(&h)
    });
    let flushed = flushes(&h);
    clock.advance(lull);
    wait_for("the idle flush", Duration::from_secs(5), || {
        flushes(&h) > flushed
    });
    h.shutdown.trigger();
    h.join.join().unwrap().unwrap();
}

/// Pins that a sink probe panicking after a successful probe turns readiness
/// off, so `/readyz` does not keep the last answer once nothing updates it.
#[test]
fn a_panicking_probe_does_not_leave_readiness_stale() {
    let health = crate::admin::HealthState::new(0, crate::admin::HealthThresholds::default());
    health.set_assignment_received(true);
    let calls = Arc::new(AtomicUsize::new(0));
    let probe: super::SinkProbeFn = {
        let calls = Arc::clone(&calls);
        Box::new(move || {
            assert_eq!(calls.fetch_add(1, Ordering::SeqCst), 0, "the probe panics");
            Box::pin(async { Ok(()) })
        })
    };
    let joined = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .start_paused(true)
        .build()
        .unwrap()
        .block_on(async {
            let probing = crate::task::spawn(
                "sink-probe",
                super::runtime::probe_sinks(probe, Arc::clone(&health)),
            );
            // Time is paused, so the 30 s recheck passes at once.
            tokio::time::timeout(Duration::from_secs(120), probing)
                .await
                .expect("the probe loop ends within the deadline")
        });
    assert!(joined.unwrap_err().is_panic());
    assert_eq!(calls.load(Ordering::SeqCst), 2);
    assert!(!health.ready(), "readiness kept the last probe's answer");
}