magnetar/

multi_topics.rs

// SPDX-License-Identifier: Apache-2.0

//! Multi-topics consumer — subscribes to N topics and merges their delivery streams.
//!
//! Mirrors Java's `MultiTopicsConsumerImpl`. The consumer is a thin coordinator over a
//! `Vec<C>` (where `C: crate::ConsumerApi`) with `receive()` returning the first message
//! ready across all underlying consumers. Cancelling the future leaves un-popped messages
//! in their respective consumer queues — see the `cancel-safe` discussion in
//! [`magnetar_runtime_tokio::Consumer::receive`].
//!
//! Engine genericity (pass-2)
//! --------------------------
//! [`MultiTopicsConsumer<C>`] is generic over `C: crate::ConsumerApi` so the same
//! coordinator drives the tokio and moonpool runtimes. The companion
//! [`MultiTopicsConsumerBuilder<'a, E>`] is generic over `E: crate::Engine` (default
//! [`crate::TokioEngine`]) and routes `.subscribe()` through the engine-generic
//! [`crate::ConsumerBuilder`] (which dispatches through
//! [`crate::SubscribeApi`]) so each per-topic child uses the engine's concrete
//! consumer type. Tokio-only methods (currently none) would live in
//! `impl MultiTopicsConsumer<magnetar_runtime_tokio::Consumer>` carve-outs.
//!
//! Dynamic membership
//! ------------------
//! The consumer set is held under a [`parking_lot::Mutex`], so callers can subscribe new
//! topics via [`MultiTopicsConsumer::add_topic`] and tear them down via
//! [`MultiTopicsConsumer::remove_topic`] without rebuilding the consumer. New topics inherit
//! every knob set on the original [`MultiTopicsConsumerBuilder`] (captured as a template
//! inside [`Inner`], mirroring `PatternConsumer`). Mirrors Java
//! `MultiTopicsConsumerImpl#subscribeAsync(String)` / `#unsubscribeAsync(String)`.
//!
//! No regex / pattern subscription (yet); callers pass an explicit topic list. Regex /
//! pattern support layers on top via a broker-side topic-list-watcher (PIP-145), which is
//! exposed by [`magnetar_proto::Connection`] but not wired through this facade — see
//! [`crate::PatternConsumer`].

use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use futures_util::future::select_all;
use magnetar_proto::{IncomingMessage, MessageId};
use parking_lot::Mutex;
use tokio::sync::Notify;

use crate::auto_update_task::{AutoUpdateTask, spawn_auto_update_task};
use crate::client::{PulsarError, SeekTarget};
use crate::consumer_template::ConsumerTemplate;
use crate::{ConsumerApi, Engine, PulsarClient, SubscribeApi};

/// Multi-topics consumer. Each contained consumer subscribes to one topic; `receive()`
/// returns the next message across the whole set.
///
/// Generic over `C: crate::ConsumerApi` — the default is the tokio runtime's
/// `Consumer`. The companion [`MultiTopicsConsumerBuilder<'a, E>`] selects the
/// engine and produces a `MultiTopicsConsumer<<E::ClientState as
/// SubscribeApi>::Consumer>` on `.subscribe()`.
#[derive(Debug)]
pub struct MultiTopicsConsumer<C: ConsumerApi = magnetar_runtime_tokio::Consumer> {
    inner: Arc<Inner<C>>,
}

#[derive(Debug)]
struct Inner<C: ConsumerApi> {
    /// Active consumer set. Held under a mutex so [`MultiTopicsConsumer::add_topic`] /
    /// [`MultiTopicsConsumer::remove_topic`] can mutate the set without rebuilding the
    /// consumer. The Vec is wrapped in an `Arc` so the many read-heavy callers
    /// (`receive`, `seek_*`, `unsubscribe`, `last_message_ids`, stats aggregators)
    /// snapshot via a cheap `Arc::clone` instead of a deep `Vec::clone`; writers use
    /// `Arc::make_mut` to copy-on-write the Vec. The mutex is never held across
    /// `.await` — readers drop the guard immediately after capturing the Arc.
    consumers: Mutex<Arc<Vec<NamedConsumer<C>>>>,
    /// Signalled whenever a child consumer is added to the set
    /// ([`MultiTopicsConsumer::add_topic`]). The push-delivery poller
    /// ([`crate::consumer_listener::spawn_wrapper_message_listener`]) races its
    /// in-flight `receive()` against this so a child discovered *after* the poller
    /// parked is swept on the next iteration (pattern-child / partition-growth
    /// inheritance, ADR-0064). `Notify` (not a channel) keeps ADR-0003 intact; it
    /// stores one permit, so a signal that arrives between two waits is not lost.
    membership_changed: Arc<Notify>,
    /// Round-robin cursor used by `receive` to record the index of the topic that produced
    /// the last message. Wrapped in a Mutex because [`MultiTopicsConsumer`] is `&self` —
    /// cloning the handle should not require mutable access.
    cursor: std::sync::atomic::AtomicUsize,
    /// Template for subscribing newly-added topics. Captures every
    /// [`crate::ConsumerBuilder`] knob the user set on the original
    /// [`MultiTopicsConsumerBuilder`].
    template: ConsumerTemplate,
    /// Optional background partition-watcher task. `Some` when the builder
    /// configured
    /// [`MultiTopicsConsumerBuilder::auto_update_partitions_interval`], `None`
    /// otherwise (default). Tracks one watched topic — the "base" topic used by a
    /// [`crate::PartitionedConsumer`] (the underlying surface). Aborts on drop.
    ///
    /// The task uses [`tokio::task::JoinHandle`] + [`tokio::sync::Notify`]
    /// directly — both engines (`TokioEngine`, `MoonpoolEngine<P>`) schedule
    /// onto tokio under the hood (see `Engine::TaskHandle = tokio::task::JoinHandle`
    /// in `engine.rs`), so this scaffolding is engine-invariant.
    auto_update: Option<Arc<AutoUpdateTask>>,
}

#[derive(Debug, Clone)]
struct NamedConsumer<C: ConsumerApi> {
    topic: String,
    consumer: C,
}

async fn republish_snapshot<T, E, F, Fut, Topic>(
    members: &Mutex<Arc<Vec<T>>>,
    topic: Topic,
    mut republish: F,
) -> Result<usize, PulsarError>
where
    T: Clone,
    E: std::fmt::Display,
    F: FnMut(T) -> Fut,
    Fut: std::future::Future<Output = Result<usize, E>>,
    Topic: Fn(&T) -> &str,
{
    let snapshot = members.lock().clone();
    let mut republished = 0usize;
    for child in snapshot.iter().cloned() {
        let child_topic = topic(&child).to_owned();
        let count = republish(child).await.map_err(|error| {
            PulsarError::Other(format!(
                "republish_dead_letters for topic {child_topic}: {error}"
            ))
        })?;
        republished = republished.saturating_add(count);
    }
    Ok(republished)
}

/// A message yielded by [`MultiTopicsConsumer::receive`], carrying the topic it came from.
#[derive(Debug)]
pub struct MultiTopicsMessage {
    /// The topic the message originated from (the same string the caller supplied to the
    /// builder).
    pub topic: String,
    /// Underlying message + payload.
    pub message: IncomingMessage,
}

impl<C: ConsumerApi + Clone> MultiTopicsConsumer<C> {
    /// Topics this consumer is currently subscribed to, in the order they were added (initial
    /// builder order followed by [`Self::add_topic`] insertions, minus any topic removed via
    /// [`Self::remove_topic`]).
    #[must_use]
    pub fn topics(&self) -> Vec<String> {
        self.inner
            .consumers
            .lock()
            .iter()
            .map(|c| c.topic.clone())
            .collect()
    }

    /// Number of underlying consumers (one per topic).
    #[must_use]
    pub fn len(&self) -> usize {
        self.inner.consumers.lock().len()
    }

