ruststream 0.7.3

//! Batch subscriber definitions: handlers that consume a whole decoded batch per invocation.
//!
//! Generated by a `&[T]`-taking `#[subscriber]` handler and mounted with
//! [`BrokerScope::include`](super::BrokerScope::include); the mount site requires the
//! source's subscriber to implement [`BatchSubscriber`](crate::BatchSubscriber) (natively, or via
//! the [`Buffered`](crate::Buffered) adapter). Mirrors the single-message pipeline:
//! [`BatchDef`] is the [`SubscriberDef`](super::SubscriberDef) counterpart, [`SliceHandler`] the
//! [`Handler`](super::Handler) counterpart, and [`TypedBatch`] the [`Typed`](super::Typed) decode
//! adapter.

#[cfg(feature = "otel")]
use std::sync::LazyLock;
use std::{fmt, future::Future, marker::PhantomData};

#[cfg(feature = "otel")]
use opentelemetry::metrics::Histogram;
#[cfg(feature = "otel")]
use opentelemetry::{KeyValue, global};
// Only the test-only `typed_batch` constructor still names the bound directly; the adapters
// reach decoding through `DecodeWith`.
#[cfg(all(test, feature = "memory", feature = "json"))]
use serde::de::DeserializeOwned;
use tracing::{error, warn};

use crate::{BuildBatchContext, IncomingMessage};

use super::context::Context;
use super::dispatch::{Delivery, Slot, Workers, settle_outcome};
use super::failure::{FailurePolicies, FailurePolicy};
use super::handle::Deserialized;
use super::handler::{HandlerOutcome, HandlerResult};
use super::input::{DecodeWith, InputKind, REPORTED};
use super::metadata::HandlerMetadata;

/// The settlement of one dispatched batch.
///
/// Returned by batch handlers (usually through [`IntoBatchResult`]) and applied by the
/// dispatcher to each message's own `ack` / `nack`.
///
/// Per-element [`HandlerOutcome`]s carry an optional `and_after` continuation each; the uniform
/// form carries at most one, which runs once after the whole batch is settled.
#[derive(Debug)]
#[non_exhaustive]
pub enum BatchResult {
    /// One outcome settles every message of the batch; its continuation (if any) runs once,
    /// after the last message is settled.
    Uniform(HandlerOutcome),
    /// Settlement `i` settles slice element `i`, each with its own optional post-settle
    /// continuation. A length mismatch with the dispatched batch is a bug in the handler: the
    /// unmatched remainder is retried (an extra redelivery beats a silently lost message) and the
    /// mismatch is logged.
    PerElement(Vec<HandlerOutcome>),
}

/// Conversion into a [`BatchResult`], so batch `#[subscriber]` handlers can return a plain
/// value.
///
/// Implemented for [`BatchResult`] (identity), [`HandlerOutcome`] / `()` / `Result<(), E>` /
/// `Result<HandlerOutcome, E>` (one outcome for the whole batch), and a per-element
/// `Vec<HandlerOutcome>` where element `i` settles slice element `i`.
pub trait IntoBatchResult {
    /// Converts `self` into the settlement the dispatcher applies.
    fn into_batch_result(self) -> BatchResult;
}

impl IntoBatchResult for BatchResult {
    fn into_batch_result(self) -> BatchResult {
        self
    }
}

impl IntoBatchResult for HandlerOutcome {
    fn into_batch_result(self) -> BatchResult {
        BatchResult::Uniform(self)
    }
}

impl IntoBatchResult for () {
    fn into_batch_result(self) -> BatchResult {
        BatchResult::Uniform(HandlerOutcome::ack())
    }
}

impl<E> IntoBatchResult for Result<(), E> {
    fn into_batch_result(self) -> BatchResult {
        BatchResult::Uniform(match self {
            Ok(()) => HandlerOutcome::ack(),
            Err(_) => HandlerOutcome::drop(),
        })
    }
}

impl<E> IntoBatchResult for Result<HandlerOutcome, E> {
    fn into_batch_result(self) -> BatchResult {
        BatchResult::Uniform(self.unwrap_or_else(|_| HandlerOutcome::drop()))
    }
}

impl IntoBatchResult for Vec<HandlerOutcome> {
    fn into_batch_result(self) -> BatchResult {
        BatchResult::PerElement(self)
    }
}

