melin-server-runtime 0.22.0

//! DPDK replication transport — smoltcp-based sender and receiver paths.
//!
//! Mirrors the kernel-TCP variants in `mod.rs` but uses `DpdkTransport`
//! (a `smoltcp` socket over DPDK queue pairs) instead of `TcpStream`.
//! The wire protocol is identical — see `protocol.rs` for the message catalogue.

#![cfg(feature = "dpdk")]

use std::io;
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tracing::{debug, error, info, warn};

use melin_app::Application;
use melin_journal::BufferedWriter;
use melin_journal::replication::ReplicationConsumer;

use super::auth::{
    AuthChallenge, AuthOutcome, AuthTransport, PolledAuthStream, authenticate_with_primary,
    generate_challenge_nonce, step_authentication,
};
use super::join_worker::{
    JoinContext, JoinRequest, JoinStallWatch, JoinStream, JoinWorker, Pump, stream_join,
};
use super::receiver_transport::{
    ControlFrameSource, FrameResult, ReceiverTransport, compact_recv_buf, run_serviced,
    streaming_loop, try_extract_frame,
};
use super::validation_worker::ValidationWorker;
use super::{
    ReceiverResult, ReplicaCursors, ReplicaPipelineHandles, ReplicationMetrics, ResumePoint,
    ResyncDecision, SentHighWater, 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 melin_app::auth::AuthorizedKeys;
use melin_transport_core::node_store::NodeStore;
use melin_transport_core::replica_presence::{FirstAck, ReplicaPresence};
use melin_transport_core::replication::PersistWatch;
use melin_transport_core::replication::catchup::CatchUpPublisher;
use melin_transport_core::replication::handoff::{
    HandoffIo, HandoffStep, LiveHandoff, PassProgress,
};
use melin_transport_core::replication::protocol::{
    Ack, Handshake, MAX_CONTROL_FRAME, PrimaryMessage, ReplicaMessage, decode_primary_message,
    decode_replica_message, encode_ack, encode_challenge, encode_handshake, encode_heartbeat,
};
use melin_transport_core::replication::validate::{
    HandshakeValidation, validate_replica_handshake_settled,
};

// ---------------------------------------------------------------------------
// DPDK ReceiverTransport implementation
// ---------------------------------------------------------------------------

/// DPDK/smoltcp-backed receiver transport for the replica side.
///
/// Wraps a `DpdkTransport` + `SocketHandle` and implements
/// [`ReceiverTransport`] so the generic [`streaming_loop`] can drive it
/// identically to the kernel io_uring path.
struct DpdkReceiverTransport<'a> {
    transport: &'a mut melin_dpdk::DpdkTransport,
    handle: melin_dpdk::SocketHandle,
    send_buf: Vec<u8>,
}

const ACK_RETRY_CAP: u32 = 32;

/// Data segments a replication socket may dispatch per egress pass.
///
/// The stack repeats its egress pass over every socket until one sends
/// nothing, so with the fan-in default of 4 a full replication window costs
/// roughly a pass per 4 segments — each pass visiting every socket on the
/// interface. 256 covers a whole in-flight window (the peer's advertised
/// window is itself bounded by its zero-copy descriptor slots, 256 of them) in
/// a single pass. The fairness the low default buys does not apply here: the
/// bulk flow and the latency-sensitive flows are on separate ports, and
/// replication latency is itself on the client's critical path through the
/// durability gate.
pub(crate) const REPL_DISPATCH_BURST: usize = 256;

/// Soft cap on the bytes one replication slot hands to the transport per tick.
///
/// Sized to roughly one in-flight window ([`REPL_DISPATCH_BURST`] segments at
/// a 1500-byte MTU), so a tick queues about what one egress pass can ship
/// rather than several passes' worth in one stall.
const MAX_REPL_BYTES_PER_TICK: usize = 128 * 1024;

impl ReceiverTransport for DpdkReceiverTransport<'_> {
    fn poll_recv(&mut self, recv_buf: &mut Vec<u8>) -> io::Result<bool> {
        self.transport.poll();
        let before = recv_buf.len();
        self.transport.recv_into_vec(self.handle, recv_buf);
        Ok(recv_buf.len() > before)
    }

    fn send_ack(&mut self, ack: &Ack) -> io::Result<bool> {
        self.send_buf.clear();
        encode_ack(ack, &mut self.send_buf);
        let mut attempts: u32 = 0;
        loop {
            if self.transport.queue_send(self.handle, &self.send_buf) {
                // Flush immediately so the ACK reaches the primary
                // without waiting for the next poll_recv iteration.
                self.transport.poll();
                return Ok(true);
            }
            attempts += 1;
            if attempts >= ACK_RETRY_CAP {
                tracing::warn!("DPDK ack send failed after {ACK_RETRY_CAP} retries");
                return Ok(false);
            }
            self.transport.poll();
            if !self.transport.is_connected(self.handle) {
                return Err(io::Error::other("replica disconnected during ack send"));
            }
        }
    }

    fn ack_in_flight(&self) -> bool {
        false
    }

    /// Both halves open: the primary's FIN ends the session (once the
    /// bytes before it are read) as its RST or the liveness deadline
    /// does, as EOF ends it on kernel TCP. See [`REPLICATION_LIVENESS`].
    fn is_connected(&mut self) -> bool {
        self.transport.is_connected(self.handle)
    }

    /// Run the stack without reading: it acknowledges what arrives (the
    /// window closing as the socket buffer fills, as a kernel's would)
    /// and answers the primary's probes, so its deadline never fires on
    /// a replica that is only waiting on its own journal.
    fn keep_link_serviced(&mut self) {
        self.transport.poll();
    }
}

/// When a replication peer counts as gone, at both ends of a DPDK link.
///
/// A kernel-TCP node's peer learns that it stopped from its kernel (EOF
/// or a reset), even when its process crashed. A DPDK node is its own TCP
/// stack, so a crashed or killed one says nothing, and without a deadline
/// its peer would wait on it for ever: a primary would keep counting a
/// replica that left (never halting, as it must when its last replica
/// leaves), and a replica would keep its link to a dead primary up, which
/// vetoes auto-promotion. Every replication socket is armed with this rule
/// once established (keep-alive probes, reset after the peer's silence;
/// see `melin_dpdk::PeerLiveness`), and closed with [`melin_dpdk::DpdkTransport::reset`]
/// so that a node that stops, or drops a link, tells its peer at once.
const REPLICATION_LIVENESS: melin_dpdk::PeerLiveness = melin_dpdk::PeerLiveness::REPLICATION;

/// How long a joining replica that has acked may leave the primary with
/// bytes for it and no room to queue them — a frame its socket refused,
/// with nothing taken since — before the primary drops it (see
/// [`JoinStallWatch`]).
///
/// Only once the replica has acked: before its first ack it may be
/// installing a snapshot, reading nothing for as long as its state takes
/// to load, and no bound on that is right for every state. Before it, a
/// join waits on the replica as the kernel-TCP sender's always does, with
/// no deadline of its own (a replica that has gone is still dropped by its
/// liveness deadline, a loaded primary's handoff still by its ring), and
/// at no cost to anyone else: neither phase of the join holds the poll
/// thread.
///
/// The liveness timeout, because to the join a replica that has taken
/// nothing for that long, past its install, is no better than one whose
/// stack has stopped answering, which the liveness deadline drops after
/// the same silence: its slot and its place in the halt gate are given
/// back, and it reconnects and resumes from what it journaled. A replica
/// alive but not reading acknowledges with a zero window, so the liveness
/// deadline itself never fires on it, and without this a join would hold
/// the slot for as long as the replica chose.
///
/// The rule the primary already applies to a streaming replica that stops
/// reading, on both transports, is eviction once its replication ring
/// fills. That rule still applies where the ring is active — the handoff,
/// once it has activated the ring — and under load fires first, acked or
/// not. This deadline covers what it cannot: the catch-up, which runs with
/// the ring inactive, and a handoff on an idle primary, whose ring never
/// fills.
const JOIN_STALL_LIMIT: std::time::Duration = REPLICATION_LIVENESS.timeout();

/// How long an end of a replication link that is closing it waits for
/// what it has already queued (a replica's final ack on shutdown and
/// promotion, a stopping primary's last stream frames) to be acknowledged
/// by the peer's stack before resetting the link, which would discard
/// it — the part of a kernel's flush before its FIN that matters here.
/// Bounded so a stop never waits on a peer that has gone; on a live link
/// it is one round trip.
const SESSION_END_LINGER: std::time::Duration = std::time::Duration::from_millis(100);

/// Close replication links and tell their peers: let the bytes queued on
/// them reach the peers (all links together, at most
/// [`SESSION_END_LINGER`]), then reset each so the peer ends the session at
/// once, and the local socket-set entry goes with it.
fn linger_then_reset(
    transport: &mut melin_dpdk::DpdkTransport,
    handles: &[melin_dpdk::SocketHandle],
) {
    let deadline = std::time::Instant::now() + SESSION_END_LINGER;
    while std::time::Instant::now() < deadline
        && handles
            .iter()
            .any(|&h| transport.is_active(h) && !transport.tx_drained(h))
    {
        transport.poll();
        std::thread::yield_now();
    }
    for &handle in handles {
        transport.reset(handle);
    }
}

/// What reading a joining replica's control frames found.
#[derive(Debug, PartialEq, Eq)]
enum JoinFrames {
    /// A frame's length prefix is zero or over [`MAX_CONTROL_FRAME`]: a
    /// replica the caller disconnects, as the streaming arm does.
    Malformed,
    /// Well formed. `acked`: an ack is kept: the replica is in its
    /// streaming loop, which arms the join's [`JoinStallWatch`].
    Read { acked: bool },
}

