//! gRPC server implementing the `WalShipping` service.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tidaldb::replication::{shard::ShardId, transport::WalSegmentPayload};
use tokio::sync::mpsc;
use tonic::{Request, Response, Status};

use crate::{
    config::GrpcTransportConfig,
    proto::{
        AppliedReport, AppliedReportAck, HeartbeatRequest, HeartbeatResponse, JoinRequest,
        JoinResponse, MemberInfo, ShipSegmentRequest, ShipSegmentResponse, SnapshotChunk,
        SnapshotFileChunk, SnapshotFileEntry, SnapshotHeader, SnapshotRequest, StreamRequest,
        TimeoutNowRequest, TimeoutNowResponse, VoteRequest, VoteResponse, WalSegmentId,
        wal_shipping_server::{WalShipping, WalShippingServer},
    },
    sources::ServingSources,
};

/// Shared per-peer applied-hint map (transport + service): the monotonic max
/// of every ship-ack hint AND every `ReportApplied` push for a peer. Both
/// inputs are durable-true (a follower's frontier only advances after its
/// own storage apply + WAL fsync).
pub(crate) type PeerAppliedMap = Arc<Mutex<HashMap<ShardId, u64>>>;

/// Fold a durable mark into the shared hint map (monotonic max; 0 ignored).
pub(crate) fn fold_peer_applied(map: &PeerAppliedMap, peer: ShardId, applied: u64) {
    if applied == 0 {
        return;
    }
    let mut hints = map
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let entry = hints.entry(peer).or_insert(0);
    if applied > *entry {
        *entry = applied;
    }
    drop(hints);
}

/// Leader-side last-seen capability bit-field per reporting peer (m11p5 §3.1).
/// Folded from `AppliedReport`s. The join/conf-change gate (stage C) reads it
/// to refuse appending kind-4 records until every voter reports the kind-4 bit.
/// Last-write-wins: a node's capabilities are fixed for a process lifetime, so
/// the freshest report is authoritative (a restart onto a downgraded binary —
/// itself a reseed-gated event — reports the new value).
pub(crate) type PeerCapabilityMap = Arc<Mutex<HashMap<ShardId, u64>>>;

/// Record a peer's reported capability bit-field (last-write-wins).
fn record_peer_capabilities(map: &PeerCapabilityMap, peer: ShardId, capabilities: u64) {
    map.lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .insert(peer, capabilities);
}

/// The major component of a `"X.Y.Z"` semver string, if parseable.
fn version_major(v: &str) -> Option<u64> {
    v.split('.').next()?.parse().ok()
}

/// Observe a heartbeat peer's build version for the m11p8 rolling-upgrade
/// handshake. Adjacent (N/N+1) versions interoperate by proto3 forward-compat;
/// a `>= 2` major-version gap is logged at WARN as an unsupported skew. Never
/// rejects — a rolling upgrade is a transient mixed-version window by design, so
/// the WARN only fires while such a window is open. An empty version is a
/// pre-m11p8 peer (version-unknown): no warning.
fn observe_peer_version(peer_region: u32, peer_version: &str) {
    if peer_version.is_empty() {
        return;
    }
    let ours = env!("CARGO_PKG_VERSION");
    if peer_version == ours {
        return;
    }
    match (version_major(peer_version), version_major(ours)) {
        (Some(theirs), Some(mine)) if theirs.abs_diff(mine) >= 2 => {
            tracing::warn!(
                peer_region,
                peer_version,
                our_version = ours,
                "rolling-upgrade version skew exceeds N/N+1 (>= 2 major versions apart); \
                 only adjacent versions are supported to interoperate"
            );
        }
        _ => {
            tracing::debug!(
                peer_region,
                peer_version,
                our_version = ours,
                "peer on a different build version (within the supported N/N+1 skew)"
            );
        }
    }
}

/// The gRPC trailer key carrying the typed catch-up refusal class (m11p5 §2.4).
/// Values: `snapshot-required` | `rejoin` | `stepping-down`. A pre-m11p5 source
/// emits no trailer; the puller treats an absent/unknown value conservatively
/// (no reseed-marker latch) — a plain election term-mismatch must NOT poison a
/// fleet with reseed markers.
pub(crate) const CATCHUP_TRAILER: &str = "x-tidal-catchup";

/// Build a `FAILED_PRECONDITION` status carrying the typed catch-up trailer
/// (m11p5 §2.4). The trailer is metadata the puller reads to classify the
/// refusal; the message stays human-readable for logs.
fn catchup_refusal(class: &'static str, message: String) -> Status {
    let mut status = Status::failed_precondition(message);
    // `class` is a fixed ASCII literal (one of three known values), so the
    // metadata-value parse never fails; on the impossible error path the
    // status still carries the human-readable message (the puller falls back
    // to today's log+retry behavior — never a panic).
    if let Ok(value) = class.parse() {
        status.metadata_mut().insert(CATCHUP_TRAILER, value);
    }
    status
}

/// Per-chunk caps for the `StreamSegments` catch-up path: small enough to
/// stay far below the codec/payload ceilings, large enough that a 100k-item
/// catch-up is a few hundred messages, not a few hundred thousand.
const STREAM_CHUNK_MAX_EVENTS: u64 = 4096;
const STREAM_CHUNK_MAX_BYTES: usize = 4 * 1024 * 1024;

/// Per-chunk byte cap for the `FetchSnapshot` file stream (1 MiB): far below
/// the codec ceiling, large enough that a multi-GiB artifact is thousands of
/// messages, not millions. Each chunk is read off `spawn_blocking`, never on
/// the reactor.
const SNAPSHOT_FILE_CHUNK_BYTES: usize = 1024 * 1024;

/// Read up to `max` bytes from `path` starting at `offset` (m11p5 snapshot
/// file chunking). Returns fewer bytes only at end-of-file (a short read of a
/// regular file is the file's tail; the caller stops once it reaches the
/// manifest size). Runs on `spawn_blocking`, never the reactor.
fn read_file_range(path: &std::path::Path, offset: u64, max: usize) -> std::io::Result<Vec<u8>> {
    use std::io::{Read, Seek, SeekFrom};
    let mut file = std::fs::File::open(path)?;
    file.seek(SeekFrom::Start(offset))?;
    let mut buf = vec![0u8; max];
    let mut filled = 0;
    // Loop to fill the buffer: a single `read` may return short for reasons
    // other than EOF, but a regular-file read on a local FS rarely does; the
    // loop makes the chunk size deterministic regardless.
    while filled < max {
        let n = file.read(&mut buf[filled..])?;
        if n == 0 {
            break; // EOF
        }
        filled += n;
    }
    buf.truncate(filled);
    Ok(buf)
}

/// The gRPC service that receives WAL segments from peers.
pub struct WalShippingService {
    inbound_tx: mpsc::Sender<WalSegmentPayload>,
    max_payload_bytes: usize,
    /// Node-side read views (m11p2): applied-seqno for ack piggybacking and
    /// WAL read-back for the catch-up stream. Absent = pre-m11p2 behavior.
    /// m11p3 adds the `applied_sink` hook for follower frontier reports.
    sources: ServingSources,
    /// Per-peer durable marks shared with the transport (m11p3): folded from
    /// `ReportApplied` pushes here and from ship acks on the client side.
    peer_applied: PeerAppliedMap,
    /// Per-peer last-seen capability bit-field (m11p5 §3.1), folded from
    /// `AppliedReport`s. Exposed to the leader's conf-change gate (stage C).
    peer_capabilities: PeerCapabilityMap,
}