/// Lowers one accepted batch return shape to the canonical batch verdict of the unified body
/// contract: `Ok(())` acks the batch, a batch-length `Err` vector settles element-wise. The
/// `#[subscriber]` expansion converts through here, so a batch body keeps its whole return
/// vocabulary while the emitted `Handle` impl returns the fixed verdict. Machinery; not part of
/// the public API.
///
/// A uniform non-ack outcome fans out per element - the failure path, the one place the batch
/// vector exists - and its continuation rides the last element, keeping the at-most-once
/// post-settle contract.
#[doc(hidden)]
pub fn batch_verdict<O: IntoBatchResult>(
    outcome: O,
    batch_len: usize,
) -> Result<(), Vec<HandlerOutcome>> {
    match outcome.into_batch_result() {
        BatchResult::Uniform(one) => {
            if one.is_ack() && !one.has_after() {
                Ok(())
            } else {
                Err(uniform_batch(one, batch_len))
            }
        }
        BatchResult::PerElement(outcomes) => Err(outcomes),
    }
}

/// Fans one uniform outcome out to a batch-length settlement vector: the status repeats per
/// element and the continuation rides the last one. The failure-path half of [`batch_verdict`],
/// also how the `#[subscriber]` expansion lowers a batch reply body's uniform `Err`. Machinery;
/// not part of the public API.
#[doc(hidden)]
#[must_use]
pub fn uniform_batch(outcome: HandlerOutcome, batch_len: usize) -> Vec<HandlerOutcome> {
    if batch_len == 0 {
        return Vec::new();
    }
    let status = outcome.outcome();
    let mut settles: Vec<HandlerOutcome> = std::iter::repeat_with(|| HandlerOutcome::from(status))
        .take(batch_len - 1)
        .collect();
    settles.push(outcome);
    settles
}

/// A handler invoked with one whole decoded batch.
///
/// The batch parameter is a slice: per-message broker handles stay with the dispatcher, which
/// settles every message of the batch according to the returned [`BatchResult`] - one uniform
/// outcome, or one outcome per element. Closures over `(&[T], &mut Context)` returning any
/// [`IntoBatchResult`] value implement it through the blanket impl below.
#[diagnostic::on_unimplemented(
    message = "`{Self}` is not a batch handler over `{T}`",
    note = "a batch handler takes (&[T], &mut Context) and returns an IntoBatchResult value; \
            per-element header contracts ride a `&[Message<H, T>]` body instead"
)]
pub(crate) trait SliceHandler<T, C = (), S = ()>: Send + Sync {
    /// Handles one decoded batch, with the per-batch [`Context`] carrying the broker's
    /// subscription-scoped context `C` and the typed app state `S`.
    fn handle_slice(
        &self,
        batch: &[T],
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = BatchResult> + Send;
}

impl<T, F, Fut, C, S> SliceHandler<T, C, S> for F
where
    F: Fn(&[T], &mut Context<'_, C, S>) -> Fut + Send + Sync,
    Fut: Future + Send,
    Fut::Output: IntoBatchResult,
{
    fn handle_slice(
        &self,
        batch: &[T],
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = BatchResult> + Send {
        // Build the inner future before the async block so the returned future does not hold
        // `&[T]` (which would demand `T: Sync` for it to be `Send`).
        let fut = (self)(batch, ctx);
        async move { fut.await.into_batch_result() }
    }
}

/// A batch handler definition produced by a `&[T]`-taking `#[subscriber]` handler.
///
/// The batch counterpart of [`SubscriberDef`](super::SubscriberDef): same metadata surface, but
/// the handler consumes `&[Input]` and the mount site drives
/// [`BatchSubscriber::batches`](crate::BatchSubscriber::batches) instead of
/// [`Subscriber::stream`](crate::Subscriber::stream).
pub trait BatchDef: Sized {
    /// The input kind of one batch element ([`Decoded<T>`](super::Decoded) for a `serde`
    /// element, [`Provided<F>`](super::Provided) for a
    /// [`Deserialized`](crate::runtime::Deserialized) one). The handler consumes a slice of the
    /// elements, `&[T]`.
    type Input: InputKind;

    /// The broker's typed subscription-scoped context the batch's handler reads by key (`()`
    /// when the handler names none). Built once per batch from its first delivery (see
    /// [`BuildBatchContext`](crate::BuildBatchContext)): a batch spans many deliveries, so only
    /// data every delivery of the subscription shares belongs here.
    type Context;

    /// The concrete handler type over batches of the element type.
    ///
    /// As for [`SubscriberDef::Handler`](super::SubscriberDef::Handler), the state-typed bound is
    /// enforced where the def is mounted, not on the trait: a batch handler that reads typed state
    /// is [`SliceHandler<T, St>`](SliceHandler) only for its declared `St`, while one that
    /// ignores state is generic over it.
    type Handler;

    /// The subscription source this handler binds to (see
    /// [`SubscriberDef::Source`](super::SubscriberDef::Source)).
    type Source;

    /// Builds the subscription source (fresh each call).
    fn source(&self) -> Self::Source;

    /// The concurrency policy for this subscriber's dispatch loop (how many batches are in
    /// flight at once). The macro fills this in from the `workers(..)` argument; the default is
    /// sequential dispatch.
    fn workers(&self) -> Workers {
        Workers::sequential()
    }

    /// The failure policy for a batch-handler panic and a per-element decode failure. The macro
    /// fills this in from the `on_failure(panic = .., decode = ..)` argument; the default fails
    /// fast on a panic and drops on a decode failure.
    fn failure_policies(&self) -> FailurePolicies {
        FailurePolicies::default()
    }

    /// An optional human description (from the handler's doc comment), for `AsyncAPI`.
    fn description(&self) -> Option<&str> {
        None
    }

    /// The element type's serialized JSON Schema, when it implements [`schemars::JsonSchema`] and
    /// the `asyncapi` feature is on. The macro fills this in; the default omits it.
    fn input_schema(&self) -> Option<String> {
        None
    }

    /// The serialized JSON Schema of the element type's header contract, when one is declared
    /// and the `asyncapi` feature is on. The default omits it.
    fn headers_schema(&self) -> Option<String> {
        None
    }

    /// The element type's [`Message`](crate::MessageInfo) name, when it implements that trait. The
    /// macro fills this in; the default omits it.
    fn message_name(&self) -> Option<&'static str> {
        None
    }

    /// The element type's [`Message`](crate::MessageInfo) description, when it implements that trait.
    /// The macro fills this in; the default omits it.
    fn message_description(&self) -> Option<&'static str> {
        None
    }

    /// Consumes the definition, returning the handler.
    fn into_handler(self) -> Self::Handler;
}

/// Builds the registration metadata for a batch definition mounted under `name`.
pub(crate) fn batch_metadata<D: BatchDef>(name: String, def: &D) -> HandlerMetadata {
    let mut meta = HandlerMetadata::raw(name).with_def_details(
        def.description(),
        def.input_schema(),
        def.headers_schema(),
        def.message_name(),
        def.message_description(),
    );
    meta.input_type = <D::Input as InputKind>::input_label();
    meta.deserialized = <D::Input as InputKind>::DESERIALIZED;
    meta
}

/// One batch as its handler takes it: what its context is built from, and what the harness
/// records of it when the handler panics before settling anything.
pub(crate) trait TakenBatch<C>: Send {
    /// The batch's context, or `None` for a batch with nothing in it: such a batch runs nothing.
    fn context(&self) -> Option<C>;

