rustis 0.20.0

use super::pub_sub_message::PubSubMessage;
use crate::{
    ClientError, Connection, ConnectionState, Error, JoinHandle, ReconnectionState, Result,
    RetryReason,
    client::{Config, Message, MessageKind},
    commands::InternalPubSubCommands,
    resp::{ClientReplyMode, CommandKind, RespResponse, StateSlot, SubscriptionType, cmd},
    spawn, timeout,
};
use bytes::Bytes;
use futures_channel::mpsc;
use futures_util::{FutureExt, select};
use smallvec::SmallVec;
use std::{
    collections::{HashMap, VecDeque},
    future::poll_fn,
    sync::Arc,
    task::Poll,
    time::Duration,
};
use tokio::{sync::broadcast, time::Instant};
use tracing::{Instrument, debug, error, info, info_span, trace, warn};

// Backpressure note: every channel in the crate — request, pub/sub and push —
// is deliberately unbounded. This is a design choice of the lock-free
// multiplexer: senders never block or await capacity, keeping the hot send path
// allocation-and-await-free and the network task's accounting simple. The
// trade-off is that memory is bounded by consumer behaviour rather than by the
// channel: a pub/sub subscriber that stops polling its stream, or a reconnect
// storm accumulating retryable traffic in `messages_to_send`, grows client-side
// memory unbounded. Callers with slow or paused consumers should drop the
// stream (or drain it) promptly; a bounded variant would trade this for
// send-path blocking and is intentionally not used here.
pub(crate) type MsgSender = tokio::sync::mpsc::UnboundedSender<Message>;
pub(crate) type MsgReceiver = tokio::sync::mpsc::UnboundedReceiver<Message>;
/// Retry-only handle the network task keeps on the message channel. It is
/// [`Weak`](tokio::sync::mpsc::WeakUnboundedSender) on purpose: holding a strong
/// sender would keep the channel open forever, so dropping the last client
/// would never end the network loop. With a weak handle the channel closes
/// naturally when the last client is dropped, and the task upgrades it only to
/// requeue a message for retry.
type WeakMsgSender = tokio::sync::mpsc::WeakUnboundedSender<Message>;
pub(crate) type ResultSender = tokio::sync::oneshot::Sender<Result<RespResponse>>;
pub(crate) type ResultReceiver = tokio::sync::oneshot::Receiver<Result<RespResponse>>;
pub(crate) type ResultsSender = tokio::sync::oneshot::Sender<Result<Vec<RespResponse>>>;
pub(crate) type ResultsReceiver = tokio::sync::oneshot::Receiver<Result<Vec<RespResponse>>>;
pub(crate) type PubSubSender = mpsc::UnboundedSender<Result<RespResponse>>;
pub(crate) type PubSubReceiver = mpsc::UnboundedReceiver<Result<RespResponse>>;
pub(crate) type PushSender = mpsc::UnboundedSender<Result<RespResponse>>;
pub(crate) type PushReceiver = mpsc::UnboundedReceiver<Result<RespResponse>>;
pub(crate) type ReconnectSender = broadcast::Sender<()>;
pub(crate) type ReconnectReceiver = broadcast::Receiver<()>;

/// Test-only observability and fault-injection hook for the send batch.
///
/// This is the queue-position primitive of the failure-path test
/// infrastructure: it lets a test deterministically force retry reasons onto a
/// message drained by [`NetworkHandler::send_messages`] and observe the retry
/// reasons every command is actually fed with. It carries no cost in shipped
/// builds because it is gated behind `cfg(test)`, so it is compiled only when
/// the crate itself is built as a test target, like the existing
/// `kill_connection_on_write` primitive.
#[cfg(test)]
#[derive(Clone, Default)]
pub(crate) struct SendBatchTestHook {
    /// Retry reasons to force onto the **first** message of each
    /// `send_messages` drain that contains at least one message. One entry is
    /// consumed per such drain; `None` leaves that drain untouched.
    inject_first_message_reasons: Arc<std::sync::Mutex<VecDeque<Option<Vec<RetryReason>>>>>,
    /// Records, in feed order, `(command name, number of retry reasons fed)`
    /// for every command actually fed to the connection.
    fed_retry_reasons: Arc<std::sync::Mutex<Vec<(String, usize)>>>,
    /// When set, the next fed command whose name matches is armed to kill the
    /// connection on its `usize`-th following read (see
    /// [`CommandBuilder::kill_connection_on_read`]). Consumed on first match,
    /// so it fires exactly once. Lets a test inject a send failure onto a
    /// command it does not build itself, such as the sink's internal
    /// `UNSUBSCRIBE`.
    kill_on_read_by_name: Arc<std::sync::Mutex<Option<(String, usize)>>>,
}

#[cfg(test)]
#[allow(
    clippy::expect_used,
    reason = "test-support code: a panic is how a test reports failure"
)]
impl SendBatchTestHook {
    pub(crate) fn new() -> Self {
        Self::default()
    }

    /// Queues retry reasons to be forced onto the first message of the next
    /// send drain (or `None` to skip that drain).
    pub(crate) fn push_injection(&self, reasons: Option<Vec<RetryReason>>) {
        self.inject_first_message_reasons
            .lock()
            .expect("send batch test hook mutex poisoned")
            .push_back(reasons);
    }

    /// Returns the recorded `(command name, number of retry reasons fed)`
    /// entries, in feed order.
    pub(crate) fn fed_retry_reasons(&self) -> Vec<(String, usize)> {
        self.fed_retry_reasons
            .lock()
            .expect("send batch test hook mutex poisoned")
            .clone()
    }

    fn take_injection(&self) -> Option<Vec<RetryReason>> {
        self.inject_first_message_reasons
            .lock()
            .expect("send batch test hook mutex poisoned")
            .pop_front()
            .flatten()
    }

    fn record_fed(&self, command_name: String, num_reasons: usize) {
        self.fed_retry_reasons
            .lock()
            .expect("send batch test hook mutex poisoned")
            .push((command_name, num_reasons));
    }