impl WalShippingService {
    /// Create a new service that forwards received segments to the given channel.
    #[must_use]
    pub(crate) const fn new(
        inbound_tx: mpsc::Sender<WalSegmentPayload>,
        max_payload_bytes: usize,
        sources: ServingSources,
        peer_applied: PeerAppliedMap,
        peer_capabilities: PeerCapabilityMap,
    ) -> Self {
        Self {
            inbound_tx,
            max_payload_bytes,
            sources,
            peer_applied,
            peer_capabilities,
        }
    }

    /// The applied-seqno ack hint for a segment from `source_shard` (0 when
    /// no applied source is wired — the proto's "unknown").
    fn applied_hint(&self, source_shard: u32) -> u64 {
        let Ok(shard) = u16::try_from(source_shard) else {
            return 0;
        };
        self.sources
            .applied
            .as_ref()
            .map_or(0, |a| a.applied_seqno(ShardId(shard)))
    }
}

// The async_trait expansion folds every handler into one item, so the lint
// must sit on the impl: ship_segment is one linear pass (size guard -> term
// fence -> convert -> enqueue) whose split would scatter the ack contract.
#[allow(clippy::too_many_lines)]
#[tonic::async_trait]
impl WalShipping for WalShippingService {
    async fn ship_segment(
        &self,
        request: Request<ShipSegmentRequest>,
    ) -> Result<Response<ShipSegmentResponse>, Status> {
        let req = request.into_inner();

        // Validate payload size.
        if req.payload.len() > self.max_payload_bytes {
            return Err(Status::resource_exhausted(format!(
                "payload {} bytes exceeds max {}",
                req.payload.len(),
                self.max_payload_bytes
            )));
        }

        // Term fence (m11p4): a stale-term ship is rejected BEFORE it can
        // enter the inbound queue — a deposed leader's traffic never reaches
        // the apply path. A current/higher claim records leader contact (and
        // any adoption persisted before this returns).
        let response_term = if let Some(hooks) = self.sources.election.get() {
            let leader_region = u16::try_from(req.leader_region)
                .map_err(|_| Status::invalid_argument("leader_region exceeds u16 range"))?;
            let first_seq = req.id.as_ref().map_or(0, |id| id.seqno);
            match hooks.observe_leader_claim(req.term, leader_region, first_seq) {
                Ok(()) => {}
                Err(crate::sources::ClaimRejection::Stale { current_term }) => {
                    return Err(Status::failed_precondition(format!(
                        "stale leadership term {} (current term {current_term}); \
                         this ship is fenced",
                        req.term
                    )));
                }
                Err(crate::sources::ClaimRejection::JoinPending) => {
                    // Transient: the next heartbeat round runs the term-join
                    // check; UNAVAILABLE keeps the sender retrying instead of
                    // quarantining the peer.
                    return Err(Status::unavailable(format!(
                        "term {} not joined yet (awaiting the leader's heartbeat); retry",
                        req.term
                    )));
                }
            }
            hooks.self_claim().0
        } else {
            0
        };

        let source_shard = req.id.as_ref().map_or(0, |id| id.shard_id);

        // Convert proto to domain type.
        let payload = WalSegmentPayload::try_from(req)
            .map_err(|e| Status::invalid_argument(e.to_string()))?;

        // The ack's applied hint: read BEFORE enqueue (the segment cannot
        // have been applied yet) — a monotonic floor of this node's durable
        // progress, always safe for the leader to fold. The ack deliberately
        // does NOT wait for this segment's apply: holding acks couples the
        // leader's ship cadence to apply latency, and one gap-parked
        // follower then throttles its own feed into a death spiral
        // (measured: total quorum collapse at 1k rps). Durable-frontier
        // freshness flows the OTHER way instead — the receiver pushes
        // `ReportApplied` once per apply round.
        let applied_seqno = self.applied_hint(source_shard);

        // Try to send on the inbound channel. On full, yield once and retry.
        // If still full, return accepted=false. The WAL is durable on the
        // leader, so the follower will catch up when capacity frees.
        match self.inbound_tx.try_send(payload) {
            Ok(()) => Ok(Response::new(ShipSegmentResponse {
                accepted: true,
                applied_seqno,
                term: response_term,
            })),
            Err(mpsc::error::TrySendError::Full(payload)) => {
                // Expected, healthy backpressure under a sustained ship burst: the
                // apply consumer (a slow 1536-dim HNSW insert path) is draining the
                // bounded inbound queue more slowly than the leader ships. We yield
                // once, retry, and on a second full reply `accepted=false` — the
                // leader's circuit breaker treats that as backpressure (NOT a
                // failure), keeps the peer alive, and re-ships from its durable WAL.
                // DEBUG, not WARN: this fires per-segment during a burst and would
                // otherwise flood the log; the leader already rate-limits the
                // matching "batch ship failing" WARN on its side.
                tracing::debug!("inbound channel full; yielding and retrying");
                tokio::task::yield_now().await;
                match self.inbound_tx.try_send(payload) {
                    Ok(()) => Ok(Response::new(ShipSegmentResponse {
                        accepted: true,
                        applied_seqno,
                        term: response_term,
                    })),
                    Err(_) => Ok(Response::new(ShipSegmentResponse {
                        accepted: false,
                        applied_seqno,
                        term: response_term,
                    })),
                }
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                Err(Status::unavailable("receiver shut down"))
            }
        }
    }

    type StreamSegmentsStream =
        tokio_stream::wrappers::ReceiverStream<Result<ShipSegmentRequest, Status>>;

