melin-server-runtime 0.21.0

//! TCP replication receiver (replica side).
//!
//! Connects to the primary, authenticates, performs catch-up / snapshot
//! recovery, and runs the streaming receive loop via [`streaming_loop`].
//! Builds the replica's local pipeline (journal + matching stage + drain
//! stages) to apply incoming events and ack durable batches back to the
//! primary.

use std::io::{self, Write};
use std::net::{SocketAddr, TcpStream};
use std::os::unix::io::RawFd;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tracing::{info, warn};

use melin_app::Application;
use melin_journal::BufferedWriter;
use melin_transport_core::node_store::NodeStore;

use super::auth::authenticate_with_primary;
use super::receiver_transport::{
    ControlFrameSource, ReceiverTransport, SessionExit, streaming_loop,
};
use super::{
    ReplicaPipelineHandles, ResumePoint, ResyncDecision, StreamStartVerdict,
    build_replica_pipeline_with_threads, handle_resync_verdict, handle_session_exit,
    journal_failed_while_disconnected, recover_replica_state, sleep_then_double_backoff,
    take_pipeline_for_promotion, teardown_replica_pipeline,
};
use crate::uring_teardown::{DrainBackoff, wake_pending_ops};
use melin_transport_core::replication::protocol::{
    Ack, Handshake, MAX_CONTROL_FRAME, MAX_DATA_FRAME, PrimaryMessage, decode_primary_message,
    encode_ack, encode_handshake, read_frame,
};

/// Force the kernel to send a TCP ACK immediately rather than holding
/// it in the delayed-ACK timer (~40 ms on Linux). Linux clears
/// `TCP_QUICKACK` after each ACK it sends, so it has to be re-armed.
///
/// A `setsockopt` is ~0.5–1 µs on the receive path's critical section,
/// so the re-arm is not unconditional: an outgoing ack frame already
/// carries the TCP ACK, and the replica sends one per receive cycle in
/// steady state. Only a cycle that produced no ack re-arms — see
/// `UringTransport::acked_since_recv`.
#[inline]
fn arm_tcp_quickack(fd: RawFd) {
    let on: libc::c_int = 1;
    // SAFETY: fd is a live socket fd owned by the caller for the
    // lifetime of this call; the option pointer is to a stack-local
    // i32 with the right size.
    let rc = unsafe {
        libc::setsockopt(
            fd,
            libc::IPPROTO_TCP,
            libc::TCP_QUICKACK,
            &on as *const libc::c_int as *const libc::c_void,
            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
        )
    };
    if rc != 0 {
        // Best-effort tuning: a kernel that refuses TCP_QUICKACK (or a
        // socket that just went away) costs delayed-ACK latency, not
        // correctness, and the caller has no better recovery than
        // continuing. Nothing to log per cycle either — it would fire on
        // every receive.
        debug_assert!(
            std::io::Error::last_os_error().raw_os_error() != Some(libc::EFAULT),
            "TCP_QUICKACK setsockopt got EFAULT — bad option pointer"
        );
    }
}

// ---------------------------------------------------------------------------
// io_uring ReceiverTransport implementation
// ---------------------------------------------------------------------------

const TOKEN_RECV: u64 = 0;
const TOKEN_SEND: u64 = 1;
const TOKEN_PROVIDE: u64 = 2;

// Multishot RECV buffer pool. 16 buffers × MAX_DATA_FRAME each.
const NUM_RECV_BUFFERS: u16 = 16;
const RECV_BUF_SIZE: usize = MAX_DATA_FRAME;
const RECV_BUF_GROUP_ID: u16 = 0;

// CQE flag bits (io_uring ABI).
const IORING_CQE_F_BUFFER: u32 = 1 << 0;
const IORING_CQE_F_MORE: u32 = 1 << 1;
const IORING_CQE_BUFFER_SHIFT: u32 = 16;

/// io_uring–backed receiver transport for kernel TCP.
///
/// Uses `IORING_OP_RECV_MULTI` against a provided buffer pool for
/// incoming frames, and single-shot SEND for acks. Owns the io_uring
/// ring, buffer pool, and ack send state.
struct UringTransport {
    ring: io_uring::IoUring,
    /// Raw socket fd, also used for the `shutdown(2)` wake in `Drop`.
    /// Always outlives the transport: the owning `TcpStream` lives in
    /// the reconnect loop, while the transport is a local of the scoped
    /// receiver thread that borrows it.
    tcp_fd: RawFd,
    // Backing storage for the io_uring provided-buffer pool. Never read
    // through this handle — the kernel accesses it via raw pointers
    // registered in ProvideBuffers, and reads go through `pool_ptr`.
    // Owned here so it outlives every RECV the kernel may land in it;
    // `Drop` quiesces the ring before letting it go.
    recv_pool: Vec<u8>,
    pool_ptr: *mut u8,
    multishot_active: bool,
    connected: bool,
    ack_buf: Vec<u8>,
    ack_offset: usize,
    ack_in_flight: bool,
    /// Newest ack accepted while a SEND was in flight — sent the moment
    /// the in-flight SEND's CQE is reaped (send-latest-on-completion).
    /// `Option<Ack>` rather than a queue: ack cursors are cumulative
    /// and monotonic, so when several acks arrive while one send is in
    /// flight only the newest pair needs the wire — intermediate values
    /// are subsumed. Exactly one SEND is ever outstanding, which keeps
    /// the TCP byte stream ordered (io_uring gives no ordering between
    /// independent SQEs, and interleaved partial sends would corrupt
    /// the frame stream).
    pending_ack: Option<Ack>,
    /// Whether an ack frame has been handed to the kernel since the last
    /// receive cycle. An outgoing data segment carries the TCP ACK with
    /// it, so an acking cycle needs no `TCP_QUICKACK` re-arm — only a
    /// silent one does. See [`arm_tcp_quickack`].
    acked_since_recv: bool,
}

impl UringTransport {
    fn new(tcp_stream: &TcpStream) -> io::Result<Self> {
        use io_uring::{IoUring, opcode, types};
        use std::os::unix::io::AsRawFd;

        let tcp_fd = tcp_stream.as_raw_fd();

        let mut ring: IoUring = IoUring::builder()
            .setup_single_issuer()
            .build(64)
            .map_err(|e| io::Error::other(format!("io_uring init failed: {e}")))?;

        ring.submitter()
            .register_files(&[tcp_fd])
            .map_err(|e| io::Error::other(format!("io_uring register_files failed: {e}")))?;

        arm_tcp_quickack(tcp_fd);

        // Pin io-wq workers to core 0.
        {
            let mut cpuset: libc::cpu_set_t = unsafe { std::mem::zeroed() };
            unsafe { libc::CPU_SET(0, &mut cpuset) };
            let _ = ring.submitter().register_iowq_aff(&cpuset);
        }

        let mut recv_pool: Vec<u8> = vec![0u8; NUM_RECV_BUFFERS as usize * RECV_BUF_SIZE];
        let pool_ptr = recv_pool.as_mut_ptr();

        // Provide the buffer pool to io_uring.
        {
            let sqe = opcode::ProvideBuffers::new(
                pool_ptr,
                RECV_BUF_SIZE as i32,
                NUM_RECV_BUFFERS,
                RECV_BUF_GROUP_ID,
                0,
            )
            .build()
            .user_data(TOKEN_PROVIDE);
            unsafe { ring.submission().push(&sqe).expect("SQ full") };
            ring.submit_and_wait(1)?;
            let cqe = ring
                .completion()
                .next()
                .ok_or_else(|| io::Error::other("no cqe after ProvideBuffers"))?;
            if cqe.result() < 0 {
                return Err(io::Error::other(format!(
                    "ProvideBuffers failed: {}",
                    cqe.result()
                )));
            }
        }

        // Submit the initial multishot RECV.
        let sqe = opcode::RecvMulti::new(types::Fd(tcp_fd), RECV_BUF_GROUP_ID)
            .build()
            .user_data(TOKEN_RECV);
        unsafe { ring.submission().push(&sqe).expect("SQ full") };

        Ok(UringTransport {
            ring,
            tcp_fd,
            recv_pool,
            pool_ptr,
            multishot_active: true,
            connected: true,
            ack_buf: Vec::with_capacity(64),
            ack_offset: 0,
            ack_in_flight: false,
            pending_ack: None,
            acked_since_recv: false,
        })
    }

    /// Encode `ack` into the send buffer and push its SEND SQE. The
    /// caller is responsible for submitting (next `ring.submit()` —
    /// either `poll_recv`'s top-of-call submit or the eager submit
    /// after a chained send). Must only be called with no SEND in
    /// flight: the buffer is pinned by `ack_in_flight` until the CQE.
    fn push_ack_send(&mut self, ack: &Ack) {
        use io_uring::{opcode, types};

        debug_assert!(!self.ack_in_flight);
        self.ack_buf.clear();
        encode_ack(ack, &mut self.ack_buf);
        let sqe = opcode::Send::new(
            types::Fixed(0),
            self.ack_buf.as_ptr(),
            self.ack_buf.len() as u32,
        )
        .build()
        .user_data(TOKEN_SEND);
        // SAFETY: `ack_buf` is owned by this transport and pinned by
        // `ack_in_flight = true` until the matching CQE is reaped. The
        // ring is single-threaded.
        unsafe { self.ring.submission().push(&sqe).expect("SQ full") };
        self.ack_in_flight = true;
        self.ack_offset = 0;
        self.acked_since_recv = true;
    }

    /// Settle the kernel-ownership flags from CQEs that were reaped out
    /// of the completion queue but will not be processed, because
    /// [`Self::poll_recv`] is bailing out on a failed session.
    ///
    /// Only the flags — the payloads are deliberately discarded, and
    /// nothing is resubmitted, since the session is over. Skipping this
    /// would leave `ack_in_flight` set for a SEND the kernel has
    /// already returned, and [`Self::drain`] would then wait out its
    /// whole deadline on a completion that can never arrive again.
    fn settle_unprocessed(&mut self, rest: &[(u64, i32, u32)]) {
        for &(token, _result, flags) in rest {
            match token {
                TOKEN_RECV if (flags & IORING_CQE_F_MORE) == 0 => self.multishot_active = false,
                TOKEN_SEND => self.ack_in_flight = false,
                _ => {}
            }
        }
    }

    /// Reap completions until the kernel holds neither buffer, bounded
    /// by [`crate::uring_teardown::DRAIN_TIMEOUT`]. `true` means the
    /// multishot RECV has terminated and any SEND has settled, so both
    /// buffers are provably free of kernel references.
    ///
    /// Unlike the sender's ring there is no in-flight *count* to drive
    /// this: multishot RECV is a single SQE that yields many CQEs and
    /// only settles on the one without `IORING_CQE_F_MORE`. So the two
    /// existing flags are the drain condition, and CQEs are classified
    /// here rather than reusing `poll_recv` — during teardown the
    /// payloads are discarded, consumed buffers are not re-provided,
    /// and neither the multishot nor a partial send is resubmitted.
    fn drain(&mut self) -> bool {
        // Flush SQEs pushed but never submitted — notably the initial
        // multishot RECV and any resubmission queued by the last
        // `poll_recv`. Until they reach the kernel no CQE is coming,
        // but the kernel takes the buffer pointers the moment it sees
        // them, so they cannot simply be abandoned in the SQ either.
        loop {
            match self.ring.submit() {
                Ok(_) => break,
                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
                Err(_) => return false,
            }
        }
        let mut backoff = DrainBackoff::new();
        while self.multishot_active || self.ack_in_flight {
            for cqe in self.ring.completion() {
                match cqe.user_data() {
                    // The kernel stops writing into the pool only once
                    // the multishot arm is retired, which it signals by
                    // clearing F_MORE. A CQE that still carries F_MORE
                    // is data landing mid-teardown: discard it and keep
                    // waiting for the terminal one.
                    TOKEN_RECV if (cqe.flags() & IORING_CQE_F_MORE) == 0 => {
                        self.multishot_active = false;
                    }
                    // Settles whether the send completed, errored, or
                    // went partial: either way the kernel is done
                    // reading `ack_buf` for this operation.
                    TOKEN_SEND => self.ack_in_flight = false,
                    // ProvideBuffers only hands the kernel addresses;
                    // it never touches pool contents, so a pending one
                    // is not a reason to keep waiting.
                    _ => {}
                }
            }
            if (self.multishot_active || self.ack_in_flight) && !backoff.wait() {
                return false;
            }
        }
        true
    }
}

impl Drop for UringTransport {
    fn drop(&mut self) {
        wake_pending_ops(self.tcp_fd);
        if !self.drain() {
            // The drain could not be proven complete, so the kernel may
            // still land a RECV in the buffer pool or read `ack_buf`
            // for an in-flight SEND. Leak both rather than hand the
            // allocator memory the kernel still touches — a session
            // that is tearing down anyway pays a bounded one-off cost.
            std::mem::forget(std::mem::take(&mut self.recv_pool));
            std::mem::forget(std::mem::take(&mut self.ack_buf));
            tracing::warn!(
                multishot_active = self.multishot_active,
                ack_in_flight = self.ack_in_flight,
                "io_uring teardown drain did not complete; leaking receiver buffers"
            );
        }
    }
}