    /// Arms the connection to be killed on the `num_reads`-th read following the
    /// next fed command named `command_name`.
    pub(crate) fn arm_kill_on_read_for(&self, command_name: &str, num_reads: usize) {
        *self
            .kill_on_read_by_name
            .lock()
            .expect("send batch test hook mutex poisoned") =
            Some((command_name.to_owned(), num_reads));
    }

    /// If the next queued kill matches `command_name`, consumes it and returns
    /// the read count to arm.
    fn take_kill_on_read_for(&self, command_name: &str) -> Option<usize> {
        let mut guard = self
            .kill_on_read_by_name
            .lock()
            .expect("send batch test hook mutex poisoned");
        if guard.as_ref().is_some_and(|(name, _)| name == command_name) {
            return guard.take().map(|(_, num_reads)| num_reads);
        }
        None
    }
}

// Why `Config::max_messages_per_wave` exists, kept next to the code that obeys
// it in `try_handle_message`.
//
// Draining the message channel until it is empty convoys the entire in-flight
// concurrency into one `writev`, so every caller waits for the whole batch to be
// written *and* answered. Capping the wave keeps a batch in flight at the server
// while the next one is being collected.
//
// The default (48) was calibrated against a live Redis over concurrency levels
// 64 → 1024 (see `RUSTIS_VS_REDIS_RS.md`, H13): the optimum is flat between 32
// and 128, 48 is within ~12% of the per-level optimum everywhere, and below 48
// in-flight messages the cap never fires, so low-concurrency behaviour is
// unchanged whatever it is set to.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Status {
    Disconnected,
    Connected,
    EnteringMonitor,
    Monitor,
    LeavingMonitor,
}

struct MessageToSend {
    pub message: Message,
}

impl MessageToSend {
    pub(crate) fn new(message: Message) -> Self {
        Self { message }
    }
}

#[derive(Debug)]
struct MessageToReceive {
    pub message: Message,
    pub num_commands: usize,
    pub pending_responses: Vec<RespResponse>,
}

impl MessageToReceive {
    pub(crate) fn new(message: Message, num_commands: usize) -> Self {
        Self {
            message,
            num_commands,
            // A batch collects exactly `num_commands` responses; size the buffer
            // once instead of letting it grow.
            pending_responses: Vec::with_capacity(num_commands),
        }
    }
}

struct PendingSubscription {
    pub channel_or_pattern: Bytes,
    pub subscription_type: SubscriptionType,
    pub sender: PubSubSender,
    /// indicates if more subscriptions will arrive in the same batch
    pub more_to_come: bool,
}

pub(crate) struct NetworkHandler {
    status: Status,
    connection: Connection,
    /// for retries
    msg_sender: WeakMsgSender,
    msg_receiver: MsgReceiver,
    messages_to_send: VecDeque<MessageToSend>,
    messages_to_receive: VecDeque<MessageToReceive>,
    pending_subscriptions: VecDeque<PendingSubscription>,
    pending_unsubscriptions: VecDeque<HashMap<Bytes, SubscriptionType>>,
    subscriptions: HashMap<Bytes, (SubscriptionType, PubSubSender)>,
    is_reply_on: bool,
    /// `CLIENT REPLY SKIP` silences the reply of the command that follows it, and
    /// only that one — unlike `OFF`, which silences the connection until `ON`.
    skip_next_reply: bool,
    /// Connection-attached state to replay when the socket is remade. Owned here
    /// and lent as `&mut` to whichever connection is being built: the network task
    /// is its only user, so no `Arc` and no lock are involved.
    connection_state: ConnectionState,
    /// Sink for client-side-caching invalidation pushes, active while the
    /// connection is in `Status::Connected`. Kept separate from `monitor_sender`
    /// so registering one push consumer cannot silently overwrite the other's
    /// slot — the two flows are routed by distinct `Status` states, so a single
    /// shared field was only ever a latent trap, not a working multiplexer.
    invalidation_sender: Option<PushSender>,
    /// Sink for MONITOR output, active while the connection is in
    /// `Status::Monitor` / `LeavingMonitor`. See `invalidation_sender`.
    monitor_sender: Option<PushSender>,
    reconnect_sender: ReconnectSender,
    auto_resubscribe: bool,
    auto_remonitor: bool,
    reconnection_state: ReconnectionState,
    /// Per-message retry cap from `Config::max_command_attempts` (`0` = unlimited).
    max_command_attempts: usize,
    /// Send-wave cap from `Config::max_messages_per_wave`.
    max_messages_per_wave: usize,
    /// Number of incoming results belonging to a message that has already been
    /// resolved, and which must therefore be dropped instead of matched.
    results_to_discard: usize,
    #[cfg(test)]
    send_batch_test_hook: Option<SendBatchTestHook>,
}

impl NetworkHandler {
    pub(crate) async fn connect(
        config: Config,
    ) -> Result<(MsgSender, JoinHandle<()>, ReconnectSender, Arc<str>)> {
        // Reject an incoherent config here rather than letting a zeroed knob
        // surface later as a stall or a rejected reply.
        config.validate()?;

        // options
        let auto_resubscribe = config.auto_resubscribe;
        let auto_remonitor = config.auto_remonitor;
        let max_command_attempts = config.max_command_attempts;
        let max_messages_per_wave = config.max_messages_per_wave;
        let reconnection_config = config.reconnection.clone();
        #[cfg(test)]
        let send_batch_test_hook = config.send_batch_test_hook.clone();

        // One registry per client: two clients built from the same `Config` must
        // not share the state either of them sets at runtime, which is exactly
        // why this is lent to the connection rather than carried by the config.
        let mut connection_state = ConnectionState::default();

        let connection = Connection::connect(config, &mut connection_state).await?;
        let (msg_sender, msg_receiver): (MsgSender, MsgReceiver) =
            tokio::sync::mpsc::unbounded_channel();
        let (reconnect_sender, _): (ReconnectSender, ReconnectReceiver) = broadcast::channel(32);
        let tag = connection.tag().to_owned();

        let mut network_handler = NetworkHandler {
            status: Status::Connected,
            connection,
            msg_sender: msg_sender.downgrade(),
            msg_receiver,
            messages_to_send: VecDeque::new(),
            messages_to_receive: VecDeque::new(),
            pending_subscriptions: VecDeque::new(),
            pending_unsubscriptions: VecDeque::new(),
            subscriptions: HashMap::new(),
            is_reply_on: true,
            skip_next_reply: false,
            connection_state,
            invalidation_sender: None,
            monitor_sender: None,
            reconnect_sender: reconnect_sender.clone(),
            auto_resubscribe,
            auto_remonitor,
            reconnection_state: ReconnectionState::new(reconnection_config),
            max_command_attempts,
            max_messages_per_wave,
            results_to_discard: 0,
            #[cfg(test)]
            send_batch_test_hook,
        };

        // Every event emitted by the network task, and by the connection code it
        // calls into, inherits this span. That is what carries the connection
        // identity, so no message below has to spell it out.
        let span = info_span!("connection", tag = %tag);

        let join_handle = spawn(
            async move {
                if let Err(e) = network_handler.network_loop().await {
                    error!("network loop ended in error: {e}");
                }
            }
            .instrument(span),
        );

        Ok((msg_sender, join_handle, reconnect_sender, tag))
    }