    /// `true` if the consumer set is currently empty (e.g. every topic has been removed).
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.inner.consumers.lock().is_empty()
    }

    /// Shared subscription name across every per-topic child. Mirrors Java
    /// `Consumer#getSubscription` at the multi-topic / partitioned scope.
    #[must_use]
    pub fn subscription(&self) -> &str {
        &self.inner.template.subscription
    }

    /// Subscribe a new per-topic child against the current consumer set. The new child
    /// inherits every knob configured on the original [`MultiTopicsConsumerBuilder`].
    /// Mirrors Java `MultiTopicsConsumerImpl#subscribeAsync(String topicName)`.
    ///
    /// Idempotent: if `topic` is already in the set the call is a no-op and returns `Ok(())`
    /// — mirrors Java's behaviour of refusing to double-subscribe the same topic.
    ///
    /// # Errors
    ///
    /// Returns the underlying subscribe error if the broker refuses the new subscription.
    /// The consumer set is left untouched on error.
    pub async fn add_topic<E>(
        &self,
        client: &PulsarClient<E>,
        topic: impl Into<String>,
    ) -> Result<(), PulsarError>
    where
        E: Engine,
        E::ClientState: SubscribeApi<Consumer = C>,
    {
        let topic = topic.into();
        // Check membership under the lock and release before awaiting — never hold the
        // mutex across an `.await`.
        let already_subscribed = self
            .inner
            .consumers
            .lock()
            .iter()
            .any(|nc| nc.topic == topic);
        if already_subscribed {
            return Ok(());
        }
        let builder = self.inner.template.apply(client.consumer(topic.clone()));
        let consumer = builder.subscribe().await?;
        // Re-check membership under the lock to handle a concurrent `add_topic(topic)` —
        // if a peer raced us, drop our newly-subscribed consumer and close it; otherwise
        // push it. The guard is released before any `.await`
        // (clippy::await_holding_lock).
        let to_close: Option<C> = {
            let mut guard = self.inner.consumers.lock();
            if guard.iter().any(|nc| nc.topic == topic) {
                Some(consumer)
            } else {
                // Copy-on-write the Vec inside the Arc so concurrent
                // readers holding an `Arc::clone` snapshot keep their
                // view; only the writer sees the new push.
                Arc::make_mut(&mut guard).push(NamedConsumer { topic, consumer });
                None
            }
        };
        if let Some(c) = to_close {
            let _ = ConsumerApi::close_owned(c).await;
        } else {
            // A new child joined the set — wake any parked wrapper-listener
            // poller so it re-snapshots and starts draining the new child.
            self.inner.membership_changed.notify_one();
        }
        Ok(())
    }

    /// Tear down the per-topic child subscribed to `topic` and remove it from the set.
    /// Mirrors Java `MultiTopicsConsumerImpl#unsubscribeAsync(String topicName)`.
    ///
    /// No-op if `topic` is not currently in the set.
    ///
    /// # Errors
    ///
    /// Returns the underlying close error from the per-topic consumer.
    pub async fn remove_topic(&self, topic: &str) -> Result<(), PulsarError> {
        // Remove under the lock, release, then close — never hold the mutex across await.
        let removed: Option<NamedConsumer<C>> = {
            let mut guard = self.inner.consumers.lock();
            let pos = guard.iter().position(|nc| nc.topic == topic);
            pos.map(|pos| Arc::make_mut(&mut guard).remove(pos))
        };
        if let Some(nc) = removed {
            ConsumerApi::close_owned(nc.consumer)
                .await
                .map_err(|e| PulsarError::Other(format!("remove_topic close: {e}")))?;
        }
        Ok(())
    }

    /// Negatively acknowledge a message. The caller supplies the topic the message came
    /// from (returned alongside the message in [`MultiTopicsMessage::topic`]) so the nack
    /// goes to the correct per-topic consumer.
    pub fn negative_ack(&self, topic: &str, message_id: MessageId) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("negative_ack: {err}")))?;
        consumer.negative_ack(message_id);
        Ok(())
    }

    /// Negatively acknowledge with an explicit per-message redelivery delay. Mirrors
    /// Java's PIP-37 backoff path at the multi-topic / partitioned scope. The caller
    /// supplies the topic the message came from so the nack routes to the correct child.
    pub fn negative_ack_with_delay(
        &self,
        topic: &str,
        message_id: MessageId,
        delay: std::time::Duration,
    ) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("negative_ack_with_delay: {err}")))?;
        consumer.negative_ack_with_delay(message_id, delay);
        Ok(())
    }

    /// Cumulative ack. The caller supplies the topic the message came from so the ack
    /// routes to the correct child. Mirrors Java
    /// `Consumer#acknowledgeCumulativeAsync(MessageId)` at the multi-topic scope.
    pub async fn ack_cumulative(
        &self,
        topic: &str,
        message_id: MessageId,
    ) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("ack_cumulative: {err}")))?;
        consumer
            .ack_cumulative(message_id)
            .await
            .map_err(|e| PulsarError::Other(format!("ack_cumulative: {e}")))
    }

    /// Fire-and-forget ack into the per-topic child's ack-grouping tracker (opt-in via
    /// `MultiTopicsConsumerBuilder::ack_group_time`). The caller supplies the topic the
    /// message came from so the ack routes to the correct child. See
    /// [`magnetar_runtime_tokio::Consumer::ack_grouped`].
    pub fn ack_grouped(&self, topic: &str, message_id: MessageId) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("ack_grouped: {err}")))?;
        consumer.ack_grouped(message_id);
        Ok(())
    }

    /// Fire-and-forget cumulative ack into the per-topic child's ack-grouping tracker.
    /// See [`Self::ack_grouped`] for the routing semantics.
    pub fn ack_grouped_cumulative(
        &self,
        topic: &str,
        message_id: MessageId,
    ) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("ack_grouped_cumulative: {err}")))?;
        consumer.ack_grouped_cumulative(message_id);
        Ok(())
    }

    /// Republish `msg` via `retry_producer` with a delay, then ack the original on the
    /// per-topic child. Mirrors Java `Consumer#reconsumeLater` at the multi-topic scope.
    /// The caller supplies the topic the message came from (returned alongside the
    /// message in [`MultiTopicsMessage::topic`]) so the ack routes to the correct child.
    pub async fn reconsume_later(
        &self,
        topic: &str,
        retry_producer: &C::Producer,
        msg: IncomingMessage,
        delay: std::time::Duration,
    ) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("reconsume_later: {err}")))?;
        consumer
            .reconsume_later(retry_producer, msg, delay)
            .await
            .map_err(|e| PulsarError::Other(format!("reconsume_later: {e}")))
    }

    /// Same as [`Self::reconsume_later`] but stamps custom properties on the republished
    /// message. Mirrors Java's properties-aware reconsumeLater overload.
    pub async fn reconsume_later_with_properties(
        &self,
        topic: &str,
        retry_producer: &C::Producer,
        msg: IncomingMessage,
        custom_properties: Vec<(String, String)>,
        delay: std::time::Duration,
    ) -> Result<(), PulsarError> {
        let consumer = self.lookup(topic).map_err(|err| {
            PulsarError::Config(format!("reconsume_later_with_properties: {err}"))
        })?;
        consumer
            .reconsume_later_with_properties(retry_producer, msg, custom_properties, delay)
            .await
            .map_err(|e| PulsarError::Other(format!("reconsume_later_with_properties: {e}")))
    }

    /// Republish every child consumer's buffered dead letters through one shared
    /// `dlq_producer` destination and return the saturating sum of republished messages.
    ///
    /// Each call snapshots membership independently when it starts. Children added later do
    /// not enter that snapshot; removing a snapshotted child does not remove it from the
    /// traversal, but [`Self::remove_topic`] may close the shared child handle and thereby
    /// affect that child's operation. The collection lock is released before the first
    /// `.await`, and children are processed sequentially in the snapshot's deterministic
    /// vector/topic order. Each child delegates to
    /// [`ConsumerApi::republish_dead_letters`], whose underlying operation confirms each
    /// replacement publication before acknowledging the original message.
    ///
    /// Cancellation stops future child work and preserves children already completed.
    /// Likewise, the first child error stops the operation immediately; prior successful
    /// children are not rolled back. Concurrent calls are not serialized and race the
    /// per-child runtime operations. Per-child counts and outcomes therefore follow the
    /// runtime's existing destructive-drain behavior; this aggregate coordinator adds no
    /// deduplication guarantee across calls. An empty snapshot returns `Ok(0)`.
    pub async fn republish_dead_letters(
        &self,
        dlq_producer: &C::Producer,
    ) -> Result<usize, PulsarError> {
        republish_snapshot(
            &self.inner.consumers,
            |child| child.topic.as_str(),
            |child| async move { child.consumer.republish_dead_letters(dlq_producer).await },
        )
        .await
    }

    /// Tell the broker to redeliver every unacked message across every child consumer.
    /// Mirrors Java `Consumer#redeliverUnacknowledgedMessages` at the multi-topic scope.
    pub fn redeliver_unacked(&self) {
        for nc in self.inner.consumers.lock().iter() {
            nc.consumer.redeliver_unacked();
        }
    }

    /// Receive the next message across any subscribed topic. The future is cancel-safe:
    /// dropping it without polling to completion leaves all unpopped messages in their
    /// respective per-consumer queues.
    pub async fn receive(&self) -> Result<MultiTopicsMessage, PulsarError> {
        // Snapshot the consumer set under the lock and release before awaiting — holding
        // the mutex across an await would serialise receive against add_topic /
        // remove_topic.
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        if snapshot.is_empty() {
            return Err(PulsarError::Config(
                "no topics subscribed to receive from".to_owned(),
            ));
        }
        if snapshot.len() == 1 {
            let nc = &snapshot[0];
            let msg = nc
                .consumer
                .receive()
                .await
                .map_err(|e| PulsarError::Other(format!("receive: {e}")))?;
            self.inner.cursor.store(0, Ordering::Relaxed);
            return Ok(MultiTopicsMessage {
                topic: nc.topic.clone(),
                message: msg,
            });
        }

        // F4: rotate the snapshot by the round-robin cursor BEFORE
        // building the futures so `select_all` does not always favour
        // index 0. Without the rotation a topic that always has a
        // message ready starves every later topic — `select_all` returns
        // the first ready future in input order. The cursor was being
        // updated but never used. The rotation gives a "start sweep at
        // cursor, wrap around" semantic that mirrors Java's
        // `MultiTopicsConsumerImpl#internalReceiveAsync` round-robin.
        let len = snapshot.len();
        let start = self.inner.cursor.load(Ordering::Relaxed) % len;
        // `rotated_indices[i] = (start + i) % len` — the order in which
        // `select_all` sees the per-topic futures. We track both the
        // futures and the original snapshot indices so a winning result
        // maps back to the right topic in `snapshot`.
        let rotated_indices: Vec<usize> = (0..len).map(|i| (start + i) % len).collect();
        let futures: Vec<_> = rotated_indices
            .iter()
            .map(|&i| snapshot[i].consumer.receive())
            .collect();
        let (result, rotated_idx, _rest) = select_all(futures).await;
        let original_idx = rotated_indices[rotated_idx];
        let topic = snapshot[original_idx].topic.clone();
        let message = result.map_err(|e| PulsarError::Other(format!("receive: {e}")))?;
        // Advance the cursor past the topic we just served so the next
        // `receive()` starts the sweep at the topic *after* this one.
        self.inner
            .cursor
            .store((original_idx + 1) % len, Ordering::Relaxed);
        Ok(MultiTopicsMessage { topic, message })
    }

    /// Acknowledge a message. The caller supplies the topic the message came from (returned
    /// alongside the message in [`MultiTopicsMessage::topic`]) so we can route the ack to
    /// the correct per-topic consumer.
    pub async fn ack(&self, topic: &str, message_id: MessageId) -> Result<(), PulsarError> {
        let consumer = self
            .lookup(topic)
            .map_err(|err| PulsarError::Config(format!("ack on multi-consumer: {err}")))?;
        consumer
            .ack(message_id)
            .await
            .map_err(|e| PulsarError::Other(format!("ack: {e}")))
    }

    /// `true` while every child consumer reports the underlying connection is up.
    /// Mirrors Java `Consumer#isConnected` at the multi-topic / partitioned scope.
    #[must_use]
    pub fn is_connected(&self) -> bool {
        let guard = self.inner.consumers.lock();
        !guard.is_empty() && guard.iter().all(|c| c.consumer.is_connected())
    }

    /// Earliest disconnect wall-clock across all child consumers. `None` if no child has
    /// ever disconnected.
    #[must_use]
    pub fn last_disconnected_timestamp(&self) -> Option<std::time::SystemTime> {
        self.inner
            .consumers
            .lock()
            .iter()
            .filter_map(|c| c.consumer.last_disconnected_timestamp())
            .min()
    }

    /// Aggregate cumulative stats across all child consumers (issue #347).
    /// Thin wrapper over [`magnetar_proto::ConsumerStats::fold`] — collects
    /// each child's `(stats(), receive_latency_histogram())` snapshot (taken
    /// under the same lock acquisition so they're consistent with each
    /// other) and folds them per that function's documented per-field rule:
    /// the six cumulative totals + `pending_batch_acks` sum; `msgs_per_sec`
    /// / `bytes_per_sec` sum as f64 (fan-in throughput); `receive_latency_max_ms`
    /// is the exact max; `receive_latency_p50_ms` / `receive_latency_p99_ms`
    /// are recomputed from a REAL merge of every child's receive-latency
    /// histogram — summing or maxing percentiles directly (the previous
    /// implementation's bug: those three fields, plus `msgs_per_sec` /
    /// `bytes_per_sec` / `pending_batch_acks`, were silently left at their
    /// `ConsumerStats::default()` zero) is not statistically sound.
    ///
    /// Applies equally to [`crate::PartitionedConsumer`] (a
    /// `MultiTopicsConsumer` type alias) since it shares this
    /// implementation.
    ///
    /// The two rate fields are populated by the client-wide sweep armed with
    /// [`crate::ClientBuilder::stats_interval`], which reaches every child
    /// because it ticks each slot on the connection rather than fanning out
    /// from here (ADR-0089 — Java's `MultiTopicsConsumerImpl.getStats()` has no
    /// fan-out either, and one clock ticking every child is what makes the f64
    /// sum well-defined). With that knob unset they stay caller-driven and
    /// therefore `0.0`; see
    /// [`magnetar_proto::consumer::ConsumerState::record_rate_window`].
    ///
    /// A child added mid-window by [`Self::add_topic`] or by partition growth
    /// is seeded at its own creation, so it contributes a full snapshot of its
    /// counters immediately but `0.0` to the rate fields for its first full
    /// interval — the window needs a baseline first. Java's recorders behave
    /// identically.
    #[must_use]
    pub fn aggregate_stats(&self) -> magnetar_proto::ConsumerStats {
        let children: Vec<_> = self
            .inner
            .consumers
            .lock()
            .iter()
            .map(|nc| (nc.consumer.stats(), nc.consumer.receive_latency_histogram()))
            .collect();
        magnetar_proto::ConsumerStats::fold(children)
    }

    /// Sum of buffered messages across every child consumer's receiver queue. Mirrors
    /// Java `Consumer#getNumMessagesInQueue` aggregated over partitions/topics.
    #[must_use]
    pub fn available_in_queue(&self) -> usize {
        self.inner
            .consumers
            .lock()
            .iter()
            .map(|c| c.consumer.available_in_queue())
            .sum()
    }

    /// Sum of outstanding broker permits across every child consumer. Mirrors Java
    /// `ConsumerBase#getAvailablePermits` aggregated over partitions/topics.
    ///
    /// Each child reports the real decrementing balance since issue #414 (ADR-0101
    /// amending ADR-0082), so the sum falls under dispatch instead of sitting pinned at
    /// the children's combined receiver-queue size.
    #[must_use]
    pub fn available_permits(&self) -> u32 {
        self.inner
            .consumers
            .lock()
            .iter()
            .map(|c| c.consumer.available_permits())
            .fold(0u32, u32::saturating_add)
    }

    /// `true` if any child consumer has received at least one message. Mirrors Java
    /// `Consumer#hasReceivedAnyMessage` at the multi-topic / partitioned scope.
    #[must_use]
    pub fn has_received_any_message(&self) -> bool {
        self.inner
            .consumers
            .lock()
            .iter()
            .any(|c| c.consumer.has_received_any_message())
    }

    /// `true` once every child consumer is closed. Mirrors Java `Consumer#isClosed` at the
    /// multi-topic / partitioned scope.
    #[must_use]
    pub fn is_closed(&self) -> bool {
        let guard = self.inner.consumers.lock();
        guard.iter().all(|c| c.consumer.is_closed())
    }

    /// Pause every child consumer. Mirrors Java `Consumer#pause` at the multi-topic scope.
    pub fn pause(&self) {
        for nc in self.inner.consumers.lock().iter() {
            nc.consumer.pause();
        }
    }

    /// Resume every child consumer.
    pub fn resume(&self) {
        for nc in self.inner.consumers.lock().iter() {
            nc.consumer.resume();
        }
    }

    /// `true` once every child consumer has reached end-of-topic. Mirrors Java
    /// `Consumer#hasReachedEndOfTopic` at the multi-topic scope.
    #[must_use]
    pub fn has_reached_end_of_topic(&self) -> bool {
        let guard = self.inner.consumers.lock();
        !guard.is_empty() && guard.iter().all(|c| c.consumer.has_reached_end_of_topic())
    }

    /// Close every underlying consumer. Returns the first error encountered; the rest are
    /// dropped (every child still gets a chance to close).
    pub async fn close(self) -> Result<(), PulsarError> {
        let inner = match Arc::try_unwrap(self.inner) {
            Ok(i) => i,
            Err(arc) => {
                // Clones outlive us; nothing safe to close concurrently.
                drop(arc);
                return Ok(());
            }
        };
        // Last Arc — `into_inner` on the Mutex yields the Arc, then we
        // try to unwrap it. `Arc::try_unwrap` succeeds because the
        // outer `arc` (the only other strong ref to `Inner`) has just
        // been dropped or is about to be.
        let consumers_arc = inner.consumers.into_inner();
        let consumers = Arc::try_unwrap(consumers_arc).unwrap_or_else(|arc| (*arc).clone());
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in consumers {
            if let Err(e) = ConsumerApi::close_owned(nc.consumer).await
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("close: {e}")));
            }
        }
        first_err
    }

    /// Unsubscribe every child subscription. Mirrors Java `Consumer#unsubscribe` at the
    /// multi-topic / partitioned scope. Returns the first error encountered; the rest are
    /// dropped (every child still gets a chance to issue its unsubscribe).
    pub async fn unsubscribe(&self, force: bool) -> Result<(), PulsarError> {
        // Snapshot under the lock and release before awaiting — never hold the mutex
        // across an `.await`.
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in snapshot.iter() {
            if let Err(e) = nc.consumer.unsubscribe(force).await
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("unsubscribe: {e}")));
            }
        }
        first_err
    }

    /// Seek every child consumer to the given publish-time deadline. Mirrors Java
    /// `Consumer#seek(long)` at the multi-topic scope.
    pub async fn seek_to_timestamp(&self, publish_time_ms: u64) -> Result<(), PulsarError> {
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in snapshot.iter() {
            if let Err(e) = nc.consumer.seek_to_timestamp(publish_time_ms).await
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("seek_to_timestamp: {e}")));
            }
        }
        first_err
    }

    /// Seek every child consumer to the earliest message. Mirrors Java
    /// `Consumer#seek(MessageId.earliest)` at the multi-topic scope.
    pub async fn seek_to_earliest(&self) -> Result<(), PulsarError> {
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in snapshot.iter() {
            if let Err(e) = nc.consumer.seek_to_earliest().await
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("seek_to_earliest: {e}")));
            }
        }
        first_err
    }

    /// Seek every child consumer to the latest (head) position. Mirrors Java
    /// `Consumer#seek(MessageId.latest)` at the multi-topic scope.
    pub async fn seek_to_latest(&self) -> Result<(), PulsarError> {
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in snapshot.iter() {
            if let Err(e) = nc.consumer.seek_to_latest().await
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("seek_to_latest: {e}")));
            }
        }
        first_err
    }

    /// Seek every child consumer to a per-topic target computed by `f`. Mirrors Java's
    /// `Consumer#seek(Function<String, Object>)` (where the function returns either a
    /// `MessageId` or a `Long` publish-time millis-since-epoch).
    ///
    /// `f` is invoked synchronously per child, in the order supplied to the builder, with
    /// the child's topic name (matching what `topics()` returns — for a
    /// [`crate::PartitionedConsumer`] this is `<topic>-partition-N`). The returned
    /// [`SeekTarget`] is then dispatched to the appropriate per-topic seek primitive.
    ///
    /// All children are attempted even if one fails; the first error encountered is
    /// returned and subsequent errors are dropped (every child still gets a chance to
    /// issue its seek). This matches the existing [`Self::seek_to_timestamp`] semantics.
    pub async fn seek_per_partition<F>(&self, mut f: F) -> Result<(), PulsarError>
    where
        F: FnMut(&str) -> SeekTarget,
    {
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut first_err: Result<(), PulsarError> = Ok(());
        for nc in snapshot.iter() {
            let target = f(nc.topic.as_str());
            let res = match target {
                SeekTarget::MessageId(id) => nc.consumer.seek_to_message(id).await,
                SeekTarget::PublishTimeMs(ts) => nc.consumer.seek_to_timestamp(ts).await,
            };
            if let Err(e) = res
                && first_err.is_ok()
            {
                first_err = Err(PulsarError::Other(format!("seek_per_partition: {e}")));
            }
        }
        first_err
    }

    /// Ask the broker for each topic's last-published message id. Returns one `(topic, id)`
    /// per child consumer, in the order they appear in the current consumer set. Mirrors
    /// Java `Consumer#getLastMessageIds` for partitioned/multi-topic consumers.
    pub async fn last_message_ids(&self) -> Result<Vec<(String, MessageId)>, PulsarError> {
        let snapshot: Arc<Vec<NamedConsumer<C>>> = self.inner.consumers.lock().clone();
        let mut out = Vec::with_capacity(snapshot.len());
        for nc in snapshot.iter() {
            let id = nc
                .consumer
                .last_message_id()
                .await
                .map_err(|e| PulsarError::Other(format!("last_message_id: {e}")))?;
            out.push((nc.topic.clone(), id));
        }
        Ok(out)
    }

    fn lookup(&self, topic: &str) -> Result<C, String> {
        self.inner
            .consumers
            .lock()
            .iter()
            .find(|c| c.topic == topic)
            .map(|c| c.consumer.clone())
            .ok_or_else(|| format!("unknown topic {topic} on multi-consumer"))
    }

    /// Returns `true` if a background partition-watcher was spawned for this
    /// consumer (i.e.
    /// [`MultiTopicsConsumerBuilder::auto_update_partitions_interval`] was set on
    /// the builder, or the surface was opened as a
    /// [`crate::PartitionedConsumer`] with
    /// [`crate::PartitionedConsumerBuilder::auto_update_partitions_interval`]).
    #[must_use]
    pub fn has_auto_update_partitions(&self) -> bool {
        self.inner.auto_update.is_some()
    }

    /// Most recent partition count observed by the background partition watcher.
    /// `None` when no watcher was configured.
    #[must_use]
    pub fn observed_partitions(&self) -> Option<u32> {
        self.inner
            .auto_update
            .as_ref()
            .map(|t| t.observed_partitions.load(Ordering::Relaxed))
    }

    /// Monotonic count of partition-change events observed by the background
    /// watcher. Returns `None` when no watcher was configured.
    #[must_use]
    pub fn partition_change_count(&self) -> Option<u64> {
        self.inner
            .auto_update
            .as_ref()
            .map(|t| t.change_count.load(Ordering::Relaxed))
    }

    /// `Arc<Notify>` signalled by the background partition-watcher on every timer
    /// tick and on every observed partition-count change driven by
    /// [`Self::refresh_partitions`]. Returns `None` when no watcher was
    /// configured. Callers may `await` `notified()` on the returned handle to
    /// react to ticks without polling [`Self::partition_change_count`].
    #[must_use]
    pub fn partitions_changed_notify(&self) -> Option<Arc<Notify>> {
        self.inner.auto_update.as_ref().map(|t| t.changed.clone())
    }

    /// Query the broker for the current partition count of the topic this
    /// consumer was opened against (the base topic, for a
    /// [`crate::PartitionedConsumer`]), and update [`Self::observed_partitions`] /
    /// [`Self::partition_change_count`] in place if the count differs from the
    /// last observation.
    ///
    /// This is the user-driven half of the
    /// [`MultiTopicsConsumerBuilder::auto_update_partitions_interval`] machinery.
    /// Returns the freshly-observed count on success, or `Ok(None)` if no watcher
    /// was configured.
    ///
    /// **Note**: this method only updates the observed count. It does *not*
    /// itself subscribe to new per-partition topics that show up after creation —
    /// the surface still subscribes to its initial set. Callers that need the
    /// expanded set should add the new per-partition topics via
    /// [`Self::add_topic`] in response to the signal.
    ///
    /// # Errors
    ///
    /// Surfaces [`PulsarError::Client`] when the broker metadata lookup fails.
    pub async fn refresh_partitions<E>(
        &self,
        client: &PulsarClient<E>,
    ) -> Result<Option<u32>, PulsarError>
    where
        E: Engine,
        E::ClientState: crate::BrokerMetadataApi,
    {
        let Some(task) = self.inner.auto_update.as_ref() else {
            return Ok(None);
        };
        let count = client.partitions_for_topic(&task.topic).await?;
        // Atomic swap-then-compare: a real change (prev != count) bumps the
        // change counter and wakes the notify. The Mutex previously held
        // here served no compound invariant — just a single u32 store.
        let prev = task.observed_partitions.swap(count, Ordering::Relaxed);
        if prev != count {
            task.change_count.fetch_add(1, Ordering::Relaxed);
            task.changed.notify_waiters();
        }
        Ok(Some(count))
    }
}