    /// The deliveries a panicking handler left unsettled, as the harness records them. Taken only
    /// while a harness watches the dispatch; a batch that lends its values keeps the copies it
    /// makes here for the record of its settlement, so each value is cloned once.
    #[cfg(feature = "testing")]
    fn unsettled(&mut self) -> Vec<crate::testing::coordinator::Delivered>;
}

// The decoding forms collect the deliveries, and a batch's context is built from its first.
impl<M, C> TakenBatch<C> for Vec<M>
where
    M: IncomingMessage,
    C: BuildBatchContext<M>,
{
    fn context(&self) -> Option<C> {
        self.first().map(C::build)
    }

    #[cfg(feature = "testing")]
    fn unsettled(&mut self) -> Vec<crate::testing::coordinator::Delivered> {
        self.iter()
            .map(|msg| crate::testing::coordinator::Delivered {
                raw: bytes::Bytes::copy_from_slice(msg.payload()),
                value: None,
                settle: None,
            })
            .collect()
    }
}

/// The dispatch-side consumer of one batch the subscription delivered: decode, run the handler,
/// settle every delivery. The batch counterpart of [`Handler`](super::Handler) at the
/// raw-message level.
pub(crate) trait BatchHandler<Batch, C = (), S = ()>: Send + Sync {
    /// The buffer the decoded batch is built in.
    ///
    /// A dispatch loop keeps one and lends it to every batch it hands over, so the slice the
    /// handler reads costs one allocation for the subscription rather than one per batch. It is
    /// `()` on the forms whose values the batch already holds or borrows, which cannot outlive
    /// the batch they came from.
    type Scratch: Default + Send;

    /// The batch as this handler takes it: its deliveries collected, or the broker's batch whole
    /// where the handler lends the values the batch holds.
    type Taken: TakenBatch<C>;

    /// Takes one batch off the subscription.
    fn take(batch: Batch) -> Self::Taken;

