spate-s3 0.4.0

//! The data plane: one [`S3Lane`] per lane, polled on a pipeline thread.
//!
//! A lane pulls byte chunks from its fetcher's channel, decompresses and
//! frames them on the pipeline thread (see [`framer`](crate::framer)),
//! assigns each record its composite offset, and hands out borrowed
//! payload batches with one [`AckRef`] each.
//!
//! These rules govern correctness:
//!
//! - **End-of-input is only decided by a `poll` that returns `Ok(None)`**
//!   after observing the channel closed with nothing buffered. The
//!   driver's poll→push→poll sequencing on one thread then guarantees the
//!   lane's final batch was fully pushed downstream before that decision
//!   records the terminal watermark.
//! - **Blocking is bounded.** When idle the lane waits on the channel via
//!   the I/O runtime with the poll timeout applied, so it never busy-spins
//!   and never parks longer than the driver allows. The poll that decides
//!   end-of-input returns at once, and from then on the lane reports
//!   [`finished`](SourceLane::finished), so the pipeline thread stops
//!   polling it.
//! - **Poison never surfaces as a poll error.** A lane `poll` error is
//!   terminal for the whole pipeline, so object-level failures (deleted,
//!   overwritten, corrupt, or unreadable objects) are reported through the
//!   [`PoisonReport`] side channel instead and the lane goes quiescent;
//!   only instance-level failures (credentials, wiring) return `Err`.
//! - **An overwrite ends the lane cleanly.** A notification unit's object
//!   overwritten under its pin is reported at once, and the lane emits the
//!   records it framed before the failure, then decides end-of-input, so
//!   the terminal watermark is one past the last record emitted.

use crate::config::Compression;
use crate::fetch::{ChunkMsg, SplitFailure};
use crate::framer::{Codec, FramerFactory, ObjectFramer};
use crate::lane_ctx::{PoisonKind, PoisonReport, SplitTracker};
use crate::metrics::S3Metrics;
use crate::offset::{MAX_RECORD_INDEX, Position};
use futures_util::future::{Either, select};
use spate_core::checkpoint::{AckIssuer, AckRef};
use spate_core::clock::tokio::Clock;
use spate_core::error::{ErrorClass, SourceError};
use spate_core::feed::FeedWaker;
use spate_core::record::{PartitionId, RawPayload};
use spate_core::source::{LaneId, PayloadBatch, SourceLane};
use spate_core::task::TaskHandle;
use std::pin::pin;
use std::sync::Arc;
use std::sync::mpsc as std_mpsc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TryRecvError;

/// The control-plane waker a lane signals when it ends or fails. It fires on
/// the first call only. End-of-input and poison share the flag, since either
/// one ends the lane. One signal suffices: the slot holds it until a park
/// consumes it, and the reader re-reads the lane's state on every pass.
struct LaneWaker {
    waker: FeedWaker,
    sent: bool,
}

impl LaneWaker {
    /// Signal the control plane. Never blocks.
    fn wake(&mut self) {
        if !self.sent {
            self.sent = true;
            self.waker.wake();
        }
    }
}

/// The object currently being framed.
struct CurrentObject {
    ordinal: u32,
    key: String,
    /// Record index the next emitted record gets.
    next_record: u64,
    /// Event time stamped on this object's records.
    event_time_ms: i64,
}

/// One framed record, owned by the lane across the batch's lifetime.
struct HeldRecord {
    offset: i64,
    event_time_ms: i64,
    bytes: Vec<u8>,
}

/// Data-plane pollable unit of the S3 source: owns one slice's chunk
/// stream and framing state.
pub struct S3Lane {
    id: LaneId,
    partition: PartitionId,
    rx: mpsc::Receiver<ChunkMsg>,
    /// The fetcher sending on `rx`, aborted when the lane drops. A fetcher in
    /// retry backoff would otherwise notice the closed channel only at its
    /// next send. It holds no receipt or ack issuer, so the abort loses
    /// nothing durable.
    _fetcher: TaskHandle<()>,
    handle: tokio::runtime::Handle,
    /// Times each poll's receive deadline.
    clock: Arc<dyn Clock>,
    issuer: AckIssuer,
    compression: Compression,
    framer: ObjectFramer,
    /// Committed resume position; consumed at the first `ObjectStart`.
    resume: Option<Position>,
    /// Records of the resume object still to discard (replayed committed
    /// records).
    pending_discard: u64,
    current: Option<CurrentObject>,
    /// The object ended but framed records are still queued; finalized
    /// once the queue drains.
    pending_end: bool,
    held: Vec<HeldRecord>,
    /// Carries the terminal watermark to the control plane at the
    /// end-of-input decision.
    tracker: Arc<SplitTracker>,
    /// Side channel for object-level failures (see the module docs).
    poison_tx: std_mpsc::Sender<PoisonReport>,
    /// Wakes the control plane. Signalled on the two edges it would
    /// otherwise only notice between waits: end-of-input, and poison.
    /// Never on the per-record path.
    waker: LaneWaker,
    /// The split hit poison: everything undelivered was discarded and
    /// every later poll idles with `Ok(None)` until the lane is retired.
    poisoned: bool,
    /// The current object was overwritten under its pin: its framed records
    /// drain, its unframed bytes are dropped, and it is not counted complete.
    truncated: bool,
    /// One past the last emitted record's offset, which becomes the terminal
    /// watermark `T` once end-of-input is observed. Starts at the resume watermark
    /// (0 fresh), so a tenancy that emits nothing terminates exactly where
    /// it began.
    watermark_candidate: i64,
    /// Sticky terminal failure: every later poll re-reports it.
    failed: Option<(ErrorClass, String)>,
    metrics: Option<S3Metrics>,
    /// Decoded bytes already counted into the metrics.
    decoded_reported: u64,
}