    async fn stream_segments(
        &self,
        request: Request<StreamRequest>,
    ) -> Result<Response<Self::StreamSegmentsStream>, Status> {
        let req = request.into_inner();
        let Some(source) = self.sources.segments.clone() else {
            // No WAL read-back wired (in-process tests, pre-m11p2 embedders):
            // fail loudly with an actionable code rather than hanging.
            return Err(Status::unimplemented(
                "StreamSegments: no segment source wired on this node",
            ));
        };

        // Term fence (m11p4, design-review C14): the source serves only a
        // SAME-TERM puller. A newer-term puller means this source is deposed
        // — it must step down, not serve its possibly-divergent tail; a
        // stale puller must rejoin the current term before pulling.
        let (chunk_term, chunk_leader_region) = if let Some(hooks) = self.sources.election.get() {
            let (current_term, self_region) = hooks.self_claim();
            if req.term > current_term {
                hooks.on_observed_term(req.term);
                // Typed refusal (m11p5 §2.4): `stepping-down` — this source is
                // deposed, not behind on data. The puller must NOT latch a
                // reseed marker; it re-discovers the current leader and
                // re-stamps its pull. A plain election term-mismatch poisoning
                // the fleet with reseed markers is the conflation hazard the
                // trailer closes.
                return Err(catchup_refusal(
                    "stepping-down",
                    format!(
                        "this source's term {current_term} is behind the puller's {}; \
                         stepping down — pull from the current leader",
                        req.term
                    ),
                ));
            }
            if req.term < current_term {
                // Typed refusal (m11p5 §2.4): `rejoin` — the puller is on a
                // stale term; it rejoins the current term and re-pulls. Not a
                // reseed condition.
                return Err(catchup_refusal(
                    "rejoin",
                    format!(
                        "puller term {} is behind the source's {current_term}; \
                         rejoin the current term before pulling",
                        req.term
                    ),
                ));
            }
            (current_term, u32::from(self_region))
        } else {
            (0, 0)
        };

        // The stream serves THIS node's own log; a request for any other
        // shard's stream is a routing error.
        let serving = source.source_shard();
        if req.shard_id != u32::from(serving.0) {
            return Err(Status::not_found(format!(
                "this node serves shard {} (requested stream for shard {})",
                serving.0, req.shard_id
            )));
        }

        // Clamp the start below the stream baseline: seqnos at or below it
        // are pre-stream history (a promoted leader's pre-promotion WAL) that
        // must never be served. Every chunk carries the baseline so the
        // receiver jumps its frontier instead of parking on a phantom gap.
        let baseline = source.stream_baseline();
        let from = req.from_seqno.max(1).max(baseline + 1);
        // Snapshot the end: the live unary push covers everything flushed
        // after this stream started.
        let end = source.flushed_seq();

        let (tx, rx) = mpsc::channel::<Result<ShipSegmentRequest, Status>>(4);
        tokio::spawn(async move {
            let mut cursor = from;
            while cursor <= end {
                let source_for_read = Arc::clone(&source);
                let read = tokio::task::spawn_blocking(move || {
                    source_for_read.collect_from(
                        cursor,
                        STREAM_CHUNK_MAX_EVENTS,
                        STREAM_CHUNK_MAX_BYTES,
                    )
                })
                .await;
                let chunks = match read {
                    Ok(Ok(chunks)) => chunks,
                    // Unservable-by-design (m11p4): the leader's WAL cannot
                    // serve this range and never will — its segments carry a
                    // format this binary cannot read (rolling-upgrade
                    // residue) or there is no durable log. FAILED_PRECONDITION
                    // tells the follower "stop expecting this stream; you
                    // need a snapshot", distinct from a retryable INTERNAL.
                    Ok(Err(crate::sources::SegmentReadError::Unavailable { detail })) => {
                        // Typed refusal (m11p5 §2.4): `snapshot-required` — the
                        // log can never serve this range; the puller latches a
                        // reseed marker and fetches a snapshot.
                        let _ = tx
                            .send(Err(catchup_refusal(
                                "snapshot-required",
                                format!(
                                    "segments not available from seq {cursor}; \
                                     snapshot required ({detail})"
                                ),
                            )))
                            .await;
                        return;
                    }
                    Ok(Err(crate::sources::SegmentReadError::Failed { detail })) => {
                        let _ = tx
                            .send(Err(Status::internal(format!(
                                "segment read-back failed at seqno {cursor}: {detail}"
                            ))))
                            .await;
                        return;
                    }
                    Err(join_err) => {
                        let _ = tx
                            .send(Err(Status::internal(format!(
                                "segment read-back task failed: {join_err}"
                            ))))
                            .await;
                        return;
                    }
                };
                if chunks.is_empty() {
                    break; // caught up to the snapshot end
                }
                // A first chunk that starts above the requested cursor (and
                // above the baseline) means the segments below it were
                // compacted away: the follower's hole cannot be served from
                // this log. Surface it honestly — a silent skip would strand
                // the follower parked forever; m11p5's snapshot transfer is
                // the designed fix for this case.
                if let Some(first) = chunks.first()
                    && first.first_seq > cursor
                {
                    // Typed refusal (m11p5 §2.4): `snapshot-required` — the
                    // history below the cursor was compacted; the puller
                    // latches a reseed marker and fetches a snapshot.
                    let _ = tx
                        .send(Err(catchup_refusal(
                            "snapshot-required",
                            format!(
                                "WAL compacted below seqno {cursor} (earliest available {}); \
                                 this follower needs a full resync (snapshot transfer, m11p5)",
                                first.first_seq
                            ),
                        )))
                        .await;
                    return;
                }
                for chunk in chunks {
                    cursor = chunk.last_seq + 1;
                    let msg = ShipSegmentRequest {
                        id: Some(WalSegmentId {
                            region_id: 0,
                            shard_id: u32::from(serving.0),
                            seqno: chunk.first_seq,
                        }),
                        payload: chunk.bytes,
                        event_count: chunk.event_count,
                        leader_last_seq: chunk.last_seq,
                        stream_baseline: baseline,
                        term: chunk_term,
                        leader_region: chunk_leader_region,
                    };
                    if tx.send(Ok(msg)).await.is_err() {
                        return; // client went away; stop reading
                    }
                }
            }
        });

        Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(
            rx,
        )))
    }

    type FetchSnapshotStream =
        tokio_stream::wrappers::ReceiverStream<Result<SnapshotChunk, Status>>;

    async fn fetch_snapshot(
        &self,
        request: Request<SnapshotRequest>,
    ) -> Result<Response<Self::FetchSnapshotStream>, Status> {
        let req = request.into_inner();
        let Some(source) = self.sources.snapshots.get().cloned() else {
            // No snapshot source wired (in-process tests, pre-m11p5 embedders).
            // §3.7: a pre-m11p5 peer answers Unimplemented; the joiner reports
            // it loudly and retries the next seed.
            return Err(Status::unimplemented(
                "FetchSnapshot: no snapshot source wired on this node",
            ));
        };

        // Term fence (m11p5 §2): a deposed source must NOT serve a snapshot of
        // its possibly-divergent state; a stale puller must rejoin first.
        // Mirrors StreamSegments exactly — same-term puller only.
        let (snap_term, leader_region) = if let Some(hooks) = self.sources.election.get() {
            let (current_term, self_region) = hooks.self_claim();
            if req.term > current_term {
                hooks.on_observed_term(req.term);
                return Err(catchup_refusal(
                    "stepping-down",
                    format!(
                        "this source's term {current_term} is behind the puller's {}; \
                         stepping down — fetch the snapshot from the current leader",
                        req.term
                    ),
                ));
            }
            if req.term < current_term {
                return Err(catchup_refusal(
                    "rejoin",
                    format!(
                        "puller term {} is behind the source's {current_term}; \
                         rejoin the current term before fetching a snapshot",
                        req.term
                    ),
                ));
            }
            (current_term, u32::from(self_region))
        } else {
            (0, 0)
        };

        // Stage (or reuse) the artifact. This call must be cheap — it reuses an
        // already-staged artifact when the live WAL still covers the request —
        // so running it on the reactor is intentional. The pin (§2.1) is taken
        // inside `stage` and released when this stream ends.
        let staging = match source.stage(req.from_seqno) {
            Ok(staging) => staging,
            Err(crate::sources::SnapshotStageError::Busy) => {
                // RETRYABLE (§2): a concurrent stage is mid-flight; the joiner
                // retries and shares it. NEVER FAILED_PRECONDITION — that would
                // mis-route the joiner into the reseed class.
                return Err(Status::unavailable(
                    "snapshot staging in progress; retry shortly",
                ));
            }
            Err(crate::sources::SnapshotStageError::Failed(detail)) => {
                return Err(Status::internal(format!(
                    "snapshot staging failed: {detail}"
                )));
            }
        };

        // A `needed=false` staging holds no pin (§2.1) — emit the header and
        // finish without touching `release`. A `needed=true` staging owns a
        // pin that the stream task releases on completion or client drop.
        let header = SnapshotHeader {
            needed: staging.needed,
            snapshot_seq: staging.snapshot_seq,
            term: snap_term,
            leader_region,
            files: staging
                .files
                .iter()
                .map(|(path, size, blake3)| SnapshotFileEntry {
                    path: path.clone(),
                    size: *size,
                    blake3: blake3.to_vec(),
                })
                .collect(),
        };

        let (tx, rx) = mpsc::channel::<Result<SnapshotChunk, Status>>(4);
        tokio::spawn(async move {
            // The pin guard: a `needed=true` staging is released exactly once
            // when this task ends, by ANY exit (header send failure, a file
            // read error, the last chunk, or the client dropping mid-stream).
            // A `needed=false` staging never took a pin, so it must not call
            // `release` (the node-side impl is idempotent, but not pinning at
            // all is cleaner). `ReleaseGuard` makes the single-release rule a
            // structural property, not a discipline scattered across returns.
            struct ReleaseGuard {
                source: Arc<dyn crate::sources::SnapshotSource>,
                armed: bool,
            }
            impl Drop for ReleaseGuard {
                fn drop(&mut self) {
                    if self.armed {
                        self.source.release();
                    }
                }
            }
            let _guard = ReleaseGuard {
                source: Arc::clone(&source),
                armed: staging.needed,
            };

            if tx
                .send(Ok(SnapshotChunk {
                    chunk: Some(crate::proto::snapshot_chunk::Chunk::Header(header)),
                }))
                .await
                .is_err()
            {
                return; // client went away before the header landed
            }
            if !staging.needed {
                return; // no files follow a no-snapshot-needed header
            }

            for (rel_path, size, _blake3) in &staging.files {
                let abs = staging.root.join(rel_path);
                let mut offset: u64 = 0;
                // Read the file in 1 MiB chunks off `spawn_blocking`. BLAKE3 is
                // NOT recomputed here — the manifest hash (computed once when
                // the artifact was staged) is the contract; the puller verifies
                // the received bytes against it end to end.
                loop {
                    let path = abs.clone();
                    let read = tokio::task::spawn_blocking(move || {
                        read_file_range(&path, offset, SNAPSHOT_FILE_CHUNK_BYTES)
                    })
                    .await;
                    let data = match read {
                        Ok(Ok(data)) => data,
                        Ok(Err(e)) => {
                            let _ = tx
                                .send(Err(Status::internal(format!(
                                    "snapshot file '{rel_path}' read failed at offset {offset}: {e}"
                                ))))
                                .await;
                            return;
                        }
                        Err(join_err) => {
                            let _ = tx
                                .send(Err(Status::internal(format!(
                                    "snapshot file read task failed: {join_err}"
                                ))))
                                .await;
                            return;
                        }
                    };
                    let read_len = data.len() as u64;
                    let next_offset = offset.saturating_add(read_len);
                    // `last` is true when this chunk reaches the manifest size,
                    // including the empty-file case (a zero-byte read at
                    // offset 0 must still emit one terminal chunk so the
                    // receiver sees the file).
                    let last = next_offset >= *size;
                    let send_ok = tx
                        .send(Ok(SnapshotChunk {
                            chunk: Some(crate::proto::snapshot_chunk::Chunk::File(
                                SnapshotFileChunk {
                                    path: rel_path.clone(),
                                    offset,
                                    data,
                                    last,
                                },
                            )),
                        }))
                        .await
                        .is_ok();
                    if !send_ok {
                        return; // client went away mid-file
                    }
                    if last {
                        break;
                    }
                    offset = next_offset;
                }
            }
        });

        Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(
            rx,
        )))
    }

    async fn heartbeat(
        &self,
        request: Request<HeartbeatRequest>,
    ) -> Result<Response<HeartbeatResponse>, Status> {
        // A heartbeat that reaches this handler proves the gRPC server is up,
        // the listener is accepting, and (under mTLS) the peer's certificate
        // was accepted — a genuine network-liveness probe. With m11p4 it is
        // also the leader's LEASE ASSERTION: the election hooks fold the
        // term/leadership claim into the failure detector and answer with
        // this node's term (a higher one is the sender's step-down signal).
        let req = request.into_inner();

        tracing::debug!(
            shard_id = req.shard_id,
            region_id = req.region_id,
            term = req.term,
            "received heartbeat",
        );

        // m11p8 rolling-upgrade handshake: observe the peer's build version.
        // N/N+1 (adjacent major) interoperate by proto3 forward-compat; a
        // >= 2-major gap is the loud signal that the skew exceeds what is
        // supported. Never a rejection — a rolling upgrade is a transient mixed
        // window by design (the warn only fires during such a window).
        observe_peer_version(req.region_id, &req.build_version);

        if let Some(hooks) = self.sources.election.get() {
            let leader_region = u16::try_from(req.leader_region)
                .map_err(|_| Status::invalid_argument("leader_region exceeds u16 range"))?;
            // The typed removed signal (m11p5 §3.3): is the heartbeat SENDER
            // (`region_id`, NOT the claimed `leader_region`) a `Removed` member
            // in THIS node's applied roster? Computed from `region_id` so a
            // removed FOLLOWER also learns it (not only when it claims
            // leadership). A bad region_id (out of u16) cannot be a roster
            // member, so it is trivially "not removed".
            let removed =
                u16::try_from(req.region_id).is_ok_and(|sender| hooks.is_removed_member(sender));
            let verdict = hooks.on_heartbeat(crate::HeartbeatContext {
                term: req.term,
                leader_region,
                stream_baseline: req.stream_baseline,
                prev_log: tidaldb::replication::LogPosition {
                    tail_term: req.prev_log_term,
                    frontier: req.prev_log_seq,
                },
                leader_last_seq: req.leader_last_seq,
            });
            return Ok(Response::new(HeartbeatResponse {
                acknowledged: true,
                term: verdict.term,
                accepted: verdict.accepted,
                // This binary's capabilities (m11p5 §3.1): the sender folds
                // this into its per-peer capability view to gate conf-changes.
                capabilities: crate::CAPABILITIES,
                removed,
            }));
        }
        // Pre-m11p4 behavior (no election driver wired): a truthful
        // reachability ack. No roster ⇒ no removed members.
        Ok(Response::new(HeartbeatResponse {
            acknowledged: true,
            term: 0,
            accepted: true,
            capabilities: crate::CAPABILITIES,
            removed: false,
        }))
    }

    async fn report_applied(
        &self,
        request: Request<AppliedReport>,
    ) -> Result<Response<AppliedReportAck>, Status> {
        let report = request.into_inner();
        let (Ok(reporter), Ok(source)) = (
            u16::try_from(report.reporter_shard),
            u16::try_from(report.source_shard),
        ) else {
            return Err(Status::invalid_argument("shard ids must fit u16"));
        };
        let reporter = ShardId(reporter);
        // A report is addressed to the stream's SOURCE; when this node's
        // serving identity is known (segment source wired), refuse a report
        // for some other node's stream — folding a mark into the wrong
        // quorum would be silent corruption.
        if let Some(segments) = &self.sources.segments
            && segments.source_shard() != ShardId(source)
        {
            return Err(Status::not_found(format!(
                "this node serves shard {} (report addressed to shard {source})",
                segments.source_shard().0
            )));
        }
        // Record the reporter's capabilities (m11p5 §3.1) BEFORE the term
        // gate: a capability bit is a term-independent fact about the
        // reporter's binary, and the conf-change gate must learn it even from
        // a reporter whose term is momentarily stale (an in-progress election
        // must not blind the leader to who is kind-4 capable).
        record_peer_capabilities(&self.peer_capabilities, reporter, report.capabilities);
        // Term fence (m11p4, design-review C4/C8/C12): a report from any
        // term but the current leadership's never reaches the hint map or
        // the commit index — a deposed leader's followers (or a delayed
        // report from a previous leadership) cannot advance commitment. The
        // sink re-checks under the commit index's own lock (the race-free
        // gate); this handler-level check keeps the HINT MAP equally clean.
        if let Some(hooks) = self.sources.election.get()
            && !hooks.report_term_acceptable(report.reporter_term)
        {
            return Err(Status::failed_precondition(format!(
                "frontier report term {} does not match the current leadership term {}; \
                 report dropped",
                report.reporter_term,
                hooks.self_claim().0
            )));
        }
        fold_peer_applied(&self.peer_applied, reporter, report.applied_seqno);
        if let Some(sink) = self.sources.applied_sink.get() {
            sink.peer_applied(reporter, report.applied_seqno, report.reporter_term);
        }
        Ok(Response::new(AppliedReportAck { acknowledged: true }))
    }

    async fn request_vote(
        &self,
        request: Request<VoteRequest>,
    ) -> Result<Response<VoteResponse>, Status> {
        let Some(hooks) = self.sources.election.get() else {
            return Err(Status::unimplemented(
                "RequestVote: no election driver on this node (pre-m11p4 binary or bare \
                 transport)",
            ));
        };
        let req = request.into_inner();
        let candidate = u16::try_from(req.candidate_region)
            .map_err(|_| Status::invalid_argument("candidate_region exceeds u16 range"))?;
        // The typed removed signal (m11p5 §3.3): is the CANDIDATE a `Removed`
        // member in this voter's applied roster? A removed node still campaigns
        // until it learns of its removal; this refusal teaches it.
        let removed = hooks.is_removed_member(candidate);
        let reply = hooks.on_vote(tidaldb::replication::VoteRpc {
            term: req.term,
            candidate: tidaldb::replication::shard::RegionId(candidate),
            log: tidaldb::replication::LogPosition {
                tail_term: req.last_log_term,
                frontier: req.last_log_seq,
            },
            prevote: req.prevote,
            transfer: req.transfer,
        });
        Ok(Response::new(VoteResponse {
            term: reply.term,
            granted: reply.granted,
            removed,
        }))
    }

    async fn timeout_now(
        &self,
        request: Request<TimeoutNowRequest>,
    ) -> Result<Response<TimeoutNowResponse>, Status> {
        let Some(hooks) = self.sources.election.get() else {
            return Err(Status::unimplemented(
                "TimeoutNow: no election driver on this node",
            ));
        };
        let req = request.into_inner();
        let leader_region = u16::try_from(req.leader_region)
            .map_err(|_| Status::invalid_argument("leader_region exceeds u16 range"))?;
        let accepted = hooks.on_timeout_now(req.term, leader_region);
        Ok(Response::new(TimeoutNowResponse { accepted }))
    }

    async fn join_cluster(
        &self,
        request: Request<JoinRequest>,
    ) -> Result<Response<JoinResponse>, Status> {
        let Some(hooks) = self.sources.join.get().cloned() else {
            // §3.7: a pre-m11p5 peer answers Unimplemented; the joiner reports
            // it loudly and retries the next seed.
            return Err(Status::unimplemented(
                "JoinCluster: no membership runtime on this node (pre-m11p5 binary or bare \
                 transport)",
            ));
        };
        let req = request.into_inner();
        let ask = crate::sources::JoinAsk {
            name: req.name,
            grpc_addr: req.grpc_addr,
            http_addr: req.http_addr,
            capabilities: req.capabilities,
        };
        // The hook BLOCKS on the bounded same-term commit wait (§3.3): run it on
        // `spawn_blocking` so the reactor is never parked. A join task panic
        // surfaces as INTERNAL rather than killing the worker.
        let outcome = tokio::task::spawn_blocking(move || hooks.join(ask))
            .await
            .map_err(|e| Status::internal(format!("join task failed: {e}")))?;
        Ok(Response::new(JoinResponse {
            accepted: outcome.accepted,
            refusal_reason: outcome.refusal_reason,
            assigned_id: u32::from(outcome.assigned_id),
            term: outcome.term,
            leader_region: outcome.leader_region,
            leader_grpc_addr: outcome.leader_grpc_addr,
            leader_http_addr: outcome.leader_http_addr,
            members: outcome
                .members
                .into_iter()
                .map(|m| MemberInfo {
                    id: u32::from(m.id),
                    name: m.name,
                    grpc_addr: m.grpc_addr,
                    http_addr: m.http_addr,
                    role: u32::from(m.role),
                })
                .collect(),
            membership_version: outcome.membership_version,
        }))
    }
}

