ruststream-rdkafka 0.7.1

//! The subscriber: a stream of Kafka deliveries from one topic subscription.

use std::fmt;
use std::future::Future;
use std::num::NonZeroUsize;
use std::pin::{Pin, pin};
use std::sync::{Arc, OnceLock};
use std::task::{Context, Poll};

#[cfg(feature = "schema-registry")]
use bytes::Bytes;
#[cfg(feature = "testing")]
use futures::future::Either;
use futures::{Stream, ready};
use rdkafka::consumer::Consumer as _;
use rdkafka::error::RDKafkaErrorCode;
#[cfg(feature = "schema-registry")]
use ruststream::IncomingMessage;
use ruststream::{BatchSubscriber, Seekable, Str, Subscriber};
use tracing::{debug, warn};

use crate::broker::ConnState;
use crate::convert;
use crate::eos::EOS_SOURCE_HEADER;
use crate::error::KafkaError;
#[cfg(feature = "testing")]
use crate::in_process::Member;
use crate::message::{KafkaMessage, Lane, Settlement};
use crate::record::{Cart, HeldRecord, LiveShared, Shared};
#[cfg(feature = "schema-registry")]
use crate::schema_registry::{SchemaPrefetch, SchemaRegistry};
use crate::seek::KafkaSeeker;
use crate::subscription::{Commit, LaneKey};
use crate::tracker::{CommitTracker, TrackedConsumer};

/// Whether librdkafka is already retrying this error by itself, making a stream error item
/// noise rather than signal. The set is deliberately small and explicit; when in doubt, the
/// error is forwarded.
fn is_transient(err: &rdkafka::error::KafkaError) -> bool {
    // A subscribed topic that does not exist (yet): pending creation is routine when the
    // broker auto-creates topics, and librdkafka keeps refreshing metadata until it appears.
    err.rdkafka_error_code() == Some(RDKafkaErrorCode::UnknownTopicOrPartition)
}

/// Where a delivery's topic name comes from.
///
/// A subscription over one literal topic knows the answer before its first record, so its
/// deliveries read the name the subscription's shared block holds and carry none of their own.
/// Reading it back out of librdkafka
/// costs a `CStr` walk and a UTF-8 validation of the same bytes on every delivery, and answers
/// what the descriptor said.
#[derive(Debug)]
pub(crate) enum DeliveredTopic {
    /// Every record of this subscription comes from this topic: one literal name, or the topic
    /// whose partitions were assigned by hand.
    One(Str),
    /// Several names or a pattern: only the record knows which topic it came from. The last one
    /// is kept because a fetch hands a consumer one topic's records at a time, so a whole run
    /// answers without minting the name again.
    PerRecord(Option<Str>),
}

impl DeliveredTopic {
    /// The name the subscription's shared block holds: its one topic, or, when the records name
    /// their own, the name the subscription was declared with.
    fn shared(&self, declared: &str) -> Str {
        match self {
            Self::One(name) => name.clone(),
            Self::PerRecord(_) => Str::from(declared),
        }
    }

    /// The name a delivery carries itself: none when the subscription reads one topic, whose
    /// name the shared block holds; otherwise the record's topic, as the shared string every
    /// delivery of that topic carries.
    fn own(&mut self, record: &HeldRecord) -> Option<Str> {
        let Self::PerRecord(last) = self else {
            return None;
        };
        let name = record.topic();
        if let Some(known) = last
            && &**known == name
        {
            return Some(known.clone());
        }
        let minted = Str::from(name);
        *last = Some(minted.clone());
        Some(minted)
    }
}