impl std::fmt::Debug for S3Lane {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("S3Lane")
            .field("id", &self.id)
            .field("partition", &self.partition)
            .finish()
    }
}

impl S3Lane {
    #[expect(
        clippy::too_many_arguments,
        reason = "assembled in one place by the source's lane builder"
    )]
    pub(crate) fn new(
        id: LaneId,
        partition: PartitionId,
        rx: mpsc::Receiver<ChunkMsg>,
        fetcher: TaskHandle<()>,
        handle: tokio::runtime::Handle,
        clock: Arc<dyn Clock>,
        issuer: AckIssuer,
        compression: Compression,
        make_framer: FramerFactory,
        resume: Option<Position>,
        tracker: Arc<SplitTracker>,
        poison_tx: std_mpsc::Sender<PoisonReport>,
        waker: FeedWaker,
        metrics: Option<S3Metrics>,
    ) -> S3Lane {
        let watermark_candidate = resume.map_or(0, |p| {
            p.encode()
                .expect("a decoded resume position always re-encodes")
        });
        S3Lane {
            id,
            partition,
            rx,
            _fetcher: fetcher,
            handle,
            clock,
            issuer,
            compression,
            framer: ObjectFramer::new(make_framer),
            resume,
            pending_discard: 0,
            current: None,
            pending_end: false,
            held: Vec::new(),
            tracker,
            poison_tx,
            waker: LaneWaker { waker, sent: false },
            poisoned: false,
            truncated: false,
            watermark_candidate,
            failed: None,
            metrics,
            decoded_reported: 0,
        }
    }

    /// Record a terminal failure and return it. Later polls re-report it.
    fn fail(&mut self, class: ErrorClass, reason: String) -> SourceError {
        self.failed = Some((class, reason.clone()));
        SourceError::Client { class, reason }
    }

    /// Report object-level poison and go quiescent. Everything undelivered
    /// is discarded; none of it was acked, so replay by the split's next
    /// owner cannot lose data. Every later poll idles with `Ok(None)` until
    /// the control plane retires the lane.
    fn poison(&mut self, kind: PoisonKind, reason: String) {
        self.held.clear();
        self.current = None;
        self.pending_end = false;
        self.pending_discard = 0;
        self.poisoned = true;
        // The receiver disappearing (shutdown) makes the report moot.
        let _ = self.poison_tx.send(PoisonReport {
            partition: self.partition,
            kind,
            reason,
            overwrite: None,
        });
        // The report is only read between control-plane waits; wake so the
        // control plane handles it at once.
        self.waker.wake();
    }

    /// Report that the current object was overwritten under its pin, and
    /// end it once its framed records drain. The fetcher sends nothing more,
    /// so the closed channel then takes the clean end-of-input path.
    fn overwritten(&mut self, overwrite: Box<crate::fetch::Overwrite>) {
        self.truncated = true;
        self.pending_end = false;
        // The receiver disappearing (shutdown) makes the report moot.
        let _ = self.poison_tx.send(PoisonReport {
            partition: self.partition,
            kind: PoisonKind::EtagDrift,
            reason: format!("\"{}\" was overwritten while it was read", overwrite.key),
            overwrite: Some(overwrite),
        });
        self.waker.wake();
    }

    /// Move framed records into `held`, assigning composite offsets, until
    /// the framer queue is empty or the batch is full. May poison the
    /// split (an object over the per-object record limit); the caller
    /// checks `self.poisoned` after every call.
    fn drain_framer(&mut self, max_records: usize) -> Result<(), SourceError> {
        while self.held.len() < max_records {
            let Some(bytes) = self.framer.pop_record() else {
                break;
            };
            if self.pending_discard > 0 {
                self.pending_discard -= 1;
                continue;
            }
            let cur = self
                .current
                .as_mut()
                .expect("framed records only exist within an object");
            if cur.next_record > MAX_RECORD_INDEX {
                // A property of the object's content: it will overflow on
                // every delivery, which is what the feed's dead-letter cap is for.
                let key = cur.key.clone();
                self.poison(
                    PoisonKind::Undecodable,
                    format!(
                        "object \"{key}\" holds more than {} records, the composite-offset \
                     limit per object",
                        MAX_RECORD_INDEX + 1
                    ),
                );
                return Ok(());
            }
            let pos = Position {
                ordinal: cur.ordinal,
                record: cur.next_record,
            };
            let offset = match pos.encode() {
                Ok(o) => o,
                Err(e) => return Err(self.fail(ErrorClass::Fatal, e.to_string())),
            };
            cur.next_record += 1;
            debug_assert!(
                offset >= self.watermark_candidate,
                "offsets must be monotonic"
            );
            self.watermark_candidate = offset + 1;
            self.held.push(HeldRecord {
                offset,
                event_time_ms: cur.event_time_ms,
                bytes,
            });
        }
        Ok(())
    }

    /// Complete a pending object end once its records have drained. May
    /// poison the split (content drift); the caller checks
    /// `self.poisoned`.
    fn finalize_object(&mut self) {
        debug_assert!(self.pending_end && self.framer.queued() == 0);
        let cur = self.current.take().expect("finalize without an object");
        self.pending_end = false;
        if self.pending_discard > 0 {
            // The object frames fewer records than were committed against
            // it; its content changed underneath the pin.
            self.poison(
                PoisonKind::EtagDrift,
                format!(
                    "object \"{}\" ended {} records short of its committed position — \
                 its content changed under the checkpoint (the key set and object \
                 contents must stay frozen for the backfill's lifetime)",
                    cur.key, self.pending_discard
                ),
            );
            return;
        }
        let counted = self.tracker.object_done();
        if let Some(m) = &self.metrics {
            m.objects_completed.increment(1);
            if counted {
                m.objects_remaining.decrement(1.0);
            }
        }
        self.report_decoded();
    }

    /// Apply one fetcher message to the framing state.
    fn on_msg(&mut self, msg: ChunkMsg) -> Result<(), SourceError> {
        match msg {
            ChunkMsg::ObjectStart {
                ordinal,
                key,
                last_modified_ms,
            } => {
                debug_assert!(
                    self.current.is_none() && !self.pending_end,
                    "ObjectStart while the previous object is open"
                );
                let codec = Codec::resolve(self.compression, &key);
                if let Err(e) = self.framer.begin_object(codec) {
                    self.poison(
                        PoisonKind::Undecodable,
                        format!("starting decode of \"{key}\": {e}"),
                    );
                    return Ok(());
                }
                // The committed watermark's record index is how many
                // records of the resume object are already committed:
                // replay them silently.
                let discard = match self.resume.take() {
                    Some(pos) if pos.ordinal == ordinal => pos.record,
                    Some(pos) => {
                        return Err(self.fail(
                            ErrorClass::Fatal,
                            format!(
                                "fetcher started at ordinal {ordinal} but the committed \
                                 resume position is ordinal {} — internal wiring bug",
                                pos.ordinal
                            ),
                        ));
                    }
                    None => 0,
                };
                self.pending_discard = discard;
                self.current = Some(CurrentObject {
                    ordinal,
                    key,
                    next_record: discard,
                    event_time_ms: last_modified_ms,
                });
                Ok(())
            }
            ChunkMsg::Chunk(bytes) => {
                if let Some(m) = &self.metrics {
                    m.bytes_read.increment(bytes.len() as u64);
                }
                if let Err(e) = self.framer.push_chunk(bytes) {
                    self.poison_decode(&e);
                }
                Ok(())
            }
            ChunkMsg::ObjectEnd => {
                if let Err(e) = self.framer.finish_object() {
                    let key = self
                        .current
                        .as_ref()
                        .map_or_else(String::new, |c| c.key.clone());
                    self.poison(
                        PoisonKind::Undecodable,
                        format!("finishing decode of \"{key}\": {e} (truncated object?)"),
                    );
                    return Ok(());
                }
                self.pending_end = true;
                Ok(())
            }
            ChunkMsg::LaneFailed(SplitFailure::Poison(kind, reason)) => {
                self.poison(kind, reason);
                Ok(())
            }
            ChunkMsg::LaneFailed(SplitFailure::Overwritten(overwrite)) => {
                self.overwritten(overwrite);
                Ok(())
            }
            ChunkMsg::LaneFailed(SplitFailure::Fatal(e)) => {
                let (class, reason) = match e {
                    SourceError::Client { class, reason } => (class, reason),
                    other => (ErrorClass::Fatal, other.to_string()),
                };
                Err(self.fail(class, reason))
            }
        }
    }

    /// Poison the split on a decode error in the current object's chunks.
    fn poison_decode(&mut self, e: &std::io::Error) {
        let key = self
            .current
            .as_ref()
            .map_or_else(String::new, |c| c.key.clone());
        self.poison(
            PoisonKind::Undecodable,
            format!("decoding \"{key}\": {e} (corrupt or truncated object?)"),
        );
    }

    /// Report the decoded-bytes delta at a batch boundary.
    fn report_decoded(&mut self) {
        if let Some(m) = &self.metrics {
            let total = self.framer.decoded_bytes();
            m.bytes_decoded
                .increment(total.saturating_sub(self.decoded_reported));
            self.decoded_reported = total;
        }
    }

    /// Signal the lane's control-plane waker as end-of-input does.
    #[cfg(test)]
    pub(crate) fn wake_control(&mut self) {
        self.waker.wake();
    }
}