/// Start the gRPC server on the given address.
///
/// Returns a `JoinHandle` that resolves when the server stops. Must be called
/// from within a tokio runtime context (it `tokio::spawn`s the serve loop).
///
/// # TLS posture (m11p7)
///
/// - `server_resolver = Some(_)` → mutual TLS. The server is served over a
///   custom `tokio-rustls` acceptor (NOT tonic's fixed `.tls_config()`) so the
///   node's certificate is hot-swappable via the resolver. The accept loop runs
///   each handshake in its own task and forwards ONLY successful streams to
///   tonic, so a foreign pod — no client cert, a cert from another CA, or a
///   plaintext probe — fails the handshake and never reaches an RPC.
/// - `server_resolver = None` and `config.insecure` → plaintext, with a loud
///   startup WARN (the trusted-loopback opt-in).
/// - `server_resolver = None` and NOT `config.insecure` → a typed error
///   (refuse to serve unauthenticated).
///
/// # Errors
///
/// Returns [`GrpcTransportError`](crate::error::GrpcTransportError) if the mTLS
/// listener cannot bind, the server config cannot be built, or TLS is unset and
/// `insecure` was not opted into.
pub(crate) fn start_server(
    config: &GrpcTransportConfig,
    inbound_tx: mpsc::Sender<WalSegmentPayload>,
    sources: ServingSources,
    peer_applied: PeerAppliedMap,
    peer_capabilities: PeerCapabilityMap,
    server_resolver: Option<Arc<crate::tls::DynamicCertResolver>>,
    shutdown: Arc<crate::transport::ShutdownSignal>,
) -> Result<
    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
    crate::error::GrpcTransportError,