impl<C: ConsumerApi> Clone for MultiTopicsConsumer<C> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

/// Push-delivery support: a [`MultiTopicsConsumer`] (and therefore a
/// [`crate::PartitionedConsumer`], a type alias) drives the wrapper listener
/// poller via its topic-fanning [`Self::receive`]. The `C: Clone` bound +
/// `Send + 'static` let the poller move a cheap `Arc`-clone of the consumer into
/// its [`tokio::spawn`]ed task. Topics added later via [`Self::add_topic`] (or by
/// a [`crate::PartitionedConsumerBuilder`] partition refresh) are picked up
/// automatically — `receive()` re-snapshots the child set every call.
impl<C> crate::consumer_listener::WrapperReceiver for MultiTopicsConsumer<C>
where
    C: ConsumerApi + Clone + Send + Sync + 'static,
{
    async fn wrapper_receive(&self) -> Result<(String, IncomingMessage), PulsarError> {
        let m = self.receive().await?;
        Ok((m.topic, m.message))
    }

    fn is_empty(&self) -> bool {
        MultiTopicsConsumer::is_empty(self)
    }

    async fn membership_changed(&self) {
        let notify = self.inner.membership_changed.clone();
        notify.notified().await;
    }
}

/// Builder for [`MultiTopicsConsumer`]. Mirrors `org.apache.pulsar.client.api.ConsumerBuilder`
/// at the multi-topic layer.
///
/// Generic over `E: crate::Engine` (default [`crate::TokioEngine`]). The
/// `.subscribe()` method dispatches through the engine-generic
/// [`crate::ConsumerBuilder`] (which routes through [`crate::SubscribeApi`]) so
/// each per-topic child uses the engine's concrete consumer type. The returned
/// [`MultiTopicsConsumer<C>`] has `C = <E::ClientState as
/// SubscribeApi>::Consumer`.
pub struct MultiTopicsConsumerBuilder<'a, E: Engine = crate::TokioEngine> {
    client: &'a PulsarClient<E>,
    topics: Vec<String>,
    subscription: Option<String>,
    consumer_name: Option<String>,
    sub_type: magnetar_proto::pb::command_subscribe::SubType,
    receiver_queue_size: usize,
    /// Issue #301: pluggable receiver-queue policy applied to every child consumer.
    receiver_queue_policy: Option<std::sync::Arc<dyn magnetar_proto::ReceiverQueuePolicy>>,
    receiver_queue_adjust_interval: Option<std::time::Duration>,
    initial_position: magnetar_proto::pb::command_subscribe::InitialPosition,
    durable: bool,
    properties: Vec<(String, String)>,
    negative_ack_redelivery_delay: Option<std::time::Duration>,
    ack_timeout: Option<std::time::Duration>,
    ack_group_time: Option<std::time::Duration>,
    dlq_policy: Option<(u32, Option<String>)>,
    max_pending_chunked_message: Option<usize>,
    auto_ack_oldest_chunked_message_on_queue_full: Option<bool>,
    expire_time_of_incomplete_chunked_message: Option<std::time::Duration>,
    read_compacted: bool,
    priority_level: Option<i32>,
    subscription_properties: Vec<(String, String)>,
    key_shared: Option<magnetar_proto::KeySharedConfig>,
    replicate_subscription_state: Option<bool>,
    force_topic_creation: Option<bool>,
    start_message_rollback_duration_sec: Option<u64>,
    auto_update_partitions_interval: Option<Duration>,
    /// Base topic recorded for the partition-watcher when this builder is driven by
    /// [`crate::PartitionedConsumerBuilder`]. `None` for direct multi-topic use —
    /// callers there pass the explicit topic list, so there is no single base topic
    /// to watch.
    auto_update_base_topic: Option<String>,
    /// Optional push-delivery callback (Java `ConsumerBuilder#messageListener` at
    /// the multi-topic scope). Set via [`Self::message_listener`] and subscribe via
    /// [`Self::subscribe_with_listener`]; the plain [`Self::subscribe`] ignores it
    /// and returns a pull-mode [`MultiTopicsConsumer`]. The callback receives the
    /// originating topic (so it can route an explicit ack to the right child).
    listener: Option<crate::consumer_listener::WrapperMessageListener>,
}

