//! Off-request-path replication shipping: per-peer windowed batch senders.
//!
//! m11p1 built this queue over the relay's in-memory log; m11p2 re-sources it
//! from the **WAL's flushed-batch feed** ([`crate::wal::feed::WalShipFeed`]) —
//! the one replicated log. A [`ShipQueue`] owns `window` dedicated OS threads
//! per peer, each of which:
//!
//! 1. claims-and-collects the next contiguous run of already-encoded flushed
//!    batches from its [`ShipSource`] (up to `max_batch_events` /
//!    `max_batch_bytes`, never past the source's flushed frontier — an event
//!    the leader has not fsynced must never reach a follower);
//! 2. ships the run's bytes verbatim with a single blocking `send_segment`
//!    call (byte-identical on the leader's disk, the wire, and the follower's
//!    apply path — no re-encode);
//! 3. on success records the run and advances the peer's contiguous **acked
//!    frontier** (folding in the follower's reported applied seqno when the
//!    transport carries one); on a transient failure parks the run for retry
//!    after `retry_backoff`.
//!
//! With `window > 1`, up to `window` runs are in flight per peer
//! concurrently; runs may complete out of order, which the receiver's
//! gap-aware `apply_range` tolerates (ahead ranges park until the gap
//! closes), and the retry queue guarantees every gap eventually closes —
//! the "N batches outstanding, acks advance the window" protocol from the
//! m11 roadmap.
//!
//! # Bounded tail + follower-pulled catch-up (m11p2)
//!
//! The feed's tail is bounded. When a peer's cursor falls behind the retained
//! tail (long partition, long downtime), the source reports
//! [`ShipCollect::Rotated`] and the sender SKIPS to the tail floor: the
//! resulting receiver-side gap triggers the follower's own catch-up pull
//! (`Transport::request_catchup` → `StreamSegments` over the leader's durable
//! WAL segments). Push covers the hot path; pull covers history — the
//! unbounded in-memory log dependency is gone.
//!
//! # Leadership activation
//!
//! Every cluster node spawns a queue at boot (its WAL feeds it regardless of
//! role — a follower's feed entries are its replicated applies), but only an
//! ACTIVE queue dispatches. Activation follows leadership: without the gate, a
//! follower would echo its replicated applies back at its peers.
//!
//! # Threading contract
//!
//! Senders are plain `std::thread`s with no ambient tokio runtime — exactly
//! what [`Transport::send_segment`]'s blocking contract requires (the gRPC
//! implementation asserts it is not called from within a runtime).
//!
//! # Pause / resume (partition + heal verbs)
//!
//! `pause(peer)` stops new dispatch to a peer (in-flight sends complete);
//! `resume(peer)` restarts dispatch, and `resume_from(peer, applied)` jumps
//! the cursor past everything the follower reports applied — heal is then
//! resume + (if the tail rotated) the follower's own pull. A **permanent**
//! transport failure (TLS/auth/codec) auto-pauses the peer and logs loudly:
//! retrying a permanent failure forever is a silent stall.

use std::{
    collections::BTreeMap,
    sync::{
        Arc, Condvar, Mutex, PoisonError, Weak,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    thread::JoinHandle,
    time::{Duration, Instant},
};

use super::{
    commit::CommitIndex,
    relay::{SignalRelay, encode_run},
    shard::ShardId,
    transport::{Transport, TransportError},
};
use crate::wal::feed::{FeedCollect, WalShipFeed};

#[cfg(feature = "metrics")]
use crate::db::metrics::cluster::ClusterMetrics;

/// Repeat-failure WARN cadence: the 1st failed attempt per outage WARNs, then
/// every Nth, with the running count attached; everything between is `debug`.
/// Recovery logs at INFO with the outage's total. Keeps a long partition from
/// flooding the log (the unbounded per-retry WARN once filled an undrained
/// 64KB stderr pipe and froze every logging thread in the process).
const FAILURE_LOG_EVERY: u64 = 50;

// ── Ship source ─────────────────────────────────────────────────────────────

/// Result of a [`ShipSource::collect`] call.
pub enum ShipCollect {
    /// Encoded batches covering exactly `[first_seq, last_seq]`, ready to
    /// ship verbatim.
    Run {
        bytes: Vec<u8>,
        first_seq: u64,
        last_seq: u64,
        event_count: u64,
    },
    /// Nothing at or above the requested seqno is flushed yet.
    NothingNew,
    /// The requested seqno predates the retained data; dispatch must resume
    /// at `resume_from` and the skipped range becomes a receiver-side gap
    /// (closed by the follower's catch-up pull).
    Rotated { resume_from: u64 },
}

/// A leader-side source of encoded, durably-flushed replication batches.
///
/// Two implementations: [`WalFeedSource`] (the m11p2 production source — the
/// WAL's flushed-batch feed) and [`SignalRelay`] (the in-process harness
/// source, whose log never rotates).
pub trait ShipSource: Send + Sync + 'static {
    /// The shard this stream ships under (stamped into every payload).
    fn source_shard(&self) -> ShardId;

    /// The contiguous flushed frontier: every seqno `<=` this is durable on
    /// the leader and eligible to ship.
    fn flushed_seq(&self) -> u64;

    /// Collect a contiguous run of encoded batches starting exactly at
    /// `from_seq`, bounded by `max_events` / `max_bytes`.
    ///
    /// # Errors
    ///
    /// A deterministic source fault (encode failure, log corruption): the
    /// caller pauses the peer loudly rather than retrying forever.
    fn collect(
        &self,
        from_seq: u64,
        max_events: u64,
        max_bytes: usize,
    ) -> crate::Result<ShipCollect>;

    /// Install a wake callback fired when new data becomes shippable. Sources
    /// without an internal flush signal (the relay) may ignore this; callers
    /// then rely on explicit [`ShipQueue::publish`] wakes.
    fn set_notify(&self, _notify: Box<dyn Fn() + Send + Sync>) {}
}