> {
    let service = WalShippingService::new(
        inbound_tx,
        config.max_payload_bytes,
        sources,
        peer_applied,
        peer_capabilities,
    );
    let addr = config.listen_addr;

    // Raise tonic's default 4 MiB codec limits to the configured max payload on
    // BOTH directions. The server is the DECODER for inbound ShipSegmentRequest
    // (and the ENCODER for the StreamSegments response stream). With the
    // default 4 MiB limit a full-size WAL segment (16 MiB by default, up to
    // `max_payload_bytes` = 64 MiB) is rejected inside the codec with a
    // "message too large" status, which the shipper would retry forever as a
    // transient `Closed`. Pin both limits to the same ceiling the application
    // already advertises and validates against (server.rs ship_segment guard,
    // transport.rs send_segment guard) so the wire path and the app agree.
    let max = config.max_payload_bytes;
    let wal_service = WalShippingServer::new(service)
        .max_decoding_message_size(max)
        .max_encoding_message_size(max);

    if let Some(resolver) = server_resolver {
        // mTLS path. Build the rustls config (CA-rooted client verifier + the
        // hot-swappable cert resolver + ALPN h2), then serve over a tokio-rustls
        // acceptor whose accept loop drops failed handshakes before tonic.
        let tls = config.tls.as_ref().ok_or_else(|| {
            crate::error::GrpcTransportError::TlsConfig(
                "server cert resolver present but TLS config absent".into(),
            )
        })?;
        let server_config = crate::tls::build_server_config(tls, resolver)?;

        // Bind synchronously so "address already in use" surfaces immediately as
        // a typed error (tonic's `.serve(addr)` binds lazily inside the future,
        // which would only fail later via the JoinHandle). Hand the std listener
        // to the task, which registers it with the reactor.
        let std_listener = std::net::TcpListener::bind(addr).map_err(|e| {
            crate::error::GrpcTransportError::Internal(format!("bind mTLS listener {addr}: {e}"))
        })?;
        std_listener.set_nonblocking(true).map_err(|e| {
            crate::error::GrpcTransportError::Internal(format!(
                "set mTLS listener non-blocking {addr}: {e}"
            ))
        })?;

        let acceptor = tokio_rustls::TlsAcceptor::from(server_config);
        let handle = tokio::spawn(serve_mtls(
            std_listener,
            acceptor,
            wal_service,
            shutdown,
            addr,
            config.handshake_timeout,
            config.max_concurrent_handshakes,
        ));
        return Ok(handle);
    }

    // Plaintext path: refuse unless explicitly opted into, and WARN loudly when
    // we do serve cleartext.
    if !config.insecure {
        return Err(crate::error::GrpcTransportError::TlsConfig(
            "TLS not configured and insecure not set".into(),
        ));
    }
    tracing::warn!(
        %addr,
        "gRPC replication listener is PLAINTEXT (insecure=true): WAL segments, election \
         traffic, snapshots, and conf-changes cross the network UNENCRYPTED and \
         UNAUTHENTICATED — a foreign pod on this network can ship segments and impersonate a \
         peer. Configure grpc_tls for mutual TLS. Acceptable ONLY on a trusted single-host / \
         loopback topology."
    );

    let mut server_builder = tonic::transport::Server::builder();
    let handle = tokio::spawn(async move {
        // Observe the serve loop's terminal Result here so a failure AFTER
        // successful startup (e.g. the listener dies, the reactor is torn down)
        // is logged at error! rather than silently swallowed when the JoinHandle
        // is dropped or aborted. The Result is still propagated out of the task
        // so callers can also inspect it via the handle.
        let result = server_builder.add_service(wal_service).serve(addr).await;
        if let Err(ref e) = result {
            tracing::error!(error = %e, %addr, "gRPC WAL-shipping serve loop terminated with error");
        } else {
            tracing::info!(%addr, "gRPC WAL-shipping serve loop stopped");
        }
        result
    });

    Ok(handle)
}