impl<E: Engine> std::fmt::Debug for MultiTopicsConsumerBuilder<'_, E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MultiTopicsConsumerBuilder")
            .field("topics", &self.topics)
            .field("subscription", &self.subscription)
            .field("sub_type", &self.sub_type)
            .finish_non_exhaustive()
    }
}

impl<'a, E: Engine> MultiTopicsConsumerBuilder<'a, E> {
    pub(crate) fn new(client: &'a PulsarClient<E>) -> Self {
        Self {
            client,
            topics: Vec::new(),
            subscription: None,
            consumer_name: None,
            sub_type: magnetar_proto::pb::command_subscribe::SubType::Exclusive,
            receiver_queue_size: 1000,
            receiver_queue_policy: None,
            receiver_queue_adjust_interval: None,
            initial_position: magnetar_proto::pb::command_subscribe::InitialPosition::Latest,
            durable: true,
            properties: Vec::new(),
            negative_ack_redelivery_delay: None,
            ack_timeout: None,
            ack_group_time: None,
            dlq_policy: None,
            max_pending_chunked_message: None,
            auto_ack_oldest_chunked_message_on_queue_full: None,
            expire_time_of_incomplete_chunked_message: None,
            read_compacted: false,
            priority_level: None,
            subscription_properties: Vec::new(),
            key_shared: None,
            replicate_subscription_state: None,
            force_topic_creation: None,
            start_message_rollback_duration_sec: None,
            auto_update_partitions_interval: None,
            auto_update_base_topic: None,
            listener: None,
        }
    }