impl SourceLane for S3Lane {
    type Batch<'a> = S3Batch<'a>;

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

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

    fn finished(&self) -> bool {
        self.tracker.terminal().is_some()
    }

    fn poll(
        &mut self,
        max_records: usize,
        timeout: Duration,
    ) -> Result<Option<S3Batch<'_>>, SourceError> {
        if let Some((class, reason)) = &self.failed {
            return Err(SourceError::Client {
                class: *class,
                reason: reason.clone(),
            });
        }
        if self.poisoned {
            // Quiescent until the control plane retires the lane; bounded
            // idle, never a busy-spin, never an error (a poll error would
            // fail the pipeline; poison must not). Real time, so a frozen
            // test clock cannot wedge a retired lane.
            std::thread::sleep(timeout);
            return Ok(None);
        }
        self.held.clear();
        let deadline = self.clock.now() + timeout;
        // Whether this poll has already tried the channel or decoded a step
        // of pending input. The deadline
        // must not short-circuit the *first* receive: with `timeout` zero
        // (the driver's head-of-line rotation) `now >= deadline` is true
        // immediately, and bailing before `try_recv` would report the lane
        // empty while a chunk sits ready in its channel, starving it for
        // as long as a sibling lane keeps the rotation fed.
        let mut recv_attempted = false;