/// Serve the gRPC service over a `tokio-rustls` acceptor with hot-swappable
/// certs (m11p7). The accept loop runs each TLS handshake in its own task and
/// forwards ONLY a successfully-handshaken stream into tonic, so a foreign pod's
/// failed handshake is dropped before any RPC is dispatched.
///
/// Lifecycle: the accept loop and the serve future both exit on the shared
/// `shutdown` latch, so a transport `Drop` (which trips the latch before
/// aborting the `JoinHandle`) tears the whole inbound path down
/// deterministically; the abort is a backstop.
async fn serve_mtls(
    std_listener: std::net::TcpListener,
    acceptor: tokio_rustls::TlsAcceptor,
    service: WalShippingServer<WalShippingService>,
    shutdown: Arc<crate::transport::ShutdownSignal>,
    addr: std::net::SocketAddr,
    handshake_timeout: std::time::Duration,
    max_concurrent_handshakes: usize,
) -> Result<(), tonic::transport::Error> {
    let listener = match tokio::net::TcpListener::from_std(std_listener) {
        Ok(listener) => listener,
        Err(e) => {
            // The listener cannot register with the reactor — the serve loop is
            // effectively dead. Logged at error!; the empty Ok lets the
            // JoinHandle finish so `serve_loop_died()` observes it (no shutdown
            // was requested), demoting this follower rather than black-holing.
            tracing::error!(%addr, error = %e, "mTLS listener registration failed");
            return Ok(());
        }
    };

    // A bounded channel of successfully-handshaken TLS streams. tonic implements
    // `Connected` for `tokio_rustls::server::TlsStream`, so the receiver stream
    // feeds `serve_with_incoming` directly and the peer's client cert surfaces in
    // request extensions for inter-node identity.
    let (conn_tx, conn_rx) = mpsc::channel::<
        Result<tokio_rustls::server::TlsStream<tokio::net::TcpStream>, std::io::Error>,
    >(128);

    let accept_shutdown = Arc::clone(&shutdown);
    // Bound the number of in-flight handshakes so a connection flood against the
    // (pre-auth) accept path cannot spawn unbounded tasks / exhaust fds.
    let handshake_limiter = Arc::new(tokio::sync::Semaphore::new(max_concurrent_handshakes));
    let accept_loop = tokio::spawn(async move {
        loop {
            let accepted = tokio::select! {
                biased;
                () = accept_shutdown.wait() => break,
                result = listener.accept() => result,
            };
            let (tcp, peer) = match accepted {
                Ok(pair) => pair,
                Err(e) => {
                    // A transient accept error (a reset during accept) is fine to
                    // retry immediately. A PERSISTENT one — most importantly
                    // EMFILE/ENFILE (fd-table exhaustion) — returns Err without
                    // consuming the pending connection, so retrying with no pause
                    // pegs a core. A brief backoff caps the spin while preserving
                    // liveness: the loop resumes the instant fds free up. (tonic's
                    // own AddrIncoming sleeps on accept errors for the same reason;
                    // the hand-rolled mTLS loop must do it explicitly.)
                    tracing::debug!(%addr, error = %e, "mTLS accept error; backing off");
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                    continue;
                }
            };
            // Load-shed under a handshake flood: if the limiter is saturated, drop
            // this connection now rather than queue an unbounded task behind it.
            let Ok(permit) = Arc::clone(&handshake_limiter).try_acquire_owned() else {
                tracing::debug!(%peer, "inbound handshake limiter saturated; dropping connection");
                drop(tcp);
                continue;
            };
            let acceptor = acceptor.clone();
            let conn_tx = conn_tx.clone();
            // Handshake off the accept path so one slow/foreign handshake cannot
            // head-of-line block the next connection.
            tokio::spawn(async move {
                // Held for the whole handshake; released (load capacity returned)
                // the instant this task ends, success or failure.
                let _permit = permit;
                match tokio::time::timeout(handshake_timeout, acceptor.accept(tcp)).await {
                    Ok(Ok(stream)) => {
                        // Channel closed = serve loop gone; drop the stream.
                        let _ = conn_tx.send(Ok(stream)).await;
                    }
                    Ok(Err(e)) => {
                        // Foreign pod (no/invalid client cert), a cert from
                        // another CA, or a non-TLS probe: the handshake fails
                        // HERE and the connection NEVER reaches tonic — the
                        // negative-test guarantee that a foreign pod cannot ship
                        // segments or call any RPC.
                        tracing::debug!(
                            %peer,
                            error = %e,
                            "rejected inbound gRPC handshake (no/invalid client cert or non-TLS probe)"
                        );
                    }
                    Err(_elapsed) => {
                        // A peer that connected but stalled the ClientHello
                        // (slowloris): abandon it so it cannot pin a task/socket.
                        tracing::debug!(%peer, "inbound gRPC handshake timed out; dropping connection");
                    }
                }
            });
        }
    });

    let serve_shutdown = Arc::clone(&shutdown);
    let result = tonic::transport::Server::builder()
        .add_service(service)
        .serve_with_incoming_shutdown(
            tokio_stream::wrappers::ReceiverStream::new(conn_rx),
            async move { serve_shutdown.wait().await },
        )
        .await;
    // Serving has ended (shutdown or error); stop the accept loop too.
    accept_loop.abort();
    if let Err(ref e) = result {
        tracing::error!(error = %e, %addr, "gRPC mTLS serve loop terminated with error");
    } else {
        tracing::info!(%addr, "gRPC mTLS serve loop stopped");
    }
    result
}