/// A consumer-group member on one topic, yielding [`KafkaMessage`] deliveries.
///
/// Created by subscribing a [`KafkaTopic`](crate::KafkaTopic) descriptor (or a bare topic name)
/// through [`KafkaBroker`](crate::KafkaBroker). The subscriber owns a dedicated librdkafka
/// consumer; dropping it closes the consumer, which leaves the group and (under auto-commit)
/// commits the final stored position. Under `Commit::Tracked` each in-flight delivery keeps
/// the consumer alive, so the close happens once the last outstanding message settles or
/// drops - do not rely on subscriber drop as an immediate group-departure barrier.
///
/// Back-pressure: polling the stream is what drives the consumer, so consuming slower simply
/// fetches slower; librdkafka's own fetch queue bounds (`queued.max.messages.kbytes` and
/// friends, settable through [`KafkaTopic::config`](crate::KafkaTopic::config)) cap local
/// buffering.
///
/// # Examples
///
/// A handler whose payload parameter is a slice reads batches, and the mount site names the
/// largest one it accepts. A batch is one delivery plus what librdkafka has already fetched, so
/// it is often smaller than that size:
///
/// ```
/// # #[cfg(feature = "json")]
/// # mod demo {
/// use ruststream_rdkafka::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Event {
///     id: u64,
/// }
///
/// #[subscriber(KafkaTopic::new("events").group("indexer").commit(Commit::Tracked))]
/// async fn index(events: &[Event]) -> HandlerOutcome {
///     tracing::info!(count = events.len(), "indexing");
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     RustStream::new(AppInfo::new("indexer", "0.1.0")).with_broker(
///         KafkaBroker::new(["localhost:9092"]),
///         |b| {
///             b.include(index.batch(nonzero!(500usize)));
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
pub struct KafkaSubscriber {
    /// What the subscription reads through.
    source: Source,
    topic: String,
    /// Where a delivery's topic name comes from; see [`DeliveredTopic`].
    delivered_topic: DeliveredTopic,
    commit: Commit,
    tracker: Arc<CommitTracker>,
    lane_key: LaneKey,
    /// Minted once, when the subscription opens. The shared block every delivery reaches holds
    /// the same handle, so a handler's context hands it out without a count of its own.
    seeker: Arc<KafkaSeeker>,
    #[cfg(feature = "schema-registry")]
    schema_registry: Option<SchemaRegistry>,
    #[cfg(feature = "schema-registry")]
    schema_prefetch: Option<SchemaPrefetch>,
    /// Whether the subscriber is inside an episode of transient consume errors; the first
    /// error of an episode warns, repeats are debug, recovery closes the episode.
    in_transient_episode: bool,
}

/// What a subscription reads through: its librdkafka consumer, or, under the `testing` feature,
/// its member of the in-process cluster.
///
/// Without the feature there is one variant, so the type is the cart itself and every `match` on
/// it is irrefutable.
enum Source {
    Kafka(Cart),
    #[cfg(feature = "testing")]
    InProcess(Arc<Member>, Arc<Shared>),
}

#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Source>() == size_of::<Cart>());

impl KafkaSubscriber {
    pub(crate) fn new(
        consumer: Arc<TrackedConsumer>,
        topic: String,
        delivered_topic: DeliveredTopic,
        commit: Commit,
        tracker: Arc<CommitTracker>,
        lane_key: LaneKey,
        connection: Arc<ConnState>,
    ) -> Self {
        let seeker = Arc::new(KafkaSeeker::new(
            Arc::clone(&consumer),
            Arc::clone(&tracker),
            connection,
            Str::from(topic.as_str()),
        ));
        let cart = Arc::new(LiveShared {
            consumer,
            shared: Arc::new(Shared::new(
                Arc::clone(&tracker),
                Arc::clone(&seeker),
                delivered_topic.shared(&topic),
            )),
        });
        Self {
            source: Source::Kafka(cart),
            topic,
            delivered_topic,
            commit,
            tracker,
            lane_key,
            seeker,
            #[cfg(feature = "schema-registry")]
            schema_registry: None,
            #[cfg(feature = "schema-registry")]
            schema_prefetch: None,
            in_transient_episode: false,
        }
    }

    /// A subscription reading through its member of the in-process cluster.
    #[cfg(feature = "testing")]
    pub(crate) fn in_process(
        member: Arc<Member>,
        topic: String,
        delivered_topic: DeliveredTopic,
        commit: Commit,
        tracker: Arc<CommitTracker>,
        lane_key: LaneKey,
        connection: Arc<ConnState>,
    ) -> Self {
        let seeker = Arc::new(KafkaSeeker::in_process(
            Arc::clone(&member),
            connection,
            Str::from(topic.as_str()),
        ));
        let shared = Arc::new(Shared::new(
            Arc::clone(&tracker),
            Arc::clone(&seeker),
            delivered_topic.shared(&topic),
        ));
        Self {
            source: Source::InProcess(member, shared),
            topic,
            delivered_topic,
            commit,
            tracker,
            lane_key,
            seeker,
            #[cfg(feature = "schema-registry")]
            schema_registry: None,
            #[cfg(feature = "schema-registry")]
            schema_prefetch: None,
            in_transient_episode: false,
        }
    }

    #[cfg(feature = "schema-registry")]
    pub(crate) fn with_schema_registry(mut self, registry: Option<SchemaRegistry>) -> Self {
        self.schema_registry = registry;
        self
    }

    #[cfg(feature = "schema-registry")]
    pub(crate) fn with_schema_prefetch(mut self, prefetch: Option<SchemaPrefetch>) -> Self {
        self.schema_prefetch = prefetch;
        self
    }

    /// Logs a transient consume error: one warning when the episode starts (the signal a
    /// human acts on in monitoring), debug for the repeats.
    fn note_transient(&mut self, err: &rdkafka::error::KafkaError) {
        if self.in_transient_episode {
            debug!(
                target: "ruststream_rdkafka",
                topic = %self.topic,
                error = %err,
                "transient consume error (repeat)",
            );
        } else {
            self.in_transient_episode = true;
            warn!(
                target: "ruststream_rdkafka",
                topic = %self.topic,
                error = %err,
                "transient consume error; librdkafka keeps retrying",
            );
        }
    }

    /// Closes a transient-error episode once deliveries flow again.
    fn note_recovered(&mut self) {
        if self.in_transient_episode {
            self.in_transient_episode = false;
            debug!(
                target: "ruststream_rdkafka",
                topic = %self.topic,
                "recovered from transient consume errors",
            );
        }
    }

    /// The cart a subscription over librdkafka reads through, for the keeper that polls it
    /// before its stream starts (see `startup`).
    pub(crate) fn cart(&self) -> Cart {
        match &self.source {
            Source::Kafka(cart) => Arc::clone(cart),
            #[cfg(feature = "testing")]
            Source::InProcess(..) => unreachable!("an in-process subscription is never kept"),
        }
    }

    /// The topic this subscriber consumes.
    #[must_use]
    pub fn topic(&self) -> &str {
        &self.topic
    }

    /// The registry middleware: transcodes a framed payload to plain JSON on the async
    /// consume path, so handlers and the default codec see JSON regardless of the wire
    /// format. A no-op without an attached registry or for non-framed payloads.
    #[cfg(feature = "schema-registry")]
    async fn transcode(&self, item: &mut KafkaMessage) {
        if let Some(registry) = &self.schema_registry
            && let Some(json) = registry
                .incoming_to_json(IncomingMessage::payload(item))
                .await
        {
            item.replace_payload(Bytes::from(json));
        }
    }

    /// Turns a fetched record into a delivery. The record travels into the delivery whole: what
    /// is read here is what a delivery cannot answer from the record later, or what a later read
    /// could no longer see.
    fn map_delivery(&mut self, record: HeldRecord) -> KafkaMessage {
        // Minted once per topic and shared from here on: the delivery carries it, and the
        // tracker opens this partition's slot under it.
        let own_topic = self.delivered_topic.own(&record);
        let topic = record.shared().topic(own_topic.as_ref());
        let partition = record.partition();
        let offset = record.offset();
        // The headers stay unread unless the delivery is asked for them - except under
        // transactional commits, where the source coordinates ride them so the reply path can
        // pair a publishing handler's reply with its consumed offset (see EosPipeline::replies);
        // stripped from every outgoing publish, so they never hit the wire.
        let headers = OnceLock::new();
        if matches!(self.commit, Commit::Transactional(_)) {
            let mut map = convert::headers_from_message(&record);
            map.insert(
                Str::from_static(EOS_SOURCE_HEADER),
                crate::eos::encode_source(topic, partition, offset),
            );
            headers.set(map).expect("the cell was just created");
        }
        // The generation is captured here, where the delivery is pulled, never where it settles:
        // that is what lets a reposition landing in between tell a delivery of the replaced read
        // position apart from one the new position produced.
        let settlement = match &self.commit {
            Commit::Auto => Settlement::Advisory,
            Commit::Tracked => Settlement::Tracked {
                slot: self.tracker.delivered(topic, partition, offset),
            },
            Commit::Transactional(_) => Settlement::Transactional {
                slot: self.tracker.delivered(topic, partition, offset),
            },
        };
        // The key itself is not built here: a subscription with no keyed lanes is never asked
        // for one, and both forms answer out of what the delivery already holds.
        let lane = match self.lane_key {
            LaneKey::RecordKey => Lane::RecordKey,
            LaneKey::Partition => Lane::Partition(OnceLock::new()),
        };
        KafkaMessage::new(
            record, headers, own_topic, partition, offset, settlement, lane,
        )
    }
}

impl Drop for KafkaSubscriber {
    /// Releases the deliveries handed back to this subscription and not delivered again: each one
    /// holds the consumer open, and the group hands their records to the partition's next owner.
    fn drop(&mut self) {
        self.tracker.attention.close();
        // A subscription closed before its stream started leaves the group rather than being
        // kept in it.
        match &self.source {
            Source::Kafka(cart) => cart.consumer.context().startup.release(),
            #[cfg(feature = "testing")]
            Source::InProcess(..) => {}
        }
    }
}

impl fmt::Debug for KafkaSubscriber {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("KafkaSubscriber")
            .field("topic", &self.topic)
            .field("commit", &self.commit)
            .finish_non_exhaustive()
    }
}