    async fn network_loop(&mut self) -> Result<()> {
        loop {
            select! {
                msg = poll_fn(|cx| self.msg_receiver.poll_recv(cx)).fuse() => {
                    if !self.try_handle_message(msg).await { break; }
                },
                result = self.connection.read().fuse() => {
                    if !self.try_handle_result(result).await { break; }
                }
            }
        }

        debug!("end of network loop");
        Ok(())
    }

    async fn try_handle_message(&mut self, mut msg: Option<Message>) -> bool {
        let mut is_channel_closed = false;
        // Messages queued since the last flush, for the wave cap below.
        let mut queued: usize = 0;

        loop {
            if let Some(msg) = msg {
                self.handle_message(msg);
                queued += 1;
            } else {
                is_channel_closed = true;
                break;
            }

            // Send in waves rather than accumulating the whole channel into
            // one write (see `Config::max_messages_per_wave`).
            if queued >= self.max_messages_per_wave {
                if self.status != Status::Disconnected {
                    self.send_messages().await;
                }
                queued = 0;
            }

            match self.msg_receiver.try_recv() {
                Ok(m) => msg = Some(m),
                Err(_) => {
                    // there are no messages available, but channel is not yet closed
                    break;
                }
            }
        }

        if self.status != Status::Disconnected {
            self.send_messages().await
        }

        !is_channel_closed
    }

    fn handle_message(&mut self, mut msg: Message) {
        trace!("[{:?}] Will handle message: {msg:?}", self.status);

        let mut collision_error = None;

        match &self.status {
            Status::Connected => {
                match &mut msg.kind {
                    MessageKind::PubSub {
                        subscription_type,
                        subscriptions,
                        ..
                    } => {
                        for (channel_or_pattern, _sender) in subscriptions.iter() {
                            if self.subscriptions.contains_key(channel_or_pattern) {
                                debug!(
                                    "[{:?}] There is already a subscription on channel `{}`",
                                    self.status,
                                    String::from_utf8_lossy(channel_or_pattern)
                                );
                                collision_error =
                                    Some(Error::Client(ClientError::AlreadySubscribed));
                                break;
                            }
                        }

                        if collision_error.is_none() {
                            let subscriptions = std::mem::take(subscriptions);
                            let num_pending_subscriptions = subscriptions.len();
                            let pending_subscriptions = subscriptions.into_iter().enumerate().map(
                                |(index, (channel_or_pattern, sender))| PendingSubscription {
                                    channel_or_pattern,
                                    subscription_type: *subscription_type,
                                    sender,
                                    more_to_come: index < num_pending_subscriptions - 1,
                                },
                            );

                            self.pending_subscriptions.extend(pending_subscriptions);
                        }
                    }
                    MessageKind::Monitor { push_sender, .. } => {
                        self.status = Status::EnteringMonitor;
                        let push_sender = push_sender.take();
                        if let Some(push_sender) = push_sender {
                            debug!("Registering MONITOR push_sender");
                            self.monitor_sender = Some(push_sender);
                        }
                    }
                    MessageKind::Invalidation { push_sender } => {
                        let push_sender = push_sender.take();
                        if let Some(push_sender) = push_sender {
                            debug!("Registering Invalidation push_sender");
                            self.invalidation_sender = Some(push_sender);
                        }
                        return; // no message to send
                    }
                    MessageKind::Single { command, .. } => {
                        if let CommandKind::Unsbuscribe(subscription_type) = command.kind() {
                            self.pending_unsubscriptions.push_back(
                                command.args().map(|a| (a, *subscription_type)).collect(),
                            );
                        }
                    }

                    _ => (),
                }

                if let Some(err) = collision_error {
                    msg.send_error(err);
                } else {
                    self.messages_to_send.push_back(MessageToSend::new(msg));
                }
            }
            Status::Disconnected => {
                if msg.retry_on_error {
                    debug!(
                        "network disconnected, queuing command: {:?}",
                        msg.commands()
                    );
                    self.messages_to_send.push_back(MessageToSend::new(msg));
                } else {
                    debug!(
                        "network disconnected, sending command in error: {:?}",
                        msg.commands()
                    );
                    msg.send_error(Error::DisconnectedByPeer);
                }
            }
            Status::EnteringMonitor => self.messages_to_send.push_back(MessageToSend::new(msg)),
            Status::Monitor => {
                for command in msg.commands() {
                    if matches!(command.kind(), CommandKind::Reset) {
                        self.status = Status::LeavingMonitor;
                    }
                }
                self.messages_to_send.push_back(MessageToSend::new(msg));
            }
            Status::LeavingMonitor => {
                self.messages_to_send.push_back(MessageToSend::new(msg));
            }
        }
    }