/// The m11p2 production [`ShipSource`]: the WAL writer's flushed-batch feed.
pub struct WalFeedSource {
    shard: ShardId,
    feed: Arc<WalShipFeed>,
}

impl WalFeedSource {
    /// Wrap a WAL ship feed as this shard's outbound stream source.
    #[must_use]
    pub const fn new(shard: ShardId, feed: Arc<WalShipFeed>) -> Self {
        Self { shard, feed }
    }
}

impl ShipSource for WalFeedSource {
    fn source_shard(&self) -> ShardId {
        self.shard
    }

    fn flushed_seq(&self) -> u64 {
        self.feed.flushed_seq()
    }

    fn collect(
        &self,
        from_seq: u64,
        max_events: u64,
        max_bytes: usize,
    ) -> crate::Result<ShipCollect> {
        Ok(match self.feed.collect(from_seq, max_events, max_bytes) {
            FeedCollect::Run {
                bytes,
                first_seq,
                last_seq,
                event_count,
            } => ShipCollect::Run {
                bytes,
                first_seq,
                last_seq,
                event_count,
            },
            FeedCollect::NothingNew => ShipCollect::NothingNew,
            FeedCollect::Rotated { tail_floor } => ShipCollect::Rotated {
                resume_from: tail_floor,
            },
        })
    }

    fn set_notify(&self, notify: Box<dyn Fn() + Send + Sync>) {
        self.feed.set_notify(notify);
    }
}

/// Harness compatibility: the m11p1 relay is a [`ShipSource`] whose in-memory
/// log never rotates (so [`ShipCollect::Rotated`] is unreachable) and whose
/// flushed frontier is the relay's durable frontier.
impl ShipSource for SignalRelay {
    fn source_shard(&self) -> ShardId {
        Self::source_shard(self)
    }

    fn flushed_seq(&self) -> u64 {
        self.durable_seq()
    }

    fn collect(
        &self,
        from_seq: u64,
        max_events: u64,
        _max_bytes: usize,
    ) -> crate::Result<ShipCollect> {
        let max = usize::try_from(max_events.clamp(1, 256)).unwrap_or(256);
        let entries = self.snapshot_range(from_seq, max);
        if entries.is_empty() {
            return Ok(ShipCollect::NothingNew);
        }
        if entries[0].seqno != from_seq {
            // Unreachable for a contiguous relay log; deterministic if it
            // ever fires, so surface it as a source fault (peer pauses).
            return Err(crate::TidalError::internal(
                "ship_collect",
                format!(
                    "relay log snapshot mismatch: requested run starts at {from_seq} but \
                     snapshot starts at {}",
                    entries[0].seqno
                ),
            ));
        }
        let first_seq = entries[0].seqno;
        let last_seq = entries.last().map_or(first_seq, |e| e.seqno);
        let event_count = entries.len() as u64;
        let bytes = encode_run(&entries)?;
        Ok(ShipCollect::Run {
            bytes,
            first_seq,
            last_seq,
            event_count,
        })
    }
}

// ── Queue configuration ─────────────────────────────────────────────────────

/// Tuning for the per-peer batch senders.
#[derive(Debug, Clone, Copy)]
pub struct ShipQueueConfig {
    /// Maximum events coalesced into one shipped run. Clamped to the WAL wire
    /// format's per-batch ceiling (256) for the relay source; the feed source
    /// ships whole pre-encoded batches, so the cap bounds the run, not any
    /// single batch.
    pub max_batch_events: usize,
    /// Maximum bytes per shipped run (bounds blob-heavy runs; a single batch
    /// larger than this still ships alone). Defaults to 16 MiB — a quarter of
    /// the transport's 64 MiB payload ceiling.
    pub max_batch_bytes: usize,
    /// In-flight batches per peer (sender threads per peer). `1` = strictly
    /// in-order shipping; higher values pipeline across the peer RTT at the
    /// cost of out-of-order arrival (which the receiver parks gap-aware).
    pub window: usize,
    /// Delay before a transiently-failed run is retried.
    pub retry_backoff: Duration,
    /// This node's region id, stamped as `leader_region` on every outbound
    /// payload (m11p4 fencing). 0 for relay-path/test users without elections.
    pub leader_region: u16,
}

impl Default for ShipQueueConfig {
    fn default() -> Self {
        Self {
            max_batch_events: 256,
            max_batch_bytes: 16 * 1024 * 1024,
            window: 4,
            retry_backoff: Duration::from_millis(100),
            leader_region: 0,
        }
    }
}

// ── Per-peer state ──────────────────────────────────────────────────────────

/// Per-peer dispatch state, guarded by one mutex per peer.
struct PeerState {
    /// Next seqno to hand to a sender (everything below is dispatched,
    /// skipped-past-rotation, or covered by the follower's reported applied).
    dispatch_next: u64,
    /// Contiguous acked frontier: every seqno `<= acked` was accepted by the
    /// peer's transport (or reported applied by the follower itself).
    acked: u64,
    /// Completed-but-not-contiguous runs, keyed by first seqno → last seqno.
    completed: BTreeMap<u64, u64>,
    /// Failed runs awaiting retry: first seqno → (last seqno, due time).
    retry: BTreeMap<u64, (u64, Instant)>,
    /// Operator/permanent-failure pause: no new dispatch while set.
    paused: bool,
    /// Consecutive failed ship attempts since the last success, for
    /// transition-based logging (see [`FAILURE_LOG_EVERY`]). A partition
    /// window otherwise emits one WARN per retry attempt per sender —
    /// tens of lines per second, unbounded for the partition's duration.
    consecutive_failures: u64,
}