    /// Test-support seam (`#[doc(hidden)]`, not part of the stable API):
    /// `true` once a push-delivery listener has been set via
    /// [`Self::message_listener`]. Lets the builder-surface guard test pin the
    /// listener → field wiring without opening a real broker connection.
    #[doc(hidden)]
    #[must_use]
    pub fn has_listener_for_test(&self) -> bool {
        self.listener.is_some()
    }

    /// Test-support seam (`#[doc(hidden)]`): the consumer name this builder will
    /// propagate to every per-topic child via
    /// [`crate::consumer_template::ConsumerTemplate`]. Lets the builder-surface
    /// guard test pin the `name()` → field plumbing without a broker.
    #[doc(hidden)]
    #[must_use]
    pub fn consumer_name_for_test(&self) -> Option<&str> {
        self.consumer_name.as_deref()
    }

    /// Register a push-delivery callback (Java `ConsumerBuilder#messageListener`
    /// at the multi-topic scope). Once set, subscribe via
    /// [`Self::subscribe_with_listener`] to start a background poller that drives
    /// [`MultiTopicsConsumer::receive`] and hands every message to `listener`,
    /// sequentially and in order. The callback receives the originating topic so
    /// it can ack against the right child via
    /// [`MultiTopicsConsumer::ack`] / [`MultiTopicsConsumer::ack_grouped`].
    ///
    /// The plain [`Self::subscribe`] ignores the listener and returns a pull-mode
    /// consumer. Pull and push are mutually exclusive (Java parity): use
    /// `subscribe_with_listener` for push and never call `receive()`, or use
    /// `subscribe` for pull. The callback **must ack explicitly** — the poller
    /// never auto-acks.
    #[must_use]
    pub fn message_listener(
        mut self,
        listener: crate::consumer_listener::WrapperMessageListener,
    ) -> Self {
        self.listener = Some(listener);
        self
    }

    /// Append a topic. Subscribing to the same topic twice yields two separate
    /// per-topic consumer sessions.
    #[must_use]
    pub fn topic(mut self, topic: impl Into<String>) -> Self {
        self.topics.push(topic.into());
        self
    }

    /// Append multiple topics from any iterable.
    #[must_use]
    pub fn topics(mut self, topics: impl IntoIterator<Item = impl Into<String>>) -> Self {
        self.topics.extend(topics.into_iter().map(Into::into));
        self
    }

    /// Mirrors `ConsumerBuilder::name`. The name is propagated verbatim to every
    /// per-topic child consumer (no per-topic suffix), so broker `topics stats`
    /// reports the same `consumerName` for each child — matching the Java client
    /// and making multi-instance consumers attributable. Default `None` lets the
    /// broker assign a name.
    #[must_use]
    pub fn name(mut self, name: impl Into<String>) -> Self {
        self.consumer_name = Some(name.into());
        self
    }

    /// Test-support seam (`#[doc(hidden)]`): the bounded-chunk-reassembly knobs
    /// this builder will propagate to every per-topic child via
    /// [`crate::consumer_template::ConsumerTemplate`]. Lets the builder-surface
    /// guard test pin the setter → field plumbing without a broker.
    #[doc(hidden)]
    #[must_use]
    pub fn chunk_knobs_for_test(
        &self,
    ) -> (Option<usize>, Option<bool>, Option<std::time::Duration>) {
        (
            self.max_pending_chunked_message,
            self.auto_ack_oldest_chunked_message_on_queue_full,
            self.expire_time_of_incomplete_chunked_message,
        )
    }

    /// Required: set the subscription name.
    #[must_use]
    pub fn subscription(mut self, name: impl Into<String>) -> Self {
        self.subscription = Some(name.into());
        self
    }

    /// Set the subscription type.
    #[must_use]
    pub fn subscription_type(
        mut self,
        sub_type: magnetar_proto::pb::command_subscribe::SubType,
    ) -> Self {
        self.sub_type = sub_type;
        self
    }

    /// Set the receiver queue size per consumer.
    #[must_use]
    pub fn receiver_queue_size(mut self, size: usize) -> Self {
        self.receiver_queue_size = size;
        self.receiver_queue_policy = None;
        self.receiver_queue_adjust_interval = None;
        self
    }

    /// Set a pluggable receiver-queue-size policy (issue #301) applied to every
    /// per-topic child. Pass [`magnetar_proto::Auto`] to let each partition's
    /// queue self-tune. Enabling a policy turns on auto-adjust with a default
    /// 5-second tick; override with [`Self::receiver_queue_adjust_interval`].
    #[must_use]
    pub fn receiver_queue_policy(
        mut self,
        policy: std::sync::Arc<dyn magnetar_proto::ReceiverQueuePolicy>,
    ) -> Self {
        self.receiver_queue_policy = Some(policy);
        if self.receiver_queue_adjust_interval.is_none() {
            self.receiver_queue_adjust_interval = Some(Duration::from_secs(5));
        }
        self
    }

    /// Override the auto-adjust tick cadence for [`Self::receiver_queue_policy`].
    #[must_use]
    pub fn receiver_queue_adjust_interval(mut self, interval: Duration) -> Self {
        self.receiver_queue_adjust_interval = Some(interval);
        self
    }

    /// Set the initial position.
    #[must_use]
    pub fn initial_position(
        mut self,
        position: magnetar_proto::pb::command_subscribe::InitialPosition,
    ) -> Self {
        self.initial_position = position;
        self
    }

    /// Toggle durability of the underlying subscriptions.
    #[must_use]
    pub fn durable(mut self, durable: bool) -> Self {
        self.durable = durable;
        self
    }

    /// Mirrors `ConsumerBuilder::property` — forwarded onto every per-topic child.
    #[must_use]
    pub fn property(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.properties.push((key.into(), value.into()));
        self
    }

    /// Mirrors `ConsumerBuilder::negative_ack_redelivery_delay`.
    #[must_use]
    pub fn negative_ack_redelivery_delay(mut self, delay: std::time::Duration) -> Self {
        self.negative_ack_redelivery_delay = Some(delay);
        self
    }

