tidaldb/tidal/src/replication/ship.rs

//! 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, RwLock, 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,
    /// Per-peer retire latch (m11p5 §3.3 `remove_peer`): set when this peer is
    /// removed from the roster. The whole-queue `shutdown` latch retires
    /// EVERY peer; this one retires a SINGLE peer's senders while the queue
    /// keeps shipping to the rest. `claim_and_collect` checks it on entry and
    /// on every wake, so a parked sender exits promptly when its peer is
    /// removed (no leaked thread).
    retired: AtomicBool,
}

/// 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>,
    /// The per-peer dispatch cells. `RwLock` because the peer SET is now
    /// mutable at runtime (m11p5 §3.3 `add_peer`/`remove_peer`), but the hot
    /// ship path is read-mostly: a sender takes a brief READ guard to clone
    /// its own cell `Arc` and immediately releases it, then operates on that
    /// clone for the whole claim→ship→record cycle WITHOUT holding the set
    /// lock. Only `add_peer`/`remove_peer` take the WRITE guard, and only to
    /// splice the Vec — never across a ship RPC or an fsync.
    ///
    /// Lock ordering (m11p1 staged-write contract): this set lock is
    /// OUTERMOST among the queue's locks and is never held while taking a
    /// per-`PeerShip` `state` mutex or the `CommitIndex` lock. A sender
    /// releases the set read guard before locking its cell's state, so the
    /// staged-write hot path never contends on the set lock for the duration
    /// of a send.
    peers: RwLock<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 peers_read(&self) -> std::sync::RwLockReadGuard<'_, Vec<Arc<PeerShip>>> {
        self.peers.read().unwrap_or_else(PoisonError::into_inner)
    }

    /// Clone the cell for `peer`, if it is still in the set. The returned
    /// `Arc` outlives the read guard so callers never operate under it.
    fn peer(&self, peer: ShardId) -> Option<Arc<PeerShip>> {
        self.peers_read()
            .iter()
            .find(|p| p.peer == peer)
            .map(Arc::clone)
    }

    /// Snapshot the current cells (clones the `Arc`s, releases the guard).
    fn peer_cells(&self) -> Vec<Arc<PeerShip>> {
        self.peers_read().iter().map(Arc::clone).collect()
    }

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

/// Build one peer's dispatch cell at the given cursor / acked frontier.
/// Shared by [`ShipQueue::spawn`] (boot) and [`ShipQueue::add_peer`] (runtime
/// conf-change) so a newly added peer is constructed byte-for-byte like a
/// boot peer.
fn new_peer_cell(peer: ShardId, dispatch_next: u64, acked: u64) -> Arc<PeerShip> {
    Arc::new(PeerShip {
        peer,
        state: Mutex::new(PeerState {
            dispatch_next,
            acked,
            completed: BTreeMap::new(),
            retry: BTreeMap::new(),
            paused: false,
            consecutive_failures: 0,
        }),
        cv: Condvar::new(),
        acked_atomic: AtomicU64::new(acked),
        retired: AtomicBool::new(false),
    })
}

/// Spawn `window` sender threads for one peer cell, returning their handles.
/// Identical construction for boot and runtime add: each thread owns its own
/// `Arc<ShipShared>` and `Arc<PeerShip>` clone and runs [`sender_loop`].
fn spawn_peer_threads(shared: &Arc<ShipShared>, cell: &Arc<PeerShip>) -> Vec<JoinHandle<()>> {
    let window = shared.config.window.max(1);
    let mut handles = Vec::with_capacity(window);
    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");
        handles.push(handle);
    }
    handles
}