impl PeerState {
    /// Advance the contiguous acked frontier across completed runs and the
    /// follower's reported applied seqno, pruning state below it.
    fn advance_acked(&mut self, reported_applied: u64) {
        if reported_applied > self.acked {
            self.acked = reported_applied;
        }
        while let Some((&first, &last)) = self.completed.first_key_value() {
            if first <= self.acked + 1 {
                self.acked = self.acked.max(last);
                self.completed.remove(&first);
            } else {
                break;
            }
        }
        // Anything at or below the acked frontier no longer needs retrying:
        // the follower already has it.
        self.retry.retain(|_, (last, _)| *last > self.acked);
    }
}

/// One peer's shared dispatch cell.
struct PeerShip {
    peer: ShardId,
    state: Mutex<PeerState>,
    cv: Condvar,
    /// Lock-free mirror of `state.acked` for status reads.
    acked_atomic: AtomicU64,
}

/// The region every outbound batch names as its leader (m11p4). Lives in the
/// queue config so [`ShipQueue::spawn`]'s signature stays stable; defaults to
/// 0 for relay-path/test users that never run elections.
impl ShipQueueConfig {
    /// This node's region id, stamped as `leader_region` on every shipped
    /// payload alongside the term.
    #[must_use]
    pub const fn with_leader_region(mut self, region: u16) -> Self {
        self.leader_region = region;
        self
    }
}

impl PeerShip {
    fn lock(&self) -> std::sync::MutexGuard<'_, PeerState> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

/// Shared core of a [`ShipQueue`].
struct ShipShared {
    source: Arc<dyn ShipSource>,
    transport: Arc<dyn Transport>,
    peers: Vec<Arc<PeerShip>>,
    config: ShipQueueConfig,
    shutdown: AtomicBool,
    /// Leadership gate: only an active queue dispatches (see module docs).
    active: AtomicBool,
    /// The leadership term every outbound batch is stamped with (m11p4
    /// fencing). 0 = the topology era; set by [`ShipQueue::activate_from`]
    /// when this node leads an elected term.
    term: AtomicU64,
    /// Quorum commit index over the peers' durable marks (m11p3). Follows
    /// the queue's leadership gate: activated/deactivated with it.
    commit: Arc<CommitIndex>,
    #[cfg(feature = "metrics")]
    metrics: Option<Arc<ClusterMetrics>>,
}

impl ShipShared {
    fn peer(&self, peer: ShardId) -> Option<&Arc<PeerShip>> {
        self.peers.iter().find(|p| p.peer == peer)
    }

    fn wake_all(&self) {
        for cell in &self.peers {
            cell.cv.notify_all();
        }
    }
}

/// Per-peer windowed batch senders draining a [`ShipSource`]'s durable log.
///
/// Spawn once per node ([`ShipQueue::spawn`]); the source's flush notify (or
/// an explicit [`publish`](Self::publish)) wakes parked senders. Threads exit
/// on [`shutdown`](Self::shutdown) (also called on `Drop`).
pub struct ShipQueue {
    shared: Arc<ShipShared>,
    threads: Vec<JoinHandle<()>>,
}

impl ShipQueue {
    /// Spawn `config.window` sender threads for every peer in `peers`
    /// (excluding the source's own shard). `active` gates dispatch: pass this
    /// node's at-boot leadership belief (an inactive queue parks until
    /// [`activate_from`](Self::activate_from)).
    ///
    /// # Panics
    ///
    /// Panics if a sender OS thread cannot be spawned — this runs once at
    /// node startup, so a thread-exhausted host fails loudly at boot.
    #[must_use]
    pub fn spawn(
        source: Arc<dyn ShipSource>,
        transport: Arc<dyn Transport>,
        peers: &[ShardId],
        config: ShipQueueConfig,
        active: bool,
        #[cfg(feature = "metrics")] metrics: Option<Arc<ClusterMetrics>>,
    ) -> Self {
        // Dispatch starts one past the source's flushed frontier: on a fresh
        // stream that is seqno 1; on a leader RESTART it means only new
        // writes push (a behind follower's gap triggers its catch-up pull
        // instead of a full-log re-ship).
        let start_next = source.flushed_seq() + 1;
        let peer_cells: Vec<Arc<PeerShip>> = peers
            .iter()
            .filter(|&&p| p != source.source_shard())
            .map(|&peer| {
                Arc::new(PeerShip {
                    peer,
                    state: Mutex::new(PeerState {
                        dispatch_next: start_next,
                        acked: 0,
                        completed: BTreeMap::new(),
                        retry: BTreeMap::new(),
                        paused: false,
                        consecutive_failures: 0,
                    }),
                    cv: Condvar::new(),
                    acked_atomic: AtomicU64::new(0),
                })
            })
            .collect();

        let commit = Arc::new(CommitIndex::new(
            &peer_cells.iter().map(|c| c.peer).collect::<Vec<_>>(),
            active,
        ));

        let shared = Arc::new(ShipShared {
            source,
            transport,
            peers: peer_cells,
            config,
            shutdown: AtomicBool::new(false),
            active: AtomicBool::new(active),
            // Boot is the topology era (term 0); an elected leadership stamps
            // its term through activate_from.
            term: AtomicU64::new(0),
            commit,
            #[cfg(feature = "metrics")]
            metrics,
        });

        // Wire the source's flush signal to the parked senders. A Weak breaks
        // the feed → notify-closure → shared → source → feed cycle, so the
        // queue's core frees as soon as the queue and the feed agree it is
        // gone.
        {
            let weak: Weak<ShipShared> = Arc::downgrade(&shared);
            shared.source.set_notify(Box::new(move || {
                if let Some(shared) = weak.upgrade() {
                    shared.wake_all();
                }
            }));
        }

        let window = config.window.max(1);
        let mut threads = Vec::with_capacity(shared.peers.len() * window);
        for cell in &shared.peers {
            for i in 0..window {
                let shared = Arc::clone(&shared);
                let cell = Arc::clone(cell);
                let handle = std::thread::Builder::new()
                    .name(format!("tidal-ship-{}-{i}", cell.peer.0))
                    .spawn(move || sender_loop(&shared, &cell))
                    .expect("spawn ship sender thread");
                threads.push(handle);
            }
        }

        Self { shared, threads }
    }