/// Keep only the newest complete ack at the front of `buf`, and a trailing
/// partial frame for the next read. `acked` whenever an ack is kept,
/// which, as one is kept across reads, is true again on every read after
/// the first (arming the stall watch is idempotent).
///
/// Serves a `Joining` slot, in both phases. The replica acks what it
/// journals of the join, and its acks must be read off the socket (left
/// unread they fill the listener's receive buffer, and the replica's acks
/// then back up in its own TX queue), but not recorded yet: the slot's
/// cursors are seeded only once the catch-up ends, and the sent high-water
/// an ack is checked against only once the handoff is over (a journal
/// pass's entries are counted when the pass ends, and the replica may ack
/// them before). The streaming arm records what is kept here, on its first
/// tick. Acks are cumulative, so the newest carries everything the older
/// ones did, and the rest are dropped.
///
/// Kept, not dropped, through the catch-up too: a replica acks only when
/// it holds something new, so its last ack of the join is never repeated.
/// On a quiet primary nothing more is journaled, and an ack dropped here
/// would leave the slot at its handshake position, with a write held for
/// the replica's ack never answered.
fn keep_last_ack(buf: &mut Vec<u8>) -> JoinFrames {
    let mut consumed = 0;
    // The newest ack's byte range in `buf`.
    let mut newest: Option<(usize, usize)> = None;
    loop {
        match try_extract_frame(&buf[consumed..], MAX_CONTROL_FRAME) {
            FrameResult::Complete(payload_start, frame_end) => {
                let payload = &buf[consumed + payload_start..consumed + frame_end];
                if matches!(decode_replica_message(payload), Ok(ReplicaMessage::Ack(_))) {
                    newest = Some((consumed, consumed + frame_end));
                }
                consumed += frame_end;
            }
            FrameResult::Incomplete => break,
            FrameResult::Oversized => return JoinFrames::Malformed,
        }
    }
    let acked = newest.is_some();
    match newest {
        Some((start, end)) => {
            let len = buf.len();
            let ack_len = end - start;
            buf.copy_within(start..end, 0);
            // `ack_len <= end <= consumed`: the partial frame moves down.
            buf.copy_within(consumed..len, ack_len);
            buf.truncate(ack_len + (len - consumed));
        }
        None => compact_recv_buf(buf, consumed),
    }
    JoinFrames::Read { acked }
}

/// End a replica's session with its primary (see [`linger_then_reset`]):
/// the primary's slot frees at once.
fn end_session(transport: &mut melin_dpdk::DpdkTransport, handle: melin_dpdk::SocketHandle) {
    linger_then_reset(transport, &[handle]);
}

/// Per-slot state for the DPDK replication sender.
enum SlotState {
    /// No replica connected on this slot.
    Idle,
    /// Replica connected; running the Ed25519 challenge/response before the
    /// handshake. Non-blocking, like `Handshaking` — the poll thread also
    /// serves client traffic and the other slot, so we never block on the
    /// replica's response (see `AuthChallenge`).
    Authenticating(melin_dpdk::SocketHandle),
    /// Replica authenticated, performing handshake.
    Handshaking(melin_dpdk::SocketHandle),
    /// Handshake validated; the slot's join worker reads what the replica
    /// lacks (journal history, or a snapshot and history) off the disk
    /// while the poll thread moves its frames into the socket a tick at a
    /// time, then the handoff into the live ring, likewise a tick at a
    /// time (see [`PendingJoin`]).
    Joining(melin_dpdk::SocketHandle),
    /// Streaming journal data to replica. Like `Joining`, counted by the
    /// halt gate: it carries the join's [`FirstAck`] (from
    /// [`ReplicaPresence::lift`]), so the slot cannot stream uncounted, and
    /// its first ack ends any halt override.
    Streaming(melin_dpdk::SocketHandle, FirstAck),
}

/// Deadline for one side of the DPDK auth exchange to make progress. On the
/// sender it bounds how long a connected-but-silent replica may hold a slot
/// before we reclaim it; on the receiver it bounds the wait for the
/// primary's challenge/result. Enforced across poll ticks (sender) or via the
/// blocking adapter's deadline (receiver), never by stalling the poll thread.
const AUTH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// In-flight handshake chain validation, running on this slot's parked
/// [`ValidationWorker`]. `validate_replica_handshake_settled` can scan a
/// full segment per attempt and sleeps between retries (~400 ms budget
/// on a divergent verdict) — far too long to run inline on the poll
/// thread, which also services client traffic and the other replica
/// slot. The slot stays `Handshaking` and polls `verdict_rx` each tick.
struct PendingValidation {
    /// The handshake that triggered validation — consumed by the
    /// post-verdict catch-up/resync flow.
    handshake: Handshake,
    /// One-shot verdict channel; the sender half lives on the
    /// validation thread.
    verdict_rx: std::sync::mpsc::Receiver<io::Result<HandshakeValidation>>,
}

/// A replica's join in progress, its disk side on this slot's parked
/// [`JoinWorker`]. The probe, the snapshot pre-flight, the snapshot and
/// seed reads and the journal history are file I/O that a large state on
/// a slow or cold disk can stretch past the replication liveness
/// deadline; inline on the poll thread they stalled client traffic and
/// the other slot, and left every replication link unanswered long
/// enough for healthy replicas to reset theirs. The slot stays `Joining`
/// and pumps the worker's frames into its socket each tick, then steps
/// the handoff into the live ring each tick.
///
/// Neither phase ever waits on the replica: a joiner that stops reading
/// costs the poll thread nothing, and, once it has acked, is dropped when
/// it has refused what the join has to send for [`JOIN_STALL_LIMIT`].
struct PendingJoin {
    phase: JoinPhase,
    /// The cursor and stream floor the handoff seeds: the handshake's
    /// position, or 0 for a divergent replica.
    stream_base: u64,
    /// Whether the replica is draining the join, once it has acked.
    stall: JoinStallWatch,
    /// The halt gate's count of this replica, raised as the join began,
    /// carried into `Streaming`: a join cannot exist uncounted.
    first_ack: FirstAck,
}

/// The two halves of a join.
enum JoinPhase {
    /// The worker's join frames, moved into the socket a tick at a time.
    CatchUp(JoinStream),
    /// Every catch-up frame is queued; the handoff into the live ring,
    /// stepped a tick at a time. `pass` is its running journal pass, on
    /// the slot's worker (see [`HandoffSocket`]).
    Handoff {
        handoff: LiveHandoff,
        pass: Option<JoinStream>,
    },
}

/// The DPDK side of a replica's handoff ([`HandoffIo`]): its socket, and
/// the slot's join worker for the journal passes, so that the poll thread
/// neither reads the disk nor waits for room in the socket. Lives for one
/// tick, and reports what the tick came to.
struct HandoffSocket<'a> {
    transport: &'a mut melin_dpdk::DpdkTransport,
    handle: melin_dpdk::SocketHandle,
    joiner: &'a JoinWorker,
    pass: &'a mut Option<JoinStream>,
    /// Bytes queued this tick, against [`MAX_REPL_BYTES_PER_TICK`].
    queued: usize,
    /// The socket refused the last offer: the handoff waits on the replica.
    blocked: bool,
    /// The worker is gone (it panicked), a bug in us: logged at `error!`.
    worker_gone: bool,
}

impl HandoffIo for HandoffSocket<'_> {
    fn start_pass(&mut self, from: u64) -> io::Result<()> {
        match self.joiner.submit(JoinRequest::HandoffPass { from }) {
            Ok(stream) => {
                *self.pass = Some(stream);
                Ok(())
            }
            Err(_) => {
                self.worker_gone = true;
                Err(io::Error::other("the replica's join worker is gone"))
            }
        }
    }

    fn pump_pass(&mut self) -> PassProgress {
        let Some(stream) = self.pass.as_mut() else {
            return PassProgress::Done(Err(io::Error::other("no handoff pass is running")));
        };
        // This tick's budget is spent: the rest is the next tick's, as the
        // streaming arm does it. Not the replica's doing.
        if self.queued >= MAX_REPL_BYTES_PER_TICK {
            self.blocked = false;
            return PassProgress::Pending;
        }
        let budget = MAX_REPL_BYTES_PER_TICK - self.queued;
        let capacity = self.transport.max_tx_queue_size(self.handle);
        let (transport, handle) = (&mut *self.transport, self.handle);
        match stream.pump(budget, capacity, |frame| {
            transport.queue_send(handle, frame)
        }) {
            Pump::Pending { queued } => {
                self.queued += queued;
                self.blocked = stream.is_blocked();
                PassProgress::Pending
            }
            Pump::Done(result) => {
                *self.pass = None;
                self.blocked = false;
                PassProgress::Done(result)
            }
            Pump::WorkerGone => {
                *self.pass = None;
                self.worker_gone = true;
                PassProgress::Done(Err(io::Error::other("the replica's join worker died")))
            }
        }
    }

    fn try_send(&mut self, frame: &[u8]) -> bool {
        // The tick's budget, as above: the streaming arm's rule, never
        // below one frame a tick.
        if self.queued >= MAX_REPL_BYTES_PER_TICK {
            self.blocked = false;
            return false;
        }
        let taken = self.transport.queue_send(self.handle, frame);
        if taken {
            self.queued += frame.len();
        }
        self.blocked = !taken;
        taken
    }
}

/// What one tick of a join came to.
enum JoinTick {
    /// More to do: `queued` bytes went to the socket, and `blocked` the
    /// socket refused what the join has to send.
    Pending { queued: usize, blocked: bool },
    /// Every catch-up frame is queued: the last sequence it streamed.
    CaughtUp(u64),
    /// The handoff is over: the slot's sent high-water.
    Live(SentHighWater),
    /// The join failed (logged): drop the replica.
    Failed,
}

/// Per-replica slot — owns its ring consumer and state machine.
struct DpdkReplicaSlot {
    state: SlotState,
    consumer: ReplicationConsumer,
    active_flag: Arc<AtomicBool>,
    evict_flag: Arc<AtomicBool>,
    recv_buf: Vec<u8>,
    send_buf: Vec<u8>,
    last_send: std::time::Instant,
    /// Warns when the replica keeps accepting entries but persists none.
    /// Fresh for each connection, from the moment it streams live.
    persist_watch: PersistWatch,
    /// Sent high-water for the current connection — the ack-sanity
    /// bound and the heartbeat sequence. Meaningless while `Idle`;
    /// re-seeded on every handshake. See `SentHighWater`.
    sent: SentHighWater,
    /// `Some` while this slot's handshake validation runs off-thread.
    pending_validation: Option<PendingValidation>,
    /// This slot's parked validation thread. Spawned once at driver
    /// construction — never from `tick`: the poll thread is pinned to an
    /// isolated core at `SCHED_FIFO` and busy-spins, so a thread created
    /// from it inherits that context and is never scheduled. See
    /// [`ValidationWorker`].
    validator: ValidationWorker,
    /// `Some` while this slot is `Joining`.
    ///
    /// Declared before `joiner`, and so dropped before it: dropping the
    /// stream cancels the join and closes the channel the worker may be
    /// blocked sending on, which the worker's drop then waits for.
    join: Option<PendingJoin>,
    /// This slot's parked join thread, spawned at driver construction for
    /// the reason `validator` is.
    joiner: JoinWorker,
    /// `Some` while this slot is `Authenticating` — the challenge we issued
    /// and its deadline. Cleared on transition out of `Authenticating`.
    auth: Option<AuthChallenge>,
}