impl Subscriber for KafkaSubscriber {
    type Message = KafkaMessage;
    type Error = KafkaError;

    /// Streams deliveries as they arrive; the stream yields an error item when the consumer
    /// fails (it does not end on its own - drop the subscriber to leave the group).
    ///
    /// Errors librdkafka is already retrying by itself are not forwarded as stream items:
    /// today that is exactly `UnknownTopicOrPartition` (a subscribed topic pending creation).
    /// Such an episode surfaces as one warning when it starts - the monitoring signal to act
    /// on - with debug lines for the repeats and for the recovery, so a topic that appears
    /// late (broker auto-creation, provisioning races) recovers without flooding the dispatch
    /// error log, while a topic that never appears leaves the warning standing. Everything
    /// else is forwarded.
    ///
    /// # Cancel safety
    ///
    /// Polling is cancel safe (the underlying `recv` is documented cancellation safe, so no
    /// delivery is lost by dropping the stream between polls), and the stream can be re-created
    /// by calling `stream` again: deliveries buffer in the consumer, not in the returned stream.
    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
        // Cloned once per stream and never per delivery: a wait owns the handles it needs, so
        // it borrows nothing from the stream that holds it.
        let cart = match &self.source {
            Source::Kafka(cart) => Arc::clone(cart),
            #[cfg(feature = "testing")]
            Source::InProcess(member, shared) => {
                let (member, shared) = (Arc::clone(member), Arc::clone(shared));
                return Either::Right(in_process::stream(self, member, shared));
            }
        };
        cart.consumer
            .context()
            .startup
            .hand_over(&cart.consumer, &self.tracker);
        let stream = KafkaStream {
            #[cfg(feature = "schema-registry")]
            registry: self.schema_registry.clone(),
            #[cfg(feature = "schema-registry")]
            prefetch: self.schema_prefetch.clone(),
            #[cfg(feature = "schema-registry")]
            finishing: None,
            tracker: Arc::clone(&self.tracker),
            sub: self,
            cart,
            waiting: None,
        };
        #[cfg(feature = "testing")]
        let stream = Either::Left(stream);
        stream
    }
}