    /// Wake parked senders: new entries are durable and shippable.
    /// Cheap (one notify per peer); the feed's flush signal normally does
    /// this automatically — explicit publish remains for relay-source users.
    pub fn publish(&self) {
        self.shared.wake_all();
    }

    /// Whether this queue currently dispatches (leadership gate).
    #[must_use]
    pub fn is_active(&self) -> bool {
        self.shared.active.load(Ordering::Acquire)
    }

    /// The quorum commit index over this queue's peers (m11p3). Follows the
    /// queue's leadership gate; `ack=quorum` writes block on its
    /// [`wait_for`](CommitIndex::wait_for).
    #[must_use]
    pub fn commit_index(&self) -> Arc<CommitIndex> {
        Arc::clone(&self.shared.commit)
    }

    /// Activate dispatch with a fresh stream baseline: every peer's cursor
    /// jumps to `baseline + 1`, parked retries clear, and acked frontiers
    /// reset to the baseline. Called when this node becomes leader (promote
    /// or election win): `baseline` is its WAL flushed frontier at promotion —
    /// everything at or below it is pre-stream history that must NOT push to
    /// peers. `term` is the leadership term every outbound batch is stamped
    /// with and the commit index is scoped to (m11p4; 0 = the topology era).
    pub fn activate_from(&self, baseline: u64, term: u64) {
        for cell in &self.shared.peers {
            {
                let mut state = cell.lock();
                state.dispatch_next = baseline + 1;
                state.acked = baseline;
                state.completed.clear();
                state.retry.clear();
                state.consecutive_failures = 0;
            }
            cell.acked_atomic.store(baseline, Ordering::Release);
        }
        self.shared.term.store(term, Ordering::Release);
        self.shared.commit.activate(baseline, term);
        self.shared.active.store(true, Ordering::Release);
        self.shared.wake_all();
        tracing::info!(baseline, term, "ship queue activated (leadership)");
    }

    /// Deactivate dispatch (this node stopped leading). In-flight sends
    /// complete; parked retries clear (the new leader's stream supersedes).
    pub fn deactivate(&self) {
        self.shared.active.store(false, Ordering::Release);
        self.shared.commit.deactivate();
        for cell in &self.shared.peers {
            cell.lock().retry.clear();
            cell.cv.notify_all();
        }
        tracing::info!("ship queue deactivated (leadership moved)");
    }

    /// Stop dispatching new batches to `peer` (in-flight sends complete).
    pub fn pause(&self, peer: ShardId) {
        if let Some(cell) = self.shared.peer(peer) {
            cell.lock().paused = true;
            cell.cv.notify_all();
        }
    }

    /// Resume dispatch to `peer` (heal path).
    pub fn resume(&self, peer: ShardId) {
        if let Some(cell) = self.shared.peer(peer) {
            cell.lock().paused = false;
            cell.cv.notify_all();
        }
    }

    /// Resume dispatch to `peer`, jumping its cursor past everything the
    /// follower reports applied (heal with a fresh follower status): retries
    /// at or below `applied` prune, and dispatch continues from
    /// `max(cursor, applied + 1)`.
    pub fn resume_from(&self, peer: ShardId, applied: u64) {
        if let Some(cell) = self.shared.peer(peer) {
            let mut state = cell.lock();
            state.paused = false;
            state.advance_acked(applied);
            if state.dispatch_next <= applied {
                state.dispatch_next = applied + 1;
            }
            cell.acked_atomic.store(state.acked, Ordering::Release);
            drop(state);
            // The resume seqno is the follower's own durable report (fetched
            // from its status), so it counts toward quorum.
            self.shared.commit.update_peer(peer, applied);
            cell.cv.notify_all();
        }
    }

    /// Whether dispatch to `peer` is currently paused.
    #[must_use]
    pub fn is_paused(&self, peer: ShardId) -> bool {
        self.shared.peer(peer).is_some_and(|c| c.lock().paused)
    }

    /// The peer's contiguous acked frontier: every seqno `<=` this was
    /// accepted by the peer's transport (or reported applied). Lock-free.
    #[must_use]
    pub fn acked_seqno(&self, peer: ShardId) -> u64 {
        self.shared
            .peer(peer)
            .map_or(0, |c| c.acked_atomic.load(Ordering::Acquire))
    }