    async fn send_messages(&mut self) {
        // The line is only worth emitting for an actual batch, and deciding that
        // needs the count, so the count is taken here rather than inside the
        // macro argument. Guarding it with `enabled!` instead would silence the
        // line for every `log`-only consumer, which the bridge exists to serve.
        //
        // The walk is bounded by `max_messages_per_wave` and each step is a
        // discriminant read; the loop below iterates the same queue and encodes
        // every command in it.
        if !self.messages_to_send.is_empty() {
            let num_commands = self
                .messages_to_send
                .iter()
                .fold(0, |sum, msg| sum + msg.message.num_commands());
            if num_commands > 1 {
                debug!("sending batch of {num_commands} commands");
            }
        }

        // Test-only: force retry reasons onto the first message of this drain so
        // a test can reproduce a redirected message ahead of unrelated ones.
        #[cfg(test)]
        if let Some(hook) = &self.send_batch_test_hook
            && !self.messages_to_send.is_empty()
            && let Some(reasons) = hook.take_injection()
            && let Some(front) = self.messages_to_send.front_mut()
        {
            front.message.retry_reasons = Some(reasons);
        }

        let start_idx = self.messages_to_receive.len();

        while let Some(message_to_send) = self.messages_to_send.pop_front() {
            let mut msg = message_to_send.message;

            // Scope the retry reasons to the current message: they must not
            // leak onto the other messages sharing this send batch.
            let mut retry_reasons = SmallVec::<[RetryReason; 10]>::new();
            let reasons = msg.retry_reasons.take();
            if let Some(reasons) = reasons {
                retry_reasons.extend(reasons);
            }

            let mut num_commands_to_receive: usize = 0;

            // Commands are fed one by one on purpose. Batching them into a single
            // call (to hoist the stream-variant `match` out of the loop and issue
            // one pre-computed reserve) was implemented and measured against a live
            // Redis: no change (long pipeline +1.3%, p=0.53). The 8 KiB write-buffer
            // flush (see `CommandEncoder::encode`) already caps the buffer, so there
            // is nothing to amortize. Keep the per-command loop.
            for command in msg.commands_mut() {
                let kind = *command.kind();

                match kind {
                    CommandKind::ClientReply(ClientReplyMode::On) => {
                        self.is_reply_on = true;
                        self.skip_next_reply = false;
                        self.connection_state.record(StateSlot::ReplyMode, command);
                    }
                    CommandKind::ClientReply(ClientReplyMode::Off) => {
                        self.is_reply_on = false;
                        self.skip_next_reply = false;
                        self.connection_state.record(StateSlot::ReplyMode, command);
                    }
                    // `SKIP` is not connection state: it is consumed by the next
                    // command and leaves the connection as it found it.
                    CommandKind::ClientReply(ClientReplyMode::Skip) => {
                        self.skip_next_reply = true;
                    }
                    CommandKind::ConnectionState(slot) => {
                        self.connection_state.record(slot, command);
                    }
                    // The server restores every connection default here, so the
                    // client's picture of the connection must go with it —
                    // including the reply mode, which `RESET` itself answers
                    // through.
                    CommandKind::Reset => {
                        self.connection_state.clear();
                        self.is_reply_on = true;
                        self.skip_next_reply = false;
                        self.subscriptions.clear();
                    }
                    _ => (),
                }

                // The registry just changed, so the copy a cluster connection
                // replays onto a joining node has to change with it. This is the
                // only place connection state is recorded, which makes it the only
                // sync point needed.
                if matches!(
                    kind,
                    CommandKind::ConnectionState(_)
                        | CommandKind::ClientReply(_)
                        | CommandKind::Reset
                ) {
                    self.connection
                        .sync_connection_state(&self.connection_state);
                }

                let expects_reply = if !self.is_reply_on {
                    false
                } else if matches!(kind, CommandKind::ClientReply(ClientReplyMode::Skip)) {
                    // `CLIENT REPLY SKIP` is not answered either.
                    false
                } else if self.skip_next_reply {
                    self.skip_next_reply = false;
                    false
                } else {
                    true
                };

                if expects_reply {
                    num_commands_to_receive += 1;
                }

                // Test-only: record the retry reasons this command is fed with,
                // so a test can assert reasons do not leak across messages.
                #[cfg(test)]
                if let Some(hook) = &self.send_batch_test_hook {
                    let command_name = String::from_utf8_lossy(command.name()).into_owned();
                    hook.record_fed(command_name.clone(), retry_reasons.len());

                    // Arm a read-side kill onto this command if a test queued one
                    // for its name, reusing the per-command countdown so the
                    // existing `feed` path picks it up.
                    if let Some(num_reads) = hook.take_kill_on_read_for(&command_name) {
                        command
                            .kill_connection_on_read
                            .store(num_reads, std::sync::atomic::Ordering::SeqCst);
                    }
                }

                if let Err(e) = self.connection.feed(command, &retry_reasons).await {
                    error!("Feed error: {e}");
                    msg.send_error(e);
                    return;
                }
            }

            if num_commands_to_receive > 0 {
                self.messages_to_receive
                    .push_back(MessageToReceive::new(msg, num_commands_to_receive));
            }
        }

        if let Err(e) = self.connection.flush().await {
            error!("Flush error: {e}");

            while self.messages_to_receive.len() > start_idx {
                if let Some(msg_to_receive) = self.messages_to_receive.pop_back() {
                    msg_to_receive.message.send_error(e.clone());
                }
            }
        }
    }

    async fn try_handle_result(&mut self, result: Option<Result<RespResponse>>) -> bool {
        let Some(result) = result else {
            return self.reconnect().await;
        };
        // A protocol decode error desynchronizes the stream; attributing it to the
        // head-of-queue message blames an innocent caller. Reconnect instead, which
        // resynchronizes the stream and purges/replays in-flight messages cleanly.
        if let Err(e) = &result
            && is_connection_level_error(e)
        {
            debug!("Connection-level read error, reconnecting: {e}");
            return self.reconnect().await;
        }
        self.handle_result(result);

        // OPTIMIZATION : Drain the next available results in the buffer
        while let Poll::Ready(result) = self.connection.try_read() {
            let Some(result) = result else {
                return self.reconnect().await;
            };
            if let Err(e) = &result
                && is_connection_level_error(e)
            {
                debug!("Connection-level read error, reconnecting: {e}");
                return self.reconnect().await;
            }
            self.handle_result(result);
        }

        true
    }