#[cfg(test)]
#[allow(clippy::unwrap_used)] // test assertions on known-good fixtures
mod tests {
    use super::*;
    use crate::{
        proto::{ShipSegmentRequest, WalSegmentId},
        sources::SegmentChunk,
    };

    fn make_request(payload_len: usize) -> ShipSegmentRequest {
        ShipSegmentRequest {
            id: Some(WalSegmentId {
                region_id: 0,
                shard_id: 0,
                seqno: 1,
            }),
            payload: vec![0xAB; payload_len],
            event_count: 1,
            leader_last_seq: 1,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        }
    }

    /// SUGGESTION (tidal-net): pin the server-side payload boundary semantics so a
    /// future change to one guard cannot silently desync from the client/codec.
    /// The contract (shared with `GrpcTransport::send_segment` in transport.rs and
    /// the codec limits in `start_server`) is **inclusive at `max_payload_bytes`**:
    /// a payload of exactly `max` is ACCEPTED, and `max + 1` is REJECTED
    /// (`len() > max`, not `>=`). This exercises the real `ship_segment` handler at
    /// both sides of the boundary.
    #[test]
    fn ship_segment_payload_boundary_is_inclusive_at_max() {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        runtime.block_on(async {
            let max = 1024usize;
            let (tx, mut rx) = mpsc::channel(4);
            let service = WalShippingService::new(
                tx,
                max,
                ServingSources::default(),
                Arc::new(Mutex::new(HashMap::new())),
                Arc::new(Mutex::new(HashMap::new())),
            );

            // Exactly max: accepted, and forwarded onto the inbound channel.
            let resp = service
                .ship_segment(Request::new(make_request(max)))
                .await
                .expect("a payload of exactly max_payload_bytes must be accepted");
            assert!(
                resp.into_inner().accepted,
                "payload == max must be accepted (inclusive boundary)"
            );
            let forwarded = rx.try_recv().expect("accepted payload reaches the channel");
            assert_eq!(forwarded.bytes.len(), max);

            // max + 1: rejected with resource_exhausted, nothing forwarded.
            let err = service
                .ship_segment(Request::new(make_request(max + 1)))
                .await
                .expect_err("a payload of max + 1 must be rejected");
            assert_eq!(
                err.code(),
                tonic::Code::ResourceExhausted,
                "over-size payload must be rejected as resource_exhausted"
            );
            assert!(
                rx.try_recv().is_err(),
                "a rejected payload must not be forwarded onto the inbound channel"
            );
        });
    }

    /// The ship ack piggybacks the applied source's CURRENT floor for the
    /// request's source shard — instantly, never waiting on the segment's
    /// own apply (m11p3 keeps acks decoupled from apply latency by design;
    /// durable freshness flows via `ReportApplied` instead).
    #[test]
    fn ship_segment_ack_carries_applied_floor() {
        struct FixedApplied;
        impl crate::sources::AppliedSource for FixedApplied {
            fn applied_seqno(&self, source_shard: ShardId) -> u64 {
                u64::from(source_shard.0) + 40
            }
        }
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        runtime.block_on(async {
            let (tx, _rx) = mpsc::channel(4);
            let sources = ServingSources {
                applied: Some(Arc::new(FixedApplied)),
                segments: None,
                ..ServingSources::default()
            };
            let service = WalShippingService::new(
                tx,
                1024,
                sources,
                Arc::new(Mutex::new(HashMap::new())),
                Arc::new(Mutex::new(HashMap::new())),
            );
            let mut req = make_request(8);
            req.id.as_mut().unwrap().shard_id = 2;
            let resp = service
                .ship_segment(Request::new(req))
                .await
                .unwrap()
                .into_inner();
            assert!(resp.accepted);
            assert_eq!(resp.applied_seqno, 42, "ack must carry shard 2's floor");
        });
    }