    /// Signal every sender to exit and join them. Idempotent.
    pub fn shutdown(&mut self) {
        self.shared.shutdown.store(true, Ordering::Release);
        // Quorum waiters must not sleep out their deadline against a queue
        // that will never ack again.
        self.shared.commit.deactivate();
        self.shared.wake_all();
        for handle in self.threads.drain(..) {
            if handle.join().is_err() {
                tracing::error!("ship sender thread panicked during shutdown");
            }
        }
    }
}

impl Drop for ShipQueue {
    fn drop(&mut self) {
        self.shutdown();
    }
}

// ── Sender machinery ────────────────────────────────────────────────────────

/// One collected unit of sender work: a contiguous seqno run plus its
/// already-encoded bytes.
struct ClaimedRun {
    first: u64,
    last: u64,
    event_count: u64,
    bytes: Vec<u8>,
}

/// Claim-and-collect the next unit of work for `cell`, or block on its
/// condvar.
///
/// Priority: due retries first (lowest first-seqno, so gaps close fastest),
/// then fresh dispatch up to the source's flushed frontier. Collection runs
/// under the peer lock so two windowed senders can never claim overlapping
/// runs; the source's collect is an in-memory tail copy (bounded by the run
/// caps). Returns `None` on shutdown.
fn claim_and_collect(shared: &ShipShared, cell: &PeerShip) -> Option<ClaimedRun> {
    let max_events = shared.config.max_batch_events.clamp(1, 65_536) as u64;
    let max_bytes = shared.config.max_batch_bytes.max(1);
    let mut state = cell.lock();
    loop {
        if shared.shutdown.load(Ordering::Acquire) {
            return None;
        }

        if shared.active.load(Ordering::Acquire) && !state.paused {
            // Due retry first (lowest first seqno = closes receiver gaps fastest).
            let now = Instant::now();
            if let Some((&first, &(last, due))) = state.retry.first_key_value()
                && due <= now
            {
                state.retry.remove(&first);
                // Cap the re-collect at the original run's seqno span so a
                // retry never grows past what was claimed (seqnos are dense,
                // so the span bounds the event count).
                let span = last.saturating_sub(first) + 1;
                match shared.source.collect(first, span, max_bytes) {
                    Ok(ShipCollect::Run {
                        bytes,
                        first_seq,
                        last_seq,
                        event_count,
                    }) => {
                        return Some(ClaimedRun {
                            first: first_seq,
                            last: last_seq,
                            event_count,
                            bytes,
                        });
                    }
                    Ok(ShipCollect::Rotated { resume_from }) => {
                        tracing::info!(
                            peer = cell.peer.0,
                            first,
                            last,
                            resume_from,
                            "ship sender: retry run rotated out of the tail; dropping \
                             (the follower's catch-up pull covers the hole)"
                        );
                        if state.dispatch_next < resume_from {
                            state.dispatch_next = resume_from;
                        }
                        continue;
                    }
                    Ok(ShipCollect::NothingNew) => continue,
                    Err(e) => {
                        pause_on_source_fault(cell, &mut state, first, last, &e);
                        continue;
                    }
                }
            }

            // Fresh dispatch, bounded by the source's flushed frontier
            // (invariant: never ship an event the leader has not fsynced).
            let flushed = shared.source.flushed_seq();
            if state.dispatch_next <= flushed {
                let from = state.dispatch_next;
                match shared.source.collect(from, max_events, max_bytes) {
                    Ok(ShipCollect::Run {
                        bytes,
                        first_seq,
                        last_seq,
                        event_count,
                    }) => {
                        state.dispatch_next = last_seq + 1;
                        return Some(ClaimedRun {
                            first: first_seq,
                            last: last_seq,
                            event_count,
                            bytes,
                        });
                    }
                    Ok(ShipCollect::Rotated { resume_from }) => {
                        tracing::info!(
                            peer = cell.peer.0,
                            from,
                            resume_from,
                            "ship sender: peer cursor fell behind the retained tail; \
                             skipping ahead (the follower's catch-up pull covers the hole)"
                        );
                        state.dispatch_next = resume_from;
                        continue;
                    }
                    Ok(ShipCollect::NothingNew) => {
                        // Raced a frontier read; fall through to park.
                    }
                    Err(e) => {
                        pause_on_source_fault(cell, &mut state, from, flushed, &e);
                        continue;
                    }
                }
            }
        }

        // Park until publish/pause/resume/shutdown, or the earliest retry due
        // time (bounded so a due retry is never missed while parked).
        let wait = state
            .retry
            .values()
            .map(|&(_, due)| due.saturating_duration_since(Instant::now()))
            .min()
            .unwrap_or(Duration::from_secs(1));
        let (guard, _timeout) = cell
            .cv
            .wait_timeout(state, wait.min(Duration::from_secs(1)))
            .unwrap_or_else(PoisonError::into_inner);
        state = guard;
    }
}

/// A deterministic source fault (encode failure / log corruption) repeats
/// identically on every retry: pause the peer loudly instead of silently
/// spinning, exactly like a permanent transport failure.
fn pause_on_source_fault(
    cell: &PeerShip,
    state: &mut PeerState,
    first: u64,
    last: u64,
    err: &crate::TidalError,
) {
    state.paused = true;
    tracing::error!(
        peer = cell.peer.0,
        first,
        last,
        error = %err,
        "ship sender: SOURCE fault collecting a run; pausing peer \
         (fix the fault then /cluster/heal to resume)"
    );
}

/// Record a successful run and advance the contiguous acked frontier, folding
/// in the follower's reported applied seqno when the transport carries one.
/// Logs recovery (at INFO) when this success ends a failure streak.
fn record_success(shared: &ShipShared, cell: &PeerShip, run: &ClaimedRun) {
    let reported = shared.transport.peer_applied_hint(cell.peer);
    // Quorum fold (m11p3): the hint is the follower's durably-applied mark
    // (the durable ship ack), the ONLY input that may advance the commit
    // index. Transport acceptance below advances `acked` for retry pruning
    // but never counts toward quorum.
    shared.commit.update_peer(cell.peer, reported);
    let recovered_after = {
        let mut state = cell.lock();
        state.completed.insert(run.first, run.last);
        state.advance_acked(reported);
        cell.acked_atomic.store(state.acked, Ordering::Release);
        std::mem::take(&mut state.consecutive_failures)
    };
    if recovered_after > 0 {
        tracing::info!(
            peer = cell.peer.0,
            failed_attempts = recovered_after,
            "ship sender: peer recovered; shipping resumed"
        );
    }
}

/// Park a failed run for retry; auto-pause the peer on a permanent failure.
/// Transition-based logging (see [`FAILURE_LOG_EVERY`]): the 1st and every
/// Nth consecutive failure WARN, the rest are `debug`.
fn record_failure(shared: &ShipShared, cell: &PeerShip, run: &ClaimedRun, err: &TransportError) {
    let permanent = matches!(
        err,
        TransportError::Permanent { .. } | TransportError::PayloadTooLarge { .. }
    );
    let (newly_paused, failure_count) = {
        let mut state = cell.lock();
        state.retry.insert(
            run.first,
            (run.last, Instant::now() + shared.config.retry_backoff),
        );
        state.consecutive_failures += 1;
        let newly_paused = if permanent && !state.paused {
            state.paused = true;
            true
        } else {
            false
        };
        (newly_paused, state.consecutive_failures)
    };
    if newly_paused {
        tracing::error!(
            peer = cell.peer.0,
            first = run.first,
            last = run.last,
            error = %err,
            "ship sender: PERMANENT transport failure; pausing peer \
             (fix config/peer then /cluster/heal to resume)"
        );
    } else if failure_count == 1 || failure_count % FAILURE_LOG_EVERY == 0 {
        tracing::warn!(
            peer = cell.peer.0,
            first = run.first,
            last = run.last,
            consecutive_failures = failure_count,
            error = %err,
            "ship sender: batch ship failing; retrying every {:?}",
            shared.config.retry_backoff
        );
    } else {
        tracing::debug!(
            peer = cell.peer.0,
            first = run.first,
            last = run.last,
            error = %err,
            "ship sender: batch ship failed; will retry"
        );
    }
}

/// The sender thread body: claim-and-collect → ship → record.
fn sender_loop(shared: &ShipShared, cell: &PeerShip) {
    while let Some(run) = claim_and_collect(shared, cell) {
        #[cfg(feature = "metrics")]
        let started = Instant::now();
        let payload = crate::replication::relay::range_payload_with_term(
            shared.source.source_shard(),
            run.first,
            run.last,
            run.event_count,
            run.bytes.clone(),
            shared.term.load(Ordering::Acquire),
            shared.config.leader_region,
        );
        match shared.transport.send_segment(cell.peer, payload) {
            Ok(()) => {
                record_success(shared, cell, &run);
                #[cfg(feature = "metrics")]
                if let Some(m) = &shared.metrics {
                    m.observe_ship(
                        cell.peer,
                        started.elapsed(),
                        run.event_count,
                        cell.acked_atomic.load(Ordering::Acquire),
                        shared.source.flushed_seq(),
                    );
                }
            }
            Err(e) => {
                // Logging lives in record_failure (transition-based, so a
                // partition window cannot flood the log with per-retry WARNs).
                #[cfg(feature = "metrics")]
                if let Some(m) = &shared.metrics {
                    m.observe_ship_failure(cell.peer);
                }
                record_failure(shared, cell, &run, &e);
            }
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::sync::Mutex as StdMutex;

    use super::*;
    use crate::{
        replication::transport::WalSegmentPayload,
        schema::{DecaySpec, EntityId, EntityKind, Schema, SchemaBuilder, Timestamp, Window},
        wal::feed::FlushedBatch,
    };

    fn view_schema() -> Schema {
        let mut b = SchemaBuilder::new();
        let _ = b
            .signal(
                "view",
                EntityKind::Item,
                DecaySpec::Exponential {
                    half_life: std::time::Duration::from_secs(7 * 24 * 3600),
                },
            )
            .windows(&[Window::AllTime])
            .velocity(false)
            .add();
        b.build().expect("schema builds")
    }

    fn leader_db() -> crate::TidalDb {
        crate::TidalDb::builder()
            .ephemeral()
            .with_schema(view_schema())
            .open()
            .expect("ephemeral leader db opens")
    }

    /// Gap-aware model receiver mirroring `ReplicationState::apply_range`:
    /// contiguous ranges advance, ahead ranges park, stale ranges no-op.
    #[derive(Default)]
    struct ModelReceiver {
        applied: u64,
        parked: BTreeMap<u64, u64>,
    }

    impl ModelReceiver {
        fn apply_range(&mut self, first: u64, last: u64) {
            if last <= self.applied {
                return; // idempotent re-ship
            }
            if first <= self.applied + 1 {
                self.applied = self.applied.max(last);
            } else {
                self.parked.insert(first, last);
            }
            // Drain any parked runs now contiguous.
            while let Some((&f, &l)) = self.parked.first_key_value() {
                if f <= self.applied + 1 {
                    self.applied = self.applied.max(l);
                    self.parked.remove(&f);
                } else {
                    break;
                }
            }
        }
    }

    /// Which error class the transport's injected failures produce.
    enum FailKind {
        Transient,
        Permanent,
    }

    /// A transport that drives a model receiver, optionally failing a
    /// configurable set of first-attempt seqno ranges, and recording every
    /// shipped batch (first, last, `event_count`). Reports the model
    /// receiver's applied seqno as its peer hint, exercising the
    /// follower-reported-acked fold.
    struct BatchTransport {
        local: ShardId,
        receiver: StdMutex<ModelReceiver>,
        shipped: StdMutex<Vec<(u64, u64, u64)>>,
        /// Fail the Nth send (1-indexed) while the counter is in this set.
        fail_sends: StdMutex<std::collections::HashSet<u64>>,
        fail_kind: StdMutex<FailKind>,
        send_counter: AtomicU64,
        /// Catch-up pull requests observed (shard, `from_seqno`).
        pulls: StdMutex<Vec<(ShardId, u64)>>,
    }

    impl BatchTransport {
        fn new() -> Self {
            Self {
                local: ShardId(1),
                receiver: StdMutex::new(ModelReceiver::default()),
                shipped: StdMutex::new(Vec::new()),
                fail_sends: StdMutex::new(std::collections::HashSet::new()),
                fail_kind: StdMutex::new(FailKind::Transient),
                send_counter: AtomicU64::new(0),
                pulls: StdMutex::new(Vec::new()),
            }
        }
        fn applied(&self) -> u64 {
            self.receiver.lock().unwrap().applied
        }
    }

    impl Transport for BatchTransport {
        fn send_segment(
            &self,
            _to: ShardId,
            payload: WalSegmentPayload,
        ) -> Result<(), TransportError> {
            let n = self.send_counter.fetch_add(1, Ordering::SeqCst) + 1;
            if self.fail_sends.lock().unwrap().remove(&n) {
                return match *self.fail_kind.lock().unwrap() {
                    FailKind::Transient => Err(TransportError::Closed),
                    FailKind::Permanent => Err(TransportError::Permanent {
                        reason: "injected permanent failure".into(),
                    }),
                };
            }
            self.receiver
                .lock()
                .unwrap()
                .apply_range(payload.id.seqno, payload.leader_last_seq);
            self.shipped.lock().unwrap().push((
                payload.id.seqno,
                payload.leader_last_seq,
                payload.event_count,
            ));
            Ok(())
        }
        fn recv_segment(&self) -> Option<WalSegmentPayload> {
            None
        }
        fn peer_applied_hint(&self, _peer: ShardId) -> u64 {
            self.applied()
        }
        fn request_catchup(&self, from_shard: ShardId, from_seqno: u64) {
            self.pulls.lock().unwrap().push((from_shard, from_seqno));
        }
        fn local_shard(&self) -> ShardId {
            self.local
        }
    }

    fn write_n(relay: &SignalRelay, db: &crate::TidalDb, n: u64) {
        for i in 1..=n {
            let staged = relay
                .stage_write(db, 0, "view", EntityId::new(i), 1.0, Timestamp::now())
                .expect("stage");
            relay.complete_write(db, staged).expect("complete");
        }
    }

    fn wait_for(deadline_ms: u64, mut check: impl FnMut() -> bool) -> bool {
        let deadline = Instant::now() + Duration::from_millis(deadline_ms);
        while Instant::now() < deadline {
            if check() {
                return true;
            }
            std::thread::sleep(Duration::from_millis(2));
        }
        check()
    }

    fn spawn_queue(
        source: Arc<dyn ShipSource>,
        transport: &Arc<BatchTransport>,
        config: ShipQueueConfig,
    ) -> ShipQueue {
        ShipQueue::spawn(
            source,
            Arc::clone(transport) as Arc<dyn Transport>,
            &[ShardId(1)],
            config,
            true,
            #[cfg(feature = "metrics")]
            None,
        )
    }

    /// Writes published to the queue are batched (fewer sends than events) and
    /// the model receiver converges to the relay's last seq.
    #[test]
    fn ships_batched_runs_and_converges() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        let queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig {
                max_batch_events: 64,
                max_batch_bytes: 1 << 20,
                window: 1,
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
        );

        write_n(&relay, &db, 100);
        queue.publish();

        assert!(
            wait_for(5_000, || transport.applied() == 100),
            "receiver must converge to 100, got {}",
            transport.applied()
        );
        assert!(wait_for(5_000, || queue.acked_seqno(ShardId(1)) == 100));

        let shipped = transport.shipped.lock().unwrap().clone();
        assert!(
            shipped.len() < 100,
            "100 events must coalesce into fewer batches, got {} sends",
            shipped.len()
        );
        let total_events: u64 = shipped.iter().map(|s| s.2).sum();
        assert_eq!(total_events, 100, "every event shipped exactly once");
    }

    /// A transient failure parks the run; the retry closes the receiver-side
    /// gap even while later runs (window > 1) already landed ahead of it.
    #[test]
    fn transient_failure_retries_and_closes_gap() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        // Fail the FIRST send: its run parks for retry while subsequent runs
        // land ahead (parked gap on the receiver).
        transport.fail_sends.lock().unwrap().insert(1);
        let queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig {
                max_batch_events: 8,
                max_batch_bytes: 1 << 20,
                window: 4,
                retry_backoff: Duration::from_millis(20),
                leader_region: 0,
            },
        );

        write_n(&relay, &db, 40);
        queue.publish();

        assert!(
            wait_for(5_000, || transport.applied() == 40),
            "retry must close the gap; receiver at {}",
            transport.applied()
        );
        assert!(wait_for(5_000, || queue.acked_seqno(ShardId(1)) == 40));
    }