    /// Hands a matched reply to its caller, waking it.
    ///
    /// Called from [`Self::receive_result`] the moment the reply is matched,
    /// before the next ready reply is parsed: on a multi-thread runtime another
    /// worker resumes the caller in parallel while this task keeps draining,
    /// which shortens first-reply latency on the critical path.
    fn dispatch_result<T>(&self, sender: tokio::sync::oneshot::Sender<T>, value: T) {
        if sender.send(value).is_err() {
            warn!("Cannot send value to caller because receiver is not there anymore");
        }
    }

    fn handle_result(&mut self, result: Result<RespResponse>) {
        match self.status {
            Status::Disconnected => (),
            Status::Connected => match &result {
                Ok(response) if response.is_push() => {
                    if let Some(response) = self.try_match_pubsub_message(result) {
                        if response.is_err() {
                            self.receive_result(response);
                        } else {
                            match &mut self.invalidation_sender {
                                Some(push_sender) => {
                                    if let Err(e) = push_sender.unbounded_send(response) {
                                        warn!("Cannot send push message result to caller: {e}");
                                    }
                                }
                                None => {
                                    warn!(
                                        "Received a push message with no sender configured: {response:?}"
                                    )
                                }
                            }
                        }
                    }
                }
                _ => {
                    self.receive_result(result);
                }
            },
            Status::EnteringMonitor => {
                self.receive_result(result);
                self.status = Status::Monitor;
            }
            Status::Monitor => match &result {
                Ok(response) if response.is_monitor() => {
                    if let Some(push_sender) = &mut self.monitor_sender
                        && let Err(e) = push_sender.unbounded_send(result)
                    {
                        warn!("Cannot send monitor result to caller: {e}");
                    }
                }
                _ => self.receive_result(result),
            },
            Status::LeavingMonitor => match &result {
                Ok(response) if response.is_monitor() => {
                    if let Some(push_sender) = &mut self.monitor_sender
                        && let Err(e) = push_sender.unbounded_send(result)
                    {
                        warn!("Cannot send monitor result to caller: {e}");
                    }
                }
                _ => {
                    self.receive_result(result);
                    self.status = Status::Connected;
                }
            },
        }
    }

    fn receive_result(&mut self, result: Result<RespResponse>) {
        // Responses owed to a message that was already resolved as a whole: the
        // commands were executed, so their replies still arrive, but there is no
        // caller left for them. Matching them would shift every subsequent
        // response by one.
        if self.results_to_discard > 0 {
            self.results_to_discard -= 1;
            debug!("discarding response of an already resolved message: {result:?}");
            return;
        }

        match self.messages_to_receive.front_mut() {
            Some(message_to_receive) => {
                trace!("message_to_receive: {:?}", message_to_receive);

                if message_to_receive.num_commands == 1 || result.is_err() {
                    if let Some(mut message_to_receive) = self.messages_to_receive.pop_front() {
                        // A batch message is sent as several independent
                        // commands, each awaiting its own response. Resolving
                        // the whole message on the error of one of them leaves
                        // the commands queued behind it without a caller, while
                        // their replies are already on their way.
                        if message_to_receive.num_commands > 1 {
                            self.results_to_discard += message_to_receive.num_commands - 1;
                        }

                        let mut should_retry = false;

                        if let Err(Error::Retry(_)) = &result {
                            should_retry = true;
                        } else if message_to_receive.message.retry_reasons.is_some() {
                            should_retry = true;
                        }

                        if should_retry {
                            if let Err(Error::Retry(reasons)) = result {
                                if let Some(retry_reasons) =
                                    &mut message_to_receive.message.retry_reasons
                                {
                                    retry_reasons.extend(reasons);
                                } else {
                                    message_to_receive.message.retry_reasons =
                                        Some(Vec::from_iter(reasons));
                                }
                            }

                            // Bound message-level retries: a command caught in a
                            // pathological redirect loop would otherwise be replayed
                            // forever. Count this attempt and fail the message with a
                            // distinct error once the cap is reached.
                            message_to_receive.message.attempts += 1;
                            if max_attempts_reached(
                                message_to_receive.message.attempts,
                                self.max_command_attempts,
                            ) {
                                debug!(
                                    "Message reached the maximum number of attempts, failing it"
                                );
                                message_to_receive.message.send_error(Error::Client(
                                    ClientError::MaxCommandAttemptsReached,
                                ));
                            }
                            // retry: upgrade the weak handle just long enough to
                            // requeue the message. A failed upgrade means every
                            // client is gone and the channel is closing, so the
                            // retry is moot.
                            else if let Some(msg_sender) = self.msg_sender.upgrade() {
                                if let Err(e) = msg_sender.send(message_to_receive.message) {
                                    error!("Cannot retry message: {e}");
                                }
                            } else {
                                debug!("Cannot retry message: channel closed");
                            }
                        } else {
                            trace!("Will respond to: {:?}", message_to_receive.message);

                            match message_to_receive.message.kind {
                                MessageKind::Single {
                                    result_sender: Some(result_sender),
                                    ..
                                }
                                | MessageKind::PubSub { result_sender, .. }
                                | MessageKind::Monitor { result_sender, .. } => {
                                    self.dispatch_result(result_sender, result);
                                }
                                MessageKind::Batch { results_sender, .. } => match result {
                                    Ok(resp_buf) => {
                                        message_to_receive.pending_responses.push(resp_buf);
                                        self.dispatch_result(
                                            results_sender,
                                            Ok(message_to_receive.pending_responses),
                                        );
                                    }
                                    Err(e) => {
                                        self.dispatch_result(results_sender, Err(e));
                                    }
                                },
                                MessageKind::Invalidation { .. }
                                | MessageKind::Single {
                                    result_sender: None,
                                    ..
                                } => {
                                    debug!("forget value {result:?}")
                                    // fire & forget
                                }
                            }
                        }
                    }
                } else {
                    match result {
                        Ok(value) => {
                            message_to_receive.pending_responses.push(value);
                            message_to_receive.num_commands -= 1;
                        }
                        Err(Error::Retry(reasons)) => {
                            if let Some(retry_reasons) =
                                &mut message_to_receive.message.retry_reasons
                            {
                                retry_reasons.extend(reasons);
                            } else {
                                message_to_receive.message.retry_reasons =
                                    Some(Vec::from_iter(reasons));
                            }
                        }
                        _ => (),
                    }
                }
            }
            None => {
                // Disconnection errors legitimately end here (no message is left
                // to carry them). An `Ok` frame with an empty in-flight queue is
                // unexpected — a mis-routed push, a buggy server/proxy, or a
                // desynchronized stream — but a network loop must never panic on
                // wire input: that would kill the sole owner of the routing state
                // and permanently wedge the client with no reconnection. Drop the
                // stray frame and log it instead.
                if result.is_ok() {
                    warn!(
                        "Dropping an unexpected response with no message awaiting it: {result:?}"
                    );
                }
            }
        }
    }