// The sender-side auth-step machinery — the `AuthTransport` surface,
// `AuthOutcome`, `AuthChallenge`, and `step_authentication` — lives in
// `super::auth` so the protocol step is shared with the kernel-TCP path,
// transport-agnostic, and unit-testable without the `dpdk` feature. Only the
// concrete transport impl belongs here.
impl AuthTransport for melin_dpdk::DpdkTransport {
    type Handle = melin_dpdk::SocketHandle;
    // Each method forwards to the inherent method of the same name (inherent
    // resolution wins, so there is no recursion), except `is_active`: a
    // replication link is open while both halves are, so a peer's FIN ends
    // the exchange as an RST does (see `REPLICATION_LIVENESS`).
    fn is_active(&mut self, handle: Self::Handle) -> bool {
        self.is_connected(handle)
    }
    fn recv_into_vec(&mut self, handle: Self::Handle, dest: &mut Vec<u8>) {
        self.recv_into_vec(handle, dest);
    }
    fn queue_send(&mut self, handle: Self::Handle, data: &[u8]) -> bool {
        self.queue_send(handle, data)
    }
    fn poll(&mut self) {
        self.poll();
    }
}

impl DpdkReplicaSlot {
    /// The single transition back to `Idle` from a connected state
    /// (`Authenticating` / `Handshaking` / `Joining` / `Streaming`). Every
    /// disconnect / reject / eviction path funnels through here so none can
    /// leak the smoltcp handle or desync the bookkeeping:
    ///
    /// - **Resets the socket** (idempotent) — tells a replica still there
    ///   that it was dropped (an RST, so it reconnects at once rather than
    ///   at its liveness deadline), and reclaims the socket whether the
    ///   handle is already removed *or* still pinned in the `SocketSet` in
    ///   `Closed`/`TimeWait`. Making the close part of "go Idle" (rather
    ///   than a step each arm must remember) is what prevents the
    ///   leak-on-disconnect class of bug.
    /// - Disengages the shared cursors **before** releasing the journal-stage
    ///   gate — ordering contract B2 (frozen replica-progress cursors would
    ///   otherwise stop the primary acking client requests even with a
    ///   healthy peer).
    /// - Decrements the halt gate **only if the replica was `Joining` or
    ///   `Streaming`** — a connection still authenticating or handshaking
    ///   never lifted it (the gate is lifted once the handshake is accepted
    ///   and the join begins, not on connect or auth), so it must not lower
    ///   it. Warns if the last streaming replica just left (the node now
    ///   halts).
    /// - Clears per-connection scratch (recv buffer, in-flight challenge,
    ///   pending handshake validation, join in progress).
    ///
    /// Callers never assign `SlotState::Idle` directly; they call this and keep
    /// only their state-specific extras (e.g. eviction's ring skip).
    fn go_idle(
        &mut self,
        slot_idx: usize,
        transport: &mut melin_dpdk::DpdkTransport,
        cursors: &ReplicaCursors,
        metrics: &ReplicationMetrics,
        replicas_connected: &ReplicaPresence,
    ) {
        if let SlotState::Authenticating(h)
        | SlotState::Handshaking(h)
        | SlotState::Joining(h)
        | SlotState::Streaming(h, _) = self.state
        {
            // The gate is lifted only once the join begins (see the
            // handshake verdict in the `Handshaking` arm); a connection
            // that drops before then never lifted it, so it must not lower
            // it.
            let was_counted =
                matches!(self.state, SlotState::Joining(_) | SlotState::Streaming(..));
            transport.reset(h);
            // Disengage cursors before the active_flag Release — contract B2.
            cursors.clear_on_disconnect(slot_idx);
            self.active_flag.store(false, Ordering::Release);
            metrics.catching_up[slot_idx].store(false, Ordering::Relaxed);
            if was_counted {
                replicas_connected.lower();
            }
        }
        self.state = SlotState::Idle;
        self.recv_buf.clear();
        self.auth = None;
        self.pending_validation = None;
        // Abandons a join in progress (the catch-up, or a handoff pass):
        // its worker stops at its next frame.
        self.join = None;
        // Nothing to release in the ring: a chunk a handoff ended mid-way
        // left uncommitted (one it was forwarding or back-filling for) is
        // simply first in it, and the `Idle` arm drains it with the rest.
    }
}

/// Step-able DPDK replication state. Owns both slot state machines and the
/// shared cursors / metrics, but does NOT own the `DpdkTransport` — it
/// reaches into one supplied by the caller per call. This shape lets the
/// primary's single DPDK poll thread drive replication alongside client
/// traffic by calling `tick()` once per poll iteration and dispatching
/// `accept_connection()` for any `AcceptedConnection` that arrives on the
/// replication listen port.
///
/// Parameterised over `A: Application` so successive `tick` calls cannot
/// disagree on the application type — the journal catch-up and snapshot
/// transfer paths both decode events through `A`, so a mid-flight switch
/// would silently corrupt the wire. The driver carries no `A`-typed
/// data, hence the `PhantomData`.
pub struct DpdkReplicationDriver<A: Application> {
    slots: [DpdkReplicaSlot; 2],
    /// Single owner of the per-replica progress cursors (per-slot
    /// acked positions and the gate's gauge pair).
    cursors: ReplicaCursors,
    replica_ready: Arc<AtomicBool>,
    replicas_connected: Arc<ReplicaPresence>,
    metrics: Arc<ReplicationMetrics>,
    batch_size: usize,
    heartbeat_interval: std::time::Duration,
    /// Node fencing state — see the kernel-TCP sender. Read to stamp the
    /// primary's epoch onto each `StreamStart` and to self-demote when a
    /// replica handshakes with a higher epoch.
    fence_state: Arc<melin_transport_core::fence::FenceState>,
    /// Operator key table. `Arc` because it's shared, read-only, with the
    /// client-auth path and the kernel-TCP sender; the driver only ever
    /// `lookup`s during a replica's challenge/response. Held by the driver
    /// (not threaded per-call) so the `Authenticating` tick arm can verify
    /// without the caller re-supplying it each poll.
    authorized_keys: Arc<AuthorizedKeys>,
    // Anchors the `A` type parameter — the struct holds no app-typed
    // state, but `tick`'s journal-catchup and snapshot-transfer paths
    // do, and we want the type system to enforce that the same `A`
    // flows through every call rather than relying on every call site
    // to spell `::<A>` consistently.
    _app: PhantomData<fn(A)>,
}