/// The stream [`KafkaSubscriber::stream`] hands out.
///
/// Written out rather than unfolded from a closure: a combinator carries the delivery out of a
/// future's output, through a tuple and into the stream's own return, and this type is 216 bytes,
/// so the compiler moved it three times per message before the caller ever saw it.
pub(crate) struct KafkaStream<'a> {
    sub: &'a mut KafkaSubscriber,
    /// Held here for the one question every turn asks it, whether anything waits ahead of the
    /// fetch queue (see [`Attention`](crate::attention::Attention)).
    tracker: Arc<CommitTracker>,
    cart: Cart,
    /// The wait, kept across polls for as long as the fetch queue stays dry: the registration
    /// `recv` makes with the consumer's queue dies with the future that made it, so re-arming it
    /// per poll would drop the wakeup it was waiting for. Boxed because an `async fn` future
    /// cannot be named, and built once per idle episode - a subscription keeping up with its
    /// topic never reaches it.
    waiting: Option<Pin<Box<dyn Future<Output = ColdTurn> + Send>>>,
    /// The registry middleware, held here rather than reached through the subscriber, so the
    /// future below owns what it reads.
    #[cfg(feature = "schema-registry")]
    registry: Option<SchemaRegistry>,
    #[cfg(feature = "schema-registry")]
    prefetch: Option<SchemaPrefetch>,
    /// The registry's own async step, for the deliveries that have one. A subscription with no
    /// registry attached never builds it.
    #[cfg(feature = "schema-registry")]
    finishing: Option<Pin<Box<dyn Future<Output = KafkaMessage> + Send>>>,
}