    fn try_match_pubsub_message(
        &mut self,
        value: Result<RespResponse>,
    ) -> Option<Result<RespResponse>> {
        if let Ok(ref_value) = &value {
            if let Ok(pub_sub_message) = PubSubMessage::try_from(ref_value) {
                match pub_sub_message {
                    PubSubMessage::Message(channel_or_pattern, _)
                    | PubSubMessage::SMessage(channel_or_pattern, _) => {
                        match self.subscriptions.get_mut(channel_or_pattern) {
                            Some((_subscription_type, pub_sub_sender)) => {
                                if let Err(e) = pub_sub_sender.unbounded_send(value) {
                                    let error_desc = e.to_string();
                                    if let Ok(ref_value) = &e.into_inner()
                                        && let Some(
                                            PubSubMessage::Message(channel_or_pattern, _)
                                            | PubSubMessage::SMessage(channel_or_pattern, _),
                                        ) = PubSubMessage::try_from(ref_value).ok()
                                    {
                                        warn!(
                                            "Cannot send pub/sub message to caller from channel `{}`: {error_desc}",
                                            String::from_utf8_lossy(channel_or_pattern)
                                        );
                                    }
                                }
                            }
                            None => {
                                error!(
                                    "Unexpected message on channel `{}`",
                                    String::from_utf8_lossy(channel_or_pattern)
                                );
                            }
                        }
                        None
                    }
                    PubSubMessage::Subscribe(channel_or_pattern)
                    | PubSubMessage::PSubscribe(channel_or_pattern)
                    | PubSubMessage::SSubscribe(channel_or_pattern) => {
                        // Peek before popping: a mismatched confirmation must not
                        // consume (and silently drop) the pending subscriber. Only
                        // pop once we know the front entry is the one being confirmed.
                        let matches = self
                            .pending_subscriptions
                            .front()
                            .is_some_and(|p| p.channel_or_pattern == channel_or_pattern);
                        if matches && let Some(pending_sub) = self.pending_subscriptions.pop_front()
                        {
                            if self
                                .subscriptions
                                .insert(
                                    pending_sub.channel_or_pattern,
                                    (pending_sub.subscription_type, pending_sub.sender),
                                )
                                .is_some()
                            {
                                return Some(Err(Error::Client(ClientError::AlreadySubscribed)));
                            }

                            if pending_sub.more_to_come {
                                return None;
                            }

                            self.receive_result(Ok(RespResponse::ok()));
                        } else {
                            error!(
                                "Unexpected subscription confirmation on channel `{}`",
                                String::from_utf8_lossy(channel_or_pattern)
                            );
                            // Surface the anomaly to the caller instead of reporting
                            // a spurious success; the pending entry is left intact.
                            self.receive_result(Err(Error::Client(
                                ClientError::UnexpectedSubscriptionConfirmation,
                            )));
                        }
                        None
                    }
                    PubSubMessage::Unsubscribe(channel_or_pattern)
                    | PubSubMessage::PUnsubscribe(channel_or_pattern)
                    | PubSubMessage::SUnsubscribe(channel_or_pattern) => {
                        self.subscriptions.remove(channel_or_pattern);
                        if let Some(remaining) = self.pending_unsubscriptions.front_mut() {
                            if remaining.len() > 1 {
                                if remaining.remove(channel_or_pattern).is_none() {
                                    error!(
                                        "Cannot find channel or pattern to remove: `{}`",
                                        String::from_utf8_lossy(channel_or_pattern)
                                    );
                                }
                                None
                            } else {
                                // last unsubscription notification received
                                let Some(mut remaining) = self.pending_unsubscriptions.pop_front()
                                else {
                                    error!(
                                        "Cannot find channel or pattern to remove: `{}`",
                                        String::from_utf8_lossy(channel_or_pattern)
                                    );
                                    return None;
                                };
                                if remaining.remove(channel_or_pattern).is_none() {
                                    error!(
                                        "Cannot find channel or pattern to remove: `{}`",
                                        String::from_utf8_lossy(channel_or_pattern)
                                    );
                                    return None;
                                }
                                self.receive_result(Ok(RespResponse::ok()));
                                None
                            }
                        } else {
                            Some(value)
                        }
                    }
                    PubSubMessage::PMessage(pattern, channel, _) => {
                        match self.subscriptions.get_mut(pattern) {
                            Some((_subscription_type, pub_sub_sender)) => {
                                if let Err(e) = pub_sub_sender.unbounded_send(value) {
                                    warn!("Cannot send pub/sub message to caller: {e}");
                                }
                            }
                            None => {
                                error!(
                                    "Unexpected message on channel `{}` for pattern `{}`",
                                    String::from_utf8_lossy(channel),
                                    String::from_utf8_lossy(pattern)
                                );
                            }
                        }
                        None
                    }
                }
            } else {
                Some(value)
            }
        } else {
            Some(value)
        }
    }