    /// Pause stops dispatch; resume catches the peer up from its acked
    /// frontier (the heal path with no operator redelivery needed).
    #[test]
    fn pause_holds_dispatch_and_resume_catches_up() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        let queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig::default(),
        );

        write_n(&relay, &db, 5);
        queue.publish();
        assert!(wait_for(5_000, || transport.applied() == 5));

        queue.pause(ShardId(1));
        write_n(&relay, &db, 5); // seqnos 6..=10 while paused
        queue.publish();
        // Give senders a moment: nothing new may ship while paused.
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(
            transport.applied(),
            5,
            "paused peer must not receive new batches"
        );

        queue.resume(ShardId(1));
        assert!(
            wait_for(5_000, || transport.applied() == 10),
            "resume must catch the peer up, got {}",
            transport.applied()
        );
    }

    /// Senders never ship past the source's flushed frontier: staged-but-not-
    /// completed writes are invisible to peers.
    #[test]
    fn never_ships_past_durable_frontier() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        let queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig::default(),
        );

        let w1 = relay
            .stage_write(&db, 0, "view", EntityId::new(1), 1.0, Timestamp::now())
            .expect("stage 1");
        let w2 = relay
            .stage_write(&db, 0, "view", EntityId::new(2), 1.0, Timestamp::now())
            .expect("stage 2");
        queue.publish();
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(
            transport.applied(),
            0,
            "staged-but-not-durable events must never ship"
        );

        relay.complete_write(&db, w1).expect("complete 1");
        queue.publish();
        assert!(wait_for(5_000, || transport.applied() == 1));

        relay.complete_write(&db, w2).expect("complete 2");
        queue.publish();
        assert!(wait_for(5_000, || transport.applied() == 2));
    }

    /// A PERMANENT transport failure auto-pauses the peer (no silent infinite
    /// retry); `resume` (the heal verb) retries the parked run and recovers.
    #[test]
    fn permanent_failure_pauses_peer_and_resume_recovers() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        // First send fails PERMANENTLY (e.g. TLS/codec class).
        transport.fail_sends.lock().unwrap().insert(1);
        *transport.fail_kind.lock().unwrap() = FailKind::Permanent;
        let queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig {
                max_batch_events: 8,
                max_batch_bytes: 1 << 20,
                window: 1,
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
        );

        write_n(&relay, &db, 4);
        queue.publish();

        assert!(
            wait_for(5_000, || queue.is_paused(ShardId(1))),
            "a permanent transport failure must auto-pause the peer"
        );
        // Paused: nothing ships even though the run is parked for retry.
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(transport.applied(), 0, "paused peer must not be retried");

        // Heal: resume retries the parked run; the transport now succeeds.
        queue.resume(ShardId(1));
        assert!(
            wait_for(5_000, || transport.applied() == 4),
            "resume must retry the parked run and converge, got {}",
            transport.applied()
        );
    }

    /// An INACTIVE queue (follower) never dispatches; activation with a
    /// baseline ships only post-baseline writes.
    #[test]
    fn inactive_queue_holds_until_activated_with_baseline() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = Arc::new(BatchTransport::new());
        let queue = ShipQueue::spawn(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            Arc::clone(&transport) as Arc<dyn Transport>,
            &[ShardId(1)],
            ShipQueueConfig::default(),
            false, // follower at boot
            #[cfg(feature = "metrics")]
            None,
        );

        write_n(&relay, &db, 5);
        queue.publish();
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(
            transport.applied(),
            0,
            "an inactive (follower) queue must not dispatch"
        );
        assert!(!queue.is_active());

        // Promote with baseline 5: pre-promote history must NOT push.
        queue.activate_from(5, 1);
        write_n(&relay, &db, 3); // seqnos 6..=8
        queue.publish();
        assert!(
            wait_for(5_000, || {
                let shipped = transport.shipped.lock().unwrap().clone();
                shipped.iter().map(|s| s.2).sum::<u64>() == 3
            }),
            "only the 3 post-baseline writes may ship"
        );
        let shipped = transport.shipped.lock().unwrap().clone();
        assert!(
            shipped.iter().all(|&(first, _, _)| first >= 6),
            "no shipped run may start at or below the baseline: {shipped:?}"
        );
    }

    /// The WAL-feed source ships pre-encoded batches verbatim, and a peer
    /// whose cursor falls behind the rotated tail SKIPS ahead (the receiver
    /// gap is the follower's pull trigger), never stalling dispatch.
    #[test]
    fn feed_source_ships_and_skips_rotated_tail() {
        use crate::wal::format::batch::{EventRecord, encode_batch};

        // A tiny feed: 12-event cap so the third 6-event batch evicts the first.
        let feed = Arc::new(WalShipFeed::new(12, 1 << 20));
        let make = |first: u64, n: u64| {
            let events: Vec<EventRecord> = (0..n)
                .map(|i| EventRecord::signal(first + i, 0, 1.0, (first + i) * 1_000))
                .collect();
            let bytes = encode_batch(&events, first, 1).expect("encode");
            FlushedBatch {
                bytes: Arc::new(bytes),
                first_seq: first,
                last_seq: first + n - 1,
                event_count: n,
            }
        };

        let source = Arc::new(WalFeedSource::new(ShardId(0), Arc::clone(&feed)));
        let transport = Arc::new(BatchTransport::new());
        // Pause the peer FIRST so the eviction below happens before any ship.
        let queue = ShipQueue::spawn(
            Arc::clone(&source) as Arc<dyn ShipSource>,
            Arc::clone(&transport) as Arc<dyn Transport>,
            &[ShardId(1)],
            ShipQueueConfig {
                max_batch_events: 64,
                max_batch_bytes: 1 << 20,
                window: 1,
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
            true,
            #[cfg(feature = "metrics")]
            None,
        );
        queue.pause(ShardId(1));

        feed.push(make(1, 6));
        feed.push(make(7, 6)); // evicts [1..6] under the 12-event cap
        feed.push(make(13, 6)); // evicts [7..12]
        assert!(feed.tail_floor() > 1, "tail must have rotated");

        queue.resume(ShardId(1));
        // The peer's cursor (1) predates the tail: it must skip ahead and
        // ship the retained batches; the receiver parks them (gap below).
        assert!(
            wait_for(5_000, || !transport.shipped.lock().unwrap().is_empty()),
            "rotated tail must not stall dispatch"
        );
        let shipped = transport.shipped.lock().unwrap().clone();
        assert!(
            shipped.iter().all(|&(first, _, _)| first > 1),
            "the rotated-out prefix must never ship from the tail: {shipped:?}"
        );
        // The receiver parked the ahead range — its applied stays 0, which is
        // exactly the state the follower-pull trigger fires on.
        assert_eq!(transport.applied(), 0, "gap must park on the receiver");
    }

    /// Shutdown joins every sender thread (no leaks, no hangs).
    #[test]
    fn shutdown_joins_senders() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let transport = Arc::new(BatchTransport::new());
        let mut queue = spawn_queue(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            &transport,
            ShipQueueConfig::default(),
        );
        queue.shutdown();
        // Idempotent.
        queue.shutdown();
    }
}