/// What one turn of the cold path resolved to.
enum ColdTurn {
    /// A record the wait took, or what the consumer reported instead of one.
    Record(Result<HeldRecord, rdkafka::error::KafkaError>),
    /// Something waits ahead of the fetch queue (a requeued delivery, a start failure); the top
    /// of the next turn serves it.
    Woken,
}

/// Waits for a record, or for something left ahead of the fetch queue: a delivery handed back
/// with `nack(true)`, or a start position the rebalance could not apply.
///
/// The waiter is enabled before the flag is read, so what lands between the two wakes this wait
/// instead of being missed - which is what makes it reach a subscription whose topic may never
/// deliver anything.
///
/// # Cancel safety
///
/// Cancel safe: dropping this future before it resolves takes no record.
async fn cold_turn(tracker: Arc<CommitTracker>, cart: Cart) -> ColdTurn {
    let mut woken = pin!(tracker.attention.waiter());
    woken.as_mut().enable();
    if tracker.attention.raised() {
        return ColdTurn::Woken;
    }
    tokio::select! {
        biased;
        () = woken => ColdTurn::Woken,
        received = HeldRecord::next(&cart) => ColdTurn::Record(received),
    }
}

/// What waits ahead of the fetch queue, once the flag says something does: a start failure
/// first, then a delivery handed back with `nack(true)`. A handed-back delivery whose partition
/// moved since (a seek, a revoke) is released instead, since the replay or the partition's next
/// owner delivers its record.
fn attended(tracker: &CommitTracker) -> Option<Result<KafkaMessage, KafkaError>> {
    if let Some(err) = tracker.attention.take_start_failure() {
        return Some(Err(err));
    }
    while let Some(delivery) = tracker.attention.take_requeued() {
        if delivery.is_current() {
            return Some(Ok(delivery));
        }
    }
    None
}