    /// Mirrors `ConsumerBuilder::ack_timeout`.
    #[must_use]
    pub fn ack_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.ack_timeout = Some(timeout);
        self
    }

    /// Mirrors `ConsumerBuilder::ack_group_time`. Applied to every per-topic child.
    #[must_use]
    pub fn ack_group_time(mut self, window: std::time::Duration) -> Self {
        self.ack_group_time = Some(window);
        self
    }

    /// Mirrors `ConsumerBuilder::dead_letter_policy`.
    #[must_use]
    pub fn dead_letter_policy(
        mut self,
        max_redeliver_count: u32,
        dead_letter_topic: Option<String>,
    ) -> Self {
        self.dlq_policy = Some((max_redeliver_count, dead_letter_topic));
        self
    }

    /// Mirrors `ConsumerBuilder::max_pending_chunked_message`.
    #[must_use]
    pub fn max_pending_chunked_message(mut self, max: usize) -> Self {
        self.max_pending_chunked_message = Some(max);
        self
    }

    /// Mirrors `ConsumerBuilder::auto_ack_oldest_chunked_message_on_queue_full`.
    #[must_use]
    pub fn auto_ack_oldest_chunked_message_on_queue_full(mut self, auto_ack: bool) -> Self {
        self.auto_ack_oldest_chunked_message_on_queue_full = Some(auto_ack);
        self
    }

    /// Mirrors `ConsumerBuilder::expire_time_of_incomplete_chunked_message`.
    #[must_use]
    pub fn expire_time_of_incomplete_chunked_message(
        mut self,
        expire: std::time::Duration,
    ) -> Self {
        self.expire_time_of_incomplete_chunked_message = Some(expire);
        self
    }

    /// Mirrors `ConsumerBuilder::read_compacted`.
    #[must_use]
    pub fn read_compacted(mut self, on: bool) -> Self {
        self.read_compacted = on;
        self
    }

    /// Mirrors `ConsumerBuilder::priority_level`.
    #[must_use]
    pub fn priority_level(mut self, level: i32) -> Self {
        self.priority_level = Some(level);
        self
    }

    /// Mirrors `ConsumerBuilder::subscription_property` — appends a (key, value) pair to
    /// every per-topic child's subscription metadata.
    #[must_use]
    pub fn subscription_property(
        mut self,
        key: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.subscription_properties
            .push((key.into(), value.into()));
        self
    }

    /// Mirrors `ConsumerBuilder::key_shared_policy`.
    #[must_use]
    pub fn key_shared_policy(mut self, cfg: magnetar_proto::KeySharedConfig) -> Self {
        self.key_shared = Some(cfg);
        self
    }

    /// Mirrors `ConsumerBuilder::replicate_subscription_state`.
    #[must_use]
    pub fn replicate_subscription_state(mut self, on: bool) -> Self {
        self.replicate_subscription_state = Some(on);
        self
    }

    /// Mirrors `ConsumerBuilder::force_topic_creation`.
    #[must_use]
    pub fn force_topic_creation(mut self, on: bool) -> Self {
        self.force_topic_creation = Some(on);
        self
    }

    /// Mirrors `ConsumerBuilder::start_message_rollback_duration`.
    #[must_use]
    pub fn start_message_rollback_duration(mut self, seconds: u64) -> Self {
        self.start_message_rollback_duration_sec = Some(seconds);
        self
    }

    /// Enable a background timer that signals every `interval`, intended to drive
    /// re-checks of the topic's partition count. Mirrors Java
    /// `ConsumerBuilder#autoUpdatePartitionsInterval`.
    ///
    /// The internal timer task signals
    /// [`MultiTopicsConsumer::partitions_changed_notify`] on every tick. Callers
    /// run [`MultiTopicsConsumer::refresh_partitions`] in response to the signal
    /// (or on their own cadence) to actually call
    /// [`PulsarClient::partitions_for_topic`].
    ///
    /// Default `None` — no timer is spawned. Pass a non-zero `Duration` to opt
    /// in. The timer is aborted when the [`MultiTopicsConsumer`] (and every
    /// clone) is dropped.
    ///
    /// Setting a zero `interval` is treated as "disable" — same as the default.
    ///
    /// **Note**: for direct multi-topic use, the watcher polls the *first* topic
    /// supplied to the builder (single watched topic is sufficient for the
    /// partitioned-consumer case which is the main use site).
    #[must_use]
    pub fn auto_update_partitions_interval(mut self, interval: Duration) -> Self {
        self.auto_update_partitions_interval = if interval.is_zero() {
            None
        } else {
            Some(interval)
        };
        self
    }

    /// Record the base topic the partition watcher should poll. Called by
    /// [`crate::PartitionedConsumerBuilder::subscribe`] so the watcher polls the
    /// base topic (e.g. `persistent://t`) rather than the first partition topic
    /// (`persistent://t-partition-0`). Crate-internal — direct multi-topic users
    /// don't need this knob; the watcher falls back to the first topic.
    #[must_use]
    pub(crate) fn auto_update_base_topic(mut self, topic: String) -> Self {
        self.auto_update_base_topic = Some(topic);
        self
    }
}

impl<E> MultiTopicsConsumerBuilder<'_, E>
where
    E: Engine,
    E::ClientState: SubscribeApi,
    <E::ClientState as SubscribeApi>::Consumer: Clone,
{
    /// Open every per-topic subscription concurrently. If any subscribe fails the others
    /// that already succeeded are torn down before the error is returned.
    pub async fn subscribe(
        self,
    ) -> Result<MultiTopicsConsumer<<E::ClientState as SubscribeApi>::Consumer>, PulsarError> {
        let mut deadline =
            crate::SubscribeApi::new_subscribe_operation_deadline(&self.client.inner);
        self.subscribe_with_deadline(&mut deadline).await
    }

    pub(crate) async fn subscribe_with_deadline(
        self,
        deadline: &mut crate::OperationDeadline,
    ) -> Result<MultiTopicsConsumer<<E::ClientState as SubscribeApi>::Consumer>, PulsarError> {
        let subscription = self
            .subscription
            .ok_or_else(|| PulsarError::Config("subscription name is required".to_owned()))?;
        if self.topics.is_empty() {
            return Err(PulsarError::Config(
                "at least one topic is required".to_owned(),
            ));
        }

        let template = ConsumerTemplate {
            subscription,
            consumer_name: self.consumer_name,
            sub_type: self.sub_type,
            receiver_queue_size: self.receiver_queue_size,
            receiver_queue_policy: self.receiver_queue_policy,
            receiver_queue_adjust_interval: self.receiver_queue_adjust_interval,
            initial_position: self.initial_position,
            durable: self.durable,
            properties: self.properties,
            negative_ack_redelivery_delay: self.negative_ack_redelivery_delay,
            ack_timeout: self.ack_timeout,
            ack_group_time: self.ack_group_time,
            dlq_policy: self.dlq_policy,
            max_pending_chunked_message: self.max_pending_chunked_message,
            auto_ack_oldest_chunked_message_on_queue_full: self
                .auto_ack_oldest_chunked_message_on_queue_full,
            expire_time_of_incomplete_chunked_message: self
                .expire_time_of_incomplete_chunked_message,
            read_compacted: self.read_compacted,
            priority_level: self.priority_level,
            subscription_properties: self.subscription_properties,
            key_shared: self.key_shared,
            replicate_subscription_state: self.replicate_subscription_state,
            force_topic_creation: self.force_topic_creation,
            start_message_rollback_duration_sec: self.start_message_rollback_duration_sec,
        };

        // Subscribe sequentially — the first failure short-circuits, and on failure we close
        // the consumers we already opened.
        let mut consumers: Vec<NamedConsumer<<E::ClientState as SubscribeApi>::Consumer>> =
            Vec::with_capacity(self.topics.len());
        for topic in &self.topics {
            let builder = template.apply(self.client.consumer(topic.clone()));
            let result = builder.subscribe_with_deadline(deadline).await;
            match result {
                Ok(c) => consumers.push(NamedConsumer {
                    topic: topic.clone(),
                    consumer: c,
                }),
                Err(e) => {
                    // Best-effort teardown of previously-opened consumers.
                    for nc in consumers {
                        let _ = ConsumerApi::close_owned(nc.consumer).await;
                    }
                    return Err(e);
                }
            }
        }

        // Spawn the partition-watcher timer iff the builder configured a non-zero
        // interval. The timer itself only emits ticks via `Notify`; callers drive
        // the actual `partitions_for_topic` call via
        // [`MultiTopicsConsumer::refresh_partitions`].
        let auto_update = self.auto_update_partitions_interval.map(|interval| {
            let watched_topic = self
                .auto_update_base_topic
                .unwrap_or_else(|| self.topics[0].clone());
            // We do not have an initial partition count for the direct multi-topic
            // case (each topic was passed explicitly); seed with 0 so the first
            // refresh always logs a change.
            spawn_auto_update_task(watched_topic, interval, 0)
        });

        Ok(MultiTopicsConsumer {
            inner: Arc::new(Inner {
                consumers: Mutex::new(Arc::new(consumers)),
                membership_changed: Arc::new(Notify::new()),
                cursor: std::sync::atomic::AtomicUsize::new(0),
                template,
                auto_update,
            }),
        })
    }
}