        loop {
            self.drain_framer(max_records)?;
            if self.pending_end && !self.poisoned && self.framer.queued() == 0 {
                self.finalize_object();
            }
            if self.truncated && self.framer.queued() == 0 {
                // The bytes after the last framed record belong to a version
                // the store no longer holds; nothing frames them.
                self.truncated = false;
                self.current = None;
            }
            if self.poisoned {
                return Ok(None);
            }
            if self.held.len() >= max_records {
                break;
            }
            // `drain_framer` returns early only on a full batch (broken
            // above) or an empty queue, and an empty queue finalizes a
            // pending object end, so no object can still be pending here.
            debug_assert!(!self.pending_end, "pending object end past the drain");
            // The deadline caps the whole poll, not just the idle wait: a
            // stream of chunks that frames no records (one enormous line,
            // whitespace floods) must still return control to the driver —
            // heartbeats and shutdown are processed between polls. Checked
            // only once a receive or a decode step has been attempted, so a
            // zero-timeout poll still decodes one step of a pending chunk, or
            // with none pending consumes one ready message (and the framing
            // pass at the top of the next iteration turns it into records),
            // before this bails.
            if recv_attempted && self.clock.now() >= deadline {
                if self.held.is_empty() {
                    return Ok(None);
                }
                break;
            }
            recv_attempted = true;
            // The rest of the last chunk is decoded before the next message
            // is taken, one bounded step per iteration, so the drain at the
            // top of the loop keeps the framer's queue short however far the
            // chunk expands.
            if self.framer.has_pending_input() {
                if let Err(e) = self.framer.pump() {
                    self.poison_decode(&e);
                    return Ok(None);
                }
                continue;
            }
            let msg = match self.rx.try_recv() {
                Ok(m) => Some(m),
                Err(TryRecvError::Empty) => {
                    if !self.held.is_empty() {
                        break; // hand over what we have instead of waiting
                    }
                    let remaining = deadline.saturating_duration_since(self.clock.now());
                    if remaining.is_zero() {
                        return Ok(None);
                    }
                    let rx = &mut self.rx;
                    let clock = &self.clock;
                    // The sleep is built inside `block_on` because a tokio
                    // timer registers with the runtime it is created in.
                    // Blocks this pipeline thread only, bounded by the poll
                    // timeout.
                    let received = self.handle.block_on(async {
                        let sleep = clock.sleep_until(deadline);
                        match select(pin!(rx.recv()), sleep).await {
                            Either::Left((received, _)) => Some(received),
                            Either::Right(_) => None,
                        }
                    });
                    match received {
                        Some(Some(m)) => Some(m),
                        Some(None) => None,
                        None => return Ok(None), // timed out idle
                    }
                }
                Err(TryRecvError::Disconnected) => None,
            };
            match msg {
                Some(m) => {
                    self.on_msg(m)?;
                    if self.poisoned {
                        return Ok(None);
                    }
                }
                None => {
                    // Channel closed. Mid-object it means the fetcher died
                    // without reporting (it always sends LaneFailed on
                    // error), so only a clean end-of-input may pass.
                    if self.current.is_some() {
                        return Err(self.fail(
                            ErrorClass::Fatal,
                            "object stream ended mid-object: the fetcher terminated \
                             unexpectedly"
                                .into(),
                        ));
                    }
                    if self.framer.queued() > 0 {
                        continue; // drain the tail first
                    }
                    if !self.held.is_empty() {
                        break; // final batch now; the decision on the next poll
                    }
                    // Nothing buffered anywhere and no more input: this
                    // poll's Ok(None) is the end-of-input decision (see
                    // module docs). Everything emitted was already handed
                    // out, so `watermark_candidate` is the terminal
                    // watermark. From here on the lane reports `finished`.
                    self.tracker.set_terminal(self.watermark_candidate);
                    // Completion is decided here, on a pipeline thread, and
                    // the control plane reads it between waits. The terminal
                    // is set first so the woken control plane sees it.
                    self.waker.wake();
                    return Ok(None);
                }
            }
        }

        self.report_decoded();
        let last = self
            .held
            .last()
            .expect("non-empty batch at emit time")
            .offset;
        let ack = self.issuer.issue(self.partition, last);
        Ok(Some(S3Batch {
            records: &self.held,
            idx: 0,
            partition: self.partition,
            ack,
        }))
    }
}

/// One poll's records, borrowed from the lane's held buffer.
pub struct S3Batch<'a> {
    records: &'a [HeldRecord],
    idx: usize,
    partition: PartitionId,
    ack: AckRef,
}