/// Runs the registry middleware over a delivery.
///
/// The prefetch first: it reads the envelope, which the transcode would have replaced with a
/// JSON document. The two are alternatives anyway.
#[cfg(feature = "schema-registry")]
async fn finish_delivery(
    registry: Option<SchemaRegistry>,
    prefetch: Option<SchemaPrefetch>,
    mut item: KafkaMessage,
) -> KafkaMessage {
    if let Some(prefetch) = &prefetch {
        prefetch
            .warm_delivery(IncomingMessage::payload(&item))
            .await;
    }
    if let Some(registry) = &registry
        && let Some(json) = registry
            .incoming_to_json(IncomingMessage::payload(&item))
            .await
    {
        item.replace_payload(Bytes::from(json));
    }
    item
}

impl KafkaStream<'_> {
    /// Hands the delivery on, through the registry middleware when one is attached.
    fn deliver(&mut self, record: HeldRecord) -> Poll<Option<Result<KafkaMessage, KafkaError>>> {
        self.sub.note_recovered();
        let item = self.sub.map_delivery(record);
        #[cfg(feature = "schema-registry")]
        if self.registry.is_some() || self.prefetch.is_some() {
            self.finishing = Some(Box::pin(finish_delivery(
                self.registry.clone(),
                self.prefetch.clone(),
                item,
            )));
            return Poll::Pending;
        }
        Poll::Ready(Some(Ok(item)))
    }
}

impl Stream for KafkaStream<'_> {
    type Item = Result<KafkaMessage, KafkaError>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        loop {
            #[cfg(feature = "schema-registry")]
            if let Some(finishing) = &mut this.finishing {
                let item = ready!(finishing.as_mut().poll(cx));
                this.finishing = None;
                return Poll::Ready(Some(Ok(item)));
            }
            // What waits ahead of the fetch queue preempts it: a start position the rebalance
            // could not apply (a subscription that was not opened where it was asked must not
            // pass for a working one), then a delivery handed back with `nack(true)`.
            if this.tracker.attention.raised()
                && let Some(item) = attended(&this.tracker)
            {
                return Poll::Ready(Some(item));
            }
            if let Some(waiting) = &mut this.waiting {
                let turn = ready!(waiting.as_mut().poll(cx));
                this.waiting = None;
                match turn {
                    // The waiter fired: the flag is read at the top of the next turn.
                    ColdTurn::Woken => continue,
                    ColdTurn::Record(Ok(record)) => {
                        if let Poll::Ready(item) = this.deliver(record) {
                            return Poll::Ready(item);
                        }
                        continue;
                    }
                    ColdTurn::Record(Err(err)) if is_transient(&err) => {
                        this.sub.note_transient(&err);
                        continue;
                    }
                    ColdTurn::Record(Err(err)) => {
                        return Poll::Ready(Some(Err(KafkaError::consume(err))));
                    }
                }
            }
            // A record already fetched needs no waiter: the failure flag above is sticky, so it
            // is read on the turn that finds nothing to deliver.
            match HeldRecord::ready(&this.cart) {
                Some(Ok(record)) => {
                    if let Poll::Ready(item) = this.deliver(record) {
                        return Poll::Ready(item);
                    }
                }
                Some(Err(err)) if is_transient(&err) => this.sub.note_transient(&err),
                Some(Err(err)) => return Poll::Ready(Some(Err(KafkaError::consume(err)))),
                // Nothing fetched, so the wait begins on the next turn of this loop.
                None => {
                    this.waiting = Some(Box::pin(cold_turn(
                        Arc::clone(&this.tracker),
                        Arc::clone(&this.cart),
                    )));
                }
            }
        }
    }
}

impl Seekable for KafkaSubscriber {
    type Seeker = KafkaSeeker;

    /// Hands out a handle for repositioning this subscription; see [`KafkaSeeker`] for the
    /// scope of a seek and what a rebalance does to it.
    fn seeker(&self) -> Self::Seeker {
        KafkaSeeker::clone(&self.seeker)
    }
}

impl BatchSubscriber for KafkaSubscriber {
    type Batch = Vec<KafkaMessage>;