/// 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>,
    /// Sender threads keyed by peer, so [`remove_peer`](Self::remove_peer) can
    /// join exactly one peer's threads (and [`add_peer`](Self::add_peer) can
    /// register a new peer's) without disturbing the rest. Behind a `Mutex`
    /// because the runtime conf-change verbs mutate it; the boot spawn and the
    /// shutdown drain are the only other touches.
    threads: Mutex<std::collections::HashMap<ShardId, 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| new_peer_cell(peer, start_next, 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: RwLock::new(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 mut threads: std::collections::HashMap<ShardId, Vec<JoinHandle<()>>> =
            std::collections::HashMap::new();
        for cell in &shared.peer_cells() {
            let handles = spawn_peer_threads(&shared, cell);
            threads.insert(cell.peer, handles);
        }

        Self {
            shared,
            threads: Mutex::new(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.peer_cells() {
            {
                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.peer_cells() {
            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))
    }

    /// The current peer set (the shards this queue ships to), sorted.
    #[must_use]
    pub fn peers(&self) -> Vec<ShardId> {
        let mut peers: Vec<ShardId> = self.shared.peers_read().iter().map(|c| c.peer).collect();
        peers.sort_unstable_by_key(|p| p.0);
        peers
    }

    /// The peer's circuit-breaker state for the self-heal gauge (m11p8): 0
    /// closed, 1 open, 2 half-open. Delegates to the transport (read-only — does
    /// not admit the breaker's half-open probe).
    #[must_use]
    pub fn peer_breaker_state(&self, peer: ShardId) -> u8 {
        self.shared.transport.peer_breaker_state(peer)
    }

    /// Add a peer to the roster at runtime (m11p5 §3.3 conf-change): construct
    /// its dispatch cell and spawn its windowed senders, inheriting the queue's
    /// CURRENT activation state. Idempotent (a re-add of an existing peer is a
    /// no-op) and a no-op for the source's own shard.
    ///
    /// Activation inheritance: if the queue is ACTIVE at term `T`, the new
    /// peer starts from the durable frontier EXACTLY as
    /// [`activate_from`](Self::activate_from) does — `dispatch_next =
    /// flushed_seq + 1`, `acked = flushed_seq` — so a freshly added voter/learner
    /// begins from the leader's current durable frontier and lets the catch-up
    /// pull cover any pre-frontier history (never a full-log re-ship). An
    /// INACTIVE queue (follower) adds the cell parked at the same cursor; it
    /// dispatches nothing until the queue activates.
    ///
    /// The COMMIT INDEX is reconfigured separately by the caller (one fenced
    /// apply path, §3.3) — `add_peer` owns only the ship cells and threads, so
    /// the role (voter vs learner) is the commit index's concern, not the
    /// queue's.
    ///
    /// Lock discipline: the write guard on the peer set is held ONLY to splice
    /// the Vec — the cell is constructed and the threads are spawned outside
    /// it, so the hot ship path's read guards are never blocked across an
    /// allocation or a thread spawn.
    pub fn add_peer(&self, peer: ShardId) {
        if peer == self.shared.source.source_shard() {
            return;
        }
        // Build the cell at the current durable frontier (active inheritance).
        let frontier = self.shared.source.flushed_seq();
        let cell = if self.shared.active.load(Ordering::Acquire) {
            // Mirror activate_from: dispatch one past the frontier, acked at it.
            new_peer_cell(peer, frontier + 1, frontier)
        } else {
            // Inactive: park at the same start the boot spawn uses.
            new_peer_cell(peer, frontier + 1, 0)
        };

        // Splice into the set under the write guard (idempotent on duplicates).
        {
            let mut set = self
                .shared
                .peers
                .write()
                .unwrap_or_else(PoisonError::into_inner);
            if set.iter().any(|c| c.peer == peer) {
                return;
            }
            set.push(Arc::clone(&cell));
        }

        // Spawn outside the set lock; register the handles per-peer.
        let handles = spawn_peer_threads(&self.shared, &cell);
        self.threads
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .insert(peer, handles);
        // Wake it so it begins draining immediately if active.
        cell.cv.notify_all();
        tracing::info!(
            peer = peer.0,
            frontier,
            active = self.shared.active.load(Ordering::Acquire),
            "ship queue: peer added (conf-change)"
        );
    }

    /// Remove a peer from the roster at runtime (m11p5 §3.3 conf-change):
    /// retire its cell, wake + JOIN its senders, then drop it. Returns whether
    /// a peer was actually removed (false = not in the set / already removed).
    ///
    /// The caller is responsible for the §3.3 delivery rule (retire only after
    /// the `Removed` record is quorum-committed AND delivered) — `remove_peer`
    /// is the mechanical retire+join. The COMMIT INDEX is reconfigured
    /// separately so the removed peer's mark stops counting toward quorum.
    ///
    /// Join discipline: the retire latch is set, the cell woken, and its
    /// threads JOINED before the cell is dropped — no leaked thread, no cell
    /// freed out from under a running sender (the join proves they exited).
    pub fn remove_peer(&self, peer: ShardId) -> bool {
        // Pull the cell out of the set under the write guard.
        let removed = {
            let mut set = self
                .shared
                .peers
                .write()
                .unwrap_or_else(PoisonError::into_inner);
            set.iter()
                .position(|c| c.peer == peer)
                .map(|pos| set.remove(pos))
        };
        let Some(cell) = removed else {
            return false;
        };
        // Retire + wake so its parked senders observe the latch and exit.
        cell.retired.store(true, Ordering::Release);
        cell.cv.notify_all();
        // Take the peer's thread handles out from under the lock BEFORE
        // joining, so the threads mutex is not held across the (blocking) join.
        let handles = self
            .threads
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(&peer);
        if let Some(handles) = handles {
            for handle in handles {
                if handle.join().is_err() {
                    tracing::error!(peer = peer.0, "ship sender thread panicked during remove");
                }
            }
        }
        // `cell` drops here, after every sender has exited.
        tracing::info!(peer = peer.0, "ship queue: peer removed (conf-change)");
        true
    }

    /// 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();
        let drained: Vec<(ShardId, Vec<JoinHandle<()>>)> = self
            .threads
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .drain()
            .collect();
        for (_peer, handles) in drained {
            for handle in handles {
                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 {
        // The whole-queue shutdown OR this single peer's retire latch
        // (m11p5 `remove_peer`) ends the loop — a removed peer's senders must
        // exit so `remove_peer` can join them without leaking a thread.
        if shared.shutdown.load(Ordering::Acquire) || cell.retired.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();
    }

    // ── m11p5 §3.3: runtime add_peer / remove_peer ───────────────────────────

    /// A transport that records shipped ranges PER PEER so a runtime-added
    /// peer's backlog ship is observable. Every send succeeds; the per-peer
    /// applied frontier mirrors the contiguous last-shipped seq.
    struct PerPeerTransport {
        local: ShardId,
        /// peer → gap-aware model receiver (mirrors `apply_range`).
        receivers: StdMutex<std::collections::HashMap<ShardId, ModelReceiver>>,
        /// peer → count of sends observed.
        sends: StdMutex<std::collections::HashMap<ShardId, u64>>,
    }

    impl PerPeerTransport {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                local: ShardId(0),
                receivers: StdMutex::new(std::collections::HashMap::new()),
                sends: StdMutex::new(std::collections::HashMap::new()),
            })
        }
        fn applied_for(&self, peer: ShardId) -> u64 {
            self.receivers
                .lock()
                .unwrap()
                .get(&peer)
                .map_or(0, |r| r.applied)
        }
        fn sends_for(&self, peer: ShardId) -> u64 {
            self.sends.lock().unwrap().get(&peer).copied().unwrap_or(0)
        }
        /// Seed a peer's applied frontier (simulates the catch-up pull having
        /// already delivered history up to `applied` before the live ship).
        fn seed_applied(&self, peer: ShardId, applied: u64) {
            self.receivers
                .lock()
                .unwrap()
                .entry(peer)
                .or_default()
                .applied = applied;
        }
    }

    impl Transport for PerPeerTransport {
        fn send_segment(
            &self,
            to: ShardId,
            payload: WalSegmentPayload,
        ) -> Result<(), TransportError> {
            self.receivers
                .lock()
                .unwrap()
                .entry(to)
                .or_default()
                .apply_range(payload.id.seqno, payload.leader_last_seq);
            *self.sends.lock().unwrap().entry(to).or_insert(0) += 1;
            Ok(())
        }
        fn recv_segment(&self) -> Option<WalSegmentPayload> {
            None
        }
        fn peer_applied_hint(&self, peer: ShardId) -> u64 {
            self.applied_for(peer)
        }
        fn request_catchup(&self, _from_shard: ShardId, _from_seqno: u64) {}
        fn local_shard(&self) -> ShardId {
            self.local
        }
    }

    /// `add_peer` mid-activation ships the existing backlog to the NEW peer:
    /// it inherits the active state and begins from the durable frontier, so
    /// everything the leader has flushed past the activation baseline reaches
    /// it (older history is the new peer's catch-up pull, not a re-ship).
    #[test]
    fn add_peer_mid_activation_ships_backlog_to_new_peer() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = PerPeerTransport::new();
        // Boot with just peer 1, active from baseline 0.
        let queue = ShipQueue::spawn(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            Arc::clone(&transport) as Arc<dyn Transport>,
            &[ShardId(1)],
            ShipQueueConfig {
                max_batch_events: 16,
                max_batch_bytes: 1 << 20,
                window: 1,
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
            true,
            #[cfg(feature = "metrics")]
            None,
        );

        // Build a backlog: peer 1 converges to 30.
        write_n(&relay, &db, 30);
        queue.publish();
        assert!(
            wait_for(5_000, || transport.applied_for(ShardId(1)) == 30),
            "the original peer must converge, got {}",
            transport.applied_for(ShardId(1))
        );

        // Add peer 2 at runtime. It inherits the ACTIVE state and starts from
        // the current durable frontier (30): new writes ship to it; older
        // history (1..=30) is its catch-up pull, which we model here by seeding
        // its applied frontier to the inherited baseline so the live ship of
        // 31..=50 lands contiguously rather than as a parked gap.
        transport.seed_applied(ShardId(2), 30);
        queue.add_peer(ShardId(2));
        assert_eq!(queue.peers(), vec![ShardId(1), ShardId(2)]);
        write_n(&relay, &db, 20); // seqnos 31..=50
        queue.publish();
        assert!(
            wait_for(5_000, || transport.applied_for(ShardId(2)) == 50),
            "the newly added peer must ship the post-add backlog, got {}",
            transport.applied_for(ShardId(2))
        );
        assert!(
            transport.sends_for(ShardId(2)) > 0,
            "the new peer must have received at least one ship"
        );
        // The original peer keeps shipping uninterrupted.
        assert!(wait_for(5_000, || transport.applied_for(ShardId(1)) == 50));
    }

    /// `add_peer` on an INACTIVE queue parks the new peer (no dispatch) until
    /// activation, then ships from the baseline — mirroring a boot peer.
    #[test]
    fn add_peer_on_inactive_queue_holds_until_activation() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = PerPeerTransport::new();
        let queue = ShipQueue::spawn(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            Arc::clone(&transport) as Arc<dyn Transport>,
            &[ShardId(1)],
            ShipQueueConfig {
                max_batch_events: 16,
                max_batch_bytes: 1 << 20,
                window: 1,
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
            false, // follower
            #[cfg(feature = "metrics")]
            None,
        );
        queue.add_peer(ShardId(2));

        write_n(&relay, &db, 10);
        queue.publish();
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(
            transport.applied_for(ShardId(2)),
            0,
            "an inactive queue must not dispatch to an added peer"
        );

        // Activate from baseline 10: only post-baseline writes ship. The
        // pre-baseline history (1..=10) is the follower's catch-up pull, modeled
        // by seeding its applied frontier to the baseline.
        transport.seed_applied(ShardId(2), 10);
        queue.activate_from(10, 1);
        write_n(&relay, &db, 5); // 11..=15
        queue.publish();
        assert!(
            wait_for(5_000, || transport.applied_for(ShardId(2)) == 15),
            "post-activation the added peer ships post-baseline writes, got {}",
            transport.applied_for(ShardId(2))
        );
    }

    /// `remove_peer` retires the cell, joins its threads (no leak), and drops
    /// it from the set. A second remove is a no-op. The remaining peer keeps
    /// shipping.
    #[test]
    fn remove_peer_joins_threads_and_keeps_the_rest_shipping() {
        let relay = Arc::new(SignalRelay::new(ShardId(0)));
        let db = leader_db();
        let transport = PerPeerTransport::new();
        let queue = ShipQueue::spawn(
            Arc::clone(&relay) as Arc<dyn ShipSource>,
            Arc::clone(&transport) as Arc<dyn Transport>,
            &[ShardId(1), ShardId(2)],
            ShipQueueConfig {
                max_batch_events: 16,
                max_batch_bytes: 1 << 20,
                window: 2, // multiple threads per peer to prove the join
                retry_backoff: Duration::from_millis(10),
                leader_region: 0,
            },
            true,
            #[cfg(feature = "metrics")]
            None,
        );

        write_n(&relay, &db, 10);
        queue.publish();
        assert!(wait_for(5_000, || transport.applied_for(ShardId(2)) == 10));

        // Remove peer 2: the call JOINS its threads — if any leaked or hung,
        // this would block forever (the test would time out, a hard failure).
        assert!(queue.remove_peer(ShardId(2)), "peer 2 was present");
        assert_eq!(queue.peers(), vec![ShardId(1)]);
        // Idempotent: a second remove finds nothing.
        assert!(!queue.remove_peer(ShardId(2)));

        // The remaining peer 1 keeps shipping new writes.
        write_n(&relay, &db, 5); // 11..=15
        queue.publish();
        assert!(
            wait_for(5_000, || transport.applied_for(ShardId(1)) == 15),
            "the surviving peer must keep shipping after a remove, got {}",
            transport.applied_for(ShardId(1))
        );
    }
}