    /// Consumes one batch, acknowledging each of its deliveries, decoding into the buffer the
    /// caller lends. Whatever the buffer holds on entry is discarded.
    fn handle_batch(
        &self,
        batch: Self::Taken,
        scratch: &mut Self::Scratch,
        ctx: &mut Context<'_, C, S>,
    ) -> impl Future<Output = ()> + Send;
}

/// Hands `handler` one batch of deliveries the way the dispatch loop does once it took them off
/// the subscription: collected, the shape every decoding batch adapter takes. The adapter tests'
/// entry point, which names the batch type the trait leaves open.
#[cfg(all(test, feature = "memory", feature = "json"))]
pub(crate) async fn handled<H, M, C, S>(
    handler: &H,
    batch: Vec<M>,
    scratch: &mut H::Scratch,
    ctx: &mut Context<'_, C, S>,
) where
    H: BatchHandler<Vec<M>, C, S, Taken = Vec<M>>,
    M: Send,
    C: Send + Sync,
    S: Send + Sync,
{
    handler.handle_batch(batch, scratch, ctx).await;
}

/// Build a [`TypedBatch`] that decodes each element with `codec` into `T` and forwards the batch
/// as `&[T]` to `inner`. The adapter tests' constructor; the mounts name the input kind
/// explicitly instead.
#[cfg(all(test, feature = "memory", feature = "json"))]
pub(crate) fn typed_batch<M, T, C, H>(
    codec: C,
    inner: H,
) -> TypedBatch<M, super::input::Decoded<T>, C, H>
where
    M: IncomingMessage,
    T: DeserializeOwned + Send + Sync + 'static,
    C: crate::codec::Codec,
{
    TypedBatch {
        codec,
        inner,
        decode: FailurePolicy::Drop,
        _phantom: PhantomData,
    }
}

/// The decode adapter for batches, the [`Typed`](super::Typed) counterpart, generic over the
/// element's input kind.
///
/// Each element decodes independently: failures are settled individually per the `decode`
/// [`FailurePolicy`] and never reach the handler; the rest are passed on as one slice. The
/// [`BatchResult`] the handler returns settles the deliveries behind that slice - uniformly, or
/// element by element.
pub struct TypedBatch<M, Input, DecodeCodec, Inner> {
    codec: DecodeCodec,
    inner: Inner,
    decode: FailurePolicy,
    _phantom: PhantomData<fn(M, Input)>,
}

impl<M, Input, DecodeCodec, Inner> TypedBatch<M, Input, DecodeCodec, Inner> {
    /// Builds the adapter for any element input kind, like [`Typed::over`](super::Typed::over).
    #[must_use]
    pub(crate) fn over(codec: DecodeCodec, inner: Inner) -> Self
    where
        Input: DecodeWith<DecodeCodec>,
    {
        Self {
            codec,
            inner,
            decode: FailurePolicy::Drop,
            _phantom: PhantomData,
        }
    }

    /// Sets the policy applied when the codec fails to decode an element.
    #[must_use]
    pub(crate) fn with_decode(mut self, decode: FailurePolicy) -> Self {
        self.decode = decode;
        self
    }
}

impl<M, Input, DecodeCodec, Inner> fmt::Debug for TypedBatch<M, Input, DecodeCodec, Inner> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TypedBatch")
            .field("decode", &self.decode)
            .finish_non_exhaustive()
    }
}

impl<Batch, M, Input, DecodeCodec, Inner, C, S> BatchHandler<Batch, C, S>
    for TypedBatch<M, Input, DecodeCodec, Inner>
where
    Batch: IntoIterator<Item = M>,
    M: IncomingMessage,
    Input: DecodeWith<DecodeCodec>,
    DecodeCodec: Send + Sync,
    Inner: SliceHandler<Input::Owned, C, S>,
    C: BuildBatchContext<M> + Send + Sync + 'static,
    S: Send + Sync,
{
    type Scratch = Vec<Input::Owned>;
    type Taken = Vec<M>;

    fn take(batch: Batch) -> Vec<M> {
        batch.into_iter().collect()
    }

    async fn handle_batch(
        &self,
        batch: Vec<M>,
        values: &mut Self::Scratch,
        ctx: &mut Context<'_, C, S>,
    ) {
        let subscription = ctx.subscription();
        let accepted = decode_batch::<M, Input, DecodeCodec, C, S>(
            batch,
            values,
            &self.codec,
            self.decode,
            ctx,
        )
        .await;
        if accepted.is_empty() {
            return;
        }
        let delivery = ctx.delivery();
        let result = self.inner.handle_slice(values, ctx).await;
        // The buffer belongs to the caller and outlives the call, so the values are dropped
        // here: a decoded value lives exactly as long as the batch it came from, as it did when
        // the vector was the call's own.
        values.clear();
        settle_batch(accepted, result, subscription, delivery).await;
    }
}