impl<A: Application> DpdkReplicationDriver<A> {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        repl_consumers: [ReplicationConsumer; 2],
        replica_slots: Arc<melin_transport_core::ReplicaSlotCursors>,
        // The node's journal, streamed to replicas, and its snapshot, sent
        // to one that needs a resync.
        store: NodeStore,
        replica_ready: Arc<AtomicBool>,
        replicas_connected: Arc<ReplicaPresence>,
        evict_flags: [Arc<AtomicBool>; 2],
        active_flags: [Arc<AtomicBool>; 2],
        metrics: Arc<ReplicationMetrics>,
        batch_size: usize,
        heartbeat_secs: u64,
        fence_state: Arc<melin_transport_core::fence::FenceState>,
        authorized_keys: Arc<AuthorizedKeys>,
    ) -> Result<Self, String> {
        let [consumer_0, consumer_1] = repl_consumers;
        let now = std::time::Instant::now();

        // Spawn both validation workers here, on the caller's thread.
        // This runs at startup, before the client poll thread pins
        // itself, which is the only moment a thread can be created with a
        // context the poll loop would never let it acquire on its own —
        // see `ValidationWorker`. A failure fails the boot cleanly
        // (nothing is running yet to leak) rather than surfacing later as
        // a replica that hangs in `Handshaking`.
        let [validator_0, validator_1] = [0usize, 1].map(|idx| {
            ValidationWorker::spawn(
                format!("repl-validate-{idx}"),
                store.journal().to_path_buf(),
                validate_replica_handshake_settled,
            )
        });
        let validator_0 =
            validator_0.map_err(|e| format!("spawn handshake validation worker 0: {e}"))?;
        let validator_1 =
            validator_1.map_err(|e| format!("spawn handshake validation worker 1: {e}"))?;
        // The join workers, likewise spawned here and never from `tick`.
        let [joiner_0, joiner_1] = [0usize, 1].map(|idx| {
            let ctx = JoinContext {
                store: store.clone(),
                fence_state: Arc::clone(&fence_state),
            };
            JoinWorker::spawn(
                format!("repl-join-{idx}"),
                move |request: &JoinRequest, publish: CatchUpPublisher<'_>, cancel: &AtomicBool| {
                    stream_join::<A::Event>(&ctx, request, publish, cancel)
                },
            )
        });
        let joiner_0 = joiner_0.map_err(|e| format!("spawn replica join worker 0: {e}"))?;
        let joiner_1 = joiner_1.map_err(|e| format!("spawn replica join worker 1: {e}"))?;

        Ok(DpdkReplicationDriver {
            slots: [
                DpdkReplicaSlot {
                    state: SlotState::Idle,
                    consumer: consumer_0,
                    active_flag: Arc::clone(&active_flags[0]),
                    evict_flag: Arc::clone(&evict_flags[0]),
                    recv_buf: Vec::with_capacity(4096),
                    send_buf: Vec::with_capacity(512 * 1024),
                    last_send: now,
                    persist_watch: PersistWatch::default(),
                    // Placeholder until the slot engages — re-seeded on
                    // every handshake.
                    sent: SentHighWater::seed(0, 0),
                    pending_validation: None,
                    validator: validator_0,
                    join: None,
                    joiner: joiner_0,
                    auth: None,
                },
                DpdkReplicaSlot {
                    state: SlotState::Idle,
                    consumer: consumer_1,
                    active_flag: Arc::clone(&active_flags[1]),
                    evict_flag: Arc::clone(&evict_flags[1]),
                    recv_buf: Vec::with_capacity(4096),
                    send_buf: Vec::with_capacity(512 * 1024),
                    last_send: now,
                    persist_watch: PersistWatch::default(),
                    // Placeholder until the slot engages — re-seeded on
                    // every handshake.
                    sent: SentHighWater::seed(0, 0),
                    pending_validation: None,
                    validator: validator_1,
                    join: None,
                    joiner: joiner_1,
                    auth: None,
                },
            ],
            cursors: ReplicaCursors::new(replica_slots, metrics.clone()),
            replica_ready,
            replicas_connected,
            metrics,
            batch_size,
            heartbeat_interval: std::time::Duration::from_secs(heartbeat_secs),
            fence_state,
            authorized_keys,
            _app: PhantomData,
        })
    }

    /// Take ownership of a freshly-accepted connection on the replication
    /// port. Assigns it to the first idle slot, or closes it if both slots
    /// are occupied (dual-repl cap). Caller is responsible for filtering by
    /// `AcceptedConnection::listen_port`.
    pub fn accept_connection(
        &mut self,
        peer: std::net::SocketAddr,
        handle: melin_dpdk::SocketHandle,
        transport: &mut melin_dpdk::DpdkTransport,
    ) {
        let idle_slot = self
            .slots
            .iter()
            .position(|s| matches!(s.state, SlotState::Idle));
        let Some(idx) = idle_slot else {
            debug!(peer = ?peer, "replica rejected — both slots occupied");
            transport.reset(handle);
            return;
        };
        // From here the slot watches its replica: a replica that stops
        // answering is reset by the stack, and every arm's link check
        // then takes the slot back to `Idle`.
        transport.set_peer_liveness(handle, &REPLICATION_LIVENESS);

        // Issue the auth challenge immediately (non-blocking): the replica
        // must sign this nonce with a key listed under the `replication` role
        // before we process its handshake. The response is verified across
        // ticks in the `Authenticating` arm — we never block the poll thread
        // on a silent replica. Mirrors the kernel-TCP sender's
        // `authenticate_replica`, sharing the nonce generation and (in tick)
        // the verification.
        let nonce = match generate_challenge_nonce() {
            Ok(n) => n,
            Err(e) => {
                warn!(peer = ?peer, error = %e, "failed to generate auth challenge — dropping replica");
                transport.reset(handle);
                return;
            }
        };
        let slot = &mut self.slots[idx];
        slot.recv_buf.clear();
        slot.send_buf.clear();
        encode_challenge(&nonce, &mut slot.send_buf);
        if !transport.queue_send(handle, &slot.send_buf) {
            warn!(peer = ?peer, slot = idx, "TX queue full sending auth challenge — dropping replica");
            transport.reset(handle);
            return;
        }
        info!(peer = ?peer, slot = idx, "replica connected via DPDK — authenticating");
        // The halt gate is NOT lifted here: a bare connection hasn't
        // proven a Replication key. It's lifted once the handshake is
        // accepted and the join begins (the `Handshaking` arm), so neither
        // an unauthenticated peer nor one refused after auth can re-enable
        // client writes.
        slot.auth = Some(AuthChallenge {
            nonce,
            deadline: std::time::Instant::now() + AUTH_TIMEOUT,
        });
        slot.state = SlotState::Authenticating(handle);
    }

    /// Drive both slots' state machines for one poll iteration. Returns
    /// `true` if at least one slot is currently active (Handshaking or
    /// Streaming) — caller can use this to decide whether to busy-spin
    /// on idle.
    pub fn tick(
        &mut self,
        transport: &mut melin_dpdk::DpdkTransport,
        shutdown: &AtomicBool,
    ) -> bool {
        // Local rebinds for readability — the body below was lifted from
        // the previous run_sender_dpdk thread, mostly verbatim, so keep
        // the variable names matching.
        let slots = &mut self.slots;
        let cursors = &self.cursors;
        let replica_ready = &self.replica_ready;
        let replicas_connected = &self.replicas_connected;
        let metrics = &self.metrics;
        let fence_state = &self.fence_state;
        let authorized_keys = &self.authorized_keys;
        let batch_size = self.batch_size;
        let heartbeat_interval = self.heartbeat_interval;

        // Check eviction flags from the journal stage.
        for (i, slot) in slots.iter_mut().enumerate() {
            let evicting =
                slot.evict_flag.load(Ordering::Acquire) && !matches!(slot.state, SlotState::Idle);
            if !evicting {
                continue;
            }
            metrics.evictions_total.fetch_add(1, Ordering::Relaxed);
            warn!(
                slot = i,
                "evicting slow replica (ring backpressure timeout, DPDK)"
            );
            // Eviction-specific: drop any unread ring entries so a reconnecting
            // replica on this slot doesn't replay pre-eviction data and stall
            // the primary's replication cursor (see tcp_sender.rs), and clear
            // the latch that fired this eviction. The close + cursor/gate/count
            // teardown is the shared `go_idle`.
            slot.consumer.skip_to_producer();
            slot.evict_flag.store(false, Ordering::Release);
            slot.go_idle(i, transport, cursors, metrics, replicas_connected);
        }

        let mut any_active = false;

        for (slot_idx, slot) in slots.iter_mut().enumerate() {
            match slot.state {
                SlotState::Idle => {
                    // Drain ring to keep it flowing. The journal stage
                    // skips inactive rings (active_flag=false), but there
                    // may be residual entries from before the flag was cleared.
                    while let Some(batch) = slot.consumer.peek() {
                        batch.commit();
                    }
                }

                SlotState::Authenticating(handle) => {
                    any_active = true;

                    // `auth` is always `Some` here — set in `accept_connection`
                    // and cleared only on transition out of `Authenticating`.
                    // The shared borrow of `slot.auth` coexists with the two
                    // `&mut` buffer borrows below because they are disjoint
                    // fields.
                    let challenge = slot
                        .auth
                        .as_ref()
                        .expect("auth challenge present while Authenticating");

                    match step_authentication(
                        transport,
                        handle,
                        challenge,
                        &mut slot.recv_buf,
                        &mut slot.send_buf,
                        authorized_keys,
                        slot_idx,
                    ) {
                        // Frame not yet complete — wait for more data next tick.
                        AuthOutcome::Pending => {}
                        AuthOutcome::Authenticated => {
                            slot.auth = None;
                            // Proven a Replication key. The halt gate waits
                            // for the handshake to be accepted.
                            slot.state = SlotState::Handshaking(handle);
                        }
                        AuthOutcome::Rejected => {
                            slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected)
                        }
                    }
                    continue;
                }

                SlotState::Handshaking(handle) => {
                    any_active = true;

                    // Check for disconnect during handshake: the
                    // replica's FIN or RST, or its liveness deadline.
                    if !transport.is_connected(handle) {
                        warn!(
                            slot = slot_idx,
                            "replica disconnected during handshake (DPDK)"
                        );
                        slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        continue;
                    }

                    // A validation verdict may be outstanding — poll it
                    // without blocking. The replica is silent between its
                    // Handshake and our response, so no frames need
                    // processing while waiting.
                    if let Some(pv) = slot.pending_validation.take() {
                        let res = match pv.verdict_rx.try_recv() {
                            Err(std::sync::mpsc::TryRecvError::Empty) => {
                                // Not settled yet — revisit next tick.
                                slot.pending_validation = Some(pv);
                                continue;
                            }
                            Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                                // The worker can only drop the sender
                                // without a verdict by panicking — our
                                // bug, so `error!`.
                                error!(
                                    slot = slot_idx,
                                    "handshake validation worker died — disconnecting"
                                );
                                slot.go_idle(
                                    slot_idx,
                                    transport,
                                    cursors,
                                    metrics,
                                    replicas_connected,
                                );
                                continue;
                            }
                            Ok(res) => res,
                        };
                        let h = pv.handshake;

                        // Chain validation verdict — see the kernel-TCP
                        // sender. A divergent replica gets a HashMismatch
                        // frame (below) and the snapshot-resync route on
                        // this same connection.
                        let divergent = match res {
                            Ok(HandshakeValidation::Ok) => false,
                            Ok(HandshakeValidation::Divergent(kind)) => {
                                // Alertable on /metrics: divergence
                                // outside an expected failover rejoin
                                // means corruption or a serious bug.
                                metrics.divergence_total.fetch_add(1, Ordering::Relaxed);
                                warn!(
                                    slot = slot_idx,
                                    last_sequence = h.last_sequence,
                                    ?kind,
                                    "replica journal divergent at handshake — \
                                     routing through snapshot resync"
                                );
                                true
                            }
                            Err(e) => {
                                warn!(slot = slot_idx, error = %e, "handshake validation failed — disconnecting");
                                slot.go_idle(
                                    slot_idx,
                                    transport,
                                    cursors,
                                    metrics,
                                    replicas_connected,
                                );
                                continue;
                            }
                        };
                        // Cursor/stream floor: a divergent replica's
                        // claimed position is meaningless (see the
                        // kernel-TCP sender) — seed from 0 like a fresh
                        // replica.
                        let stream_base = if divergent { 0 } else { h.last_sequence };

                        // The join's disk work — the catch-up probe, then
                        // either `StreamStart` and journal history, or the
                        // pre-flight, the resync verdict and the snapshot
                        // — goes to this slot's parked worker; the
                        // `Joining` arm moves its frames into the socket.
                        // A submission, not a spawn, for the reason the
                        // validation is (see `ValidationWorker`).
                        match slot.joiner.submit(JoinRequest::Replica {
                            last_sequence: h.last_sequence,
                            divergent,
                        }) {
                            Ok(stream) => {
                                // Handshake accepted, join begun — only now
                                // does this connection lift the halt gate
                                // (lowered by `go_idle` on teardown): a
                                // replica refused before here never counted.
                                slot.join = Some(PendingJoin {
                                    phase: JoinPhase::CatchUp(stream),
                                    stream_base,
                                    stall: JoinStallWatch::default(),
                                    first_ack: replicas_connected.lift(),
                                });
                                slot.state = SlotState::Joining(handle);
                            }
                            Err(e) => {
                                // `error!`: the worker only disappears by
                                // panicking, a bug in us.
                                error!(slot = slot_idx, error = %e, "cannot start the replica's join — disconnecting");
                                slot.go_idle(
                                    slot_idx,
                                    transport,
                                    cursors,
                                    metrics,
                                    replicas_connected,
                                );
                            }
                        }
                        continue;
                    }

                    // Try to read handshake frame.
                    transport.recv_into_vec(handle, &mut slot.recv_buf);

                    match try_extract_frame(&slot.recv_buf, MAX_CONTROL_FRAME) {
                        FrameResult::Complete(payload_start, frame_end) => {
                            let payload = &slot.recv_buf[payload_start..frame_end];
                            match decode_replica_message(payload) {
                                Ok(ReplicaMessage::Handshake(h)) => {
                                    info!(
                                        slot = slot_idx,
                                        last_sequence = h.last_sequence,
                                        replica_epoch = h.epoch,
                                        "replica handshake received (DPDK)"
                                    );

                                    // Fence: a replica with a higher epoch
                                    // means we are a superseded ex-primary —
                                    // self-demote and drop the connection.
                                    // Policy (latch + shutdown) lives on
                                    // `FenceState`; see the kernel-TCP
                                    // sender for the rationale.
                                    let our_epoch = fence_state.epoch();
                                    if let Some(first_latch) =
                                        fence_state.fence_if_superseded(h.epoch, shutdown)
                                    {
                                        if first_latch {
                                            error!(
                                                slot = slot_idx,
                                                replica_epoch = h.epoch,
                                                our_epoch,
                                                "fenced: a replica advertises a higher epoch — \
                                                 this primary has been superseded; self-demoting \
                                                 and shutting down (DPDK)"
                                            );
                                        }
                                        slot.go_idle(
                                            slot_idx,
                                            transport,
                                            cursors,
                                            metrics,
                                            replicas_connected,
                                        );
                                        continue;
                                    }

                                    metrics.catching_up[slot_idx].store(true, Ordering::Relaxed);

                                    // The handshake frame is consumed here;
                                    // chain validation is handed to this slot's
                                    // parked worker (it can scan a full segment
                                    // per attempt and sleeps between retries —
                                    // blocking tick() would freeze client
                                    // traffic and the other slot). The slot
                                    // stays Handshaking; the verdict is picked
                                    // up by the pending_validation poll at the
                                    // top of this arm.
                                    //
                                    // Deliberately a submission, not a spawn:
                                    // this is the pinned, busy-spinning poll
                                    // thread, and a thread created here would
                                    // inherit its core and SCHED_FIFO priority
                                    // and never run at all — the handshake would
                                    // hang forever. See `ValidationWorker`.
                                    compact_recv_buf(&mut slot.recv_buf, frame_end);
                                    match slot.validator.submit(h.clone()) {
                                        Ok(verdict_rx) => {
                                            slot.pending_validation = Some(PendingValidation {
                                                handshake: h,
                                                verdict_rx,
                                            });
                                        }
                                        Err(e) => {
                                            // `error!`, not `warn!`: the
                                            // worker only disappears by
                                            // panicking, which is a bug in
                                            // us, not bad input from the
                                            // replica.
                                            error!(slot = slot_idx, error = %e, "cannot validate handshake — disconnecting");
                                            slot.go_idle(
                                                slot_idx,
                                                transport,
                                                cursors,
                                                metrics,
                                                replicas_connected,
                                            );
                                        }
                                    }
                                }
                                Ok(ReplicaMessage::Ack(_)) => {
                                    warn!(
                                        slot = slot_idx,
                                        "expected Handshake, got Ack — disconnecting"
                                    );
                                    slot.go_idle(
                                        slot_idx,
                                        transport,
                                        cursors,
                                        metrics,
                                        replicas_connected,
                                    );
                                }
                                Err(e) => {
                                    warn!(slot = slot_idx, error = %e, "failed to decode handshake — disconnecting");
                                    slot.go_idle(
                                        slot_idx,
                                        transport,
                                        cursors,
                                        metrics,
                                        replicas_connected,
                                    );
                                }
                            }
                        }
                        FrameResult::Oversized => {
                            warn!(slot = slot_idx, "oversized handshake frame — disconnecting");
                            slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        }
                        FrameResult::Incomplete => {} // Wait for more data.
                    }
                }

                SlotState::Joining(handle) => {
                    any_active = true;

                    // The replica's FIN or RST, or its liveness deadline.
                    if !transport.is_connected(handle) {
                        warn!(
                            slot = slot_idx,
                            "replica disconnected during catch-up (DPDK)"
                        );
                        slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        continue;
                    }

                    let join = slot.join.as_mut().expect("join in progress while Joining");

                    // Read the replica's acks off the socket, the newest
                    // kept for the streaming arm (see `keep_last_ack`).
                    transport.recv_into_vec(handle, &mut slot.recv_buf);
                    match keep_last_ack(&mut slot.recv_buf) {
                        JoinFrames::Malformed => {
                            warn!(
                                slot = slot_idx,
                                "malformed (zero-length or oversized) frame from replica during catch-up — disconnecting"
                            );
                            slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                            continue;
                        }
                        // The replica is past whatever it does before
                        // streaming (a snapshot install): from now on it
                        // reads, and refusing for the limit is a stall.
                        JoinFrames::Read { acked: true } => join.stall.arm(),
                        JoinFrames::Read { acked: false } => {}
                    }

                    // One tick of the join: as much as fits in the socket,
                    // at most a tick's worth (the bound the streaming arm
                    // keeps, for the same reason: this thread is also client
                    // ingress). Never waits on the disk or the replica.
                    let now = std::time::Instant::now();
                    let tick = match &mut join.phase {
                        // The worker's join frames.
                        JoinPhase::CatchUp(stream) => {
                            let capacity = transport.max_tx_queue_size(handle);
                            match stream.pump(MAX_REPL_BYTES_PER_TICK, capacity, |frame| {
                                transport.queue_send(handle, frame)
                            }) {
                                Pump::Pending { queued } => JoinTick::Pending {
                                    queued,
                                    blocked: stream.is_blocked(),
                                },
                                // Every frame of the join is queued, in order.
                                Pump::Done(Ok(end)) => JoinTick::CaughtUp(end),
                                Pump::Done(Err(e)) => {
                                    warn!(slot = slot_idx, error = %e, "catch-up/snapshot failed — disconnecting");
                                    JoinTick::Failed
                                }
                                Pump::WorkerGone => {
                                    // `error!`: the worker only dies by
                                    // panicking, a bug in us.
                                    error!(
                                        slot = slot_idx,
                                        "replica join worker died — disconnecting"
                                    );
                                    JoinTick::Failed
                                }
                            }
                        }
                        // The handoff into live streaming: activates the
                        // ring, re-reads from the journal the entries that
                        // fell into the activation window, then drains the
                        // ring into sequence-contiguity (back-filling from
                        // disk while a skipped entry hasn't flushed) before
                        // going live — `bridge_catchup_to_live`'s steps and
                        // decisions, taken a tick at a time: its journal
                        // passes run on the slot's worker, and a ring chunk
                        // the socket has no room for stays held in the ring
                        // until it does. See
                        // `melin_transport_core::replication::handoff`.
                        JoinPhase::Handoff { handoff, pass } => {
                            let mut io = HandoffSocket {
                                transport: &mut *transport,
                                handle,
                                joiner: &slot.joiner,
                                pass,
                                queued: 0,
                                blocked: false,
                                worker_gone: false,
                            };
                            match handoff.step(&mut slot.consumer, &slot.active_flag, &mut io, now)
                            {
                                Ok(HandoffStep::Pending) => JoinTick::Pending {
                                    queued: io.queued,
                                    blocked: io.blocked,
                                },
                                Ok(HandoffStep::Live(sent)) => JoinTick::Live(sent),
                                Err(e) if io.worker_gone => {
                                    // `error!`: the worker only goes by
                                    // panicking, a bug in us.
                                    error!(slot = slot_idx, error = %e, "replica join worker died during the handoff — disconnecting");
                                    JoinTick::Failed
                                }
                                Err(e) => {
                                    warn!(slot = slot_idx, error = %e, "catch-up handoff failed — disconnecting");
                                    JoinTick::Failed
                                }
                            }
                        }
                    };

                    match tick {
                        JoinTick::Pending { queued, blocked } => {
                            if queued > 0 {
                                // Flush now, as the streaming arm does.
                                transport.poll();
                            }
                            let stalled = join.stall.observe(queued > 0, blocked, now);
                            if stalled >= JOIN_STALL_LIMIT {
                                warn!(
                                    slot = slot_idx,
                                    stalled_ms = stalled.as_millis() as u64,
                                    "replica stopped reading its catch-up or handoff — disconnecting"
                                );
                                slot.go_idle(
                                    slot_idx,
                                    transport,
                                    cursors,
                                    metrics,
                                    replicas_connected,
                                );
                            }
                        }
                        JoinTick::CaughtUp(catchup_end) => {
                            transport.poll();
                            // Engage this slot's cursors and seed the gauge
                            // pair BEFORE the handoff flips active so a
                            // reader that observes active=true also observes
                            // a non-zero cursor pair — see `ReplicaCursors`
                            // for the ordering contract. The handoff's first
                            // step, next tick, flips it.
                            cursors.seed_on_handshake(slot_idx, join.stream_base);
                            join.phase = JoinPhase::Handoff {
                                handoff: LiveHandoff::new(join.stream_base, catchup_end),
                                pass: None,
                            };
                            // A catch-up that ended is progress. The watch
                            // stays armed if the replica has acked.
                            join.stall.restart();
                        }
                        JoinTick::Live(sent) => {
                            transport.poll();
                            super::log_replica_live(slot_idx, &sent);
                            // The sent high-water: heartbeats and the
                            // ack-sanity bound.
                            slot.sent = sent;
                            let first_ack = slot
                                .join
                                .take()
                                .expect("join in progress while Joining")
                                .first_ack;
                            slot.last_send = now;
                            slot.persist_watch = PersistWatch::default();
                            replica_ready.store(true, Ordering::Release);
                            metrics.catching_up[slot_idx].store(false, Ordering::Relaxed);
                            slot.state = SlotState::Streaming(handle, first_ack);
                        }
                        JoinTick::Failed => {
                            slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        }
                    }
                }

                SlotState::Streaming(handle, _) => {
                    any_active = true;

                    // 1. Process acks (non-blocking).
                    transport.recv_into_vec(handle, &mut slot.recv_buf);
                    let mut consumed = 0;
                    let mut ack_error = false;
                    // Whether an ack was recorded this tick. The latency
                    // gauge is stored once after the loop rather than per
                    // ack: `last_send` does not move inside it, so a later
                    // ack's store would only overwrite an earlier one, and
                    // this poll thread is also client ingress.
                    let mut acked = false;
                    // When this tick's acks arrived, read once, on the first.
                    let mut ack_received: Option<std::time::Instant> = None;
                    loop {
                        let remaining = &slot.recv_buf[consumed..];
                        match try_extract_frame(remaining, MAX_CONTROL_FRAME) {
                            FrameResult::Complete(payload_start, frame_end) => {
                                let payload = &remaining[payload_start..frame_end];
                                if let Ok(ReplicaMessage::Ack(ack)) =
                                    decode_replica_message(payload)
                                {
                                    if cursors.record_ack(slot_idx, &ack, slot.sent.get()).is_err()
                                    {
                                        // Eviction on violation: reuse the ack-error
                                        // teardown below (the store already logged
                                        // the violation at error level).
                                        ack_error = true;
                                        break;
                                    }
                                    acked = true;
                                    let received =
                                        *ack_received.get_or_insert_with(std::time::Instant::now);
                                    if let Some(event) = slot.persist_watch.observe(&ack, received)
                                    {
                                        event.log(slot_idx);
                                    }
                                }
                                consumed += frame_end;
                            }
                            FrameResult::Oversized => {
                                warn!(
                                    slot = slot_idx,
                                    "oversized ack frame from replica — disconnecting"
                                );
                                ack_error = true;
                                break;
                            }
                            FrameResult::Incomplete => break,
                        }
                    }
                    compact_recv_buf(&mut slot.recv_buf, consumed);
                    if acked {
                        // Same measure as the TCP sender: time since the
                        // last send (data or heartbeat) to the ack.
                        metrics.ack_latency_us[slot_idx].store(
                            slot.last_send.elapsed().as_micros() as u64,
                            Ordering::Relaxed,
                        );
                        // The replica took the stream: its first ack ends
                        // any halt override (see `ReplicaPresence`). Always
                        // `Streaming` in this arm; the match only reaches
                        // the token.
                        if let SlotState::Streaming(_, first_ack) = &mut slot.state {
                            first_ack.acked(replicas_connected);
                        }
                    }
                    if ack_error {
                        slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        continue;
                    }

                    // 2. Send data batches. Ring chunks are wire-ready
                    //    `InputBatch` frames produced by the journal stage
                    //    — the sender is a passthrough. Pre-check the
                    //    per-socket TX queue: if we commit a batch from
                    //    the ring but `queue_send` rejects it (TX full),
                    //    the data is gone from the ring without ever
                    //    reaching the replica — the replica never acks,
                    //    its acked-progress cursor stalls, and the
                    //    response gate freezes every client. We saw
                    //    this exact symptom on dpdk-dual-repl.
                    let max_tx = transport.max_tx_queue_size(handle);
                    let used = transport.tx_queue_bytes(handle);
                    let mut available = max_tx.saturating_sub(used);

                    slot.send_buf.clear();
                    let mut batches_sent = 0;
                    let mut bytes_this_tick = 0usize;
                    let mut tx_overflow = false;
                    while batches_sent < batch_size {
                        // Bound the bytes handed to the transport in one tick.
                        // The poll thread is also the ingress thread, so every
                        // microsecond it spends copying a replication burst
                        // into the socket and building mbufs is time no client
                        // packet is being received; a full 512 KiB queue is a
                        // single stall of that length. Spreading the same
                        // bytes over more ticks costs nothing — the driver is
                        // ticked several times per outer iteration.
                        //
                        // Never below one batch: a batch larger than the cap
                        // must still make progress, or the stream wedges.
                        if batches_sent > 0 && bytes_this_tick >= MAX_REPL_BYTES_PER_TICK {
                            break;
                        }
                        // A batch a previous tick found no room for is
                        // peeked again: the `break`s below leave it
                        // uncommitted, and only a commit moves past it.
                        let Some(batch) = slot.consumer.peek() else {
                            break;
                        };
                        let data_len = batch.data().len();
                        if data_len > available {
                            // Batch doesn't fit — flush what we have so
                            // far, poll to drain the wire, then re-check
                            // with a clean slate.
                            if !slot.send_buf.is_empty() {
                                metrics.bytes_sent[slot_idx]
                                    .fetch_add(slot.send_buf.len() as u64, Ordering::Relaxed);
                                if !transport.queue_send(handle, &slot.send_buf) {
                                    slot.send_buf.clear();
                                    tx_overflow = true;
                                    break;
                                }
                                slot.send_buf.clear();
                            }
                            transport.poll();
                            let used = transport.tx_queue_bytes(handle);
                            available = max_tx.saturating_sub(used);
                            if data_len > available {
                                break;
                            }
                        }
                        slot.send_buf.extend_from_slice(batch.data());
                        let end_sequence = batch.meta().end_sequence;
                        batch.commit();
                        slot.sent.advance(end_sequence);
                        batches_sent += 1;
                        bytes_this_tick += data_len;
                        available = available.saturating_sub(data_len);
                    }

                    if !tx_overflow && !slot.send_buf.is_empty() {
                        metrics.bytes_sent[slot_idx]
                            .fetch_add(slot.send_buf.len() as u64, Ordering::Relaxed);
                        if !transport.queue_send(handle, &slot.send_buf) {
                            tx_overflow = true;
                        }
                    }
                    if tx_overflow {
                        slot.send_buf.clear();
                        warn!(
                            slot = slot_idx,
                            "TX overflow on replica socket — disconnecting"
                        );
                        slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        continue;
                    }
                    if !slot.send_buf.is_empty() {
                        // Flush immediately so replication data hits the
                        // wire without waiting for the next outer-loop
                        // poll. Without this, a client-traffic burst
                        // starves the replication TX path: the TxQueue
                        // fills, the driver backs off, the ring overflows,
                        // and the response gate freezes every client.
                        transport.poll();
                        slot.last_send = std::time::Instant::now();
                    }

                    // 3. Heartbeat if idle.
                    if batches_sent == 0 && slot.last_send.elapsed() >= heartbeat_interval {
                        slot.send_buf.clear();
                        encode_heartbeat(slot.sent.get(), &mut slot.send_buf);
                        transport.queue_send(handle, &slot.send_buf);
                        slot.last_send = std::time::Instant::now();
                    }

                    // 4. Check for disconnect: the replica's FIN (its acks
                    //    above already read) or RST, or its liveness
                    //    deadline — a replica that stopped answering.
                    if !transport.is_connected(handle) {
                        warn!(slot = slot_idx, "replica disconnected (DPDK)");
                        slot.go_idle(slot_idx, transport, cursors, metrics, replicas_connected);
                        continue;
                    }

                    // Eviction is handled by the journal-stage evict_flag check
                    // at the top of the loop (lines 3254+). No timeout-based
                    // eviction here — peek() returning None means the
                    // consumer caught up, not that it's slow.
                }
            }
        }

        any_active
    }

    /// Close every replica's link, for a primary that is stopping: what is
    /// already queued on them gets a bounded chance to arrive, and then
    /// the replicas learn at once that it has gone (a kernel-TCP primary's
    /// kernel flushes and tells them the same way), rather than at their
    /// liveness deadline. See `linger_then_reset`. Consumes the driver:
    /// it is the poll loop's last act.
    ///
    /// Only the sockets: the halt gate, cursors and gauges are left as they
    /// are, since lowering the gate here would log the last replica leaving
    /// a node that is itself leaving.
    pub fn close_all_links(self, transport: &mut melin_dpdk::DpdkTransport) {
        // A `Vec`: built once, on the way out, so its allocation is
        // immaterial.
        let handles: Vec<_> = self
            .slots
            .iter()
            .filter_map(|slot| match slot.state {
                SlotState::Authenticating(h)
                | SlotState::Handshaking(h)
                | SlotState::Joining(h)
                | SlotState::Streaming(h, _) => Some(h),
                _ => None,
            })
            .collect();
        linger_then_reset(transport, &handles);
    }
}