impl<E> MultiTopicsConsumerBuilder<'_, E>
where
    E: Engine,
    E::ClientState: SubscribeApi,
    <E::ClientState as SubscribeApi>::Consumer: Clone + Send + Sync + 'static,
{
    /// Subscribe every per-topic child and start a push-delivery poller over the
    /// resulting [`MultiTopicsConsumer`], returning the owning
    /// [`crate::MessageListenerHandle`]. Mirrors Java's
    /// `ConsumerBuilder#messageListener(...)` + `subscribe()` at the multi-topic /
    /// partitioned scope.
    ///
    /// The poller delivers messages sequentially and in order across every
    /// subscribed topic, handing the callback the originating topic and message,
    /// and does **not** auto-ack (the callback acks explicitly via
    /// [`MultiTopicsConsumer::ack`]). It stops cleanly when the consumer set is
    /// drained or the returned handle is dropped. Topics added later (via
    /// [`MultiTopicsConsumer::add_topic`]) are picked up automatically.
    ///
    /// Because the consumer is moved into the poller, there is no handle left to
    /// call `receive()` on — the listener owns delivery (Java's "no `receive()`
    /// with a `messageListener`" rule).
    ///
    /// # Errors
    /// - [`PulsarError::Config`] if no listener was set via [`Self::message_listener`].
    /// - any subscribe error from [`Self::subscribe`].
    pub async fn subscribe_with_listener(
        self,
    ) -> Result<crate::MessageListenerHandle, PulsarError> {
        let Some(listener) = self.listener.clone() else {
            return Err(PulsarError::Config(
                "subscribe_with_listener() requires a listener — \
                 call message_listener(...) first (or use subscribe() for pull mode)"
                    .to_owned(),
            ));
        };
        let consumer = self.subscribe().await?;
        Ok(crate::consumer_listener::spawn_wrapper_message_listener(
            consumer, listener,
        ))
    }
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;
    use std::sync::atomic::{AtomicBool, AtomicUsize};

    use magnetar_proto::MessageId;
    use tokio::sync::Barrier;

    use super::*;
    use crate::SeekTarget;

    fn empty_template() -> ConsumerTemplate {
        ConsumerTemplate {
            subscription: "sub".to_owned(),
            consumer_name: None,
            sub_type: magnetar_proto::pb::command_subscribe::SubType::Exclusive,
            receiver_queue_size: 1000,
            receiver_queue_policy: None,
            receiver_queue_adjust_interval: None,
            initial_position: magnetar_proto::pb::command_subscribe::InitialPosition::Latest,
            durable: true,
            properties: Vec::new(),
            negative_ack_redelivery_delay: None,
            ack_timeout: None,
            ack_group_time: None,
            dlq_policy: None,
            max_pending_chunked_message: None,
            auto_ack_oldest_chunked_message_on_queue_full: None,
            expire_time_of_incomplete_chunked_message: None,
            read_compacted: false,
            priority_level: None,
            subscription_properties: Vec::new(),
            key_shared: None,
            replicate_subscription_state: None,
            force_topic_creation: None,
            start_message_rollback_duration_sec: None,
        }
    }

    /// Mutex round-trip: build an `Inner` with no consumers and verify the dynamic-membership
    /// helpers (`topics`, `len`, `is_empty`, `lookup`) operate consistently against the
    /// `Mutex<Vec<NamedConsumer>>` and that the template-stored subscription name is
    /// reachable via [`MultiTopicsConsumer::subscription`] even with an empty set.
    #[test]
    fn empty_inner_is_consistent() {
        let inner: Arc<Inner<magnetar_runtime_tokio::Consumer>> = Arc::new(Inner {
            consumers: Mutex::new(Arc::new(Vec::new())),
            membership_changed: Arc::new(Notify::new()),
            cursor: std::sync::atomic::AtomicUsize::new(0),
            template: empty_template(),
            auto_update: None,
        });
        let consumer: MultiTopicsConsumer<magnetar_runtime_tokio::Consumer> = MultiTopicsConsumer {
            inner: inner.clone(),
        };
        assert_eq!(consumer.len(), 0);
        assert!(consumer.is_empty());
        assert!(consumer.topics().is_empty());
        assert_eq!(consumer.subscription(), "sub");
        let lookup = consumer.lookup("missing");
        assert!(lookup.is_err());
        // Cloning the handle shares the same Inner.
        let cloned = consumer.clone();
        assert!(cloned.is_empty());
        assert_eq!(cloned.subscription(), "sub");
    }

    #[test]
    fn template_clone_preserves_settings() {
        let mut t = empty_template();
        t.properties.push(("k".to_owned(), "v".to_owned()));
        t.subscription_properties
            .push(("sk".to_owned(), "sv".to_owned()));
        let clone = t.clone();
        assert_eq!(clone.subscription, "sub");
        assert_eq!(clone.properties, vec![("k".to_owned(), "v".to_owned())]);
        assert_eq!(
            clone.subscription_properties,
            vec![("sk".to_owned(), "sv".to_owned())]
        );
    }

    #[derive(Clone)]
    struct TestRepublish {
        topic: &'static str,
        count: usize,
    }

    #[tokio::test]
    async fn republish_stops_at_first_error_with_partial_progress_and_topic_context() {
        let members = Mutex::new(Arc::new(vec![
            TestRepublish {
                topic: "orders-0",
                count: 3,
            },
            TestRepublish {
                topic: "orders-1",
                count: 5,
            },
            TestRepublish {
                topic: "orders-2",
                count: 7,
            },
        ]));
        let started = Arc::new(Mutex::new(Vec::new()));
        let completed = Arc::new(Mutex::new(Vec::new()));

        let error = republish_snapshot(
            &members,
            |child| child.topic,
            |child| {
                let started = started.clone();
                let completed = completed.clone();
                async move {
                    started.lock().push(child.topic);
                    if child.topic == "orders-1" {
                        return Err("broker rejected publish");
                    }
                    completed.lock().push((child.topic, child.count));
                    Ok(child.count)
                }
            },
        )
        .await
        .expect_err("the second child must stop orchestration");

        let PulsarError::Other(message) = error else {
            panic!("child failure must surface as an engine error");
        };
        assert_eq!(
            message,
            "republish_dead_letters for topic orders-1: broker rejected publish"
        );
        assert_eq!(*started.lock(), vec!["orders-0", "orders-1"]);
        assert_eq!(*completed.lock(), vec![("orders-0", 3)]);
    }

    #[tokio::test]
    async fn republish_count_saturates_at_usize_max() {
        let members = Mutex::new(Arc::new(vec![
            TestRepublish {
                topic: "orders-0",
                count: usize::MAX,
            },
            TestRepublish {
                topic: "orders-1",
                count: 1,
            },
        ]));

        let republished = republish_snapshot(
            &members,
            |child| child.topic,
            |child| async move { Ok::<_, &'static str>(child.count) },
        )
        .await
        .expect("republish must succeed");

        assert_eq!(republished, usize::MAX);
    }

    #[tokio::test]
    async fn republish_uses_membership_snapshot_taken_before_first_await() {
        let members = Arc::new(Mutex::new(Arc::new(vec![
            TestRepublish {
                topic: "orders-0",
                count: 2,
            },
            TestRepublish {
                topic: "orders-1",
                count: 3,
            },
        ])));
        let first_started = Arc::new(Notify::new());
        let release_first = Arc::new(Notify::new());
        let completed = Arc::new(Mutex::new(Vec::new()));

        let task = tokio::spawn({
            let members = members.clone();
            let first_started = first_started.clone();
            let release_first = release_first.clone();
            let completed = completed.clone();
            async move {
                republish_snapshot(
                    &members,
                    |child| child.topic,
                    |child| {
                        let first_started = first_started.clone();
                        let release_first = release_first.clone();
                        let completed = completed.clone();
                        async move {
                            if child.topic == "orders-0" {
                                first_started.notify_one();
                                release_first.notified().await;
                            }
                            completed.lock().push(child.topic);
                            Ok::<_, &'static str>(child.count)
                        }
                    },
                )
                .await
            }
        });

        first_started.notified().await;
        {
            let mut current = members
                .try_lock()
                .expect("membership lock must not be held across child work");
            let current = Arc::make_mut(&mut current);
            current.remove(1);
            current.push(TestRepublish {
                topic: "orders-2",
                count: 11,
            });
        }
        release_first.notify_one();

        assert_eq!(
            task.await
                .expect("task must not panic")
                .expect("snapshot republish must succeed"),
            5
        );
        assert_eq!(*completed.lock(), vec!["orders-0", "orders-1"]);
    }

    struct CancellationProbe(Arc<AtomicBool>);

    impl Drop for CancellationProbe {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    #[tokio::test]
    async fn cancelling_republish_preserves_completed_children_and_stops_later_work() {
        let members = Arc::new(Mutex::new(Arc::new(vec![
            TestRepublish {
                topic: "orders-0",
                count: 2,
            },
            TestRepublish {
                topic: "orders-1",
                count: 3,
            },
            TestRepublish {
                topic: "orders-2",
                count: 5,
            },
        ])));
        let second_started = Arc::new(Notify::new());
        let never_release = Arc::new(Notify::new());
        let completed = Arc::new(Mutex::new(Vec::new()));
        let in_flight_dropped = Arc::new(AtomicBool::new(false));

        let task = tokio::spawn({
            let members = members.clone();
            let second_started = second_started.clone();
            let never_release = never_release.clone();
            let completed = completed.clone();
            let in_flight_dropped = in_flight_dropped.clone();
            async move {
                republish_snapshot(
                    &members,
                    |child| child.topic,
                    |child| {
                        let second_started = second_started.clone();
                        let never_release = never_release.clone();
                        let completed = completed.clone();
                        let in_flight_dropped = in_flight_dropped.clone();
                        async move {
                            if child.topic == "orders-1" {
                                let _probe = CancellationProbe(in_flight_dropped);
                                second_started.notify_one();
                                never_release.notified().await;
                            }
                            completed.lock().push(child.topic);
                            Ok::<_, &'static str>(child.count)
                        }
                    },
                )
                .await
            }
        });

        second_started.notified().await;
        task.abort();
        let join_error = task
            .await
            .expect_err("aborted orchestration must be cancelled");

        assert!(join_error.is_cancelled());
        assert!(in_flight_dropped.load(Ordering::SeqCst));
        assert_eq!(*completed.lock(), vec!["orders-0"]);
    }

    #[tokio::test]
    async fn concurrent_republish_calls_overlap_on_independent_snapshots() {
        let members = Arc::new(Mutex::new(Arc::new(vec![
            TestRepublish {
                topic: "orders-0",
                count: 2,
            },
            TestRepublish {
                topic: "orders-1",
                count: 3,
            },
        ])));
        let overlap = Arc::new(Barrier::new(2));
        let first_calls = Arc::new(AtomicUsize::new(0));
        let second_calls = Arc::new(AtomicUsize::new(0));

        let run = || {
            let members = members.clone();
            let overlap = overlap.clone();
            let first_calls = first_calls.clone();
            let second_calls = second_calls.clone();
            tokio::spawn(async move {
                republish_snapshot(
                    &members,
                    |child| child.topic,
                    |child| {
                        let overlap = overlap.clone();
                        let first_calls = first_calls.clone();
                        let second_calls = second_calls.clone();
                        async move {
                            if child.topic == "orders-0" {
                                first_calls.fetch_add(1, Ordering::SeqCst);
                                overlap.wait().await;
                            } else {
                                second_calls.fetch_add(1, Ordering::SeqCst);
                            }
                            Ok::<_, &'static str>(child.count)
                        }
                    },
                )
                .await
            })
        };

        let first = run();
        let second = run();
        let (first, second) = tokio::time::timeout(Duration::from_secs(2), async {
            tokio::join!(first, second)
        })
        .await
        .expect("concurrent calls must not serialize behind the membership lock");

        let first = first
            .expect("first task must not panic")
            .expect("first republish must succeed");
        let second = second
            .expect("second task must not panic")
            .expect("second republish must succeed");
        assert_eq!(first, 5);
        assert_eq!(second, 5);
        assert_eq!(first_calls.load(Ordering::SeqCst), 2);
        assert_eq!(second_calls.load(Ordering::SeqCst), 2);
    }

    /// Mirror of the dispatch arm inside [`super::MultiTopicsConsumer::seek_per_partition`].
    /// Records the routing decision per topic instead of issuing a real seek so the routing
    /// logic can be exercised without spinning up a broker.
    fn dispatch<F>(topics: &[&str], mut f: F) -> Vec<(String, DispatchKind)>
    where
        F: FnMut(&str) -> SeekTarget,
    {
        topics
            .iter()
            .map(|t| {
                let kind = match f(t) {
                    SeekTarget::MessageId(id) => DispatchKind::MessageId(id),
                    SeekTarget::PublishTimeMs(ts) => DispatchKind::PublishTimeMs(ts),
                };
                ((*t).to_owned(), kind)
            })
            .collect()
    }

    #[derive(Debug, PartialEq, Eq)]
    enum DispatchKind {
        MessageId(MessageId),
        PublishTimeMs(u64),
    }

    #[test]
    fn seek_per_partition_routes_each_topic_via_closure() {
        let topics = [
            "persistent://public/default/orders-partition-0",
            "persistent://public/default/orders-partition-1",
            "persistent://public/default/orders-partition-2",
        ];

        // Track what topics the closure was called with — mirrors Java's
        // Function<String, Object> semantics where each partition gets its own decision.
        let seen = RefCell::new(Vec::<String>::new());
        let mid = MessageId::EARLIEST;
        let decisions = dispatch(&topics, |t| {
            seen.borrow_mut().push(t.to_owned());
            if t.ends_with("-partition-1") {
                SeekTarget::PublishTimeMs(1_700_000_000_000)
            } else {
                SeekTarget::MessageId(mid)
            }
        });

        // Closure was called exactly once per topic, in builder order.
        assert_eq!(seen.borrow().len(), 3);
        assert_eq!(seen.borrow()[0], topics[0]);
        assert_eq!(seen.borrow()[1], topics[1]);
        assert_eq!(seen.borrow()[2], topics[2]);

        // Routing: partition-0 / partition-2 -> MessageId seek; partition-1 -> timestamp seek.
        assert_eq!(decisions.len(), 3);
        assert_eq!(decisions[0].1, DispatchKind::MessageId(mid));
        assert_eq!(
            decisions[1].1,
            DispatchKind::PublishTimeMs(1_700_000_000_000)
        );
        assert_eq!(decisions[2].1, DispatchKind::MessageId(mid));
    }

    #[test]
    fn seek_target_enum_variants_are_constructible() {
        let by_id = SeekTarget::MessageId(MessageId::LATEST);
        let by_ts = SeekTarget::PublishTimeMs(42);
        // PartialEq + Copy: derived impls round-trip without surprises.
        assert_eq!(by_id, SeekTarget::MessageId(MessageId::LATEST));
        assert_ne!(by_id, by_ts);
    }

    /// F4 — pure rotation math used by [`MultiTopicsConsumer::receive`].
    /// The production code computes `(start + i) % len` for `i in 0..len`,
    /// where `start = cursor % len`. The cursor is then advanced past
    /// the winning index. This helper mirrors the same shape so we can
    /// exercise the starvation-defence guarantee without spinning up
    /// real consumers.
    fn rotated(start: usize, len: usize) -> Vec<usize> {
        (0..len).map(|i| (start + i) % len).collect()
    }

    /// F4 — every starting cursor value yields a permutation of
    /// `0..len` (each topic appears exactly once). This is the
    /// non-starvation invariant: even if `select_all` always returns
    /// the first ready future, no topic is permanently locked out;
    /// every topic gets first crack on `len` consecutive `receive()`
    /// calls.
    #[test]
    fn rotated_indices_are_a_permutation_of_all_topics() {
        for len in 1..=8 {
            for start in 0..len {
                let v = rotated(start, len);
                assert_eq!(v.len(), len);
                let mut sorted = v.clone();
                sorted.sort_unstable();
                assert_eq!(
                    sorted,
                    (0..len).collect::<Vec<_>>(),
                    "rotated(start={start}, len={len}) must be a permutation; got {v:?}"
                );
            }
        }
    }

    /// F4 — the rotation puts the cursor's topic first. With the cursor
    /// advancing past the winner, a steady stream of "every topic
    /// always ready" produces a uniform sweep across topics rather
    /// than starving the topics with higher indices.
    #[test]
    fn cursor_advance_visits_every_topic_in_round_robin() {
        // Simulate the steady-state: every topic always has a message
        // ready, so `select_all` returns index 0 of the rotated list,
        // which is the cursor's current head. Each call advances the
        // cursor by one.
        let len = 3_usize;
        let mut cursor = 0_usize;
        let mut winners = Vec::with_capacity(len * 2);
        for _ in 0..len * 2 {
            let start = cursor % len;
            let order = rotated(start, len);
            // Steady state: first in the rotated list always wins (it's
            // always ready). The production code maps that back to the
            // original index via `rotated_indices[rotated_idx]`.
            let original_idx = order[0];
            winners.push(original_idx);
            cursor = (original_idx + 1) % len;
        }
        // Two full sweeps in order: 0, 1, 2, 0, 1, 2.
        assert_eq!(winners, vec![0, 1, 2, 0, 1, 2]);
    }

    /// F4 — counter-test for the regression. Without the rotation,
    /// `select_all` always returns index 0 of a fixed-order snapshot,
    /// so the steady-state winner is always topic 0 and topics 1..N
    /// starve. This simulates the buggy behaviour and asserts it would
    /// have failed the round-robin expectation — locking in the
    /// regression.
    #[test]
    fn without_rotation_first_topic_starves_the_rest() {
        // Buggy: ignore the cursor, always take index 0 in builder
        // order. The cursor is updated to the winner+1 but never read,
        // so it has no effect.
        let len = 3_usize;
        // Bug shape: every call picks index 0 (the snapshot order is
        // fixed; `select_all` always returns the first ready future).
        let winners: Vec<usize> = vec![0; len * 2];
        assert_eq!(
            winners,
            vec![0, 0, 0, 0, 0, 0],
            "regression witness: pre-F4 behaviour starved topics 1 and 2"
        );
    }
}