impl ReceiverTransport for UringTransport {
    fn poll_recv(&mut self, recv_buf: &mut Vec<u8>) -> io::Result<bool> {
        use io_uring::{opcode, types};

        if !self.connected {
            return Err(io::Error::other("not connected"));
        }

        // Submit pending SQEs (ProvideBuffers re-provisions, Send
        // remainders, multishot resubmissions).
        let pending = self.ring.submission().len();
        if pending > 0 {
            self.ring.submit()?;
        }

        // Drain CQEs.
        let mut cqes: [(u64, i32, u32); 16] = [(0, 0, 0); 16];
        let mut cqe_count = 0;
        {
            let mut cq = self.ring.completion();
            while cqe_count < cqes.len() {
                match cq.next() {
                    Some(cqe) => {
                        cqes[cqe_count] = (cqe.user_data(), cqe.result(), cqe.flags());
                        cqe_count += 1;
                    }
                    None => break,
                }
            }
        }

        let mut any_recv = false;
        // Set when a queued ack is chained onto a completed SEND below —
        // it must hit the wire within THIS call, not ride the next
        // poll_recv's top-of-call submit one loop iteration later.
        let mut submit_chained_ack = false;

        for (idx, &(token, result, flags)) in cqes[..cqe_count].iter().enumerate() {
            // Every early return below abandons the CQEs after `idx`,
            // which have already been reaped out of the CQ — hence the
            // `settle_unprocessed` calls: without them teardown would
            // wait on a completion that has come and gone.
            match token {
                TOKEN_RECV => {
                    if (flags & IORING_CQE_F_MORE) == 0 {
                        self.multishot_active = false;
                    }
                    if result < 0 {
                        if result == -libc::ENOBUFS {
                            tracing::debug!("recv multishot ENOBUFS — pool exhausted");
                            continue;
                        }
                        self.connected = false;
                        self.settle_unprocessed(&cqes[idx + 1..cqe_count]);
                        return Err(io::Error::other(format!(
                            "primary disconnected (recv returned {result})"
                        )));
                    }
                    if result == 0 {
                        self.connected = false;
                        self.settle_unprocessed(&cqes[idx + 1..cqe_count]);
                        return Err(io::Error::other("primary disconnected (recv returned 0)"));
                    }
                    if (flags & IORING_CQE_F_BUFFER) == 0 {
                        self.connected = false;
                        self.settle_unprocessed(&cqes[idx + 1..cqe_count]);
                        return Err(io::Error::other("recv cqe missing F_BUFFER flag"));
                    }

                    let n = result as usize;
                    let buf_id = (flags >> IORING_CQE_BUFFER_SHIFT) as usize;
                    // SAFETY: buf_id from kernel CQE for a buffer we
                    // provided from recv_pool; kernel wrote n bytes.
                    let buf_ptr = unsafe { self.pool_ptr.add(buf_id * RECV_BUF_SIZE) };
                    let slice = unsafe { std::slice::from_raw_parts(buf_ptr, n) };
                    recv_buf.extend_from_slice(slice);
                    any_recv = true;

                    // Re-provide the consumed buffer.
                    let provide_sqe = opcode::ProvideBuffers::new(
                        buf_ptr,
                        RECV_BUF_SIZE as i32,
                        1,
                        RECV_BUF_GROUP_ID,
                        buf_id as u16,
                    )
                    .build()
                    .user_data(TOKEN_PROVIDE);
                    unsafe { self.ring.submission().push(&provide_sqe).expect("SQ full") };
                }

                TOKEN_PROVIDE if result < 0 => {
                    tracing::debug!("ProvideBuffers re-provision failed: {result}");
                }

                TOKEN_SEND => {
                    if result < 0 {
                        self.connected = false;
                        // This CQE is the kernel handing `ack_buf` back,
                        // failure or not — record that before bailing.
                        self.ack_in_flight = false;
                        self.settle_unprocessed(&cqes[idx + 1..cqe_count]);
                        return Err(io::Error::other(format!(
                            "ack send error (returned {result})"
                        )));
                    }
                    let sent = result as usize;
                    self.ack_offset += sent;
                    if self.ack_offset >= self.ack_buf.len() {
                        self.ack_buf.clear();
                        self.ack_offset = 0;
                        self.ack_in_flight = false;
                        // Send-latest-on-completion: an ack accepted
                        // while this send was in flight goes out now
                        // instead of waiting a full loop iteration for
                        // the next flush. Chained only after FULL
                        // completion — a partial send's remainder keeps
                        // exclusive ownership of the byte stream.
                        if let Some(ack) = self.pending_ack.take() {
                            self.push_ack_send(&ack);
                            submit_chained_ack = true;
                        }
                    } else {
                        // Partial send — resubmit remainder.
                        let sqe = opcode::Send::new(
                            types::Fixed(0),
                            self.ack_buf[self.ack_offset..].as_ptr(),
                            (self.ack_buf.len() - self.ack_offset) as u32,
                        )
                        .build()
                        .user_data(TOKEN_SEND);
                        unsafe { self.ring.submission().push(&sqe).expect("SQ full") };
                    }
                }

                _ => {}
            }
        }

        // Resubmit multishot if terminated.
        if !self.multishot_active {
            let sqe = opcode::RecvMulti::new(types::Fd(self.tcp_fd), RECV_BUF_GROUP_ID)
                .build()
                .user_data(TOKEN_RECV);
            unsafe { self.ring.submission().push(&sqe).expect("SQ full") };
            self.multishot_active = true;
        }

        // Flush a chained ack to the kernel before returning (also
        // carries any multishot resubmission pushed above).
        if submit_chained_ack {
            self.ring.submit()?;
        }

        if any_recv {
            // One decision per receive cycle rather than a setsockopt per
            // RECV CQE — and none at all while the replica is acking,
            // since each ack frame is an outgoing segment that carries
            // the TCP ACK. A cycle that produces no ack re-arms, so a
            // stream that goes app-level silent waits at most one cycle
            // (not a delayed-ACK timer) for its next immediate ACK.
            if self.acked_since_recv {
                self.acked_since_recv = false;
            } else {
                arm_tcp_quickack(self.tcp_fd);
            }
        }

        Ok(any_recv)
    }

    fn send_ack(&mut self, ack: &Ack) -> io::Result<bool> {
        if self.ack_in_flight {
            // Coalesce: overwrite any previously queued value — the
            // cursors are cumulative, so the newest pair subsumes
            // everything before it. Sent on the in-flight SEND's CQE.
            self.pending_ack = Some(*ack);
            return Ok(true);
        }
        self.push_ack_send(ack);
        Ok(true)
    }

    fn ack_in_flight(&self) -> bool {
        // Pending counts as in flight: the drain paths use this to mean
        // "everything offered has reached the wire", and the flush gate
        // in `streaming_loop` recomputes a fresh ack next iteration
        // anyway.
        self.ack_in_flight || self.pending_ack.is_some()
    }

    fn is_connected(&mut self) -> bool {
        self.connected
    }
}

// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------

/// Kernel-TCP control-frame source: a blocking `read_frame` on the
/// replication stream. Drives the shared [`receive_chunked_body`] (and,
/// later, the rest of the resync transfer). See [`ControlFrameSource`].
struct TcpFrameSource<'a> {
    reader: &'a mut TcpStream,
}

impl ControlFrameSource for TcpFrameSource<'_> {
    fn next_frame(
        &mut self,
        max_size: usize,
    ) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
        Ok(read_frame(self.reader, max_size)?)
    }
}

/// Outcome of `run_receiver`: `None` = clean shutdown, `Some` = promotion.
pub type ReceiverResult<A, W> = Result<Option<(A, W)>, Box<dyn std::error::Error>>;