/// The batch adapter of the self-deserializing input kind: no codec runs; each element of the
/// batch constructs itself from its delivery's payload
/// ([`Deserialized`](crate::runtime::Deserialized)), borrowing it.
///
/// An element whose construction fails is settled by the `decode` [`FailurePolicy`] and never
/// reaches the handler, exactly as a codec decode failure on the typed path; the rest reach the
/// body as one batch, and the returned [`BatchResult`] settles the deliveries behind it.
pub struct DeserializedBatch<M, F, Inner> {
    inner: Inner,
    decode: FailurePolicy,
    _phantom: PhantomData<fn(M, F)>,
}

impl<M, F, Inner> DeserializedBatch<M, F, Inner> {
    /// Builds the adapter over the batch handler, like [`TypedBatch::over`].
    #[must_use]
    pub(crate) fn over(inner: Inner) -> Self {
        Self {
            inner,
            decode: FailurePolicy::Drop,
            _phantom: PhantomData,
        }
    }

    /// Sets the policy applied when an element's construction fails.
    #[must_use]
    pub(crate) fn with_decode(mut self, decode: FailurePolicy) -> Self {
        self.decode = decode;
        self
    }
}

impl<M, F, Inner> fmt::Debug for DeserializedBatch<M, F, Inner> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DeserializedBatch")
            .field("decode", &self.decode)
            .finish_non_exhaustive()
    }
}

impl<Batch, M, F, Inner, C, S> BatchHandler<Batch, C, S> for DeserializedBatch<M, F, Inner>
where
    Batch: IntoIterator<Item = M>,
    M: IncomingMessage,
    F: Deserialized + Send + Sync + 'static,
    Inner: for<'p> SliceHandler<F::Output<'p>, C, S>,
    C: BuildBatchContext<M> + Send + Sync + 'static,
    S: Send + Sync,
{
    // The constructed values borrow their deliveries' payloads (see the loop below), so a
    // buffer that outlived the batch could not hold them: this form builds its values vector per
    // batch, and the borrow is what forces it.
    type Scratch = ();
    type Taken = Vec<M>;

    fn take(batch: Batch) -> Vec<M> {
        batch.into_iter().collect()
    }

    async fn handle_batch(&self, batch: Vec<M>, _scratch: &mut (), ctx: &mut Context<'_, C, S>) {
        let subscription = ctx.subscription();
        if batch.is_empty() {
            return;
        }
        let delivery = ctx.delivery();
        // The constructed values borrow the deliveries' payloads, so the deliveries stay owned
        // by `batch` and the values are dropped before anything settles. Rejections are settled
        // after the batch runs, which keeps the accepted values contiguous with no second pass.
        let mut values = Vec::with_capacity(batch.len());
        // (index into `batch`, its settlement); empty on the happy path, so nothing allocates.
        let mut rejected: Vec<(usize, HandlerResult)> = Vec::new();
        for (index, msg) in batch.iter().enumerate() {
            if let Some(outcome) = reported(msg, std::any::type_name::<F>(), self.decode, ctx) {
                rejected.push((index, outcome));
                continue;
            }
            match F::from_payload(msg.payload()) {
                Ok(value) => values.push(value),
                Err(err) => {
                    warn!(
                        target: "ruststream::dispatch",
                        subscription = %subscription,
                        message_type = std::any::type_name::<F>(),
                        error = %err,
                        "payload construction failed",
                    );
                    let outcome =
                        rejection(&err, "batch payload construction failed", self.decode, ctx);
                    rejected.push((index, outcome));
                }
            }
        }
        #[cfg(feature = "otel")]
        if !values.is_empty() {
            record_batch_size(subscription, values.len());
        }
        let result = if values.is_empty() {
            BatchResult::PerElement(Vec::new())
        } else {
            self.inner.handle_slice(&values, ctx).await
        };
        drop(values);
        settle_split_batch(batch, rejected, result, subscription, delivery).await;
    }
}