    /// m11p3: a follower's `ReportApplied` folds its durable mark into the
    /// shared hint map AND the late-bound applied sink (the quorum input);
    /// stale reports never regress; a report addressed to another node's
    /// stream is refused.
    #[test]
    fn report_applied_folds_marks_and_feeds_the_sink() {
        struct RecordingSink(Mutex<Vec<(ShardId, u64)>>);
        impl crate::sources::AppliedSink for RecordingSink {
            fn peer_applied(&self, peer: ShardId, applied: u64, _reporter_term: u64) {
                self.0.lock().unwrap().push((peer, applied));
            }
        }
        struct FixedSegments;
        impl crate::sources::SegmentSource for FixedSegments {
            fn source_shard(&self) -> ShardId {
                ShardId(0)
            }
            fn stream_baseline(&self) -> u64 {
                0
            }
            fn flushed_seq(&self) -> u64 {
                0
            }
            fn collect_from(
                &self,
                _from_seq: u64,
                _max_events: u64,
                _max_bytes: usize,
            ) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
                Ok(vec![])
            }
        }
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        runtime.block_on(async {
            let sink = Arc::new(RecordingSink(Mutex::new(Vec::new())));
            let sources = ServingSources {
                applied: None,
                segments: Some(Arc::new(FixedSegments)),
                ..ServingSources::default()
            };
            sources.set_applied_sink(Arc::clone(&sink) as Arc<dyn crate::sources::AppliedSink>);
            let map: PeerAppliedMap = Arc::new(Mutex::new(HashMap::new()));
            let (tx, _rx) = mpsc::channel(4);
            let service = WalShippingService::new(
                tx,
                1024,
                sources,
                Arc::clone(&map),
                Arc::new(Mutex::new(HashMap::new())),
            );

            let report = |reporter, source, applied| AppliedReport {
                reporter_shard: reporter,
                source_shard: source,
                applied_seqno: applied,
                reporter_term: 0,
                capabilities: crate::CAP_KIND4_MEMBERSHIP,
            };
            service
                .report_applied(Request::new(report(2, 0, 9)))
                .await
                .expect("report for this node's stream is accepted");
            // A stale (lower) report folds into the sink but must not
            // regress the hint map.
            service
                .report_applied(Request::new(report(2, 0, 5)))
                .await
                .unwrap();
            assert_eq!(map.lock().unwrap().get(&ShardId(2)), Some(&9));
            assert_eq!(
                sink.0.lock().unwrap().as_slice(),
                &[(ShardId(2), 9), (ShardId(2), 5)],
                "the sink receives every report verbatim (it owns monotonicity)"
            );

            // Wrong stream: refused loudly.
            let err = service
                .report_applied(Request::new(report(2, 7, 11)))
                .await
                .expect_err("a report addressed to another node's stream is a routing bug");
            assert_eq!(err.code(), tonic::Code::NotFound);
        });
    }

    /// m11p2: the catch-up stream serves chunks from the segment source in
    /// order, stamps every message with the stream baseline, and stops at the
    /// snapshot end.
    #[test]
    fn stream_segments_serves_chunks_with_baseline() {
        struct FakeSegments;
        impl crate::sources::SegmentSource for FakeSegments {
            fn source_shard(&self) -> ShardId {
                ShardId(3)
            }
            fn stream_baseline(&self) -> u64 {
                10
            }
            fn flushed_seq(&self) -> u64 {
                20
            }
            fn collect_from(
                &self,
                from_seq: u64,
                _max_events: u64,
                _max_bytes: usize,
            ) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
                // Two five-seqno chunks: [11..15], [16..20].
                if from_seq <= 15 {
                    Ok(vec![SegmentChunk {
                        bytes: vec![1; 8],
                        first_seq: from_seq,
                        last_seq: 15,
                        event_count: 15 - from_seq + 1,
                    }])
                } else if from_seq <= 20 {
                    Ok(vec![SegmentChunk {
                        bytes: vec![2; 8],
                        first_seq: from_seq,
                        last_seq: 20,
                        event_count: 20 - from_seq + 1,
                    }])
                } else {
                    Ok(vec![])
                }
            }
        }

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        runtime.block_on(async {
            use tokio_stream::StreamExt;
            let (tx, _rx) = mpsc::channel(4);
            let sources = ServingSources {
                applied: None,
                segments: Some(Arc::new(FakeSegments)),
                ..ServingSources::default()
            };
            let service = WalShippingService::new(
                tx,
                1024,
                sources,
                Arc::new(Mutex::new(HashMap::new())),
                Arc::new(Mutex::new(HashMap::new())),
            );

            // Request from seqno 1: the baseline (10) clamps the start to 11.
            let resp = service
                .stream_segments(Request::new(StreamRequest {
                    shard_id: 3,
                    from_seqno: 1,
                    term: 0,
                }))
                .await
                .expect("stream must open");
            let mut stream = resp.into_inner();
            let mut msgs = Vec::new();
            while let Some(msg) = stream.next().await {
                msgs.push(msg.expect("chunk"));
            }
            assert_eq!(msgs.len(), 2, "two chunks cover [11..20]");
            assert_eq!(msgs[0].id.as_ref().unwrap().seqno, 11, "baseline clamps");
            assert_eq!(msgs[0].leader_last_seq, 15);
            assert_eq!(msgs[0].stream_baseline, 10);
            assert_eq!(msgs[1].id.as_ref().unwrap().seqno, 16);
            assert_eq!(msgs[1].leader_last_seq, 20);

            // A request for the WRONG shard's stream is NOT_FOUND.
            let err = service
                .stream_segments(Request::new(StreamRequest {
                    shard_id: 9,
                    from_seqno: 1,
                    term: 0,
                }))
                .await
                .expect_err("wrong shard must be refused");
            assert_eq!(err.code(), tonic::Code::NotFound);
        });
    }

    /// m11p4: a source whose log can never serve the range (segment format
    /// unknown after a rolling upgrade / no durable WAL) surfaces as
    /// `FAILED_PRECONDITION` naming the snapshot remedy — distinct from the
    /// retryable `INTERNAL` a transient read failure produces.
    #[test]
    fn stream_segments_unavailable_is_failed_precondition() {
        struct UnservableSegments;
        impl crate::sources::SegmentSource for UnservableSegments {
            fn source_shard(&self) -> ShardId {
                ShardId(0)
            }
            fn stream_baseline(&self) -> u64 {
                0
            }
            fn flushed_seq(&self) -> u64 {
                100
            }
            fn collect_from(
                &self,
                _from_seq: u64,
                _max_events: u64,
                _max_bytes: usize,
            ) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
                Err(crate::sources::SegmentReadError::Unavailable {
                    detail: "WAL segment format unknown: wal-…001.seg".into(),
                })
            }
        }

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap();
        runtime.block_on(async {
            use tokio_stream::StreamExt;
            let (tx, _rx) = mpsc::channel(4);
            let sources = ServingSources {
                applied: None,
                segments: Some(Arc::new(UnservableSegments)),
                ..ServingSources::default()
            };
            let service = WalShippingService::new(
                tx,
                1024,
                sources,
                Arc::new(Mutex::new(HashMap::new())),
                Arc::new(Mutex::new(HashMap::new())),
            );

            let resp = service
                .stream_segments(Request::new(StreamRequest {
                    shard_id: 0,
                    from_seqno: 5,
                    term: 0,
                }))
                .await
                .expect("the stream opens; the failure arrives as the first message");
            let mut stream = resp.into_inner();
            let first = stream
                .next()
                .await
                .expect("one error message")
                .expect_err("unavailable must be an error, not a chunk");
            assert_eq!(first.code(), tonic::Code::FailedPrecondition);
            assert!(
                first
                    .message()
                    .contains("segments not available from seq 5")
                    && first.message().contains("snapshot required"),
                "the status must carry the structured remedy: {}",
                first.message()
            );
        });
    }
}