    /// Nested inside the connection span, so everything a reconnection does —
    /// purging in-flight messages, replaying subscriptions, failing what
    /// exhausted its retry budget — is grouped under one identifiable unit
    /// instead of being interleaved with ordinary traffic.
    #[tracing::instrument(name = "reconnect", skip_all)]
    async fn reconnect(&mut self) -> bool {
        debug!("reconnecting...");
        let old_status = self.status;
        self.status = Status::Disconnected;

        // The responses we were waiting to discard died with the connection;
        // keeping the count would discard legitimate responses afterwards.
        self.results_to_discard = 0;

        // A `SKIP` waiting for the command it silences died with the connection too.
        self.skip_next_reply = false;

        // Purge every non-retryable message, wherever it sits in the queue,
        // and keep the retryable ones in order. A prefix-only purge would leave
        // a non-retryable message behind a retryable one, and it would then be
        // replayed on reconnect — double-executing a command whose caller
        // opted out of retries.
        // A reconnection replay is also a retry attempt: count it and fail a message
        // that has exhausted its budget instead of replaying it once more.
        let max_command_attempts = self.max_command_attempts;

        let mut retained_to_receive = VecDeque::with_capacity(self.messages_to_receive.len());
        while let Some(mut message_to_receive) = self.messages_to_receive.pop_front() {
            if !message_to_receive.message.retry_on_error {
                message_to_receive
                    .message
                    .send_error(Error::DisconnectedByPeer);
            } else {
                message_to_receive.message.attempts += 1;
                if max_attempts_reached(message_to_receive.message.attempts, max_command_attempts) {
                    message_to_receive
                        .message
                        .send_error(Error::Client(ClientError::MaxCommandAttemptsReached));
                } else {
                    retained_to_receive.push_back(message_to_receive);
                }
            }
        }
        self.messages_to_receive = retained_to_receive;

        let mut retained_to_send = VecDeque::with_capacity(self.messages_to_send.len());
        while let Some(mut message_to_send) = self.messages_to_send.pop_front() {
            if !message_to_send.message.retry_on_error {
                message_to_send
                    .message
                    .send_error(Error::DisconnectedByPeer);
            } else {
                message_to_send.message.attempts += 1;
                if max_attempts_reached(message_to_send.message.attempts, max_command_attempts) {
                    message_to_send
                        .message
                        .send_error(Error::Client(ClientError::MaxCommandAttemptsReached));
                } else {
                    retained_to_send.push_back(message_to_send);
                }
            }
        }
        self.messages_to_send = retained_to_send;

        loop {
            if let Some(delay) = self.reconnection_state.next_delay() {
                debug!("Waiting {delay} ms before reconnection");

                // keep on receiving new message during the delay
                let start = Instant::now();
                // A pathologically large reconnection delay would overflow the
                // monotonic clock; cap it rather than panicking the network task.
                let end = start
                    .checked_add(Duration::from_millis(delay))
                    .unwrap_or_else(|| start + Duration::from_secs(3600));
                loop {
                    let delay = end.duration_since(Instant::now());
                    let result =
                        timeout(delay, poll_fn(|cx| self.msg_receiver.poll_recv(cx))).await;
                    if let Ok(msg) = result {
                        if !self.try_handle_message(msg).await {
                            return false;
                        }
                    } else {
                        // delay has expired
                        break;
                    }
                }
            } else {
                warn!("Max reconnection attempts reached");
                while let Some(message_to_receive) = self.messages_to_receive.pop_front() {
                    message_to_receive
                        .message
                        .send_error(Error::DisconnectedByPeer);
                }
                while let Some(message_to_send) = self.messages_to_send.pop_front() {
                    message_to_send
                        .message
                        .send_error(Error::DisconnectedByPeer);
                }
                return false;
            }

            if let Err(e) = self.connection.reconnect(&mut self.connection_state).await {
                error!("Failed to reconnect: {e:?}");
                continue;
            }

            // The new connection was restored from the same registry, so the
            // mirror the send loop uses has to be derived from it rather than
            // left at whatever the dead connection ended on.
            self.is_reply_on = self.connection_state.is_reply_on();

            if self.auto_resubscribe
                && let Err(e) = self.auto_resubscribe().await
            {
                error!("Failed to reconnect: {e:?}");
                continue;
            }

            if self.auto_remonitor
                && let Err(e) = self.auto_remonitor(old_status).await
            {
                error!("Failed to reconnect: {e:?}");
                continue;
            }

            if let Err(e) = self.reconnect_sender.send(()) {
                debug!("Cannot send reconnect notification to clients: {e}");
            }

            // Restore the connection status before replaying in-flight
            // messages so that they are routed through `handle_message`,
            // exactly as fresh messages and the retry path are.
            if let Status::Monitor | Status::EnteringMonitor = old_status {
                if self.monitor_sender.is_some() {
                    self.status = Status::Monitor;
                } else {
                    self.status = Status::Connected;
                }
            } else {
                self.status = Status::Connected;
            }

            // Replay every in-flight message through `handle_message` rather
            // than pushing it straight into `messages_to_send`. This rebuilds
            // the pub/sub bookkeeping (`pending_subscriptions` /
            // `pending_unsubscriptions`) for the replayed messages exactly as
            // the retry path does. Bypassing it would replay, for instance, an
            // UNSUBSCRIBE without a matching `pending_unsubscriptions` entry:
            // its confirmation push would then go unmatched, the stale message
            // would keep its slot in the receive queue, and every subsequent
            // response would be shifted by one, permanently. Messages already
            // sent but awaiting a reply are replayed before the ones still
            // queued, preserving the original global send order.
            let to_replay: Vec<Message> = std::mem::take(&mut self.messages_to_receive)
                .into_iter()
                .map(|message_to_receive| message_to_receive.message)
                .chain(
                    std::mem::take(&mut self.messages_to_send)
                        .into_iter()
                        .map(|message_to_send| message_to_send.message),
                )
                .collect();
            for message in to_replay {
                self.handle_message(message);
            }

            self.send_messages().await;

            info!("reconnected!");
            self.reconnection_state.reset_attempts();
            return true;
        }
    }