/// DPDK variant of the replication receiver. Uses a `DpdkTransport` (smoltcp)
/// to connect to the primary via DPDK instead of kernel TCP.
///
/// Includes reconnection with exponential backoff (1s → 30s) and snapshot
/// transfer support — matching the TCP receiver's feature set.
///
/// The protocol is identical to `run_receiver` — same wire format, same
/// fsync-then-ack-then-replay pattern. Only the I/O primitives differ.
///
/// `transport` is borrowed, not consumed: on a promotion the node goes on
/// to serve as a DPDK primary on the same transport. Every session this
/// loop opens is reset before it returns, so the caller gets the
/// transport back holding no connection of the receiver's.
#[allow(clippy::too_many_arguments)]
pub fn run_receiver_dpdk<A>(
    transport: &mut melin_dpdk::DpdkTransport,
    primary_ip: std::net::Ipv4Addr,
    primary_port: u16,
    signing_key: &ed25519_dalek::SigningKey,
    // As the kernel-TCP `run_receiver`'s.
    store: &NodeStore,
    shutdown: &AtomicBool,
    // The handles this loop shares with the admin endpoint and the
    // control-plane raft driver — see [`super::ReplicaControlPlane`].
    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: 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,
{
    // Recover local state from journal whenever any 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,
            &control.lineage_ack_policy,
            sizing,
        )?;
    // Fence epoch now reflects the recovered journal; seed the advertised
    // sequence from the same recovery, then let the raft driver trust
    // this node's tip. See the kernel-TCP `run_receiver`.
    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;
    journal_tip.advance(melin_transport_core::WireSeq::new(last_sequence));
    tip_ready.store(true, Ordering::Release);

    // Exponential backoff for reconnection: 1s → 2s → 4s → … → 30s max.
    // Reset to 1s when a session heard the primary speak (data or
    // heartbeat) — see `handle_session_exit`.
    let mut backoff = std::time::Duration::from_secs(1);

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

    // Reusable buffers — survive across reconnections.
    let mut send_buf = Vec::with_capacity(64);
    let mut recv_buf: Vec<u8> = Vec::with_capacity(4096);
    // Ephemeral port counter for outbound connections. Each reconnect uses
    // a different local port to avoid smoltcp TIME_WAIT conflicts.
    let mut local_port: u16 = 40000;

    // Live pipeline state — built once on first connect (or after a snapshot
    // transfer), persists across `Disconnected` reconnects so we don't pay
    // the journal-recover + thread-spawn + warm-up cost on every drop.
    // None = no pipeline yet (first iteration, or just torn down for
    // snapshot transfer); Some = running pipeline with threads + atomics
    // we can read for the next reconnect handshake.
    let mut pipeline: Option<ReplicaPipelineHandles<A, BufferedWriter<A::Event>>> = None;

    // --- Outer reconnect loop ---
    //
    // Each iteration: connect → handshake → (snapshot rebuild?) →
    // (build pipeline if absent) → stream. On disconnect the pipeline
    // stays live — we just refresh handshake state from its atomics and
    // reconnect. Only `Promote` / `Shutdown` / snapshot-transfer / fatal
    // error tear it down.
    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);

        // Refresh handshake state from the running pipeline, if any.
        // The (last_sequence, chain_hash) pair must come from ONE
        // FsyncState snapshot — a torn pair read from two sources while
        // the journal stage keeps flushing trips the primary's
        // handshake chain validation (false divergence; see the
        // kernel-TCP receiver). Taken once the journal covers what the
        // last session published (see `settled_resume_point`).
        if let Some(p) = pipeline.as_ref() {
            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 — as the
                // kernel-TCP receiver does.
                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: the last one's socket was
                        // closed by its own exit dispatch.
                        || {},
                        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);
        }

        info!(
            primary_ip = %primary_ip,
            primary_port,
            "connecting to primary as replica (DPDK)"
        );

        // Seed the primary's MAC into smoltcp's neighbor cache. Without
        // this the replica has no way to address its first frame: an
        // acceptor learns the peer MAC from the inbound packet, but an
        // initiator has to know it before anything goes out. ARP cannot
        // supply it — an SR-IOV VF does not receive broadcast, and in
        // bifurcated mode the `rte_flow` rule matches on IPv4 source
        // address so ARP is never steered into DPDK at all. Either way
        // the SYN never ships and the replica spins on "failed to
        // connect (DPDK)" with no error to show for it.
        //
        // `--dpdk-peer-mac` wins when supplied; otherwise this falls back
        // to the SR-IOV `02:00:<ip>` convention, which holds only on
        // ports whose MAC dpdk-setup.sh assigned. Re-seeded on every
        // reconnect so an expired neighbour entry cannot strand a
        // reconnecting replica.
        let mac_source = transport.seed_peer_neighbor(primary_ip);
        info!(
            primary_ip = %primary_ip,
            mac_source = mac_source.as_str(),
            "seeded primary MAC into the neighbour cache"
        );
        // Drain the injected ARP reply through smoltcp so the neighbor
        // cache is populated BEFORE connect_to() runs. Without this poll
        // smoltcp's connect tries to resolve ARP from an empty cache,
        // queues a broadcast request that the PF drops, and the SYN
        // never ships.
        transport.poll();

        // Connect to primary via smoltcp.
        // Replication streams large journal batches; use bigger RX buffer
        // so smoltcp can advertise a window large enough to sustain throughput.
        const REPL_RX_BUF: usize = 512 * 1024;
        const REPL_TX_BUF: usize = 64 * 1024;
        const REPL_TX_QUEUE: usize = 64 * 1024;
        let dial_port = local_port;
        local_port = local_port.wrapping_add(1).max(40000);
        let handle = match transport.connect_to_with_buffers(
            primary_ip,
            primary_port,
            dial_port,
            REPL_RX_BUF,
            REPL_TX_BUF,
            REPL_TX_QUEUE,
            REPL_DISPATCH_BURST,
        ) {
            Ok(handle) => handle,
            // The stack would not open the socket (an endpoint it cannot
            // address): a dial that failed before its SYN, on the same
            // retry path as one that got no answer. No socket was added,
            // so there is nothing to reset.
            Err(e) => {
                warn!(
                    primary_ip = %primary_ip,
                    primary_port,
                    local_port = dial_port,
                    error = ?e,
                    backoff_secs = backoff.as_secs(),
                    "the stack refused to open a connection to the primary (DPDK) — retrying"
                );
                sleep_then_double_backoff(&mut backoff, shutdown, promote);
                continue;
            }
        };

        // Poll until TCP handshake completes (with timeout).
        let connect_start = std::time::Instant::now();
        const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
        let connected = loop {
            // Route shutdown and promotion through the `!connected` arm
            // and the loop top — returning from here would leak the
            // pipeline threads and the smoltcp socket. Promotion is
            // checked here because a dial to a primary that has gone is
            // answered by nothing (a stopped DPDK node has no stack to
            // refuse it), so waiting out the connect timeout would delay
            // the failover by as much.
            if shutdown.load(Ordering::Relaxed) || promote.is_requested() {
                break false;
            }
            transport.poll();
            if transport.is_connected(handle) {
                break true;
            }
            if !transport.is_active(handle) || connect_start.elapsed() >= CONNECT_TIMEOUT {
                break false;
            }
            std::thread::yield_now();
        };

        if !connected {
            // A shutdown racing the connect window is a clean exit, not
            // degraded operation — don't warn.
            if shutdown.load(Ordering::Relaxed) {
                debug!("connect aborted — shutting down (DPDK)");
            } else if promote.is_requested() {
                debug!("connect aborted — promoting (DPDK)");
            } else {
                warn!(
                    backoff_secs = backoff.as_secs(),
                    "failed to connect to primary (DPDK) — retrying"
                );
            }
            // Reset rather than drop: a primary that accepted after our
            // timeout frees the connection at once.
            transport.reset(handle);
            sleep_then_double_backoff(&mut backoff, shutdown, promote);
            continue;
        }
        info!("connected to primary (DPDK)");
        // From here the link watches the primary: one that stops
        // answering is reset by the stack, which every wait below (auth,
        // handshake, transfer, streaming) reads as the session's end.
        // Armed now rather than at connect, so the connect keeps its own
        // timeout.
        transport.set_peer_liveness(handle, &REPLICATION_LIVENESS);

        // Authenticate BEFORE the handshake — mirrors the kernel-TCP receiver.
        // The primary issues a Challenge first and will not accept our
        // Handshake until it verifies our signature. Runs the SHARED
        // `authenticate_with_primary` over a blocking adapter that drives the
        // smoltcp poll loop, so TCP and DPDK can never diverge on the auth flow.
        recv_buf.clear();
        let auth_result = {
            let mut auth_stream = PolledAuthStream {
                transport: &mut *transport,
                handle,
                recv_buf: &mut recv_buf,
                shutdown,
                deadline: std::time::Instant::now() + AUTH_TIMEOUT,
            };
            authenticate_with_primary(&mut auth_stream, signing_key)
        };
        if let Err(e) = auth_result {
            transport.reset(handle);
            // A shutdown racing the auth window surfaces here as an auth error
            // — that is a clean exit, not degraded operation, so don't warn
            // (the loop falls through to the shutdown check after the backoff
            // sleep). Promotion never aborts the auth stream, so it can't reach
            // this path spuriously; only shutdown can.
            if shutdown.load(Ordering::Relaxed) {
                debug!("authentication aborted — shutting down (DPDK)");
            } else {
                warn!(
                    error = %e,
                    backoff_secs = backoff.as_secs(),
                    "authentication failed (DPDK) — retrying"
                );
            }
            sleep_then_double_backoff(&mut backoff, shutdown, promote);
            continue;
        }
        info!("authenticated with primary (DPDK)");
        // 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);

        // Send handshake. Advertise our fencing epoch so a stale primary
        // self-demotes when it sees we are ahead (see `crate::fence`).
        send_buf.clear();
        let handshake = Handshake {
            last_sequence,
            chain_hash,
            epoch: fence_state.epoch(),
        };
        encode_handshake(&handshake, &mut send_buf);
        transport.queue_send(handle, &send_buf);
        send_buf.clear();

        // Read protocol response (StreamStart / NeedSnapshot / HashMismatch).
        // Helper macro: shut the pipeline down before bubbling up a fatal
        // error from the handshake. Borrows `pipeline` directly so we don't
        // leak the threads on the way out, and resets the link so the
        // primary frees our slot now, not at its liveness deadline.
        macro_rules! fatal_err_dpdk {
            ($msg:expr) => {{
                transport.reset(handle);
                if let Some(p) = pipeline.take() {
                    let _ = teardown_replica_pipeline::<A, BufferedWriter<A::Event>>(p);
                }
                return Err($msg);
            }};
        }
        recv_buf.clear();
        // `None` from the loop = failure path (disconnect or snapshot
        // error) → reconnect. `Some(lineage)` = StreamStart received (or a
        // resync that re-seeded and validated its own post-snapshot
        // StreamStart inline, via `handle_resync_verdict`).
        // `(segment start, anchor, genesis length)`.
        let stream_lineage: Option<(u64, [u8; 32], Option<u64>)> = 'handshake: loop {
            if shutdown.load(Ordering::Relaxed) {
                // Route through the loop top (via the `None` check
                // below) rather than duplicating its teardown here.
                transport.reset(handle);
                break 'handshake None;
            }
            transport.poll();
            transport.recv_into_vec(handle, &mut recv_buf);

            match try_extract_frame(&recv_buf, MAX_CONTROL_FRAME) {
                FrameResult::Complete(payload_start, frame_end) => {
                    let payload = &recv_buf[payload_start..frame_end];
                    let response = match decode_primary_message(payload) {
                        Ok(response) => response,
                        Err(e) => fatal_err_dpdk!(e.into()),
                    };
                    compact_recv_buf(&mut recv_buf, frame_end);
                    match response {
                        PrimaryMessage::StreamStart {
                            start_sequence,
                            segment_start_sequence,
                            anchor_hash,
                            genesis_entries,
                            epoch,
                        } => {
                            // (A resync rebase adopts the primary's epoch
                            // inside `handle_resync_verdict` instead.)
                            match super::accept_stream_start(
                                super::StreamStart {
                                    start_sequence,
                                    lineage: (segment_start_sequence, anchor_hash, genesis_entries),
                                    epoch,
                                },
                                "dpdk",
                                &pipeline,
                                &mut journal_writer,
                                &fence_state,
                                control,
                            ) {
                                Ok(StreamStartVerdict::Follow(lineage)) => {
                                    break 'handshake Some(lineage);
                                }
                                Ok(StreamStartVerdict::StalePrimary) => {
                                    // Close before sleeping — a handle left
                                    // open here leaks its smoltcp socket on
                                    // every refusal round (the reconnect
                                    // allocates a fresh one) and holds the
                                    // stale primary's connection through the
                                    // backoff.
                                    transport.reset(handle);
                                    sleep_then_double_backoff(&mut backoff, shutdown, promote);
                                    break 'handshake None; // caught by the None check below
                                }
                                Err(e) => fatal_err_dpdk!(e),
                            }
                        }
                        ref resync @ (PrimaryMessage::NeedSnapshot
                        | PrimaryMessage::HashMismatch) => {
                            let divergent = matches!(resync, PrimaryMessage::HashMismatch);
                            let decision = handle_resync_verdict(
                                divergent,
                                &mut DpdkFrameSource {
                                    transport: &mut *transport,
                                    handle,
                                    recv_buf: &mut recv_buf,
                                    shutdown,
                                },
                                &mut pipeline,
                                &mut app,
                                &mut journal_writer,
                                store,
                                &fence_state,
                                control,
                                &mut last_sequence,
                                &mut chain_hash,
                            );
                            match decision {
                                Ok(ResyncDecision::Ready {
                                    segment_start_sequence,
                                    anchor_hash,
                                    genesis_entries,
                                    resume_sequence,
                                }) => {
                                    // DPDK resumes streaming from `last_sequence`
                                    // (the TCP path uses a separate `session_start`).
                                    last_sequence = resume_sequence;
                                    break 'handshake Some((
                                        segment_start_sequence,
                                        anchor_hash,
                                        genesis_entries,
                                    ));
                                }
                                Ok(ResyncDecision::Retry) => {
                                    transport.reset(handle);
                                    sleep_then_double_backoff(&mut backoff, shutdown, promote);
                                    break 'handshake None; // caught by the None check below
                                }
                                Err(e) => fatal_err_dpdk!(e),
                            }
                        }
                        other => {
                            fatal_err_dpdk!(format!("unexpected response: {other:?}").into());
                        }
                    }
                }
                FrameResult::Oversized => {
                    fatal_err_dpdk!("oversized frame from primary during handshake".into());
                }
                FrameResult::Incomplete => {}
            }

            if !transport.is_connected(handle) {
                warn!("disconnected from primary during handshake (DPDK)");
                transport.reset(handle);
                sleep_then_double_backoff(&mut backoff, shutdown, promote);
                break None; // trigger reconnect via the None check below
            }
            std::thread::yield_now();
        };

        // `None` means the handshake loop exited via a failure path
        // (disconnect or snapshot error) — reconnect.
        let Some((lineage_start, lineage_anchor, lineage_genesis)) = stream_lineage else {
            continue;
        };

        // Create journal for fresh replica (first connection only).
        //
        // Gate on `pipeline.is_none()` rather than `journal_writer.is_none()`:
        // the writer is moved into the pipeline on first connect and never
        // returned, so on every subsequent reconnect `journal_writer` is
        // `None` even though a live writer is still mid-stream inside the
        // pipeline. `pipeline.is_none()` distinguishes "true first connect or
        // post-snapshot rebuild" from "reconnect against an existing
        // pipeline" — the latter must not recreate the journal file.
        //
        // The StreamStart lineage carries the segment header identity
        // (starting sequence + chain anchor); 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, for the
        // promotion check (see the kernel-TCP receiver).
        if pipeline.is_none() && journal_writer.is_none() {
            let writer = match BufferedWriter::create_continuing(
                store.journal(),
                lineage_start,
                lineage_anchor,
                lineage_genesis,
            ) {
                Ok(writer) => writer,
                Err(e) => {
                    // Leaving: tell the primary now (see `REPLICATION_LIVENESS`).
                    transport.reset(handle);
                    return Err(e.into());
                }
            };
            app = Some(A::default());
            journal_writer = Some(writer);
        }

        // --- Build pipeline if absent ---
        //
        // Built once on first connect, or after a snapshot transfer tore
        // the previous one down. On disconnect the pipeline lives, so
        // this branch is skipped.
        if pipeline.is_none() {
            // If we still have no state after all the handshake logic,
            // reconnect — on a fresh socket, so this one is reset first
            // (left open, it would hold a slot on the primary).
            if app.is_none() || journal_writer.is_none() {
                transport.reset(handle);
                continue;
            }
            let cur_app = app.take().expect("application initialized");
            let cur_writer = journal_writer.take().expect("journal_writer initialized");

            // On a post-snapshot rebuild this thread is pinned to
            // `cores.reader`, at real-time priority on an isolated core.
            // The builder creates its own threads in their own placement,
            // but it also runs application code here (`prefault`,
            // `clone_via_snapshot`), and a thread that code spawns would
            // inherit this context and never run. Unpinned first; the
            // receiver pins itself again after the build.
            if let Err(e) = melin_app::affinity::clear_affinity() {
                tracing::warn!(error = e, "failed to clear receiver affinity before spawn");
            }

            pipeline = match 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,
            ) {
                Ok(p) => Some(p),
                Err(e) => {
                    // Leaving: tell the primary now (see `REPLICATION_LIVENESS`).
                    transport.reset(handle);
                    return Err(e);
                }
            };

            // Pipeline children are spawned and self-pinned. Now safe to
            // pin the receive thread — mirrors the primary's reader pin
            // so the thread producing input-ring entries from the network
            // isn't migrated across L3s mid-batch.
            melin_app::affinity::pin_thread("receiver", cores.reader.core);
        }

        // --- Streaming session (transport-agnostic) ---
        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;
            let mut dpdk_transport = DpdkReceiverTransport {
                transport: &mut *transport,
                handle,
                send_buf: std::mem::take(&mut send_buf),
            };
            // `last_sequence` is the session's resume point — updated to
            // the snapshot sequence after a transfer, so it anchors the
            // contiguity gate on both negotiation paths.
            let r = streaming_loop::<DpdkReceiverTransport<'_>, A::Event>(
                &mut dpdk_transport,
                input_producer,
                journal_cursor,
                shutdown,
                control,
                pipeline_depth,
                cores.reader.wait,
                last_sequence,
                std::mem::take(&mut recv_buf),
                None,
                stream_marks,
                journal_failed,
            );
            send_buf = dpdk_transport.send_buf;
            r
        };

        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,
            // End the smoltcp session on every exit, a stop or a promotion
            // included. The TCP connection to the primary may still be
            // healthy (a locally detected divergence, a shutdown), so
            // without the reset the primary's slot stays occupied until
            // its liveness deadline — a repair handshake refused, or a
            // stopped replica still counted, meanwhile. Closing also
            // returns the socket entry to the socket set; each reconnect
            // allocates a fresh one, so skipping it leaks one entry per
            // disconnect.
            || end_session(transport, handle),
            sizing,
        );
        if let Some(r) = after.adopt(
            &mut app,
            &mut journal_writer,
            &mut last_sequence,
            &mut chain_hash,
        ) {
            return r;
        }
    }
}