    /// Streams non-empty batches natively: each waits for one delivery, then drains what
    /// librdkafka has already fetched, up to `size` messages in total. The batch never carries
    /// more than the registration's `batch(n)` asked for, and carries fewer whenever the fetch
    /// queue holds less; how much librdkafka keeps queued locally stays a consumer setting
    /// (`queued.max.messages.kbytes` and friends, settable through
    /// [`KafkaTopic::config`](crate::KafkaTopic::config)). A consumer error inside an open batch
    /// yields the batch first; the error (if it persists) surfaces on the next poll.
    ///
    /// # Cancel safety
    ///
    /// Same guarantees as [`Subscriber::stream`]: cancel safe between polls, no delivery is
    /// lost by dropping the stream.
    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, <Self as Subscriber>::Error>> + Send + '_ {
        let size = size.get();
        // See `stream`: the consumer is held beside the subscriber, not through it.
        let cart = match &self.source {
            Source::Kafka(cart) => Arc::clone(cart),
            #[cfg(feature = "testing")]
            Source::InProcess(member, shared) => {
                let (member, shared) = (Arc::clone(member), Arc::clone(shared));
                return Either::Right(in_process::batches(self, member, shared, size));
            }
        };
        cart.consumer
            .context()
            .startup
            .hand_over(&cart.consumer, &self.tracker);
        let tracker = Arc::clone(&self.tracker);
        let batches = futures::stream::unfold(
            (self, tracker, cart),
            move |(sub, tracker, cart)| async move {
                // Wait for the batch's first delivery, or for what waits ahead of the fetch
                // queue - see `stream` for why the wait is a select and not a check.
                let first = loop {
                    let woken = tracker.attention.waiter();
                    match attended(&tracker) {
                        Some(Ok(delivery)) => break delivery,
                        Some(Err(err)) => {
                            drop(woken);
                            return Some((Err(err), (sub, tracker, cart)));
                        }
                        None => {}
                    }
                    let received = tokio::select! {
                        biased;
                        () = woken => continue,
                        received = HeldRecord::next(&cart) => received,
                    };
                    match received {
                        Ok(record) => break sub.map_delivery(record),
                        Err(err) if is_transient(&err) => sub.note_transient(&err),
                        Err(err) => {
                            return Some((Err(KafkaError::consume(err)), (sub, tracker, cart)));
                        }
                    }
                };
                sub.note_recovered();

                let mut batch = Vec::with_capacity(size.min(64));
                batch.push(first);
                // Drain what is already fetched, stopping at the batch size; recv is cancel safe,
                // so dropping the probe future loses nothing.
                while batch.len() < size {
                    let Some(result) = HeldRecord::ready(&cart) else {
                        break;
                    };
                    match result {
                        Ok(record) => {
                            let item = sub.map_delivery(record);
                            batch.push(item);
                        }
                        Err(err) if is_transient(&err) => sub.note_transient(&err),
                        // Yield what was collected; a persistent error re-surfaces on the next
                        // batch's first recv.
                        Err(_) => break,
                    }
                }
                #[cfg(feature = "schema-registry")]
                for item in &mut batch {
                    sub.transcode(item).await;
                }
                Some((Ok(batch), (sub, tracker, cart)))
            },
        );
        #[cfg(feature = "testing")]
        let batches = Either::Left(batches);
        batches
    }
}

/// The in-process arm of the subscriber: the same deliveries, fetched from the subscription's
/// member of the in-process cluster.
#[cfg(feature = "testing")]
mod in_process {
    use std::sync::Arc;

    use futures::Stream;

    use super::{KafkaSubscriber, attended};
    use crate::error::KafkaError;
    use crate::in_process::{InProcessRecord, Member};
    use crate::message::KafkaMessage;
    use crate::record::{HeldRecord, Shared};
    use crate::tracker::CommitTracker;