#[allow(clippy::too_many_arguments)]
pub fn run_receiver<A>(
    primary_addr: SocketAddr,
    // The node's journal and snapshot: recovered from, written by the
    // replica's shadow stage, and replaced (archived first) by a resync.
    store: &NodeStore,
    signing_key: &ed25519_dalek::SigningKey,
    shutdown: &AtomicBool,
    // The handles this loop shares with the admin endpoint and the
    // control-plane raft driver — see [`super::ReplicaControlPlane`]
    // for each field's contract.
    control: &super::ReplicaControlPlane,
    snapshot_interval_ms: u64,
    cores: crate::layout::PipelineCores,
    // Passed straight to the replica's journal stage; see
    // `build_replica_pipeline_with_threads`.
    staging_mode: melin_journal::StagingMode,
    group_commit_delay: std::time::Duration,
    pipeline_depth: usize,
    fence_state: std::sync::Arc<melin_transport_core::fence::FenceState>,
    // Applied to every instance this loop builds a pipeline around; see
    // `build_replica_pipeline_with_threads`.
    sizing: &A::Sizing,
) -> ReceiverResult<A, BufferedWriter<A::Event>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    let super::ReplicaControlPlane {
        promote,
        tip_ready,
        journal_tip,
        primary_link_up,
        // Set by `accept_stream_start`, which takes `control` whole.
        primary_observed: _,
        lineage_ack_policy,
        pipeline_healthy,
    } = control;
    // Recover whenever any journal segment survives — live OR archived;
    // fresh replicas get `(None, None, 0, zeros)`. See
    // `recover_replica_state` for the lineage rules.
    let (mut app, mut journal_writer, mut last_sequence, mut chain_hash) =
        recover_replica_state::<A, BufferedWriter<A::Event>>(
            store,
            &fence_state,
            lineage_ack_policy,
            sizing,
        )?;
    // The fence epoch now reflects the recovered journal (a fresh replica
    // legitimately recovers epoch 0). Seed the advertised sequence from
    // the same recovery before releasing the flag, so the first tip the
    // raft driver trusts is already the recovered high-water mark.
    // Release-store so the driver's Acquire-load sees both.
    journal_tip.advance(melin_transport_core::WireSeq::new(last_sequence));
    tip_ready.store(true, Ordering::Release);

    let mut backoff = std::time::Duration::from_secs(1);

    // Consecutive mid-stream divergence resyncs this process has
    // attempted — see `MAX_INPROCESS_DIVERGENCE_RESYNCS`.
    let mut divergence_resyncs: u32 = 0;

    let mut send_buf = Vec::with_capacity(64);
    let mut pipeline: Option<ReplicaPipelineHandles<A, BufferedWriter<A::Event>>> = None;

    // --- Outer reconnect loop ---
    loop {
        // No authenticated session while (re)connecting — the raft
        // driver may auto-promote from here on if it wins an election.
        primary_link_up.store(false, Ordering::Release);

        if let Some(p) = pipeline.as_ref() {
            // The handshake pair (last_sequence, chain_hash) must come
            // from ONE FsyncState snapshot — the journal stage keeps
            // flushing while we reconnect, and the primary's handshake
            // validation recomputes its chain at exactly the sequence
            // we claim. Reading the sequence and the hash from two
            // separate sources would tear under load, and a torn pair
            // is indistinguishable from divergence (false resync of a
            // healthy replica). Taken once the journal covers what the
            // last session published (see `settled_resume_point`).
            match p
                .settled_resume_point(|| shutdown.load(Ordering::Relaxed) || promote.is_requested())
            {
                ResumePoint::Settled(fsync_state) => {
                    last_sequence = fsync_state.journal_seq.get();
                    chain_hash = fsync_state.chain_hash;
                }
                // Handled just below, before any handshake.
                ResumePoint::Interrupted => {}
                // Torn down here rather than dialing the primary with a
                // pipeline that cannot take the session.
                ResumePoint::JournalFailed => {
                    let after = handle_session_exit(
                        journal_failed_while_disconnected(),
                        &mut pipeline,
                        &mut divergence_resyncs,
                        &mut backoff,
                        last_sequence,
                        store,
                        &fence_state,
                        lineage_ack_policy,
                        shutdown,
                        promote,
                        // No session is open.
                        || {},
                        sizing,
                    );
                    // A fatal exit never keeps the pipeline, so going
                    // round again dials only after a resync or not at all.
                    if let Some(r) = after.adopt(
                        &mut app,
                        &mut journal_writer,
                        &mut last_sequence,
                        &mut chain_hash,
                    ) {
                        return r;
                    }
                    continue;
                }
            }
        }

        // Sole authority for shutdown/promote while disconnected: every
        // backoff arm just sleeps (the sleep observes both flags) and
        // continues back here, so teardown/promotion handling exists
        // exactly once.
        if shutdown.load(Ordering::Relaxed) {
            if let Some(p) = pipeline.take() {
                let _ = teardown_replica_pipeline::<A, BufferedWriter<A::Event>>(p);
            }
            return Ok(None);
        }
        if promote.is_requested() {
            info!("promotion triggered while disconnected");
            return take_pipeline_for_promotion(&mut pipeline, &mut app, &mut journal_writer);
        }

        // --- Connect and authenticate ---
        info!(primary = %primary_addr, "connecting to primary as replica");

        let stream = match TcpStream::connect(primary_addr) {
            Ok(s) => s,
            Err(e) => {
                warn!(
                    error = %e,
                    backoff_secs = backoff.as_secs(),
                    "failed to connect to primary — retrying"
                );
                sleep_then_double_backoff(&mut backoff, shutdown, promote);
                continue;
            }
        };
        if let Err(e) = stream.set_nodelay(true) {
            warn!(error = %e, "failed to set TCP_NODELAY on replica receive socket");
        }
        {
            use std::os::unix::io::AsRawFd;
            let busy_poll: libc::c_int = 50;
            let rc = unsafe {
                libc::setsockopt(
                    stream.as_raw_fd(),
                    libc::SOL_SOCKET,
                    libc::SO_BUSY_POLL,
                    &busy_poll as *const libc::c_int as *const libc::c_void,
                    std::mem::size_of::<libc::c_int>() as libc::socklen_t,
                )
            };
            if rc != 0 {
                let err = std::io::Error::last_os_error();
                warn!(error = %err, "failed to set SO_BUSY_POLL on replica receive socket");
            }
        }
        stream.set_read_timeout(Some(std::time::Duration::from_secs(5)))?;

        let mut reader = stream.try_clone()?;
        let mut tcp_writer = stream;

        // `reader`/`tcp_writer` are clones of one socket; auth is sequential so
        // a single handle works. Use `reader` (carries the read timeout above).
        if let Err(e) = authenticate_with_primary(&mut reader, signing_key) {
            // A read timeout means no challenge arrived: the target
            // holds the port but is not serving — typically a
            // not-yet-promoted replica (the listener is bound from boot)
            // or a primary mid-start. Only other errors suggest a real
            // credential problem. SO_RCVTIMEO surfaces as WouldBlock on
            // Linux; match TimedOut too for portability.
            if matches!(
                e.kind(),
                std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
            ) {
                warn!(error = %e, "no auth challenge from primary (not serving yet?) — retrying");
            } else {
                warn!(error = %e, "authentication failed — retrying");
            }
            // Drop the socket before sleeping — held through the
            // backoff it would sit in the target's accept backlog, and
            // a just-promoted primary would waste a replica slot (and
            // its 10s auth timeout) on the zombie.
            drop(reader);
            drop(tcp_writer);
            // Back off before redialing — without the sleep this loop
            // hammers a primary that keeps refusing us.
            sleep_then_double_backoff(&mut backoff, shutdown, promote);
            continue;
        }
        info!("authenticated with primary");
        // The primary is demonstrably alive from here until this
        // session ends — the raft driver must not auto-promote past it.
        primary_link_up.store(true, Ordering::Release);

        // --- Handshake ---
        // Advertise our fencing epoch. If we are ahead of the primary it
        // sees this and self-demotes (it's a stale ex-primary); see
        // `crate::fence` and the sender's handshake handling.
        let handshake = Handshake {
            last_sequence,
            chain_hash,
            epoch: fence_state.epoch(),
        };
        send_buf.clear();
        encode_handshake(&handshake, &mut send_buf);
        tcp_writer.write_all(&send_buf)?;
        tcp_writer.flush()?;
        send_buf.clear();

        // --- Protocol negotiation ---
        // `session_start` is the resume point this streaming session
        // continues from — the value that anchors the receiver's
        // sequence-contiguity gate. Derived from local knowledge (our
        // own handshake, or the verified snapshot), never from the
        // wire.
        let response_frame = read_frame(&mut reader, MAX_CONTROL_FRAME)?;
        let response = decode_primary_message(&response_frame)?;
        let (stream_lineage, session_start) = match response {
            PrimaryMessage::StreamStart {
                start_sequence,
                segment_start_sequence,
                anchor_hash,
                genesis_entries,
                epoch,
            } => match super::accept_stream_start(
                super::StreamStart {
                    start_sequence,
                    lineage: (segment_start_sequence, anchor_hash, genesis_entries),
                    epoch,
                },
                "tcp",
                &pipeline,
                &mut journal_writer,
                &fence_state,
                control,
            )? {
                StreamStartVerdict::Follow(lineage) => (lineage, last_sequence),
                // Disconnect and retry with backoff; the operator's logs
                // flag the misdirected `--replica-of`.
                StreamStartVerdict::StalePrimary => {
                    // Drop the socket before sleeping — see the auth
                    // failure arm above.
                    drop(reader);
                    drop(tcp_writer);
                    sleep_then_double_backoff(&mut backoff, shutdown, promote);
                    continue;
                }
            },
            ref resync @ (PrimaryMessage::NeedSnapshot | PrimaryMessage::HashMismatch) => {
                let divergent = matches!(resync, PrimaryMessage::HashMismatch);
                // `map_err` drops `Send + Sync` by coercion — the `?` `From`
                // bridge doesn't apply directly across the two boxed trait
                // objects (`Box<dyn Error + Send + Sync>` → `Box<dyn Error>`).
                let decision = handle_resync_verdict(
                    divergent,
                    &mut TcpFrameSource {
                        reader: &mut reader,
                    },
                    &mut pipeline,
                    &mut app,
                    &mut journal_writer,
                    store,
                    &fence_state,
                    control,
                    &mut last_sequence,
                    &mut chain_hash,
                )
                .map_err(|e| -> Box<dyn std::error::Error> { e })?;
                match decision {
                    ResyncDecision::Ready {
                        segment_start_sequence,
                        anchor_hash,
                        genesis_entries,
                        resume_sequence,
                    } => (
                        (segment_start_sequence, anchor_hash, genesis_entries),
                        resume_sequence,
                    ),
                    ResyncDecision::Retry => {
                        // Drop the socket before sleeping — see the auth
                        // failure arm above.
                        drop(reader);
                        drop(tcp_writer);
                        sleep_then_double_backoff(&mut backoff, shutdown, promote);
                        continue;
                    }
                }
            }
            _ => {
                return Err(format!("unexpected response: {response:?}").into());
            }
        };

        // --- Create journal for fresh replica ---
        // The StreamStart lineage gives the segment header identity
        // (starting sequence + chain anchor) the primary's own journal
        // lineage began with; creating the local segment from the same
        // identity makes the replica's segment byte-identical to the
        // primary's, and adopted `Rotate` boundaries keep it that way
        // across rotations (bitwise mirror). The header records the
        // lineage's genesis length too, so it survives this replica's
        // restarts and is there to check at promotion even with the
        // primary gone.
        if pipeline.is_none() && journal_writer.is_none() {
            let (lineage_start, lineage_anchor, lineage_genesis) = stream_lineage;
            let writer = BufferedWriter::create_continuing(
                store.journal(),
                lineage_start,
                lineage_anchor,
                lineage_genesis,
            )?;
            app = Some(A::default());
            journal_writer = Some(writer);
        }

        // --- Build pipeline if absent ---
        if pipeline.is_none() {
            let cur_app = app.take().expect("application initialized");
            let cur_writer = journal_writer.take().expect("journal_writer initialized");
            pipeline = Some(build_replica_pipeline_with_threads::<A>(
                cur_app,
                cur_writer,
                cores,
                staging_mode,
                snapshot_interval_ms,
                store,
                group_commit_delay,
                Arc::clone(&fence_state),
                lineage_ack_policy,
                Arc::clone(pipeline_healthy),
                sizing,
            )?);
        }

        // --- Streaming session ---
        let result = {
            let p = pipeline.as_mut().expect("pipeline must exist by here");
            let input_producer = &mut p.input_producer;
            let journal_cursor = p.journal_cursor.as_ref();
            let stream_marks = &p.stream_marks;
            let journal_failed = &p.journal_failed;
            std::thread::scope(|s| {
                let receiver = || {
                    melin_app::affinity::pin_thread("replica-receiver", cores.reader.core);
                    let mut transport = match UringTransport::new(&tcp_writer) {
                        Ok(t) => t,
                        Err(e) => {
                            tracing::error!(error = %e, "UringTransport init failed");
                            // Synthetic result: streaming never started,
                            // so `heard_from_primary` stays false and the
                            // reconnect keeps its escalated backoff — this
                            // failure is local (io_uring setup), and a
                            // persistent one must not redial at 1 Hz.
                            return super::receiver_transport::StreamingResult {
                                exit: SessionExit::Disconnected,
                                heard_from_primary: false,
                            };
                        }
                    };
                    streaming_loop::<UringTransport, A::Event>(
                        &mut transport,
                        input_producer,
                        journal_cursor,
                        shutdown,
                        control,
                        pipeline_depth,
                        cores.reader.wait,
                        session_start,
                        Vec::with_capacity(MAX_DATA_FRAME + 4),
                        None,
                        stream_marks,
                        journal_failed,
                    )
                };
                // Created in its own placement, not this thread's — see
                // `melin_app::affinity::spawn_placed`.
                let handle = melin_app::affinity::with_child_context(
                    "replica-receiver",
                    cores.reader.core,
                    || {
                        std::thread::Builder::new()
                            .name("replica-receiver".into())
                            .spawn_scoped(s, receiver)
                    },
                )
                .expect("spawn replica-receiver thread");
                handle.join().expect("replica-receiver thread panicked")
            })
        };

        primary_link_up.store(false, Ordering::Release);
        let after = handle_session_exit(
            result,
            &mut pipeline,
            &mut divergence_resyncs,
            &mut backoff,
            last_sequence,
            store,
            &fence_state,
            lineage_ack_policy,
            shutdown,
            promote,
            // Kernel TCP needs no explicit close — the `TcpStream` is
            // dropped on the next loop turn when a fresh connection
            // replaces it.
            || {},
            sizing,
        );
        if let Some(r) = after.adopt(
            &mut app,
            &mut journal_writer,
            &mut last_sequence,
            &mut chain_hash,
        ) {
            return r;
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::uring_teardown::DRAIN_TIMEOUT;
    use std::io::Read;
    use std::net::TcpListener;
    use std::time::{Duration, Instant};

    use melin_transport_core::replication::protocol::{ReplicaMessage, decode_replica_message};
    use melin_wire_protocol::framing::split_frame_limited;

    /// The store of a node whose snapshot is derived from `journal`, as a
    /// node without `--snapshot-path` resolves it. For the tests that run
    /// a receiver on a real journal, which a `no-persist` build has none
    /// of.
    #[cfg(not(feature = "no-persist"))]
    fn store(journal: &std::path::Path) -> NodeStore {
        NodeStore::resolve(journal.to_path_buf(), None).expect("store")
    }

    /// Connected localhost pair: (transport side, peer side).
    fn socket_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let client = TcpStream::connect(addr).unwrap();
        let (server, _) = listener.accept().unwrap();
        (client, server)
    }

    fn ack(seq: u64) -> Ack {
        Ack {
            acked_sequence: seq,
            in_memory_sequence: seq,
        }
    }

    /// Drive `poll_recv` until every accepted ack has reached the wire.
    fn flush_acks(transport: &mut UringTransport) {
        let mut recv_buf = Vec::new();
        let deadline = Instant::now() + Duration::from_secs(5);
        while transport.ack_in_flight() {
            transport.poll_recv(&mut recv_buf).unwrap();
            assert!(Instant::now() < deadline, "ack send never completed");
            std::hint::spin_loop();
        }
    }

    /// Read `expect` length-prefixed ack frames from the peer socket.
    fn read_acks(peer: &mut TcpStream, expect: usize) -> Vec<Ack> {
        peer.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
        let mut buf = Vec::new();
        let mut chunk = [0u8; 256];
        let mut acks = Vec::new();
        while acks.len() < expect {
            let n = peer.read(&mut chunk).expect("peer read");
            assert!(n > 0, "peer closed before all acks arrived");
            buf.extend_from_slice(&chunk[..n]);
            // Parse complete frames.
            while let Some(frame) = split_frame_limited(&buf, MAX_CONTROL_FRAME).unwrap() {
                match decode_replica_message(frame.payload).expect("decodable frame") {
                    ReplicaMessage::Ack(a) => acks.push(a),
                    other => panic!("expected Ack frame, got {other:?}"),
                }
                let consumed = frame.consumed();
                buf.drain(..consumed);
            }
        }
        assert!(buf.is_empty(), "trailing bytes after expected acks");
        acks
    }

    #[test]
    fn coalesces_to_latest_ack_while_send_in_flight() {
        let (client, mut peer) = socket_pair();
        let mut transport = UringTransport::new(&client).unwrap();

        // First ack goes straight onto the wire path; the next two are
        // accepted while it is in flight (the flag clears only when the
        // CQE is reaped in poll_recv, so this is deterministic) and
        // coalesce — only the newest survives.
        assert!(transport.send_ack(&ack(1)).unwrap());
        assert!(transport.ack_in_flight());
        assert!(transport.send_ack(&ack(2)).unwrap());
        assert!(transport.send_ack(&ack(3)).unwrap());

        flush_acks(&mut transport);

        // Exactly two frames, in cursor order: ack(2) was subsumed.
        let acks = read_acks(&mut peer, 2);
        let seqs: Vec<u64> = acks.iter().map(|a| a.acked_sequence).collect();
        assert_eq!(seqs, vec![1, 3]);
    }

    #[test]
    fn sequential_acks_all_reach_the_wire_in_order() {
        let (client, mut peer) = socket_pair();
        let mut transport = UringTransport::new(&client).unwrap();

        for seq in 1..=3 {
            assert!(transport.send_ack(&ack(seq)).unwrap());
            flush_acks(&mut transport);
        }

        let acks = read_acks(&mut peer, 3);
        let seqs: Vec<u64> = acks.iter().map(|a| a.acked_sequence).collect();
        assert_eq!(seqs, vec![1, 2, 3]);
    }

    /// The multishot RECV armed at construction stays armed for the
    /// whole session, so every ordinary session exit drops the
    /// transport with the kernel still holding pointers into the buffer
    /// pool. `Drop` has to quiesce it — and promptly: a drain that
    /// cannot settle only gives up on the deadline, so finishing well
    /// inside it is what proves the buffers were released rather than
    /// leaked.
    #[test]
    fn drop_quiesces_the_multishot_recv() {
        let (client, _peer) = socket_pair();
        let mut transport = UringTransport::new(&client).unwrap();
        // `new` only pushes the multishot RECV; the first poll submits
        // it. Teardown from that state is what a real session does.
        let mut recv_buf = Vec::new();
        transport.poll_recv(&mut recv_buf).unwrap();
        assert!(transport.multishot_active);

        let started = Instant::now();
        drop(transport);
        let elapsed = started.elapsed();
        assert!(
            elapsed < DRAIN_TIMEOUT / 4,
            "teardown took {elapsed:?}, close to the {DRAIN_TIMEOUT:?} deadline"
        );
    }

    /// A failing session bails out of `poll_recv` part-way through a
    /// batch, abandoning CQEs that were already reaped out of the
    /// completion queue. Their kernel-ownership flags still have to be
    /// recorded: otherwise teardown waits out its whole deadline for a
    /// SEND completion that was consumed here and can never arrive
    /// again, and then leaks the buffer pool.
    #[test]
    fn abandoned_cqes_still_settle_kernel_ownership() {
        let (client, _peer) = socket_pair();
        let mut transport = UringTransport::new(&client).unwrap();
        let mut recv_buf = Vec::new();
        transport.poll_recv(&mut recv_buf).unwrap();

        // Stands in for a SEND whose CQE landed in the batch behind a
        // RECV CQE that failed the session.
        transport.ack_in_flight = true;
        transport.settle_unprocessed(&[(TOKEN_SEND, 13, 0)]);
        assert!(
            !transport.ack_in_flight,
            "an abandoned SEND completion must release `ack_buf`"
        );

        // A RECV that still carries F_MORE leaves the arm live — the
        // kernel may write into the pool again, so teardown must wait.
        transport.settle_unprocessed(&[(TOKEN_RECV, 128, IORING_CQE_F_MORE)]);
        assert!(
            transport.multishot_active,
            "F_MORE must leave the multishot armed"
        );
        // Only the terminal completion retires it.
        transport.settle_unprocessed(&[(TOKEN_RECV, 0, 0)]);
        assert!(!transport.multishot_active);

        // The real arm submitted by `poll_recv` above is still live in
        // the kernel; restore the flag the assertion just cleared so
        // `Drop` drains it instead of freeing the pool underneath it.
        transport.multishot_active = true;
    }

    /// Same, for the other kernel-visible buffer: an ack SEND pushed
    /// but never submitted, so the drain has to flush the SQ before it
    /// can wait for anything.
    #[test]
    fn drop_quiesces_a_pending_ack_send() {
        let (client, _peer) = socket_pair();
        let mut transport = UringTransport::new(&client).unwrap();
        assert!(transport.send_ack(&ack(7)).unwrap());
        assert!(transport.ack_in_flight());

        let started = Instant::now();
        drop(transport);
        let elapsed = started.elapsed();
        assert!(
            elapsed < DRAIN_TIMEOUT / 4,
            "teardown took {elapsed:?}, close to the {DRAIN_TIMEOUT:?} deadline"
        );
    }

    /// Scripted-primary scaffolding shared by the end-to-end receiver
    /// tests below: a no-op application over a trivial event, the
    /// primary-side frame helpers, and common fixtures. Gated like its
    /// consumers so a `no-persist` build carries no dead test code.
    #[cfg(not(feature = "no-persist"))]
    mod scripted {
        use super::*;
        use melin_app::{AppEvent, Application, ApplyCtx, CodecError, QueryCtx, RejectReason};

        // Re-export so the sibling test modules can name the generic
        // slot type without reaching into transport-core themselves.
        pub(super) use melin_transport_core::pipeline::{InputSlot, RequestId};

        #[derive(Debug, Clone, Copy, PartialEq)]
        pub(super) struct EvtAdd(pub(super) u64);

        impl AppEvent for EvtAdd {
            const MAX_ENCODED_SIZE: usize = 8;

            fn encoded_size(&self) -> usize {
                8
            }
            fn encode(&self, buf: &mut [u8]) -> usize {
                buf[..8].copy_from_slice(&self.0.to_le_bytes());
                8
            }
            fn decode(buf: &[u8]) -> Result<Self, CodecError> {
                if buf.len() < 8 {
                    return Err(CodecError::Truncated);
                }
                Ok(EvtAdd(u64::from_le_bytes(buf[..8].try_into().expect("8"))))
            }
            fn is_query(&self) -> bool {
                false
            }
        }

        #[derive(Debug, Clone, Copy)]
        pub(super) struct Rpt;

        #[derive(Default)]
        pub(super) struct App;

        impl Application for App {
            type Event = EvtAdd;
            type Report = Rpt;
            type QueryResponse = Rpt;
            type Sizing = ();
            const APP_VERSION: u16 = 1;

            fn apply(&mut self, _e: EvtAdd, _ctx: &ApplyCtx, _out: &mut Vec<Rpt>) {}
            fn query(&self, _e: EvtAdd, _ctx: &QueryCtx) -> Option<Rpt> {
                None
            }
            fn tick(&mut self, _now_ns: u64, _out: &mut Vec<Rpt>) {}
            fn build_reject(_e: &EvtAdd, _r: RejectReason) -> Rpt {
                Rpt
            }
            fn snapshot<W: std::io::Write>(&self, _w: &mut W) -> std::io::Result<()> {
                Ok(())
            }
            fn restore<R: std::io::Read>(_r: &mut R) -> std::io::Result<Self> {
                Ok(App)
            }
        }

        /// Deterministic replica signing-key bytes; [`replica_auth`]
        /// builds the matching authorized-keys table.
        pub(super) const REPLICA_KEY: [u8; 32] = [0xFC; 32];

        /// The authorized-keys table listing [`REPLICA_KEY`]'s public
        /// half.
        pub(super) fn replica_auth() -> melin_app::auth::AuthorizedKeys {
            let key = ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY);
            let pub_b64 = base64::Engine::encode(
                &base64::engine::general_purpose::STANDARD,
                key.verifying_key().to_bytes(),
            );
            melin_app::auth::AuthorizedKeys::parse::<melin_app::auth::NoRoles>(&format!(
                "replication {pub_b64} test-replica\n"
            ))
            .expect("parse keys")
        }

        /// Read one length-prefixed `ReplicaMessage` frame.
        pub(super) fn read_replica_msg(stream: &mut TcpStream) -> ReplicaMessage {
            let mut len_buf = [0u8; 4];
            stream.read_exact(&mut len_buf).expect("frame length");
            let len = u32::from_le_bytes(len_buf) as usize;
            let mut payload = vec![0u8; len];
            stream.read_exact(&mut payload).expect("frame payload");
            decode_replica_message(&payload).expect("decodable replica frame")
        }

        /// Skip frames until an `Ack` at or past `seq` arrives.
        pub(super) fn wait_for_ack(stream: &mut TcpStream, seq: u64) {
            loop {
                if let ReplicaMessage::Ack(a) = read_replica_msg(stream)
                    && a.acked_sequence >= seq
                {
                    return;
                }
            }
        }

        /// Accept with a deadline so a replica that fails to (re)connect
        /// fails the test instead of hanging it.
        pub(super) fn accept_within(listener: &std::net::TcpListener, secs: u64) -> TcpStream {
            let deadline = Instant::now() + Duration::from_secs(secs);
            loop {
                match listener.accept() {
                    Ok((s, _)) => {
                        s.set_read_timeout(Some(Duration::from_secs(10)))
                            .expect("read timeout");
                        return s;
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                        assert!(Instant::now() < deadline, "replica never connected");
                        std::thread::sleep(Duration::from_millis(10));
                    }
                    Err(e) => panic!("accept failed: {e}"),
                }
            }
        }
    }

    // -----------------------------------------------------------------
    // In-process divergence resync — end to end against a scripted
    // primary. Needs hash-chain (divergence is a chain verdict) and
    // real persistence (the resync re-derives state from disk).
    // -----------------------------------------------------------------
    #[cfg(all(feature = "hash-chain", not(feature = "no-persist")))]
    mod divergence_resync {
        use super::super::super::auth::authenticate_replica;
        use super::scripted::*;
        use super::*;
        use melin_journal::{
            AckPolicyByte, AckPolicyChange, BufferedWriter, JournalEvent, JournalWrite,
        };
        use melin_transport_core::cursors::WireSeq;
        use melin_transport_core::replication::catchup::{lineage_origin, snapshot_transfer_with};
        use melin_transport_core::replication::protocol::{
            PrimaryMessage, decode_primary_message, encode_hash_mismatch, encode_rotate,
            encode_stream_start,
        };
        use std::sync::Arc;
        use std::sync::atomic::AtomicBool;

        /// The policy the forked replica journals and the primary never
        /// did.
        const DIVERGENT_POLICY: AckPolicyByte = AckPolicyByte(1);

        /// The policy the primary's snapshot lineage records: what the
        /// resync must rebase the replica onto.
        const PRIMARY_POLICY: AckPolicyByte = AckPolicyByte(0);

        /// The primary's snapshot lineage policy, whole: every field set
        /// apart from the forked record's, so adopting it whole shows.
        const PRIMARY_LINEAGE_POLICY: melin_transport_core::lineage::LineagePolicy =
            melin_transport_core::lineage::LineagePolicy {
                change: AckPolicyChange {
                    policy: PRIMARY_POLICY,
                    unwitnessed: true,
                },
                wire_seq: WireSeq::new(3),
                previous: Some(AckPolicyByte(2)),
            };

        /// A shadow snapshot interval short enough for a test to wait on.
        const SHADOW_INTERVAL_MS: u64 = 50;

        /// A scripted primary brings a fresh replica up, streams one
        /// event, then announces a rotation with a deliberately wrong
        /// tail hash. The replica's journal stage detects divergence —
        /// `run_receiver` must repair it WITHOUT exiting: tear the
        /// pipeline down, re-derive its handshake position from disk,
        /// reconnect, take the HashMismatch verdict, archive the forked
        /// lineage, and complete the snapshot + segment-seed resync,
        /// then resume streaming on the rebased lineage.
        #[test]
        fn mid_stream_divergence_resyncs_in_process() {
            let dir = tempfile::tempdir().expect("tempdir");

            // --- The scripted primary's journal + snapshot, served by
            // the real sender-side transfer code in session 2. Entries
            // 1..=5, rotation after 2, snapshot at 4 (mid-live-segment).
            let primary_journal = dir.path().join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&primary_journal).expect("create");
            let mut chain_at_4 = [0u8; 32];
            for v in 1..=5u64 {
                w.append(&JournalEvent::App(EvtAdd(v))).expect("append");
                if v == 2 {
                    w.rotate_segment().expect("rotate");
                }
                if v == 4 {
                    chain_at_4 = w.chain_hash().expect("chain");
                }
            }
            drop(w);
            // The snapshot's lineage carries a policy of its own, distinct
            // from the forked one, so the rebase's adoption of it is
            // observable (the archive step alone clears the cell to `None`).
            melin_transport_core::snapshot::save::<App>(
                &App,
                WireSeq::new(4),
                chain_at_4,
                melin_transport_core::lineage::Lineage {
                    epoch: 0,
                    ack_policy: Some(PRIMARY_LINEAGE_POLICY),
                },
                Some(0),
                store(&primary_journal).snapshot(),
            )
            .expect("save snapshot");
            let origin = lineage_origin(&primary_journal).expect("lineage");
            let (lineage_start, lineage_anchor, lineage_genesis) = (
                origin.starting_sequence,
                origin.anchor_hash,
                origin.genesis_entries,
            );

            // --- Auth: one replica key, authorized.
            let authorized_keys = replica_auth();

            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            // --- Replica under test.
            let replica_journal = dir.path().join("replica.journal");
            let replica_snapshot_check = store(&replica_journal).snapshot().to_path_buf();
            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let cores = crate::layout::PipelineCores::unpinned();
            let replica = {
                let journal = replica_journal.clone();
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<bool, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        // Short, so the rebuilt pipeline's shadow writes a
                        // snapshot of its own after the resync (see the
                        // lineage check below).
                        SHADOW_INTERVAL_MS,
                        cores,
                        melin_journal::StagingMode::ZeroFill,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    // ReceiverResult's error is !Send — stringify for join().
                    .map(|state| state.is_none())
                    .map_err(|e| e.to_string())
                })
            };

            // --- Session 1: fresh sync, one event, then the poisoned
            // rotation announce.
            let mut s1 = accept_within(&listener, 30);
            let mut s1r = s1.try_clone().expect("clone");
            authenticate_replica(&mut s1r, &authorized_keys).expect("auth 1");
            match read_replica_msg(&mut s1) {
                ReplicaMessage::Handshake(h) => {
                    assert_eq!(h.last_sequence, 0, "fresh replica handshake")
                }
                other => panic!("expected Handshake, got {other:?}"),
            }
            let mut buf = Vec::new();
            encode_stream_start(
                0,
                lineage_start,
                lineage_anchor,
                lineage_genesis,
                0,
                &mut buf,
            );
            s1.write_all(&buf).expect("StreamStart 1");
            buf.clear();
            // The forked entry is an ack-policy change the primary's
            // lineage does not hold (its entry 1 is an application
            // event): the resync must drop it with the rest of the fork.
            melin_transport_core::replication_wire::encode_input_batch(
                &[InputSlot::<EvtAdd> {
                    connection_id: 0,
                    request_id: RequestId(0),
                    key_hash: 0,
                    sequence: 1,
                    timestamp_ns: 1,
                    event: JournalEvent::AckPolicyChange(AckPolicyChange {
                        policy: DIVERGENT_POLICY,
                        unwitnessed: false,
                    }),
                    // `()` without latency-trace, a timestamp with it.
                    publish_ts: Default::default(),
                    recv_ts: Default::default(),
                }],
                &mut buf,
            )
            .expect("encode InputBatch");
            s1.write_all(&buf).expect("InputBatch");
            wait_for_ack(&mut s1, 1);

            // Wrong tail hash at the announced boundary — the replica's
            // local chain at sequence 1 cannot match this.
            buf.clear();
            encode_rotate(1, &[0xEE; 32], &mut buf);
            s1.write_all(&buf).expect("Rotate");

            // --- Session 2: the SAME process reconnects with its
            // position re-derived from the forked on-disk journal.
            let mut s2 = accept_within(&listener, 30);
            let mut s2r = s2.try_clone().expect("clone");
            authenticate_replica(&mut s2r, &authorized_keys).expect("auth 2");
            match read_replica_msg(&mut s2) {
                ReplicaMessage::Handshake(h) => assert_eq!(
                    h.last_sequence, 1,
                    "post-divergence handshake must carry the recovered journal position"
                ),
                other => panic!("expected Handshake, got {other:?}"),
            }
            // The journal stage's divergence failure latched the health
            // mirror before the teardown that led to this reconnect, and
            // the replacement pipeline is not built until after the resync
            // verdict — so at this deterministic point the replica must be
            // reporting unhealthy.
            assert!(
                !control.pipeline_healthy.load(Ordering::Acquire),
                "the divergence must have latched the health mirror false"
            );
            // The handshake position was re-derived by recovering the
            // forked journal, and the lineage with it: its change is the
            // policy this replica holds until the resync replaces it.
            assert_eq!(
                control.lineage_ack_policy.load(),
                Some(melin_transport_core::lineage::LineagePolicy {
                    change: AckPolicyChange {
                        policy: DIVERGENT_POLICY,
                        unwitnessed: false,
                    },
                    wire_seq: WireSeq::new(1),
                    previous: None,
                }),
                "recovery must store the forked journal's policy"
            );
            buf.clear();
            encode_hash_mismatch(&mut buf);
            s2.write_all(&buf).expect("HashMismatch");
            let transfer_shutdown = AtomicBool::new(false);
            let mut publish = |b: &[u8]| -> std::io::Result<()> {
                s2.write_all(b)?;
                s2.flush()
            };
            // Real sender-side resync: SnapshotBegin → chunks →
            // SegmentSeedBegin → seed → StreamStart → catch-up (entry 5).
            let end = snapshot_transfer_with::<EvtAdd>(
                &store(&primary_journal),
                &mut publish,
                &transfer_shutdown,
            )
            .expect("snapshot transfer");
            assert_eq!(
                end,
                melin_transport_core::replication::catchup::CatchUpResult::Ok(5)
            );
            let mut s2_acks = s2.try_clone().expect("clone");
            wait_for_ack(&mut s2_acks, 5);

            // Streaming resumed on the rebuilt pipeline — the rebuild
            // must have reset the health mirror.
            assert!(
                control.pipeline_healthy.load(Ordering::Acquire),
                "the resync rebuild must reset the health mirror true"
            );
            // The rebase took the primary's snapshot lineage whole: its
            // policy is in force, and the forked change is gone, not kept
            // as the newest value seen.
            assert_eq!(
                control.lineage_ack_policy.load(),
                Some(PRIMARY_LINEAGE_POLICY),
                "the resync must adopt the primary snapshot's policy and drop the divergent tail's"
            );
            // The rebuilt pipeline's shadow starts from that cell: its
            // own snapshot, past the transferred one (at 4), records the
            // adopted policy although the stream since carried no change.
            let deadline = std::time::Instant::now() + Duration::from_secs(30);
            let shadow_lineage = loop {
                // No snapshot past 4 yet (or none at all) is "not yet".
                if let Ok((_, header)) =
                    melin_transport_core::snapshot::load::<App>(&replica_snapshot_check)
                    && header.sequence == 5
                {
                    break header.lineage;
                }
                assert!(
                    std::time::Instant::now() < deadline,
                    "the rebuilt shadow never snapshotted entry 5"
                );
                std::thread::sleep(Duration::from_millis(10));
            };
            assert_eq!(
                shadow_lineage.ack_policy,
                Some(PRIMARY_LINEAGE_POLICY),
                "the replica's shadow must seed its lineage from the adopted policy"
            );

            // --- Clean shutdown; the receiver must return Ok(None) —
            // the divergence never escaped as a process-fatal error.
            shutdown.store(true, Ordering::Relaxed);
            let result = replica.join().expect("replica thread panicked");
            assert_eq!(result, Ok(true), "receiver must exit cleanly via shutdown");

            // The forked lineage was archived (never deleted), and the
            // live journal is the re-seeded one.
            assert!(
                dir.path().join("replica.journal.divergent.0").exists(),
                "divergent lineage must be archived"
            );
            assert!(replica_journal.exists(), "re-seeded live journal");
        }

        /// A resync whose snapshot transfer drops mid-flight must retry
        /// as a fresh replica with NO leftover in-memory state. On the
        /// in-process divergence repair path `recover_replica_state`
        /// leaves `app`/`journal_writer` populated; the resync arm
        /// archives the live journal (renaming it aside) before the
        /// transfer. If the transfer then fails and those handles are not
        /// nulled, the stale writer — its backing file now under the
        /// `.divergent.<n>` archive — survives the fresh-replica create
        /// gate (`journal_writer.is_none()` false) and is rebuilt into the
        /// next pipeline, so the live journal is never recreated and the
        /// new session streams onto an archived-away inode. This pins the
        /// fix (mirrors the DPDK receiver): after the failed transfer the
        /// retry creates a fresh live journal from the StreamStart lineage.
        ///
        /// The transfer fails AFTER the snapshot was received but before
        /// the segment seed (the primary drops), so the snapshot's lineage
        /// must not be adopted: a fresh replica holds no lineage, so no
        /// ack policy.
        #[test]
        fn resync_transfer_failure_rebuilds_clean_not_over_stale_writer() {
            let dir = tempfile::tempdir().expect("tempdir");

            // Scripted primary: entries 1..=3 and a snapshot at 2 whose
            // lineage records a policy. Session 2 sends that snapshot and
            // drops before the seed; session 3 streams from genesis on
            // the same lineage identity.
            let primary_journal = dir.path().join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&primary_journal).expect("create");
            let mut chain_at_2 = [0u8; 32];
            for v in 1..=3u64 {
                w.append(&JournalEvent::App(EvtAdd(v))).expect("append");
                if v == 2 {
                    chain_at_2 = w.chain_hash().expect("chain");
                }
            }
            drop(w);
            melin_transport_core::snapshot::save::<App>(
                &App,
                WireSeq::new(2),
                chain_at_2,
                melin_transport_core::lineage::Lineage {
                    epoch: 0,
                    ack_policy: Some(PRIMARY_LINEAGE_POLICY),
                },
                Some(0),
                store(&primary_journal).snapshot(),
            )
            .expect("save snapshot");
            let origin = lineage_origin(&primary_journal).expect("lineage");
            let (lineage_start, lineage_anchor, lineage_genesis) = (
                origin.starting_sequence,
                origin.anchor_hash,
                origin.genesis_entries,
            );

            let authorized_keys = replica_auth();

            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            let replica_journal = dir.path().join("replica.journal");
            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let cores = crate::layout::PipelineCores::unpinned();
            let replica = {
                let journal = replica_journal.clone();
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<bool, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        3_600_000,
                        cores,
                        melin_journal::StagingMode::ZeroFill,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    .map(|state| state.is_none())
                    .map_err(|e| e.to_string())
                })
            };

            // Helper: stream a single event (seq 1) over an established
            // session and wait for its ack.
            let stream_one = |s: &mut TcpStream, buf: &mut Vec<u8>| {
                buf.clear();
                encode_stream_start(0, lineage_start, lineage_anchor, lineage_genesis, 0, buf);
                s.write_all(buf).expect("StreamStart");
                buf.clear();
                melin_transport_core::replication_wire::encode_input_batch(
                    &[InputSlot::<EvtAdd> {
                        connection_id: 0,
                        request_id: RequestId(0),
                        key_hash: 0,
                        sequence: 1,
                        timestamp_ns: 1,
                        event: JournalEvent::App(EvtAdd(1)),
                        publish_ts: Default::default(),
                        recv_ts: Default::default(),
                    }],
                    buf,
                )
                .expect("encode InputBatch");
                s.write_all(buf).expect("InputBatch");
            };

            // --- Session 1: fresh sync, event 1, poisoned rotation —
            // the journal stage detects divergence and the receiver
            // repairs in-process (`recover_replica_state` repopulates
            // `app`/`journal_writer`).
            let mut buf = Vec::new();
            let mut s1 = accept_within(&listener, 30);
            let mut s1r = s1.try_clone().expect("clone");
            authenticate_replica(&mut s1r, &authorized_keys).expect("auth 1");
            match read_replica_msg(&mut s1) {
                ReplicaMessage::Handshake(h) => assert_eq!(h.last_sequence, 0, "fresh"),
                other => panic!("expected Handshake, got {other:?}"),
            }
            stream_one(&mut s1, &mut buf);
            wait_for_ack(&mut s1, 1);
            buf.clear();
            encode_rotate(1, &[0xEE; 32], &mut buf);
            s1.write_all(&buf).expect("Rotate");

            // --- Session 2: reconnect at the recovered position (seq 1),
            // verdict HashMismatch (archives the live journal aside), send
            // the snapshot, then DROP the connection before the segment
            // seed — the transfer fails after the snapshot is received but
            // before the seed, so its lineage must not be adopted, and the
            // receiver must retry as a fresh replica.
            let mut s2 = accept_within(&listener, 30);
            let mut s2r = s2.try_clone().expect("clone");
            authenticate_replica(&mut s2r, &authorized_keys).expect("auth 2");
            match read_replica_msg(&mut s2) {
                ReplicaMessage::Handshake(h) => assert_eq!(
                    h.last_sequence, 1,
                    "reconnect carries the recovered forked position"
                ),
                other => panic!("expected Handshake, got {other:?}"),
            }
            buf.clear();
            encode_hash_mismatch(&mut buf);
            s2.write_all(&buf).expect("HashMismatch");
            s2.flush().expect("flush");
            {
                let transfer_shutdown = AtomicBool::new(false);
                // Every publish is one whole frame (length prefix, then
                // payload): refuse the seed's, so the snapshot is the last
                // thing the replica receives.
                let mut publish = |b: &[u8]| -> std::io::Result<()> {
                    if matches!(
                        decode_primary_message(&b[4..]),
                        Ok(PrimaryMessage::SegmentSeedBegin { .. })
                    ) {
                        return Err(std::io::Error::other("scripted drop before the seed"));
                    }
                    s2.write_all(b)?;
                    s2.flush()
                };
                snapshot_transfer_with::<EvtAdd>(
                    &store(&primary_journal),
                    &mut publish,
                    &transfer_shutdown,
                )
                .expect_err("the scripted primary drops before the seed");
            }
            // Both fds reference the same socket (try_clone dups) — drop
            // both to send FIN so the replica's SegmentSeedBegin read hits
            // EOF.
            drop(s2);
            drop(s2r);

            // --- Session 3: the receiver reconnects as a FRESH replica
            // (the archived lineage's position is meaningless, so
            // `last_sequence` is 0). It must recreate a live journal from
            // the StreamStart lineage, NOT rebuild over the stale writer.
            let mut s3 = accept_within(&listener, 30);
            let mut s3r = s3.try_clone().expect("clone");
            authenticate_replica(&mut s3r, &authorized_keys).expect("auth 3");
            match read_replica_msg(&mut s3) {
                ReplicaMessage::Handshake(h) => assert_eq!(
                    h.last_sequence, 0,
                    "after a failed transfer the retry handshakes as fresh"
                ),
                other => panic!("expected Handshake, got {other:?}"),
            }
            // The archive cleared the cell before the transfer began, a
            // failed transfer adopts nothing, and no recovery runs on the
            // Retry path, so this read is deterministic.
            assert_eq!(
                control.lineage_ack_policy.load(),
                None,
                "a failed transfer must not adopt the snapshot's policy"
            );
            stream_one(&mut s3, &mut buf);
            wait_for_ack(&mut s3, 1);

            shutdown.store(true, Ordering::Relaxed);
            let result = replica.join().expect("replica thread panicked");
            assert_eq!(result, Ok(true), "receiver exits cleanly via shutdown");

            // The forked lineage stays archived, and a fresh live journal
            // was recreated from the lineage. With the bug, the stale
            // writer would be reused and this path would not exist (its
            // inode stranded under `.divergent.0`).
            assert!(
                dir.path().join("replica.journal.divergent.0").exists(),
                "forked lineage archived, never deleted"
            );
            assert!(
                replica_journal.exists(),
                "fresh live journal recreated after the failed transfer"
            );
        }

        /// The in-process repair budget is one per process lifetime: a
        /// second mid-stream divergence is systematic (corruption or a
        /// serious bug that the first re-seed did not cure) and the
        /// receiver must exit hard instead of looping — each repair
        /// cycle archives a full journal copy, and a repair loop would
        /// fill the disk while masking the underlying fault.
        #[test]
        fn second_mid_stream_divergence_exits_hard() {
            let dir = tempfile::tempdir().expect("tempdir");

            let primary_journal = dir.path().join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&primary_journal).expect("create");
            let mut chain_at_4 = [0u8; 32];
            for v in 1..=5u64 {
                w.append(&JournalEvent::App(EvtAdd(v))).expect("append");
                if v == 2 {
                    w.rotate_segment().expect("rotate");
                }
                if v == 4 {
                    chain_at_4 = w.chain_hash().expect("chain");
                }
            }
            drop(w);
            melin_transport_core::snapshot::save::<App>(
                &App,
                WireSeq::new(4),
                chain_at_4,
                melin_transport_core::lineage::Lineage::GENESIS,
                Some(0),
                store(&primary_journal).snapshot(),
            )
            .expect("save snapshot");
            let origin = lineage_origin(&primary_journal).expect("lineage");
            let (lineage_start, lineage_anchor, lineage_genesis) = (
                origin.starting_sequence,
                origin.anchor_hash,
                origin.genesis_entries,
            );

            let authorized_keys = replica_auth();

            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            let replica_journal = dir.path().join("replica.journal");
            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let cores = crate::layout::PipelineCores::unpinned();
            let replica = {
                let journal = replica_journal.clone();
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<bool, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        3_600_000,
                        cores,
                        melin_journal::StagingMode::ZeroFill,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    .map(|state| state.is_none())
                    .map_err(|e| e.to_string())
                })
            };

            // Session 1: fresh sync, one event, poisoned rotation.
            let mut s1 = accept_within(&listener, 30);
            let mut s1r = s1.try_clone().expect("clone");
            authenticate_replica(&mut s1r, &authorized_keys).expect("auth 1");
            match read_replica_msg(&mut s1) {
                ReplicaMessage::Handshake(h) => assert_eq!(h.last_sequence, 0),
                other => panic!("expected Handshake, got {other:?}"),
            }
            let mut buf = Vec::new();
            encode_stream_start(
                0,
                lineage_start,
                lineage_anchor,
                lineage_genesis,
                0,
                &mut buf,
            );
            s1.write_all(&buf).expect("StreamStart 1");
            buf.clear();
            melin_transport_core::replication_wire::encode_input_batch(
                &[InputSlot::<EvtAdd> {
                    connection_id: 0,
                    request_id: RequestId(0),
                    key_hash: 0,
                    sequence: 1,
                    timestamp_ns: 1,
                    event: JournalEvent::App(EvtAdd(1)),
                    publish_ts: Default::default(),
                    recv_ts: Default::default(),
                }],
                &mut buf,
            )
            .expect("encode InputBatch");
            s1.write_all(&buf).expect("InputBatch");
            wait_for_ack(&mut s1, 1);
            buf.clear();
            encode_rotate(1, &[0xEE; 32], &mut buf);
            s1.write_all(&buf).expect("Rotate 1");

            // Session 2: in-process repair (the one allowed), then a
            // SECOND poisoned rotation after streaming resumes.
            let mut s2 = accept_within(&listener, 30);
            let mut s2r = s2.try_clone().expect("clone");
            authenticate_replica(&mut s2r, &authorized_keys).expect("auth 2");
            match read_replica_msg(&mut s2) {
                ReplicaMessage::Handshake(h) => assert_eq!(h.last_sequence, 1),
                other => panic!("expected Handshake, got {other:?}"),
            }
            buf.clear();
            encode_hash_mismatch(&mut buf);
            s2.write_all(&buf).expect("HashMismatch");
            let transfer_shutdown = AtomicBool::new(false);
            {
                let mut publish = |b: &[u8]| -> std::io::Result<()> {
                    s2.write_all(b)?;
                    s2.flush()
                };
                snapshot_transfer_with::<EvtAdd>(
                    &store(&primary_journal),
                    &mut publish,
                    &transfer_shutdown,
                )
                .expect("snapshot transfer");
            }
            buf.clear();
            melin_transport_core::replication_wire::encode_input_batch(
                &[InputSlot::<EvtAdd> {
                    connection_id: 0,
                    request_id: RequestId(0),
                    key_hash: 0,
                    sequence: 6,
                    timestamp_ns: 6,
                    event: JournalEvent::App(EvtAdd(6)),
                    publish_ts: Default::default(),
                    recv_ts: Default::default(),
                }],
                &mut buf,
            )
            .expect("encode InputBatch");
            s2.write_all(&buf).expect("InputBatch 6");
            let mut s2_acks = s2.try_clone().expect("clone");
            wait_for_ack(&mut s2_acks, 6);
            buf.clear();
            encode_rotate(6, &[0xEE; 32], &mut buf);
            s2.write_all(&buf).expect("Rotate 2");

            // No third connection: the receiver must give up and exit
            // with an error naming the recurrence.
            let result = replica.join().expect("replica thread panicked");
            let err = result.expect_err("second divergence must exit hard");
            assert!(
                err.contains("divergence recurred"),
                "error must name the recurrence: {err}"
            );
        }

        // -------------------------------------------------------------
        // Promotion teardown helper (`take_pipeline_for_promotion`) — the
        // no-live-pipeline decision branches. With no pipeline to tear
        // down it hands back the warm state held in the receiver's locals
        // (a clean promotion) or errors when none is present. Colocated
        // here for the `App` + `BufferedWriter` fixtures above; the
        // Clean-pipeline branch needs a live pipeline (failover IT).
        // -------------------------------------------------------------
        type PromoteHandles =
            crate::replication::ReplicaPipelineHandles<App, BufferedWriter<EvtAdd>>;

        #[test]
        fn promotion_hands_back_local_state_when_present() {
            let dir = tempfile::tempdir().unwrap();
            let mut pipeline: Option<PromoteHandles> = None;
            let mut app = Some(App);
            let mut journal_writer =
                Some(BufferedWriter::<EvtAdd>::create(&dir.path().join("p.journal")).unwrap());

            let result = crate::replication::take_pipeline_for_promotion(
                &mut pipeline,
                &mut app,
                &mut journal_writer,
            );

            assert!(matches!(result, Ok(Some(_))), "warm state must be promoted");
            assert!(
                app.is_none() && journal_writer.is_none(),
                "state must be moved into the result"
            );
        }

        #[test]
        fn promotion_errs_when_no_local_state() {
            let mut pipeline: Option<PromoteHandles> = None;
            let mut app: Option<App> = None;
            let mut journal_writer: Option<BufferedWriter<EvtAdd>> = None;

            let result = crate::replication::take_pipeline_for_promotion(
                &mut pipeline,
                &mut app,
                &mut journal_writer,
            );

            // Not `expect_err` — the Ok payload (App) isn't `Debug`.
            let err = match result {
                Err(e) => e,
                Ok(_) => panic!("a promote with nothing to promote must error"),
            };
            assert!(
                err.to_string().contains("no local state available"),
                "{err}"
            );
        }

        #[test]
        fn promotion_errs_on_partial_state() {
            // Application present, writer missing — not a usable hand-off.
            let mut pipeline: Option<PromoteHandles> = None;
            let mut app = Some(App);
            let mut journal_writer: Option<BufferedWriter<EvtAdd>> = None;

            let result = crate::replication::take_pipeline_for_promotion(
                &mut pipeline,
                &mut app,
                &mut journal_writer,
            );

            assert!(result.is_err(), "partial state is not a valid promotion");
        }
    }

    // -----------------------------------------------------------------
    // Shutdown during reconnect backoff — end to end against a scripted
    // primary. Needs real persistence (the pipeline journals the one
    // streamed event); no chain verdicts involved, so no hash-chain.
    // -----------------------------------------------------------------
    #[cfg(not(feature = "no-persist"))]
    mod backoff_shutdown {
        use super::super::super::auth::authenticate_replica;
        use super::scripted::*;
        use super::*;
        use melin_journal::{BufferedWriter, JournalEvent, JournalWrite};
        use melin_transport_core::replication::catchup::lineage_origin;
        use melin_transport_core::replication::protocol::encode_stream_start;
        use std::sync::Arc;
        use std::sync::atomic::AtomicBool;

        /// Names of every live thread in this process, via
        /// `/proc/self/task` (Linux-only, like the runtime itself).
        fn live_thread_names() -> Vec<String> {
            std::fs::read_dir("/proc/self/task")
                .expect("read /proc/self/task")
                .filter_map(|entry| {
                    // A thread may exit between readdir and the comm
                    // read — skipping it is the correct outcome.
                    let entry = entry.ok()?;
                    let name = std::fs::read_to_string(entry.path().join("comm")).ok()?;
                    Some(name.trim().to_string())
                })
                .collect()
        }

        /// Set in the child process the test below runs its body in.
        const ISOLATED: &str = "MELIN_TEST_BACKOFF_SHUTDOWN_ISOLATED";

        /// Shutdown while the receiver waits out a reconnect backoff must
        /// tear the live pipeline down — join every stage thread — rather
        /// than abandon it. Pins the fix for the connect-failure arm that
        /// returned without teardown (stages left detached mid-run, the
        /// journal writer never closed).
        ///
        /// The check reads every thread in the process by name, so it holds
        /// only where no other test shares the process: under plain `cargo
        /// test`, other tests' pipelines run stages of the same names, and
        /// the check failed on them. So the body runs in a child process of
        /// its own, this test binary filtered to this one test (nextest
        /// already gives each test a process; there it costs one re-exec).
        #[test]
        fn shutdown_during_reconnect_backoff_joins_pipeline_threads() {
            if std::env::var_os(ISOLATED).is_some() {
                return backoff_shutdown_body();
            }
            // The test's path as libtest names it: the module path without
            // the crate's name.
            let module = module_path!();
            let path = module.split_once("::").map_or(module, |(_, rest)| rest);
            let test = format!("{path}::shutdown_during_reconnect_backoff_joins_pipeline_threads");
            let output = std::process::Command::new(std::env::current_exe().expect("current_exe"))
                .args(["--exact", &test, "--test-threads=1", "--nocapture"])
                .env(ISOLATED, "1")
                .output()
                .expect("run the test in a process of its own");
            let stdout = String::from_utf8_lossy(&output.stdout);
            let stderr = String::from_utf8_lossy(&output.stderr);
            // A filter that matched nothing would also exit 0: require the
            // one test to have run and passed.
            assert!(
                output.status.success() && stdout.contains("1 passed"),
                "the isolated run failed ({}):\n{stdout}\n{stderr}",
                output.status
            );
        }

        fn backoff_shutdown_body() {
            let dir = tempfile::tempdir().expect("tempdir");

            // Minimal primary journal — only its lineage identity is
            // needed for the StreamStart the fresh replica accepts.
            let primary_journal = dir.path().join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&primary_journal).expect("create");
            w.append(&JournalEvent::App(EvtAdd(1))).expect("append");
            drop(w);
            let origin = lineage_origin(&primary_journal).expect("lineage");
            let (lineage_start, lineage_anchor, lineage_genesis) = (
                origin.starting_sequence,
                origin.anchor_hash,
                origin.genesis_entries,
            );

            let authorized_keys = replica_auth();
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let replica = {
                let journal = dir.path().join("replica.journal");
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<bool, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        3_600_000,
                        crate::layout::PipelineCores::unpinned(),
                        melin_journal::StagingMode::ZeroFill,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    // ReceiverResult's error is !Send — stringify for join().
                    .map(|state| state.is_none())
                    .map_err(|e| e.to_string())
                })
            };

            // Session 1: bring the pipeline up for real — stream one
            // event and wait for its ack (acked ⇒ journaled ⇒ the stage
            // threads exist).
            let mut s1 = accept_within(&listener, 30);
            let mut s1r = s1.try_clone().expect("clone");
            authenticate_replica(&mut s1r, &authorized_keys).expect("auth");
            match read_replica_msg(&mut s1) {
                ReplicaMessage::Handshake(h) => {
                    assert_eq!(h.last_sequence, 0, "fresh replica handshake")
                }
                other => panic!("expected Handshake, got {other:?}"),
            }
            let mut buf = Vec::new();
            encode_stream_start(
                0,
                lineage_start,
                lineage_anchor,
                lineage_genesis,
                0,
                &mut buf,
            );
            s1.write_all(&buf).expect("StreamStart");
            buf.clear();
            melin_transport_core::replication_wire::encode_input_batch(
                &[InputSlot::<EvtAdd> {
                    connection_id: 0,
                    request_id: RequestId(0),
                    key_hash: 0,
                    sequence: 1,
                    timestamp_ns: 1,
                    event: JournalEvent::App(EvtAdd(1)),
                    publish_ts: Default::default(),
                    recv_ts: Default::default(),
                }],
                &mut buf,
            )
            .expect("encode InputBatch");
            s1.write_all(&buf).expect("InputBatch");
            wait_for_ack(&mut s1, 1);

            // Kill the primary completely — both socket halves and the
            // listener — so every reconnect attempt is refused and the
            // receiver settles into the connect-failure backoff.
            drop(s1);
            drop(s1r);
            drop(listener);
            // Past the 1s post-disconnect backoff, inside the 2s
            // connect-failure backoff sleep. (Any other point in the
            // reconnect loop must satisfy the assertions too — this
            // timing just targets the arm that historically leaked.)
            std::thread::sleep(Duration::from_millis(1500));

            shutdown.store(true, Ordering::Relaxed);
            let deadline = Instant::now() + Duration::from_secs(10);
            while !replica.is_finished() {
                assert!(
                    Instant::now() < deadline,
                    "receiver did not exit after shutdown"
                );
                std::thread::sleep(Duration::from_millis(20));
            }
            let result = replica.join().expect("replica thread panicked");
            assert_eq!(result, Ok(true), "receiver must exit cleanly via shutdown");

            // The stage threads are named at spawn, and a leaked stage
            // spins forever — absence within the deadline proves the
            // teardown joined them. This process runs no other test (see
            // above), so every stage thread here is the receiver's.
            let deadline = Instant::now() + Duration::from_secs(10);
            loop {
                let names = live_thread_names();
                if !names
                    .iter()
                    .any(|n| n == "journal-seq" || n == "matching" || n == "drain")
                {
                    break;
                }
                assert!(
                    Instant::now() < deadline,
                    "pipeline threads still alive after shutdown: {names:?}"
                );
                std::thread::sleep(Duration::from_millis(50));
            }
        }
    }

    // -----------------------------------------------------------------
    // The resume point a reconnecting replica hands the primary, end to
    // end against a scripted primary. Needs real persistence: the claim
    // is the journal's durable position.
    // -----------------------------------------------------------------
    #[cfg(not(feature = "no-persist"))]
    mod reconnect_resume {
        use super::super::super::auth::authenticate_replica;
        use super::scripted::*;
        use super::*;
        use melin_journal::{BufferedWriter, JournalEvent, JournalWrite};
        use melin_transport_core::replication::catchup::lineage_origin;
        use melin_transport_core::replication::protocol::encode_stream_start;
        use std::sync::Arc;
        use std::sync::atomic::AtomicBool;

        /// A primary journal holding one entry: its path and lineage
        /// identity, for the `StreamStart` a replica accepts.
        fn primary_journal(
            dir: &std::path::Path,
        ) -> (std::path::PathBuf, melin_journal::FileHeaderInfo) {
            let path = dir.join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&path).expect("create");
            w.append(&JournalEvent::App(EvtAdd(1))).expect("append");
            drop(w);
            let lineage = lineage_origin(&path).expect("lineage");
            (path, lineage)
        }

        /// Run the receiver against `addr` on `dir/replica.journal`,
        /// under `group_commit_delay`, until `shutdown`.
        fn spawn_replica(
            dir: &std::path::Path,
            addr: std::net::SocketAddr,
            group_commit_delay: Duration,
            shutdown: &Arc<AtomicBool>,
        ) -> std::thread::JoinHandle<Result<bool, String>> {
            let journal = dir.join("replica.journal");
            let shutdown = Arc::clone(shutdown);
            std::thread::spawn(move || -> Result<bool, String> {
                run_receiver::<App>(
                    addr,
                    &store(&journal),
                    &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                    &shutdown,
                    &crate::replication::ReplicaControlPlane::new(),
                    3_600_000,
                    crate::layout::PipelineCores::unpinned(),
                    melin_journal::StagingMode::ZeroFill,
                    group_commit_delay,
                    64,
                    Arc::new(melin_transport_core::fence::FenceState::new(0)),
                    &(),
                )
                // ReceiverResult's error is !Send: stringify for join().
                .map(|state| state.is_none())
                .map_err(|e| e.to_string())
            })
        }

        /// Accept the replica's next connection, authenticate it and
        /// return its handshake with the open stream.
        fn next_handshake(
            listener: &std::net::TcpListener,
        ) -> (TcpStream, TcpStream, crate::replication::Handshake) {
            let mut s = accept_within(listener, 30);
            let mut sr = s.try_clone().expect("clone");
            authenticate_replica(&mut sr, &replica_auth()).expect("auth");
            match read_replica_msg(&mut s) {
                ReplicaMessage::Handshake(h) => (s, sr, h),
                other => panic!("expected Handshake, got {other:?}"),
            }
        }

        /// Answer a handshake: stream from `start` on the given lineage.
        fn stream_start(
            stream: &mut TcpStream,
            start: u64,
            lineage: melin_journal::FileHeaderInfo,
        ) {
            let mut buf = Vec::new();
            encode_stream_start(
                start,
                lineage.starting_sequence,
                lineage.anchor_hash,
                lineage.genesis_entries,
                0,
                &mut buf,
            );
            stream.write_all(&buf).expect("StreamStart");
        }

        /// One streamed entry at `sequence`, as the primary frames it.
        fn entry(sequence: u64) -> InputSlot<EvtAdd> {
            InputSlot::<EvtAdd> {
                connection_id: 0,
                request_id: RequestId(0),
                key_hash: 0,
                sequence,
                timestamp_ns: sequence,
                event: JournalEvent::App(EvtAdd(sequence)),
                publish_ts: Default::default(),
                recv_ts: Default::default(),
            }
        }

        fn send_entries(stream: &mut TcpStream, sequences: &[u64]) {
            let slots: Vec<_> = sequences.iter().map(|&s| entry(s)).collect();
            let mut buf = Vec::new();
            melin_transport_core::replication_wire::encode_input_batch(&slots, &mut buf)
                .expect("encode InputBatch");
            stream.write_all(&buf).expect("InputBatch");
        }

        fn stop(replica: std::thread::JoinHandle<Result<bool, String>>, shutdown: &AtomicBool) {
            shutdown.store(true, Ordering::Relaxed);
            let deadline = Instant::now() + Duration::from_secs(10);
            while !replica.is_finished() {
                assert!(
                    Instant::now() < deadline,
                    "receiver did not exit after shutdown"
                );
                std::thread::sleep(Duration::from_millis(20));
            }
            let result = replica.join().expect("replica thread panicked");
            assert_eq!(result, Ok(true), "receiver must exit cleanly via shutdown");
        }

        /// A session ends on a stream gap with two received entries still
        /// unwritten: the group-commit delay holds them past the reconnect
        /// backoff. The next handshake claims them, once written, rather
        /// than the older durable position, from which the primary would
        /// resend them and the replica would take them a second time.
        #[test]
        fn a_reconnect_claims_entries_received_before_the_drop() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (_, lineage) = primary_journal(dir.path());
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            // Longer than the 1 s backoff after a stream gap, with room
            // to spare: without the wait, the reconnect lands while
            // entries 2 and 3 are still unwritten.
            let delay = Duration::from_millis(2_500);
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                delay,
                &shutdown,
            );

            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "fresh replica handshake");
            stream_start(&mut s1, 0, lineage);
            // Entry 1 made durable first, so the durable position the
            // handshake could wrongly claim is 1, not the fresh zero.
            send_entries(&mut s1, &[1]);
            wait_for_ack(&mut s1, 1);
            // Entries 2 and 3 are taken, then 5 breaks contiguity: the
            // session ends without waiting for the ones it took.
            send_entries(&mut s1, &[2, 3, 5]);

            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(
                h2.last_sequence, 3,
                "the reconnect must claim the entries received before the drop"
            );
            // Streaming again, so shutdown finds the receiver in a session
            // rather than waiting on the primary's answer.
            stream_start(&mut s2, h2.last_sequence, lineage);
            stop(replica, &shutdown);
        }

        /// A failed snapshot transfer backs off before redialing, and the
        /// socket must be closed before that sleep, not after it: held
        /// through the backoff, it ties up a primary slot on a dead session.
        #[test]
        fn a_failed_resync_transfer_closes_the_socket_before_backing_off() {
            use melin_transport_core::replication::protocol::encode_need_snapshot;

            let dir = tempfile::tempdir().expect("tempdir");
            let (_, lineage) = primary_journal(dir.path());
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                Duration::ZERO,
                &shutdown,
            );

            let (mut s1, _s1r, _) = next_handshake(&listener);
            // NeedSnapshot, then anything but the SnapshotBegin the
            // transfer expects: the transfer fails and the receiver retries.
            let mut buf = Vec::new();
            encode_need_snapshot(&mut buf);
            encode_need_snapshot(&mut buf);
            s1.write_all(&buf).expect("NeedSnapshot x2");

            // The first reconnect backoff is 1 s; the close must land well
            // inside it.
            s1.set_read_timeout(Some(Duration::from_millis(700)))
                .expect("read timeout");
            let mut byte = [0u8; 1];
            match s1.read(&mut byte) {
                Ok(0) => {}
                Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => {}
                other => panic!("socket still open during the backoff: {other:?}"),
            }

            // The retry dials again, as a fresh replica.
            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(h2.last_sequence, 0, "retry handshakes as fresh");
            stream_start(&mut s2, 0, lineage);
            stop(replica, &shutdown);
        }

        /// The journal stage dies after a session has ended, while the
        /// reconnect waits for it to cover the ring: entries 2 and 3 are
        /// held by the group-commit delay past the backoff that follows a
        /// stream gap, and the chain check announced between them carries
        /// a hash the replica's chain cannot match, so the stage fails
        /// before any of them is handed to the disk. The loop must take
        /// the fatal path there (here, the in-process divergence resync)
        /// before dialing, so the next session runs on the pipeline
        /// rebuilt from disk and journals what it is sent. Dialing with
        /// the dead pipeline instead would claim the same position (entry
        /// 1 was durable before the failure) but hand the session a
        /// journal stage that can take nothing: it would exit at once and
        /// never acknowledge entry 2.
        #[cfg(feature = "hash-chain")]
        #[test]
        fn a_journal_that_dies_while_disconnected_is_torn_down_before_dialing() {
            use melin_transport_core::replication::protocol::encode_chain_check;

            let dir = tempfile::tempdir().expect("tempdir");
            let (_, lineage) = primary_journal(dir.path());
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            // Longer than the 1 s backoff after a stream gap, with room
            // to spare, so the failure lands inside the reconnect's wait.
            let delay = Duration::from_millis(2_500);
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                delay,
                &shutdown,
            );

            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "fresh replica handshake");
            stream_start(&mut s1, 0, lineage);
            send_entries(&mut s1, &[1]);
            wait_for_ack(&mut s1, 1);
            // Entry 2, a chain check at 2 no replica chain can match,
            // entry 3, then 5: the gap ends the session at once, with 2
            // and 3 still unwritten.
            send_entries(&mut s1, &[2]);
            let mut buf = Vec::new();
            encode_chain_check(2, &[0xEE; 32], &mut buf);
            s1.write_all(&buf).expect("ChainCheck");
            send_entries(&mut s1, &[3]);
            send_entries(&mut s1, &[5]);

            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(
                h2.last_sequence, 1,
                "the first dial after the journal died claims what is on disk"
            );
            // The discriminating step: only a rebuilt pipeline journals
            // and acknowledges entry 2 (after the group-commit delay).
            // The dead one would end the session unacknowledged, failing
            // the read below on EOF or its timeout.
            stream_start(&mut s2, h2.last_sequence, lineage);
            send_entries(&mut s2, &[2]);
            wait_for_ack(&mut s2, 2);
            stop(replica, &shutdown);
        }

        /// A replica built over a journal that already holds an entry
        /// reconnects after a session that journaled nothing (a quiet
        /// primary). Its handshake claims that journal, exactly as the
        /// first one did, not an empty one.
        #[test]
        fn a_reconnect_after_a_quiet_session_claims_the_recovered_journal() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (primary, lineage) = primary_journal(dir.path());
            // The replica holds what the primary holds.
            std::fs::copy(&primary, dir.path().join("replica.journal")).expect("copy");
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                Duration::ZERO,
                &shutdown,
            );

            let (mut s1, s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 1, "recovered replica handshake");
            stream_start(&mut s1, 1, lineage);
            // Nothing streamed: the primary goes away.
            drop(s1);
            drop(s1r);

            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(
                (h2.last_sequence, h2.chain_hash),
                (h1.last_sequence, h1.chain_hash),
                "the reconnect must claim the journal the pipeline was built over"
            );
            // Streaming again, so shutdown finds the receiver in a session.
            stream_start(&mut s2, h2.last_sequence, lineage);
            stop(replica, &shutdown);
        }

        /// The audit's scenario end to end: an entry damaged on the way
        /// still decodes — as a different event — and its CRC is the only
        /// thing that can tell. The replica must refuse it before applying,
        /// journaling or acknowledging it, drop the session, and reconnect
        /// claiming only what it verified; the intact resend is what lands
        /// in its journal.
        #[test]
        fn a_damaged_entry_is_refused_and_fetched_again() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (_, lineage) = primary_journal(dir.path());
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                Duration::ZERO,
                &shutdown,
            );

            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "fresh replica handshake");
            stream_start(&mut s1, 0, lineage);
            send_entries(&mut s1, &[1]);
            wait_for_ack(&mut s1, 1);

            // Entry 2 with one payload bit flipped: `EvtAdd(2)` arrives as
            // a perfectly decodable `EvtAdd(3)`.
            let mut buf = Vec::new();
            melin_transport_core::replication_wire::encode_input_batch(&[entry(2)], &mut buf)
                .expect("encode InputBatch");
            let payload_at = buf.len() - melin_journal::codec::CRC_SIZE - 8;
            buf[payload_at] ^= 0x01;
            s1.write_all(&buf).expect("damaged InputBatch");

            // Nothing acknowledges entry 2 — not in memory, not on disk —
            // before the replica drops the session.
            loop {
                let mut len_buf = [0u8; 4];
                if std::io::Read::read_exact(&mut s1, &mut len_buf).is_err() {
                    break;
                }
                let mut payload = vec![0u8; u32::from_le_bytes(len_buf) as usize];
                if std::io::Read::read_exact(&mut s1, &mut payload).is_err() {
                    break;
                }
                if let Ok(ReplicaMessage::Ack(a)) = decode_replica_message(&payload) {
                    assert!(
                        a.in_memory_sequence < 2 && a.acked_sequence < 2,
                        "the damaged entry was acknowledged: {a:?}"
                    );
                }
            }

            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(
                h2.last_sequence, 1,
                "the reconnect must claim only the entries that passed the check"
            );
            stream_start(&mut s2, 1, lineage);
            send_entries(&mut s2, &[2]);
            wait_for_ack(&mut s2, 2);
            stop(replica, &shutdown);

            let mut reader =
                melin_journal::JournalReader::<EvtAdd>::open(&dir.path().join("replica.journal"))
                    .expect("open replica journal");
            let mut journaled = Vec::new();
            while let Some(e) = reader.next_entry().expect("read entry") {
                if let JournalEvent::App(EvtAdd(v)) = e.event {
                    journaled.push(v);
                }
            }
            assert_eq!(journaled, vec![1, 2], "only the intact entry is journaled");
        }

        /// `lineage`, as a primary whose genesis took `genesis_entries`
        /// entries would advertise it.
        fn with_genesis(
            lineage: melin_journal::FileHeaderInfo,
            genesis_entries: Option<u64>,
        ) -> melin_journal::FileHeaderInfo {
            melin_journal::FileHeaderInfo {
                genesis_entries,
                ..lineage
            }
        }

        fn replica_genesis(dir: &std::path::Path) -> Option<u64> {
            melin_journal::segment::read_header_info(&dir.join("replica.journal"))
                .expect("replica journal header")
                .genesis_entries
        }

        /// A fresh replica writes the genesis length the primary
        /// advertises into the journal it creates, and keeps it: a
        /// restarted replica recovers it from its own header and follows
        /// a primary advertising the same.
        #[test]
        fn a_replica_learns_the_genesis_length_and_keeps_it_across_restarts() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (_, base) = primary_journal(dir.path());
            let lineage = with_genesis(base, Some(3));
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(dir.path(), addr, Duration::ZERO, &shutdown);
            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "fresh replica handshake");
            stream_start(&mut s1, 0, lineage);
            send_entries(&mut s1, &[1, 2]);
            wait_for_ack(&mut s1, 2);
            assert_eq!(replica_genesis(dir.path()), Some(3), "learned at creation");
            stop(replica, &shutdown);

            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(dir.path(), addr, Duration::ZERO, &shutdown);
            let (mut s2, _s2r, h2) = next_handshake(&listener);
            assert_eq!(h2.last_sequence, 2, "restarted replica handshake");
            stream_start(&mut s2, 2, lineage);
            send_entries(&mut s2, &[3]);
            wait_for_ack(&mut s2, 3);
            stop(replica, &shutdown);
            assert_eq!(
                replica_genesis(dir.path()),
                Some(3),
                "kept across a restart"
            );
        }

        /// A replica holding an empty journal from before the genesis
        /// length was recorded — copied from a primary that stopped
        /// before journaling anything — learns the length when the
        /// upgraded primary, having journaled its genesis under the same
        /// anchor, advertises it. Left unknown, its promotion check would
        /// be skipped for the very partial genesis it guards.
        #[test]
        fn an_empty_old_format_replica_journal_learns_the_genesis_length() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (_, base) = primary_journal(dir.path());
            assert_eq!(base.starting_sequence, 1);
            drop(
                melin_journal::BufferedWriter::<EvtAdd>::create_continuing(
                    &dir.path().join("replica.journal"),
                    1,
                    base.anchor_hash,
                    None,
                )
                .expect("old-format empty replica journal"),
            );
            assert_eq!(replica_genesis(dir.path()), None);
            let lineage = with_genesis(base, Some(3));
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                Duration::ZERO,
                &shutdown,
            );

            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "empty replica handshake");
            stream_start(&mut s1, 0, lineage);
            send_entries(&mut s1, &[1, 2]);
            wait_for_ack(&mut s1, 2);
            stop(replica, &shutdown);
            assert_eq!(replica_genesis(dir.path()), Some(3));
        }

        /// A replica whose journal records one genesis length refuses a
        /// primary advertising another — a lineage error, never an
        /// overwrite — and exits with the reason.
        #[test]
        fn a_primary_advertising_another_genesis_length_is_refused() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (primary, lineage) = primary_journal(dir.path());
            assert_eq!(lineage.genesis_entries, Some(0));
            // The replica holds what the primary holds, genesis length
            // included.
            std::fs::copy(&primary, dir.path().join("replica.journal")).expect("copy");
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let shutdown = Arc::new(AtomicBool::new(false));
            let replica = spawn_replica(
                dir.path(),
                listener.local_addr().expect("addr"),
                Duration::ZERO,
                &shutdown,
            );

            let (mut s1, _s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 1);
            stream_start(&mut s1, 1, with_genesis(lineage, Some(4)));
            let result = replica.join().expect("replica thread panicked");
            let err = result.expect_err("the replica must refuse the lineage");
            assert!(err.contains("refusing to follow the primary"), "{err}");
            assert_eq!(replica_genesis(dir.path()), Some(0), "never overwritten");
        }

        /// The protected case end to end: a brand-new cluster whose
        /// primary dies while its first replica is still copying the
        /// genesis. The replica learned the genesis length from the
        /// primary — it is configured with none (the receiver takes no
        /// genesis at all) — and the promotion the operator requests next
        /// is refused on it.
        #[test]
        fn a_replica_promoted_mid_genesis_is_refused() {
            let dir = tempfile::tempdir().expect("tempdir");
            let (_, base) = primary_journal(dir.path());
            let lineage = with_genesis(base, Some(3));
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");
            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let replica = {
                let journal = dir.path().join("replica.journal");
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<Option<String>, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        3_600_000,
                        crate::layout::PipelineCores::unpinned(),
                        melin_journal::StagingMode::ZeroFill,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    // The promotion path's check, on the writer the
                    // receiver hands over.
                    .map(|promoted| {
                        promoted.map(
                            |(_, writer)| match crate::server::check_promotable(&writer) {
                                Ok(()) => "promotable".to_string(),
                                Err(e) => e.to_string(),
                            },
                        )
                    })
                    .map_err(|e| e.to_string())
                })
            };

            let (mut s1, s1r, h1) = next_handshake(&listener);
            assert_eq!(h1.last_sequence, 0, "fresh replica handshake");
            stream_start(&mut s1, 0, lineage);
            send_entries(&mut s1, &[1, 2]);
            wait_for_ack(&mut s1, 2);
            // The primary dies two entries into a three-entry genesis.
            drop(s1);
            drop(s1r);
            assert!(
                control
                    .promote
                    .request(crate::promotion::PromotionRequest::MANUAL),
                "promotion filed"
            );

            let verdict = replica
                .join()
                .expect("replica thread panicked")
                .expect("the receiver hands its state over")
                .expect("promoted, not shut down");
            assert!(verdict.contains("refusing promotion"), "{verdict}");
            assert!(verdict.contains("holds 2 entries"), "{verdict}");
        }
    }

    // -----------------------------------------------------------------
    // Boundary adoption under both staging modes — end to end against a
    // scripted primary. Needs hash-chain (the boundary announce carries
    // a chain value the replica verifies before rotating) and real
    // persistence (the rotation archives the on-disk segment).
    // -----------------------------------------------------------------
    #[cfg(all(feature = "hash-chain", not(feature = "no-persist")))]
    mod rotation_adoption {
        use super::super::super::auth::authenticate_replica;
        use super::scripted::*;
        use super::*;
        use melin_journal::{BufferedWriter, JournalEvent, JournalWrite};
        use melin_transport_core::replication::catchup::lineage_origin;
        use melin_transport_core::replication::protocol::{encode_rotate, encode_stream_start};
        use std::sync::Arc;
        use std::sync::atomic::AtomicBool;

        /// Payloads of every entry in the journal at `path`, in order.
        fn payloads(path: &std::path::Path) -> Vec<u64> {
            let mut reader = melin_journal::JournalReader::<EvtAdd>::open(path).expect("open");
            let mut out = Vec::new();
            while let Some(entry) = reader.next_entry().expect("read entry") {
                if let JournalEvent::App(EvtAdd(v)) = entry.event {
                    out.push(v);
                }
            }
            out
        }

        /// A scripted primary streams two events, announces the rotation
        /// boundary with the *correct* tail hash, and streams two more.
        /// The replica must verify the boundary, rotate — adopting the
        /// segment its preparer staged in `mode` — and keep acking on
        /// the new segment, leaving the sealed one archived.
        ///
        /// Run under both modes: everything observable (archive layout,
        /// entry placement across the boundary, acks, clean shutdown)
        /// must be identical, or the staging mode is not a tuning knob
        /// but a behaviour change.
        fn boundary_adoption_rotates_and_streams_on(mode: melin_journal::StagingMode) {
            // Shrink the prealloc chunk so the replica's preparer stages
            // a few MiB (chunk + margin) instead of 256 MiB per test.
            // The guard holds a process-wide lock, so the two mode
            // variants serialise rather than race the global override.
            let _prealloc_guard =
                melin_journal::test_utils::PreallocOverrideGuard::new(1024 * 1024);
            let dir = tempfile::tempdir().expect("tempdir");

            // Scripted primary journal: entries 1..=2 carrying the same
            // stamps as the InputSlots below (deterministic timestamps,
            // zero key-hash/request-seq), so its chain value at the
            // boundary equals the replica's — journals are byte-for-byte
            // mirrors, and `append`'s wall-clock stamp would fork them.
            let primary_journal = dir.path().join("primary.journal");
            let mut w = BufferedWriter::<EvtAdd>::create(&primary_journal).expect("create");
            for v in 1..=2u64 {
                w.batch_append_with_ts(&JournalEvent::App(EvtAdd(v)), v, 0)
                    .expect("append");
            }
            w.flush_batch_sync().expect("flush");
            let chain_at_2 = w.chain_hash().expect("chain");
            drop(w);
            let origin = lineage_origin(&primary_journal).expect("lineage");
            let (lineage_start, lineage_anchor, lineage_genesis) = (
                origin.starting_sequence,
                origin.anchor_hash,
                origin.genesis_entries,
            );

            let authorized_keys = replica_auth();
            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
            listener.set_nonblocking(true).expect("nonblocking");
            let addr = listener.local_addr().expect("addr");

            let replica_journal = dir.path().join("replica.journal");
            let shutdown = Arc::new(AtomicBool::new(false));
            let control = crate::replication::ReplicaControlPlane::new();
            let replica = {
                let journal = replica_journal.clone();
                let shutdown = Arc::clone(&shutdown);
                let control = control.clone();
                std::thread::spawn(move || -> Result<bool, String> {
                    run_receiver::<App>(
                        addr,
                        &store(&journal),
                        &ed25519_dalek::SigningKey::from_bytes(&REPLICA_KEY),
                        &shutdown,
                        &control,
                        3_600_000,
                        crate::layout::PipelineCores::unpinned(),
                        mode,
                        Duration::ZERO,
                        64,
                        Arc::new(melin_transport_core::fence::FenceState::new(0)),
                        &(),
                    )
                    // ReceiverResult's error is !Send — stringify for join().
                    .map(|state| state.is_none())
                    .map_err(|e| e.to_string())
                })
            };

            let mut s1 = accept_within(&listener, 30);
            let mut s1r = s1.try_clone().expect("clone");
            authenticate_replica(&mut s1r, &authorized_keys).expect("auth");
            match read_replica_msg(&mut s1) {
                ReplicaMessage::Handshake(h) => {
                    assert_eq!(h.last_sequence, 0, "fresh replica handshake")
                }
                other => panic!("expected Handshake, got {other:?}"),
            }
            let mut buf = Vec::new();
            encode_stream_start(
                0,
                lineage_start,
                lineage_anchor,
                lineage_genesis,
                0,
                &mut buf,
            );
            s1.write_all(&buf).expect("StreamStart");

            let send_events = |s: &mut TcpStream, range: std::ops::RangeInclusive<u64>| {
                let slots: Vec<InputSlot<EvtAdd>> = range
                    .map(|v| InputSlot {
                        connection_id: 0,
                        request_id: RequestId(0),
                        key_hash: 0,
                        sequence: v,
                        timestamp_ns: v,
                        event: JournalEvent::App(EvtAdd(v)),
                        // `()` without latency-trace, a timestamp with it.
                        publish_ts: Default::default(),
                        recv_ts: Default::default(),
                    })
                    .collect();
                let mut buf = Vec::new();
                melin_transport_core::replication_wire::encode_input_batch(&slots, &mut buf)
                    .expect("encode InputBatch");
                s.write_all(&buf).expect("InputBatch");
            };

            send_events(&mut s1, 1..=2);
            wait_for_ack(&mut s1, 2);

            // Wait for the preparer's staged segment to reach full size
            // (the overridden prealloc chunk + the 8 MiB margin) so the
            // boundary below adopts it rather than racing staging into
            // the synchronous fallback — adoption is the path under
            // test. Both modes must produce it; a preparer that never
            // does is itself a failure.
            let staging = {
                // `<live>.next-staging` — the preparer's sibling-file
                // convention (crate-private, so spelled out here the way
                // the `.divergent.<n>` archive name is above).
                let mut s = replica_journal.clone().into_os_string();
                s.push(".next-staging");
                std::path::PathBuf::from(s)
            };
            let staged_len: u64 = (1 + 8) * 1024 * 1024;
            let deadline = Instant::now() + Duration::from_secs(10);
            loop {
                if std::fs::metadata(&staging).is_ok_and(|m| m.len() == staged_len) {
                    break;
                }
                assert!(
                    Instant::now() < deadline,
                    "preparer produced no full-size staged segment in {mode:?} mode"
                );
                std::thread::sleep(Duration::from_millis(10));
            }
            // Length is the last thing staging changes before the final
            // sync and the slot hand-off; give those a moment so the
            // rotation's `take()` finds the segment parked. (Losing this
            // race anyway is not a failure: the rotation falls back to
            // the synchronous path and every assertion below still
            // holds — the staged segment is then simply adopted at a
            // later boundary.)
            std::thread::sleep(Duration::from_millis(300));

            // The boundary, with the correct tail hash — verified
            // against the replica's own chain, then adopted.
            buf.clear();
            encode_rotate(2, &chain_at_2, &mut buf);
            s1.write_all(&buf).expect("Rotate");

            // Streaming continues on the new segment. The journal stage
            // holds entries past a pending boundary until the rotation
            // applies, so this ack proves the adoption completed.
            send_events(&mut s1, 3..=4);
            wait_for_ack(&mut s1, 4);

            shutdown.store(true, Ordering::Relaxed);
            let result = replica.join().expect("replica thread panicked");
            assert_eq!(result, Ok(true), "receiver must exit cleanly via shutdown");

            // The sealed segment was archived at the first archive slot
            // holding exactly the pre-boundary entries; the live segment
            // carries the stream from the boundary on.
            let archived = dir.path().join("replica.journal.000001");
            assert!(archived.exists(), "sealed segment must be archived");
            assert_eq!(payloads(&archived), vec![1, 2]);
            assert_eq!(payloads(&replica_journal), vec![3, 4]);
        }

        #[test]
        fn boundary_adoption_zero_fill() {
            boundary_adoption_rotates_and_streams_on(melin_journal::StagingMode::ZeroFill);
        }

        #[test]
        fn boundary_adoption_allocate() {
            boundary_adoption_rotates_and_streams_on(melin_journal::StagingMode::Allocate);
        }
    }
}