/// DPDK control-frame source: polls the smoltcp transport and extracts
/// one length-prefixed frame per call. Drives the shared
/// [`receive_chunked_body`] and the resync prologue reads
/// (`SnapshotBegin` / `SegmentSeedBegin`). See [`ControlFrameSource`].
struct DpdkFrameSource<'a> {
    transport: &'a mut melin_dpdk::DpdkTransport,
    handle: melin_dpdk::SocketHandle,
    recv_buf: &'a mut Vec<u8>,
    shutdown: &'a AtomicBool,
}

impl ControlFrameSource for DpdkFrameSource<'_> {
    fn next_frame(
        &mut self,
        max_size: usize,
    ) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
        loop {
            if self.shutdown.load(Ordering::Relaxed) {
                return Err("shutdown during transfer".into());
            }
            self.transport.poll();
            self.transport.recv_into_vec(self.handle, self.recv_buf);

            match try_extract_frame(self.recv_buf, max_size) {
                FrameResult::Complete(payload_start, frame_end) => {
                    let payload = self.recv_buf[payload_start..frame_end].to_vec();
                    compact_recv_buf(self.recv_buf, frame_end);
                    return Ok(payload);
                }
                FrameResult::Oversized => {
                    return Err("oversized frame during transfer".into());
                }
                FrameResult::Incomplete => {}
            }

            // A frame arriving in the same poll as the FIN is returned
            // above before we observe the disconnect here (the link stays
            // up while unread bytes remain).
            if !self.transport.is_connected(self.handle) {
                return Err("disconnected during transfer".into());
            }
            std::thread::yield_now();
        }
    }

    /// The work runs on a helper thread while this one keeps polling the
    /// stack (without reading), so the primary's probes and data are
    /// answered — its window closing as the socket buffer fills — however
    /// long the work takes.
    fn serviced<R: Send>(&mut self, work: impl FnOnce() -> R + Send) -> R {
        run_serviced(work, || {
            self.transport.poll();
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A length-prefixed frame with a `len`-byte payload.
    fn frame(len: usize) -> Vec<u8> {
        let mut f = (len as u32).to_le_bytes().to_vec();
        f.resize(4 + len, 0xAB);
        f
    }

    fn ack(acked_sequence: u64, in_memory_sequence: u64) -> Vec<u8> {
        let mut buf = Vec::new();
        encode_ack(
            &Ack {
                acked_sequence,
                in_memory_sequence,
            },
            &mut buf,
        );
        buf
    }

    /// The newest ack survives, ahead of a partial frame; older acks and
    /// frames that are not acks go. Kept across reads, so the newest of
    /// all of them is what the streaming arm records.
    #[test]
    fn keep_last_ack_keeps_the_newest_ack_and_a_partial_frame() {
        let mut buf = ack(1, 2);
        buf.extend(frame(8));
        buf.extend(ack(3, 4));
        buf.extend(frame(8));
        let partial = ack(5, 6);
        buf.extend_from_slice(&partial[..5]);

        let acked = JoinFrames::Read { acked: true };
        assert_eq!(keep_last_ack(&mut buf), acked);
        let mut expected = ack(3, 4);
        expected.extend_from_slice(&partial[..5]);
        assert_eq!(buf, expected);

        // The partial ack completes, and is the newest from then on.
        buf.extend_from_slice(&partial[5..]);
        assert_eq!(keep_last_ack(&mut buf), acked);
        assert_eq!(buf, ack(5, 6));
        assert_eq!(keep_last_ack(&mut buf), acked, "idempotent");
        assert_eq!(buf, ack(5, 6));
    }

    #[test]
    fn keep_last_ack_with_no_ack_keeps_only_a_partial_frame() {
        let none = JoinFrames::Read { acked: false };
        let mut buf = frame(8);
        buf.extend_from_slice(&frame(8)[..3]);
        assert_eq!(keep_last_ack(&mut buf), none);
        assert_eq!(buf, frame(8)[..3]);

        let mut buf = Vec::new();
        assert_eq!(keep_last_ack(&mut buf), none);
        assert!(buf.is_empty());

        // Too short for a length prefix: left as it is.
        let mut buf = vec![5, 0];
        assert_eq!(keep_last_ack(&mut buf), none);
        assert_eq!(buf, [5, 0]);

        // A partial frame that completes as something other than an ack
        // goes on the next read.
        let partial = frame(12);
        let mut buf = partial[..7].to_vec();
        assert_eq!(keep_last_ack(&mut buf), none);
        buf.extend_from_slice(&partial[7..]);
        assert_eq!(keep_last_ack(&mut buf), none);
        assert!(buf.is_empty());
    }

    /// The replica's only ack of a join, arriving in pieces, is reported
    /// once complete and then kept, read after read, with nothing newer
    /// behind it: a replica on a quiet primary acks the end of its catch-up
    /// once, and the streaming arm must still find that ack when the join
    /// is over, however many reads the rest of the join took.
    #[test]
    fn keep_last_ack_keeps_the_only_ack_until_the_join_is_over() {
        let only = ack(1_560, 1_560);
        let mut buf = only[..5].to_vec();
        assert_eq!(keep_last_ack(&mut buf), JoinFrames::Read { acked: false });
        buf.extend_from_slice(&only[5..]);
        for read in 0..3 {
            assert_eq!(
                keep_last_ack(&mut buf),
                JoinFrames::Read { acked: true },
                "read {read}"
            );
            assert_eq!(buf, only, "read {read}");
        }
    }

    #[test]
    fn keep_last_ack_rejects_an_oversized_or_empty_frame() {
        let mut buf = ack(1, 1);
        buf.extend(((MAX_CONTROL_FRAME + 1) as u32).to_le_bytes());
        assert_eq!(keep_last_ack(&mut buf), JoinFrames::Malformed);

        let mut buf = 0u32.to_le_bytes().to_vec();
        assert_eq!(keep_last_ack(&mut buf), JoinFrames::Malformed);
    }

    /// The deadline on a joiner that stops reading is the liveness
    /// timeout, and long enough that a refusal or two is not a stall.
    #[test]
    fn the_join_stall_limit_is_the_liveness_timeout() {
        assert_eq!(JOIN_STALL_LIMIT, REPLICATION_LIVENESS.timeout());
        assert!(JOIN_STALL_LIMIT >= std::time::Duration::from_secs(1));
    }
}