impl std::fmt::Debug for S3Batch<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("S3Batch")
            .field("records", &self.records.len())
            .field("idx", &self.idx)
            .finish()
    }
}

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

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::testutil::TestLineFramer;
    use spate_core::checkpoint::Checkpointer;
    use spate_core::clock::tokio::{SystemClock, TestClock};
    use spate_core::feed::wake_channel;

    fn runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap()
    }

    struct LaneRig {
        lane: S3Lane,
        tx: Option<mpsc::Sender<ChunkMsg>>,
        tracker: Arc<SplitTracker>,
        poison_rx: std_mpsc::Receiver<PoisonReport>,
        _rt: tokio::runtime::Runtime,
    }

    fn rig(resume: Option<Position>) -> LaneRig {
        rig_with(resume, wake_channel().0, || Arc::new(SystemClock))
    }

    /// A rig whose lane times its receive deadline on a frozen clock.
    fn frozen_rig() -> (LaneRig, Arc<TestClock>) {
        let mut frozen = None;
        let rig = rig_with(None, wake_channel().0, || {
            let clock = TestClock::frozen();
            frozen = Some(Arc::clone(&clock));
            clock
        });
        (rig, frozen.expect("the rig built the clock"))
    }

    /// A rig over a lane built with `waker` and the clock `make_clock`
    /// returns, which runs inside the rig's runtime so a tokio clock reads
    /// that runtime's time.
    fn rig_with(
        resume: Option<Position>,
        waker: FeedWaker,
        make_clock: impl FnOnce() -> Arc<dyn Clock>,
    ) -> LaneRig {
        let rt = runtime();
        let clock = {
            let _entered = rt.enter();
            make_clock()
        };
        let (tx, rx) = mpsc::channel(64);
        let checkpointer = Checkpointer::new();
        let tracker = Arc::new(SplitTracker::new());
        let (poison_tx, poison_rx) = std_mpsc::channel();
        // The rig sends the chunks itself, so the lane's fetcher only waits.
        let fetcher = spate_core::task::spawn_on("s3-fetcher", std::future::pending(), rt.handle());
        let lane = S3Lane::new(
            LaneId(0),
            PartitionId(0),
            rx,
            fetcher,
            rt.handle().clone(),
            clock,
            checkpointer.handle(),
            Compression::Auto,
            Arc::new(|| Box::new(TestLineFramer::new(1 << 20))),
            resume,
            Arc::clone(&tracker),
            poison_tx,
            waker,
            None,
        );
        LaneRig {
            lane,
            tx: Some(tx),
            tracker,
            poison_rx,
            _rt: rt,
        }
    }

    fn send(rig: &LaneRig, msg: ChunkMsg) {
        rig.tx.as_ref().unwrap().try_send(msg).unwrap();
    }

    fn start(rig: &LaneRig, ordinal: u32, key: &str) {
        send(
            rig,
            ChunkMsg::ObjectStart {
                ordinal,
                key: key.into(),
                last_modified_ms: 1_000,
            },
        );
    }

    /// Poll and collect `(offset, bytes)` pairs of the returned batch.
    fn poll_batch(lane: &mut S3Lane, max: usize) -> Option<Vec<(i64, Vec<u8>)>> {
        let batch = lane
            .poll(max, Duration::from_millis(50))
            .expect("poll must succeed");
        batch.map(|mut b| {
            let mut out = Vec::new();
            while let Some(p) = b.next_payload() {
                out.push((p.offset, p.bytes.to_vec()));
            }
            out
        })
    }

    /// A small zstd chunk that decodes to a million two-byte records leaves
    /// at most one decode step's records queued after each poll, and every
    /// record is still framed.
    #[test]
    fn a_compressed_chunk_queues_one_step_of_records_per_poll() {
        let lines = 1usize << 20;
        let chunk = zstd::encode_all(&b"a\n".repeat(lines)[..], 19).unwrap();
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson.zst");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from(chunk)));
        send(&r, ChunkMsg::ObjectEnd);
        let mut framed = 0;
        while framed < lines {
            framed += poll_batch(&mut r.lane, 512).expect("records remain").len();
            let queued_bytes = 2 * r.lane.framer.queued() as u64;
            assert!(
                queued_bytes <= crate::framer::PUMP_DECODED_BYTES / 2,
                "{queued_bytes} decoded bytes queued after {framed} records"
            );
        }
        assert_eq!(framed, lines);
    }

    #[test]
    fn frames_records_with_composite_offsets_across_objects() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\na2\n")));
        send(&r, ChunkMsg::ObjectEnd);
        start(&r, 1, "p/b.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"b1\n")));
        send(&r, ChunkMsg::ObjectEnd);
        let records = poll_batch(&mut r.lane, 512).unwrap();
        let expect = |ord: u32, rec: u64| {
            Position {
                ordinal: ord,
                record: rec,
            }
            .encode()
            .unwrap()
        };
        assert_eq!(
            records,
            vec![
                (expect(0, 0), b"a1".to_vec()),
                (expect(0, 1), b"a2".to_vec()),
                (expect(1, 0), b"b1".to_vec()),
            ],
            "a batch may span objects; offsets stay monotonic"
        );
    }

    #[test]
    fn terminal_watermark_is_recorded_on_the_poll_after_the_final_batch() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"only\n")));
        send(&r, ChunkMsg::ObjectEnd);
        r.tx.take(); // close the channel: input exhausted
        assert!(!r.lane.finished());
        let records = poll_batch(&mut r.lane, 512).unwrap();
        assert_eq!(records.len(), 1);
        assert!(
            r.tracker.terminal().is_none(),
            "no terminal while the final batch is being handed out"
        );
        assert!(!r.lane.finished(), "finished while the final batch is out");
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(r.lane.finished(), "not finished after the deciding poll");
        let expected = Position {
            ordinal: 0,
            record: 1,
        }
        .encode()
        .unwrap();
        assert_eq!(
            r.tracker.terminal(),
            Some(expected),
            "terminal = one past the last emitted record, decided by the None poll"
        );
    }

    #[test]
    fn empty_input_terminates_at_the_resume_watermark() {
        // A tenancy that emits nothing (resume exactly at end-of-input)
        // must terminate exactly where it began.
        let resume = Position {
            ordinal: 2,
            record: 3,
        };
        let mut r = rig(Some(resume));
        r.tx.take(); // nothing to read
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert_eq!(r.tracker.terminal(), Some(resume.encode().unwrap()));
        assert!(r.lane.finished());
    }

    #[test]
    fn resume_discards_the_committed_record_count() {
        // Watermark E(0, 2): two records of object 0 are committed.
        let mut r = rig(Some(Position {
            ordinal: 0,
            record: 2,
        }));
        start(&r, 0, "p/a.ndjson");
        send(
            &r,
            ChunkMsg::Chunk(bytes::Bytes::from_static(b"r0\nr1\nr2\nr3\n")),
        );
        send(&r, ChunkMsg::ObjectEnd);
        let records = poll_batch(&mut r.lane, 512).unwrap();
        let expect = |rec: u64| {
            Position {
                ordinal: 0,
                record: rec,
            }
            .encode()
            .unwrap()
        };
        assert_eq!(
            records,
            vec![(expect(2), b"r2".to_vec()), (expect(3), b"r3".to_vec())],
            "replayed committed records are discarded; indexes continue"
        );
    }

    #[test]
    fn resume_exactly_at_object_end_advances_cleanly() {
        // Watermark E(0, 2) with the object holding exactly two records.
        let mut r = rig(Some(Position {
            ordinal: 0,
            record: 2,
        }));
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"r0\nr1\n")));
        send(&r, ChunkMsg::ObjectEnd);
        start(&r, 1, "p/b.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"next\n")));
        send(&r, ChunkMsg::ObjectEnd);
        let records = poll_batch(&mut r.lane, 512).unwrap();
        assert_eq!(
            records,
            vec![(
                Position {
                    ordinal: 1,
                    record: 0
                }
                .encode()
                .unwrap(),
                b"next".to_vec()
            )]
        );
    }

    /// A compressed object that fails to decode poisons the split, whether
    /// the failure comes in the chunk's first decode step or in a later one.
    #[test]
    fn an_undecodable_compressed_object_poisons_the_split() {
        let mut late = Vec::new();
        let mut enc = flate2::write::GzEncoder::new(&mut late, flate2::Compression::best());
        std::io::Write::write_all(&mut enc, &b"a\n".repeat(1 << 16)).unwrap();
        enc.finish().unwrap();
        late.extend([0xFF; 64]);
        for chunk in [vec![0xFF; 64], late] {
            let mut r = rig(None);
            start(&r, 0, "p/a.ndjson.gz");
            send(&r, ChunkMsg::Chunk(bytes::Bytes::from(chunk)));
            send(&r, ChunkMsg::ObjectEnd);
            while poll_batch(&mut r.lane, 512).is_some() {}
            let report = r.poison_rx.try_recv().expect("a poison report");
            assert!(
                matches!(report.kind, PoisonKind::Undecodable),
                "{:?}",
                report.kind
            );
            assert!(
                report.reason.contains("decoding \"p/a.ndjson.gz\""),
                "{}",
                report.reason
            );
        }
    }

    #[test]
    fn resume_object_shorter_than_committed_poisons_the_split() {
        // Watermark says 3 records are committed; the object now has 1 —
        // content drift. Object-level, so it must not error the poll.
        let mut r = rig(Some(Position {
            ordinal: 0,
            record: 3,
        }));
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"only\n")));
        send(&r, ChunkMsg::ObjectEnd);
        assert!(poll_batch(&mut r.lane, 512).is_none());
        let report = r.poison_rx.try_recv().expect("a poison report");
        assert!(report.reason.contains("short"), "{}", report.reason);
        assert!(!r.lane.finished(), "a poisoned lane is not finished");
        // Quiescent thereafter: no error, no duplicate report, no
        // terminal watermark (the split did not finish).
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(r.poison_rx.try_recv().is_err(), "poison reports once");
        assert!(r.tracker.terminal().is_none());
        assert!(!r.lane.finished());
    }

    #[test]
    fn fetcher_poison_goes_quiescent_and_discards_undelivered_records() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\n")));
        send(
            &r,
            ChunkMsg::LaneFailed(SplitFailure::Poison(
                PoisonKind::NotFound,
                "object vanished".into(),
            )),
        );
        // The framed-but-unacked record must not be delivered past the
        // poison: replay by the next owner would then duplicate it
        // harmlessly, but delivering it here while reporting failure
        // would tangle the split's accounting.
        assert!(poll_batch(&mut r.lane, 512).is_none());
        let report = r.poison_rx.try_recv().expect("a poison report");
        assert!(report.reason.contains("vanished"), "{}", report.reason);
        assert!(!r.lane.finished(), "a poisoned lane is not finished");
    }

    /// Pins the feed lane's signal at end-of-input: none while records
    /// remain, one when the lane decides end-of-input, and none on a later
    /// poll that decides it again.
    #[test]
    fn end_of_input_wakes_the_control_plane() {
        let (waker, park) = wake_channel();
        let mut r = rig_with(None, waker, || Arc::new(SystemClock));
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\na2\n")));
        send(&r, ChunkMsg::ObjectEnd);
        r.tx.take();

        assert_eq!(poll_batch(&mut r.lane, 512).map(|b| b.len()), Some(2));
        assert!(!park.park(Duration::ZERO), "signalled while records remain");
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(r.tracker.terminal().is_some());
        assert!(r.lane.finished());
        assert!(park.park(Duration::ZERO), "no signal at end-of-input");
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(
            !park.park(Duration::ZERO),
            "signalled again after end-of-input"
        );
        assert!(r.lane.finished(), "finished turned back after a re-poll");
    }

    /// Pins the feed lane's signal on poison: the poll that poisons the lane
    /// signals the control plane.
    #[test]
    fn poison_wakes_the_control_plane() {
        let (waker, park) = wake_channel();
        let mut r = rig_with(None, waker, || Arc::new(SystemClock));
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\n")));
        send(
            &r,
            ChunkMsg::LaneFailed(SplitFailure::Poison(
                PoisonKind::NotFound,
                "object vanished".into(),
            )),
        );
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(r.poison_rx.try_recv().is_ok(), "a poison report");
        assert!(park.park(Duration::ZERO), "no signal on poison");
    }

    #[test]
    fn fetcher_fatal_failure_is_terminal() {
        let mut r = rig(None);
        send(
            &r,
            ChunkMsg::LaneFailed(SplitFailure::Fatal(SourceError::Client {
                class: ErrorClass::Fatal,
                reason: "access denied".into(),
            })),
        );
        let err = r.lane.poll(512, Duration::from_millis(50)).unwrap_err();
        assert!(err.to_string().contains("access denied"), "{err}");
        let again = r.lane.poll(512, Duration::from_millis(50)).unwrap_err();
        assert!(again.to_string().contains("access denied"), "sticky");
        assert!(r.poison_rx.try_recv().is_err(), "fatal is not poison");
        assert!(!r.lane.finished(), "a failed lane is not finished");
    }

    #[test]
    fn mid_object_disconnect_is_fatal() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"partial")));
        r.tx.take(); // fetcher gone without ObjectEnd or LaneFailed
        // First poll returns no framed data: "partial" has no newline and
        // the object never ends, so nothing is emittable.
        let err = r.lane.poll(512, Duration::from_millis(50)).unwrap_err();
        assert!(err.to_string().contains("mid-object"), "{err}");
        assert!(
            r.tracker.terminal().is_none(),
            "a dead lane never terminates"
        );
        assert!(!r.lane.finished());
    }

    #[test]
    fn batch_respects_max_records_and_continues() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(
            &r,
            ChunkMsg::Chunk(bytes::Bytes::from_static(b"1\n2\n3\n4\n5\n")),
        );
        send(&r, ChunkMsg::ObjectEnd);
        let first = poll_batch(&mut r.lane, 2).unwrap();
        assert_eq!(first.len(), 2);
        let second = poll_batch(&mut r.lane, 2).unwrap();
        assert_eq!(second.len(), 2);
        let third = poll_batch(&mut r.lane, 2).unwrap();
        assert_eq!(third.len(), 1);
        assert_eq!(third[0].1, b"5".to_vec());
    }

    /// Bounds every wait for a poll thread to reach its idle wait.
    const ARMED: Duration = Duration::from_secs(10);

    /// Moves the clock past every deadline when the test panics, so the
    /// thread scope's join does not wait on a poll parked on the frozen
    /// clock.
    struct AdvanceOnPanic<'a>(&'a TestClock);

    impl Drop for AdvanceOnPanic<'_> {
        fn drop(&mut self) {
            if std::thread::panicking() {
                self.0.advance(Duration::from_secs(3600));
            }
        }
    }

    /// Pins that an idle poll waits on the lane's clock until its timeout,
    /// then returns `Ok(None)` without deciding end-of-input.
    #[test]
    fn an_idle_poll_parks_until_the_clock_reaches_its_timeout() {
        let (mut r, clock) = frozen_rig();
        let timeout = Duration::from_millis(60);
        let deadline = clock.now() + timeout;
        let lane = &mut r.lane;
        std::thread::scope(|scope| {
            let polling = scope.spawn(move || lane.poll(512, timeout).map(|b| b.is_none()));
            let _release = AdvanceOnPanic(&clock);
            spate_test::wait_until(ARMED, "the poll arms its deadline", || {
                clock.has_slept_until(deadline)
            });
            assert!(!polling.is_finished(), "returned before the clock moved");
            clock.advance(timeout);
            let idle = polling.join().expect("the poll thread").expect("poll");
            assert!(idle, "an idle poll returned a batch");
        });
        assert!(r.tracker.terminal().is_none(), "idle is not end-of-input");
    }

    /// Pins that a zero-timeout poll still takes one ready message, so the
    /// next poll frames the chunk behind it, and that neither poll waits.
    #[test]
    fn a_zero_timeout_poll_still_takes_a_ready_chunk() {
        let (mut r, _clock) = frozen_rig();
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a\n")));
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        let batch = r.lane.poll(512, Duration::ZERO).unwrap();
        let mut batch = batch.expect("the chunk's record");
        assert_eq!(batch.next_payload().map(|p| p.bytes), Some(b"a".as_slice()));
        assert!(batch.next_payload().is_none(), "one record");
    }

    /// Pins that the deadline cap reads the lane's clock: real time past the
    /// deadline does not end a poll while that clock stands still.
    #[test]
    fn the_deadline_cap_reads_the_lanes_clock() {
        let (mut r, clock) = frozen_rig();
        let timeout = Duration::from_millis(1);
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"ab")));
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"c\n")));
        let deadline = clock.now() + timeout;
        spate_test::wait_until(ARMED, "real time passes the deadline", || {
            tokio::time::Instant::now() > deadline
        });
        let batch = r.lane.poll(512, timeout).unwrap();
        let mut batch = batch.expect("the joined record");
        assert_eq!(
            batch.next_payload().map(|p| p.bytes),
            Some(b"abc".as_slice())
        );
    }

    /// Pins that the idle wait's remaining time reads the lane's clock: a
    /// poll whose deadline real time has passed still parks on that clock.
    #[test]
    fn the_idle_wait_reads_the_lanes_clock() {
        let (mut r, clock) = frozen_rig();
        let timeout = Duration::from_millis(1);
        let deadline = clock.now() + timeout;
        spate_test::wait_until(ARMED, "real time passes the deadline", || {
            tokio::time::Instant::now() > deadline
        });
        let lane = &mut r.lane;
        std::thread::scope(|scope| {
            let polling = scope.spawn(move || lane.poll(512, timeout).map(|b| b.is_none()));
            let _release = AdvanceOnPanic(&clock);
            spate_test::wait_until(ARMED, "the poll arms its deadline", || {
                clock.has_slept_until(deadline)
            });
            clock.advance(timeout);
            let idle = polling.join().expect("the poll thread").expect("poll");
            assert!(idle, "an idle poll returned a batch");
        });
    }

    /// Pins that the deadline caps a poll whose chunks frame no record: the
    /// poll returns after one receive past the deadline, leaving the next
    /// chunk queued.
    #[test]
    fn the_deadline_caps_a_poll_that_frames_nothing() {
        let (mut r, _clock) = frozen_rig();
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"ab")));
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"c\n")));
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        let tx = r.tx.as_ref().unwrap();
        assert_eq!(tx.max_capacity() - tx.capacity(), 1, "b\"c\\n\" was taken");
        let batch = r.lane.poll(512, Duration::ZERO).unwrap();
        let mut batch = batch.expect("the joined record");
        assert_eq!(
            batch.next_payload().map(|p| p.bytes),
            Some(b"abc".as_slice())
        );
    }

    /// Pins that the deadline caps a poll that decodes a compressed chunk
    /// framing no record: a zero-timeout poll decodes one bounded step and
    /// returns with the rest of the chunk pending.
    #[test]
    fn a_zero_timeout_poll_decodes_one_step_of_a_blank_compressed_chunk() {
        let (mut r, _clock) = frozen_rig();
        let blank = zstd::encode_all(&vec![b'\n'; 16 << 20][..], 19).unwrap();
        start(&r, 0, "p/a.ndjson.zst");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from(blank)));
        send(&r, ChunkMsg::ObjectEnd);
        // One poll takes the object start, the next takes the chunk.
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        for _ in 0..3 {
            let before = r.lane.framer.decoded_bytes();
            assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
            let decoded = r.lane.framer.decoded_bytes() - before;
            assert!(
                r.lane.framer.has_pending_input(),
                "a zero-timeout poll decoded the rest of the chunk"
            );
            assert!(
                decoded < 2 * crate::framer::PUMP_DECODED_BYTES,
                "one zero-timeout poll decoded {decoded} bytes"
            );
        }
    }

    /// Pins that the channel closing while a poll waits on the clock ends the
    /// lane: the poll returns `Ok(None)` and records the terminal watermark.
    #[test]
    fn closing_the_channel_while_a_poll_waits_ends_the_lane() {
        let (mut r, clock) = frozen_rig();
        start(&r, 0, "p/a.ndjson");
        send(&r, ChunkMsg::Chunk(bytes::Bytes::from_static(b"a\n")));
        send(&r, ChunkMsg::ObjectEnd);
        assert_eq!(poll_batch(&mut r.lane, 512).map(|b| b.len()), Some(1));
        let timeout = Duration::from_millis(60);
        let deadline = clock.now() + timeout;
        let tx = r.tx.take();
        let lane = &mut r.lane;
        std::thread::scope(|scope| {
            let polling = scope.spawn(move || lane.poll(512, timeout).map(|b| b.is_none()));
            let _release = AdvanceOnPanic(&clock);
            spate_test::wait_until(ARMED, "the poll arms its deadline", || {
                clock.has_slept_until(deadline)
            });
            drop(tx);
            let idle = polling.join().expect("the poll thread").expect("poll");
            assert!(idle, "a closed lane returned a batch");
        });
        assert!(r.tracker.terminal().is_some(), "the close ended the lane");
    }

    fn overwrite() -> ChunkMsg {
        ChunkMsg::LaneFailed(SplitFailure::Overwritten(Box::new(
            crate::fetch::Overwrite {
                key: "p/a.ndjson".into(),
                etag: Some("\"new\"".into()),
                size: 10,
                last_modified_ms: 2_000,
            },
        )))
    }

    /// Pins that an overwrite with nothing framed reports the overwrite,
    /// never poisons, and ends the lane at its start.
    #[test]
    fn an_overwrite_with_nothing_framed_ends_at_the_start() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(&r, overwrite());
        drop(r.tx.take());
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert_eq!(r.tracker.terminal(), Some(0));
        assert!(r.lane.finished());
        let report = r.poison_rx.try_recv().expect("an overwrite report");
        assert_eq!(report.overwrite.expect("the overwrite").key, "p/a.ndjson");
        assert!(r.poison_rx.try_recv().is_err(), "one report");
    }

    /// Pins that an overwrite keeps the records framed before it, drops the
    /// unterminated tail, and ends the lane one past the last record
    /// emitted, with every later poll idle.
    #[test]
    fn an_overwrite_emits_the_framed_records_and_ends_after_them() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(
            &r,
            ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\na2\na3")),
        );
        send(&r, overwrite());
        drop(r.tx.take());
        let records = poll_batch(&mut r.lane, 512).expect("the framed records");
        let bytes: Vec<&[u8]> = records.iter().map(|(_, b)| b.as_slice()).collect();
        assert_eq!(bytes, [b"a1".as_slice(), b"a2".as_slice()]);
        assert!(
            r.tracker.terminal().is_none(),
            "decided before the batch was pushed"
        );
        assert!(!r.lane.finished());
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert_eq!(r.tracker.terminal(), Some(records[1].0 + 1));
        assert!(r.lane.finished());
        assert!(poll_batch(&mut r.lane, 512).is_none());
        assert!(poll_batch(&mut r.lane, 512).is_none());
        let report = r.poison_rx.try_recv().expect("an overwrite report");
        assert!(report.overwrite.is_some());
    }

    /// Pins that records framed before an overwrite drain across polls when
    /// they exceed one batch.
    #[test]
    fn an_overwrite_drains_its_framed_records_across_batches() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(
            &r,
            ChunkMsg::Chunk(bytes::Bytes::from_static(b"a1\na2\na3\n")),
        );
        send(&r, overwrite());
        drop(r.tx.take());
        assert_eq!(poll_batch(&mut r.lane, 2).expect("two records").len(), 2);
        assert!(!r.lane.finished());
        assert_eq!(poll_batch(&mut r.lane, 2).expect("one record").len(), 1);
        assert!(!r.lane.finished());
        assert!(poll_batch(&mut r.lane, 2).is_none());
        assert!(r.tracker.terminal().is_some());
        assert!(r.lane.finished());
    }

    /// Pins that a lane draining its framer across batches after its channel
    /// closed reports `finished` only on the `Ok(None)` after the last batch.
    /// Regression for #1478.
    #[test]
    fn finished_waits_for_the_batch_that_empties_the_framer() {
        let mut r = rig(None);
        start(&r, 0, "p/a.ndjson");
        send(
            &r,
            ChunkMsg::Chunk(bytes::Bytes::from_static(b"1\n2\n3\n4\n5\n")),
        );
        send(&r, ChunkMsg::ObjectEnd);
        r.tx.take();
        for expected in [2, 2, 1] {
            assert_eq!(poll_batch(&mut r.lane, 2).map(|b| b.len()), Some(expected));
            assert!(!r.lane.finished(), "finished with records still framed");
        }
        assert!(poll_batch(&mut r.lane, 2).is_none());
        assert!(r.lane.finished());
    }

    /// Pins that an idle lane, whose channel is open with nothing on it, does
    /// not report `finished`.
    #[test]
    fn an_idle_lane_is_not_finished() {
        let mut r = rig(None);
        assert!(r.lane.poll(512, Duration::ZERO).unwrap().is_none());
        assert!(!r.lane.finished());
    }
}