/// Settles a batch whose rejected elements were deferred past the handler call: each rejected
/// index settles by its recorded outcome, and the accepted remainder settles by the handler's
/// [`BatchResult`] in order, exactly as [`settle_batch`] applies it.
async fn settle_split_batch<M, C>(
    batch: Vec<M>,
    rejected: Vec<(usize, HandlerResult)>,
    result: BatchResult,
    subscription: &str,
    delivery: &Delivery<C>,
) where
    M: IncomingMessage,
    C: BuildBatchContext<M> + Send + Sync + 'static,
{
    if rejected.is_empty() {
        return settle_batch(batch, result, subscription, delivery).await;
    }
    let accepted_len = batch.len() - rejected.len();
    let per_element = match result {
        // Fan the uniform outcome out over the accepted elements only; the rejects keep their
        // own settlements.
        BatchResult::Uniform(outcome) => uniform_batch(outcome, accepted_len),
        BatchResult::PerElement(results) => {
            if results.len() != accepted_len {
                error!(
                    target: "ruststream::dispatch",
                    subscription = %subscription,
                    expected = accepted_len,
                    returned = results.len(),
                    "per-element outcome count does not match the batch; \
                     retrying the unmatched remainder",
                );
            }
            results
        }
    };
    // The body saw the accepted elements, so they are the call the harness records. A batch that
    // refused every element never reached the body, and records nothing.
    #[cfg(feature = "testing")]
    let mut log = (accepted_len > 0).then(|| {
        BatchLog::of(
            batch
                .iter()
                .enumerate()
                .filter(|(index, _)| rejected.binary_search_by_key(index, |&(at, _)| at).is_err())
                .map(|(_, msg)| msg),
            LentValues::default(),
        )
    });
    let mut accepted_results = per_element.into_iter();
    let mut rejected = rejected.into_iter().peekable();
    for (index, msg) in batch.into_iter().enumerate() {
        if rejected.peek().is_some_and(|(at, _)| *at == index) {
            let (_, outcome) = rejected.next().expect("peeked");
            settle_outcome(
                &mut Slot::new(msg),
                outcome,
                subscription,
                delivery,
                C::build as fn(&M) -> C,
            )
            .await;
            continue;
        }
        let mut result = accepted_results
            .next()
            .unwrap_or_else(HandlerOutcome::retry);
        let after = result.take_after();
        #[cfg(feature = "testing")]
        if let Some(log) = log.as_mut() {
            log.settled(result.outcome());
        }
        settle_outcome(
            &mut Slot::new(msg),
            result.outcome(),
            subscription,
            delivery,
            C::build as fn(&M) -> C,
        )
        .await;
        if let Some(after) = after {
            delivery.spawn_after(after);
        }
    }
    #[cfg(feature = "testing")]
    if let Some(log) = log {
        log.record(subscription);
    }
}

/// Applies one [`BatchResult`] to the accepted deliveries behind it - uniformly, or element by
/// element, spawning each per-element continuation on the tracked set after its settle.
pub(crate) async fn settle_batch<M, C>(
    accepted: Vec<M>,
    result: BatchResult,
    subscription: &str,
    delivery: &Delivery<C>,
) where
    M: IncomingMessage,
    C: BuildBatchContext<M> + Send + Sync + 'static,
{
    settle_lent_batch(
        accepted,
        result,
        subscription,
        delivery,
        LentValues::default(),
    )
    .await;
}

/// What the harness records of the values a batch lent its handler, one per accepted delivery
/// in order: empty where the batch lent none, and nothing at all outside the `testing` feature.
#[cfg(feature = "testing")]
pub(crate) type LentValues = Vec<crate::testing::coordinator::RecordedValue>;

/// See the `testing` counterpart: without the harness there is nothing to record.
#[cfg(not(feature = "testing"))]
pub(crate) type LentValues = ();

/// [`settle_batch`] for a batch that lent its handler the values its deliveries carry: the
/// harness records each value beside the delivery it belongs to.
pub(crate) async fn settle_lent_batch<M, C>(
    accepted: Vec<M>,
    result: BatchResult,
    subscription: &str,
    delivery: &Delivery<C>,
    lent: LentValues,
) where
    M: IncomingMessage,
    C: BuildBatchContext<M> + Send + Sync + 'static,
{
    // Every batch form funnels its batch through here, which is the one place that knows both
    // the deliveries and the settlements they got; the harness reads the batch off it.
    #[cfg(feature = "testing")]
    let mut batch = BatchLog::of(&accepted, lent);
    #[cfg(not(feature = "testing"))]
    let () = lent;
    match result {
        BatchResult::Uniform(mut outcome) => {
            let after = outcome.take_after();
            let status = outcome.outcome();
            for msg in accepted {
                #[cfg(feature = "testing")]
                batch.settled(status);
                settle_outcome(
                    &mut Slot::new(msg),
                    status,
                    subscription,
                    delivery,
                    C::build as fn(&M) -> C,
                )
                .await;
            }
            // The one uniform continuation runs after the whole batch is settled, on the
            // tracked set so a graceful shutdown drains it (at-most-once, like the
            // per-element ones below).
            if let Some(after) = after {
                delivery.spawn_after(after);
            }
        }
        BatchResult::PerElement(results) => {
            if results.len() != accepted.len() {
                error!(
                    target: "ruststream::dispatch",
                    subscription = %subscription,
                    expected = accepted.len(),
                    returned = results.len(),
                    "per-element outcome count does not match the batch; \
                     retrying the unmatched remainder",
                );
            }
            let mut results = results.into_iter();
            for msg in accepted {
                // An unmatched message gets retried: an extra redelivery beats losing it.
                let mut result = results.next().unwrap_or_else(HandlerOutcome::retry);
                let after = result.take_after();
                #[cfg(feature = "testing")]
                batch.settled(result.outcome());
                settle_outcome(
                    &mut Slot::new(msg),
                    result.outcome(),
                    subscription,
                    delivery,
                    C::build as fn(&M) -> C,
                )
                .await;
                // The continuation runs after this element is settled, on the tracked set so a
                // graceful shutdown drains it. At-most-once: a lost or panicking continuation
                // never redelivers the already-settled message.
                if let Some(after) = after {
                    delivery.spawn_after(after);
                }
            }
        }
    }
    #[cfg(feature = "testing")]
    batch.record(subscription);
}