    async fn auto_resubscribe(&mut self) -> Result<()> {
        // Drop every pending unsubscription first, emitting nothing. On a fresh
        // connection the server is subscribed to nothing, so a pending
        // unsubscription has already achieved its goal. Removing the channels
        // from `subscriptions` up front also prevents the resubscribe loop
        // below from restoring subscriptions the caller was in the middle of
        // cancelling.
        for map in self.pending_unsubscriptions.drain(..) {
            for channel_or_pattern in map.into_keys() {
                self.subscriptions.remove(&channel_or_pattern);
            }
        }

        if !self.subscriptions.is_empty() {
            for (channel_or_pattern, (subscription_type, _)) in &self.subscriptions {
                match subscription_type {
                    SubscriptionType::Channel => {
                        self.connection.subscribe(channel_or_pattern).await?;
                    }
                    SubscriptionType::Pattern => {
                        self.connection.psubscribe(channel_or_pattern).await?;
                    }
                    SubscriptionType::ShardChannel => {
                        self.connection.ssubscribe(channel_or_pattern).await?;
                    }
                }
            }
        }

        if !self.pending_subscriptions.is_empty() {
            for pending_sub in self.pending_subscriptions.drain(..) {
                match pending_sub.subscription_type {
                    SubscriptionType::Channel => {
                        self.connection
                            .subscribe(pending_sub.channel_or_pattern.clone())
                            .await?;
                    }
                    SubscriptionType::Pattern => {
                        self.connection
                            .psubscribe(pending_sub.channel_or_pattern.clone())
                            .await?;
                    }
                    SubscriptionType::ShardChannel => {
                        self.connection
                            .ssubscribe(pending_sub.channel_or_pattern.clone())
                            .await?;
                    }
                }

                self.subscriptions.insert(
                    pending_sub.channel_or_pattern,
                    (pending_sub.subscription_type, pending_sub.sender),
                );
            }
        }

        Ok(())
    }

    async fn auto_remonitor(&mut self, old_status: Status) -> Result<()> {
        if let Status::Monitor | Status::EnteringMonitor = old_status {
            self.connection.send(&cmd("MONITOR").into()).await?;
        }

        Ok(())
    }
}

/// Whether an error surfaced by `connection.read()` is a connection-level failure
/// (a protocol decode error, a transport/IO error, or end of stream) rather than a
/// per-message one.
///
/// A decode error desynchronizes the byte stream, so it belongs to the connection,
/// not to whichever caller happens to sit at the head of the receive queue; it must
/// trigger a reconnect (clean purge + replay) instead of being dispatched as that
/// caller's result. Per-message errors that legitimately arrive here — a cluster
/// `Error::Retry` (ASK/MOVED) and a `Error::Redis` command error from a failing
/// shard — must be delivered to the caller, so this is a positive allow-list of the
/// framing/transport errors: anything unlisted is treated as per-message, which is
/// the safe default (a stray error reaches one caller instead of churning the whole
/// connection).
#[inline]
fn is_connection_level_error(error: &Error) -> bool {
    match error {
        Error::IO(_) | Error::EOF => true,
        Error::Client(client_error) => matches!(
            client_error,
            ClientError::CannotParseInteger
                | ClientError::CannotParseDouble
                | ClientError::CannotParseBulkString
                | ClientError::CannotParseBulkError
                | ClientError::CannotParseVerbatimString
                | ClientError::CannotParseBoolean
                | ClientError::CannotParseMap
                | ClientError::CannotParseSequence
                | ClientError::UnknownRespTag(_)
                | ClientError::BulkLengthTooLarge
                | ClientError::CollectionLengthTooLarge
                | ClientError::MaxNestingDepthExceeded
                | ClientError::VerbatimStringTooShort
        ),
        _ => false,
    }
}

/// Whether a message that has been attempted `attempts` times has reached the
/// configured per-message cap. `cap == 0` means unlimited (the default), matching
/// the historical behavior of never bounding retries at the message level.
#[inline]
fn max_attempts_reached(attempts: usize, cap: usize) -> bool {
    cap != 0 && attempts >= cap
}

#[cfg(test)]
mod tests {
    #![allow(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::panic,
        clippy::unreachable,
        clippy::indexing_slicing,
        reason = "test code: a panic is how a test reports failure"
    )]
    use super::{is_connection_level_error, max_attempts_reached};

    #[test]
    fn zero_cap_is_unlimited() {
        assert!(!max_attempts_reached(1, 0));
        assert!(!max_attempts_reached(1_000_000, 0));
    }

    #[test]
    fn cap_reached_at_or_above_limit() {
        assert!(!max_attempts_reached(2, 3));
        assert!(max_attempts_reached(3, 3));
        assert!(max_attempts_reached(4, 3));
    }
    use crate::{ClientError, Error, RedisError, RedisErrorKind};

    #[test]
    fn per_message_errors_are_not_connection_level() {
        // Cluster redirection and a failing-shard Redis error must reach the
        // caller, not tear down the connection.
        assert!(!is_connection_level_error(
            &Error::Retry(Default::default())
        ));
        assert!(!is_connection_level_error(&Error::Redis(RedisError {
            kind: RedisErrorKind::NoPerm,
            description: "no permission".to_owned(),
        })));
        // A caller-side client error is not a stream desync either.
        assert!(!is_connection_level_error(&Error::Client(
            ClientError::CrossSlot
        )));
    }

    #[test]
    fn decode_and_transport_errors_are_connection_level() {
        assert!(is_connection_level_error(&Error::Client(
            ClientError::CannotParseInteger
        )));
        assert!(is_connection_level_error(&Error::Client(
            ClientError::UnknownRespTag('?')
        )));
        assert!(is_connection_level_error(&Error::Client(
            ClientError::MaxNestingDepthExceeded
        )));
        assert!(is_connection_level_error(&Error::EOF));
    }
}