    /// What one wait of the in-process stream resolved to.
    enum Turn {
        /// What waited ahead of the member's records (see [`attended`]).
        Attended(Result<KafkaMessage, KafkaError>),
        /// A record the member fetched, or what it reported instead.
        Fetched(Result<InProcessRecord, KafkaError>),
    }

    /// Serves what waits ahead of the member's records, or waits for either, as the live stream
    /// does.
    async fn turn(member: &Arc<Member>, tracker: &CommitTracker) -> Turn {
        loop {
            let woken = tracker.attention.waiter();
            if let Some(item) = attended(tracker) {
                return Turn::Attended(item);
            }
            tokio::select! {
                biased;
                () = woken => {}
                fetched = member.next() => return Turn::Fetched(fetched),
            }
        }
    }

    /// Turns a fetched record into a delivery.
    fn deliver(
        sub: &mut KafkaSubscriber,
        shared: &Arc<Shared>,
        record: InProcessRecord,
    ) -> KafkaMessage {
        sub.map_delivery(HeldRecord::InProcess {
            shared: Arc::clone(shared),
            record: Box::new(record),
        })
    }

    /// Runs the registry middleware over a delivery, as the live stream does.
    #[cfg(feature = "schema-registry")]
    async fn finish(sub: &KafkaSubscriber, item: KafkaMessage) -> KafkaMessage {
        super::finish_delivery(
            sub.schema_registry.clone(),
            sub.schema_prefetch.clone(),
            item,
        )
        .await
    }

    pub(super) fn stream(
        sub: &mut KafkaSubscriber,
        member: Arc<Member>,
        shared: Arc<Shared>,
    ) -> impl Stream<Item = Result<KafkaMessage, KafkaError>> + Send + '_ {
        futures::stream::unfold((sub, member, shared), async move |(sub, member, shared)| {
            let item = match turn(&member, &shared.tracker).await {
                Turn::Attended(item) => item,
                Turn::Fetched(Ok(record)) => {
                    let item = deliver(sub, &shared, record);
                    #[cfg(feature = "schema-registry")]
                    let item = finish(sub, item).await;
                    Ok(item)
                }
                Turn::Fetched(Err(err)) => Err(err),
            };
            Some((item, (sub, member, shared)))
        })
    }

    pub(super) fn batches(
        sub: &mut KafkaSubscriber,
        member: Arc<Member>,
        shared: Arc<Shared>,
        size: usize,
    ) -> impl Stream<Item = Result<Vec<KafkaMessage>, KafkaError>> + Send + '_ {
        // The size rides the state: an async closure that captured it could not be called again.
        futures::stream::unfold(
            (sub, member, shared, size, None::<KafkaError>),
            async move |(sub, member, shared, size, failed)| {
                if let Some(err) = failed {
                    return Some((Err(err), (sub, member, shared, size, None)));
                }
                let first = match turn(&member, &shared.tracker).await {
                    Turn::Attended(Ok(delivery)) => delivery,
                    Turn::Fetched(Ok(record)) => deliver(sub, &shared, record),
                    Turn::Attended(Err(err)) | Turn::Fetched(Err(err)) => {
                        return Some((Err(err), (sub, member, shared, size, None)));
                    }
                };
                let mut batch = Vec::with_capacity(size.min(64));
                batch.push(first);
                let mut failed = None;
                while batch.len() < size {
                    match member.fetch() {
                        Some(Ok(record)) => batch.push(deliver(sub, &shared, record)),
                        // A failure inside an open batch yields the batch first, as the live
                        // subscriber does, and surfaces on the next poll.
                        Some(Err(err)) => {
                            failed = Some(err);
                            break;
                        }
                        None => break,
                    }
                }
                #[cfg(feature = "schema-registry")]
                let batch = {
                    let mut finished = Vec::with_capacity(batch.len());
                    for item in batch {
                        finished.push(finish(sub, item).await);
                    }
                    finished
                };
                Some((Ok(batch), (sub, member, shared, size, failed)))
            },
        )
    }
}