/// The batch a settle is applying, captured for the harness: the payloads before the
/// deliveries are consumed, then one settlement per element as it is applied.
///
/// The settlements are what release the harness's in-flight count, and the record is only
/// complete once the last of them is applied - so the log holds a count of its own for as long
/// as it lives, and a quiescence wait cannot return between the final settle and the record.
#[cfg(feature = "testing")]
struct BatchLog {
    deliveries: Vec<crate::testing::coordinator::Delivered>,
    settled: usize,
    _pending: crate::testing::coordinator::PendingRecord,
}

#[cfg(feature = "testing")]
impl BatchLog {
    fn of<'m, M: IncomingMessage + 'm>(
        accepted: impl IntoIterator<Item = &'m M>,
        lent: LentValues,
    ) -> Self {
        let mut lent = lent.into_iter();
        Self {
            deliveries: accepted
                .into_iter()
                .map(|msg| crate::testing::coordinator::Delivered {
                    raw: bytes::Bytes::copy_from_slice(msg.payload()),
                    value: lent.next(),
                    settle: None,
                })
                .collect(),
            settled: 0,
            _pending: crate::testing::coordinator::PendingRecord::new(),
        }
    }

    fn settled(&mut self, status: HandlerResult) {
        if let Some(one) = self.deliveries.get_mut(self.settled) {
            one.settle = Some(status);
        }
        self.settled += 1;
    }

    fn record(self, subscription: &str) {
        crate::testing::coordinator::record_batch(subscription, self.deliveries);
    }
}

/// The `ruststream.batch.size` histogram: decoded batch sizes, per destination. Batch handlers
/// bypass the per-message otel layer (the documented middleware exception), so the metric is
/// recorded at the batch decode choke point every batch form passes through instead.
///
/// Built off the process-global meter, not an `Otel` handle. Until
/// `Otel::builder().init()` installs the SDK's global provider the global meter is
/// a no-op, so an app without the exporters records nothing; the instrument binds to whatever
/// provider is global at the first dispatched batch, and the documented wiring runs `init()`
/// before `app.start()`.
#[cfg(feature = "otel")]
static BATCH_SIZE: LazyLock<Histogram<u64>> = LazyLock::new(|| {
    global::meter("ruststream")
        .u64_histogram("ruststream.batch.size")
        .with_unit("{message}")
        .with_description("Decoded batch sizes handed to batch handlers.")
        .build()
});

/// Records one decoded batch's size under its destination.
#[cfg(feature = "otel")]
pub(crate) fn record_batch_size(destination: &str, len: usize) {
    // Batches beyond u64 elements do not exist.
    #[allow(clippy::cast_possible_truncation)]
    BATCH_SIZE.record(
        len as u64,
        &[KeyValue::new(
            "messaging.destination.name",
            destination.to_owned(),
        )],
    );
}

/// Decodes each element of one raw batch independently into `values`, returning the deliveries
/// behind them: failures are settled per the `decode` [`FailurePolicy`] and never reach the
/// handler; the rest pass through, each decoded value paired with its delivery (`values[i]`
/// decodes element `i` of the returned batch). A `fail_fast` decode policy tears the service down
/// via `ctx` and drops the offending element so it is not requeued into the failure.
///
/// `values` is the caller's buffer, emptied before it is filled: a buffer that already has the
/// room for the batch reserves nothing, which is what keeps a batch off the allocator.
///
/// Takes `&mut Context` (rather than `&Context`) so the future stays `Send`: `Context` carries
/// post-settle hooks (boxed `Send` futures that are not `Sync`), so holding a shared `&Context`
/// across an `.await` would wrongly require `Context: Sync`.
// The `&mut` is for Send-ness, not mutation (only `&self` methods are called), so the lint fires.
#[allow(clippy::needless_pass_by_ref_mut)]
pub(crate) async fn decode_batch<M, Input, DecodeCodec, C, S>(
    batch: Vec<M>,
    values: &mut Vec<Input::Owned>,
    codec: &DecodeCodec,
    decode: FailurePolicy,
    ctx: &mut Context<'_, C, S>,
) -> Vec<M>
where
    M: IncomingMessage,
    Input: DecodeWith<DecodeCodec>,
    DecodeCodec: Sync,
    C: BuildBatchContext<M> + Send + Sync + 'static,
    S: Send + Sync,
{
    let subscription = ctx.subscription();
    // Taken before the loop: the settle below awaits, and a borrow of `ctx` held across it would
    // demand `Context: Sync` (see the signature's note).
    let delivery = ctx.delivery();
    values.clear();
    values.reserve(batch.len());
    // (index into `batch`, its settlement). Empty while every element decodes, and an empty
    // `Vec` holds no buffer: a batch that decodes whole is handed on exactly as it was
    // delivered, so the accepted deliveries cost nothing to carry.
    let mut rejected: Vec<(usize, HandlerResult)> = Vec::new();
    for (index, msg) in batch.iter().enumerate() {
        if let Some(outcome) = reported(msg, Input::input_label(), decode, ctx) {
            rejected.push((index, outcome));
            continue;
        }
        match Input::decode(codec, msg) {
            Ok(value) => values.push(value),
            Err(err) => {
                warn!(
                    target: "ruststream::dispatch",
                    subscription = %subscription,
                    message_type = Input::input_label(),
                    error = %err,
                    "codec decode failed",
                );
                rejected.push((index, rejection(&err, "batch decode failed", decode, ctx)));
            }
        }
    }
    // Only batches that reach a handler count: a fully-rejected batch is short-circuited by the
    // callers and would skew the size distribution with zeros.
    #[cfg(feature = "otel")]
    if !values.is_empty() {
        record_batch_size(subscription, values.len());
    }
    if rejected.is_empty() {
        return batch;
    }
    // A rejected element is settled before the handler runs, as it always was, and the accepted
    // remainder is collected on the way past it.
    let mut accepted = Vec::with_capacity(batch.len() - rejected.len());
    let mut rejected = rejected.into_iter().peekable();
    for (index, msg) in batch.into_iter().enumerate() {
        if rejected.peek().is_some_and(|(at, _)| *at == index) {
            let (_, outcome) = rejected.next().expect("peeked");
            settle_outcome(
                &mut Slot::new(msg),
                outcome,
                subscription,
                delivery,
                <C as BuildBatchContext<M>>::build as fn(&M) -> C,
            )
            .await;
            continue;
        }
        accepted.push(msg);
    }
    accepted
}

/// The settlement of a batch element its broker reports as undecodable
/// ([`IncomingMessage::decode_error`]), logged with the subscription, the element type and the
/// reported error; `None` where the broker reports nothing and the lane reads the element itself.
///
/// Every batch lane asks here before it reads an element, so the handler never sees one its broker
/// refused. A broker that keeps the default reports nothing, and the check folds away.
pub(crate) fn reported<M, C, S>(
    msg: &M,
    message_type: &str,
    decode: FailurePolicy,
    ctx: &Context<'_, C, S>,
) -> Option<HandlerResult>
where
    M: IncomingMessage,
{
    let err = msg.decode_error()?;
    warn!(
        target: "ruststream::dispatch",
        subscription = %ctx.subscription(),
        message_type,
        error = %err,
        "{}",
        REPORTED,
    );
    Some(rejection(err, "batch decode failed", decode, ctx))
}

/// The settlement of one delivery the handler will never see, per the subscriber's decode policy.
/// Shared by the payload decode, the header contract and a delivery its broker could not read,
/// single or in a batch, which are the same class of bad external input. Not `async`: the outcome
/// is decided before the delivery is settled, so the borrowed error never crosses an await.
pub(crate) fn rejection<C, S>(
    err: &impl fmt::Display,
    reason: &str,
    decode: FailurePolicy,
    ctx: &Context<'_, C, S>,
) -> HandlerResult {
    match decode {
        FailurePolicy::FailFast => {
            ctx.fail_fast(&format!("{reason}: {err}"));
            HandlerResult::drop()
        }
        other => other.settlement().unwrap_or_else(HandlerResult::drop),
    }
}

mod carried;

pub(crate) use carried::{CarriedBatch, SourceCarriesBatch};

#[cfg(all(test, feature = "memory", feature = "json"))]
mod tests;