melin-server-runtime 0.20.0

//! DPDK transport integration — single poll thread for NIC I/O + TCP.
//!
//! Replaces both the io_uring reader and the response stage's socket
//! writes. A single DPDK poll thread owns all NIC I/O:
//!
//! - **Inbound**: `rx_burst` → smoltcp → frame decode → disruptor publish
//! - **Outbound**: response SPSC → per-connection TX queue → smoltcp → `tx_burst`
//! - **Tick**: cadence comparison between bursts → `JournalEvent::Tick { now_ns }`
//!   onto the same input ring (see `run_dpdk_poll` for the single-poll-thread
//!   invariant assumed here).
//!
//! The response stage still runs on its own pinned thread for cursor
//! gating and encoding, but instead of calling `write_all` on kernel
//! sockets, it pushes encoded frames into a lock-free SPSC queue per
//! connection. The DPDK poll thread drains these into smoltcp sockets.
//!
//! # Auth handshake
//!
//! New connections start in `AuthState::WaitingForResponse` — the poll loop
//! sends a Challenge frame and waits for the ChallengeResponse. Auth is
//! non-blocking: bytes accumulate in `parse_buf` across poll iterations
//! until a complete frame arrives. Connections that don't complete auth
//! within `AUTH_TIMEOUT` are dropped. A failed attempt is answered with
//! `AuthFailed` and the connection is closed once the peer has it.
//!
//! # Thread model
//!
//! ```text
//! Core N:   DPDK poll thread  (rx_burst, smoltcp, frame decode, tx_burst)
//! Core 1:   Journal stage     (unchanged)
//! Core 2:   Matching stage    (unchanged)
//! Core 3:   Response stage    (encodes to SPSC queues instead of kernel sockets)
//! ```

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::time::{Duration, Instant};

use rustc_hash::FxHashMap;

use melin_app::Application;
use melin_app::auth::{AuthorizedKeys, ClientRole, KeyRole, RoleId};
use melin_app::unix_epoch_nanos;
use melin_dpdk::transport::DpdkTransport;
use melin_pipeline::ring;
use melin_transport_core::pipeline::InputSlot;
use melin_transport_core::tick::TickSchedule;
#[cfg(feature = "latency-trace")]
use melin_transport_core::trace::mono_trace_ns;
use melin_wire_protocol::control::ConnectionId;
use melin_wire_protocol::control::TransportResponse;
use melin_wire_protocol::control_codec;
use melin_wire_protocol::framing::split_frame_limited;
use rand::RngExt;

use melin_dpdk::SocketHandle;
use tracing::{debug, warn};

use crate::dpdk_response::{ControlEvent, TxFrame};
use crate::halt::{HaltGate, RefusalSender};
use crate::policy_control::AckPolicyPublisher;

use crate::client_auth::MAX_AUTH_FRAME;
use crate::client_frames::MAX_FRAME_SIZE;

/// Auth handshake timeout. Connections that don't complete auth within
/// this window are dropped.
const AUTH_TIMEOUT: Duration = Duration::from_secs(5);

/// Auth handshake state machine. Drives the Challenge → ChallengeResponse
/// → ServerReady flow non-blockingly across poll iterations.
enum AuthState {
    /// Challenge frame has been queued for sending. Waiting for the
    /// ChallengeResponse frame from the client.
    WaitingForResponse {
        /// The nonce sent in the Challenge. Needed to verify the signature.
        nonce: [u8; 32],
        /// When the connection was accepted. Used for timeout.
        accepted_at: Instant,
    },
    /// Auth completed successfully. Connection is ready for requests.
    Authenticated { role: ClientRole<RoleId> },
    /// Auth failed and `AuthFailed` has been queued. Terminal: nothing
    /// the client sends is read any more, so it gets one attempt per
    /// connection, as on the io_uring path.
    ///
    /// The connection lingers only until the peer has acknowledged the
    /// `AuthFailed` frame, because `DpdkTransport::close` aborts and
    /// removes the socket without another dispatch — closing at once
    /// would discard the reason unsent. `AUTH_TIMEOUT`, still counted
    /// from accept, bounds the linger if the peer never acknowledges.
    Rejected {
        /// When the connection was accepted. Used for timeout.
        accepted_at: Instant,
    },
}

/// Per-connection state in the DPDK poll thread.
struct ConnectionState {
    connection_id: ConnectionId,
    addr: SocketAddr,
    handle: SocketHandle,
    auth: AuthState,
    /// FxHash of the client's Ed25519 public key. Set after auth,
    /// copied into every InputSlot as the submitting key's identity.
    key_hash: u64,
    /// Incremental frame parsing state: accumulates bytes until a
    /// complete length-prefixed frame is available.
    parse_buf: Vec<u8>,
    /// Last time data was received from this connection. Used for
    /// idle timeout (drops connections that stop sending).
    last_activity: Instant,
}

/// Run the DPDK poll loop.
///
/// This replaces the io_uring reader. It accepts connections, drives
/// auth handshakes, parses frames, publishes events to the disruptor,
/// and drains the TX channel from the response stage into smoltcp sockets.
/// When `tick_cadence` is `Some`, this thread also generates the application's
/// clock ticks via a wall-clock comparison between NIC bursts.
///
/// Called from a dedicated OS thread pinned to its own core.
///
/// **Single-poll-thread invariant.** Today `dpdk_num_queues` returns at
/// most 2 (one client queue + one replication-sender queue), so there is
/// always exactly one client poll thread. If we ever add multi-queue RSS
/// (multiple client poll threads), every poll thread would emit ticks
/// concurrently — re-introducing the multi-producer ordering race that
/// embedding tick into the poll thread was designed to eliminate. Gate
/// tick emission to a single thread (e.g. `thread_id == 0`) before
/// scaling out client queues.
///
/// Top-level thread entry point — the wide arg list mirrors transport
/// state owned elsewhere; bundling into a config struct adds indirection
/// without simplifying.
#[allow(clippy::too_many_arguments)]
pub fn run_dpdk_poll<A: Application>(
    mut transport: DpdkTransport,
    mut producer: ring::Producer<InputSlot<A::Event>>,
    decoder: crate::reader::RequestDecoderArc<A>,
    // While it refuses writes, they are answered through `refusals`
    // instead of published — see `crate::halt`.
    halt: HaltGate,
    mut refusals: RefusalSender<A::Report>,
    // The admin `ACK-POLICY` command's requests, journaled by this thread
    // as the ring's one producer — see `crate::policy_control`.
    mut ack_policy: AckPolicyPublisher,
    control_tx: mpsc::Sender<ControlEvent>,
    mut tx_rx: melin_pipeline::spsc::Consumer<TxFrame>,
    shutdown: &AtomicBool,
    authorized_keys: Arc<AuthorizedKeys>,
    connection_timeout: Option<Duration>,
    tick_cadence: Option<Duration>,
    max_connections: u64,
    active_connections: Arc<std::sync::atomic::AtomicU64>,
    thread_id: u8,
    // Optional replication driver. When `Some`, this thread owns BOTH
    // client traffic on the client port and replication traffic on the
    // configured replication listen port. Connections accepted on the
    // replication port are handed to the driver; per-iteration the
    // driver's `tick()` advances each replica slot's state machine
    // (handshake, journal catch-up, snapshot transfer, live streaming,
    // ack processing). Replaces the separate replication-sender thread —
    // see `feat/dpdk-single-queue` for the rationale (RSS routing on
    // iavf made the previous multi-queue split unworkable).
    repl_driver: Option<crate::replication::DpdkReplicationDriver<A>>,
    // TCP port the replication driver listens on. Used to filter
    // `AcceptedConnection::listen_port` so client connections go to the
    // client handler and replication connections to the driver. Ignored
    // when `repl_driver` is None.
    repl_listen_port: u16,
) {
    let mut repl_driver = repl_driver;
    // Per-connection state indexed by `SocketHandle::index()`. Dense
    // `Vec<Option<_>>` beats a HashMap on the DPDK hot path: the HashMap
    // hashing + probe cost was the top remaining hotspot after the
    // clock-read cleanup. MAX_CONNECTIONS matches the SocketSet capacity,
    // so handle indices are always in-bounds.
    let mut connections: Vec<Option<ConnectionState>> =
        (0..melin_dpdk::MAX_CONNECTIONS).map(|_| None).collect();
    // Reverse lookup from response-stage connection_id to SocketHandle.
    // FxHash instead of SipHash — u64 keys, no HashDoS surface internally.
    let mut id_to_handle: FxHashMap<u64, SocketHandle> =
        FxHashMap::with_capacity_and_hasher(256, Default::default());
    let mut next_connection_id: u64 = 1;
    // Occupied-slot count for the max_connections gate. Cheaper than
    // scanning `connections` on every accept.
    let mut connection_count: usize = 0;
    // Exclusive upper bound of the `connections` range that has ever
    // been used. Lets the per-poll loop scan just the active range
    // instead of all MAX_CONNECTIONS slots — skipping ~1000 None slots
    // per poll was a chunk of `run_dpdk_poll` self-time. smoltcp reuses
    // closed slab indices before extending, so this grows to the steady-
    // state watermark and stays there.
    let mut conn_range_end: usize = 0;

    // Reader-stage histogram. Single sample per published frame
    // covering recv_ts → batch.push_with completion (decode + auth
    // dispatch + slot construction + push). DPDK has no separate
    // publish-call histogram because `batch.push_with` is a slot
    // assignment in a pre-allocated batch — the work is dominated by
    // the surrounding decode, which is what `ingest` measures.
    #[cfg(feature = "latency-trace")]
    let mut publish_rec =
        melin_transport_core::trace::register_stage("dpdk: publish (decode → disruptor publish)");
    #[cfg(feature = "tick-to-trade")]
    let mut ingest_rec =
        melin_transport_core::trace::register_stage("reader: ingest (recv_ts → publish complete)");

    // Pre-allocated parse buffer pool. Avoids heap allocation on accept
    // by recycling buffers from disconnected connections.
    let mut parse_buf_pool: Vec<Vec<u8>> = (0..256)
        .map(|_| Vec::with_capacity(MAX_FRAME_SIZE + 4))
        .collect();

    // RNG for auth nonces. `rand::rng()` is a CSPRNG (ChaCha-based,
    // seeded and periodically reseeded from OS entropy), as
    // `client_auth::verify_client` requires: a predictable nonce would let
    // a response be obtained in advance — signed by the client ahead of
    // the challenge — and presented when that nonce is issued. It
    // generates in user space, so the poll thread never blocks on a
    // syscall per nonce.
    let mut rng = rand::rng();

    // Tick generator state. The DPDK poll thread is a tight busy-spin
    // loop, so unlike the io_uring reader it does not need a timeout
    // primitive — a wall-clock comparison between bursts is enough.
    // `tick_check_interval` amortises `Instant::now()` over many poll
    // iterations: at typical ~5M iterations/sec, checking every 4096
    // iterations stays well within microseconds of the deadline while
    // making the per-iteration overhead unmeasurable.
    //
    // The same counter paces the check for an `ACK-POLICY` request: a
    // request waits at most a few thousand polls, microseconds, and the
    // poll thread never parks, so it needs no wakeup.
    let tick_enabled = tick_cadence.is_some();
    let cadence = tick_cadence.unwrap_or(Duration::ZERO);
    let mut tick_schedule = TickSchedule::new(cadence, Instant::now());
    let mut tick_check_counter: u32 = 0;
    const TICK_CHECK_INTERVAL: u32 = 4096;

    // Auth / idle-timeout checks call `Instant::now()` via `elapsed()`,
    // which showed up as ~13% of the poll core in perf. The timeouts are
    // ~seconds, so firing the check once per ~1M polls (~100ms at 10M
    // polls/sec) is still orders of magnitude tighter than the deadline.
    let mut slow_check_counter: u32 = 0;
    const SLOW_CHECK_INTERVAL: u32 = 1024 * 1024;
    if tick_enabled {
        tracing::info!(
            cadence_ms = cadence.as_millis() as u64,
            thread_id,
            "tick generator integrated into DPDK poll thread"
        );
    }

    // Outer-loop wall-time histogram, gated on at least one byte received
    // this iteration. Idle iterations would otherwise drown the percentiles
    // in ~100ns samples; what we want is "how long does a poll cycle take
    // when there's actual work to do" — which is the cycle an in-flight
    // request experiences. Registered with the global stats registry; the
    // /stats-dump endpoint snapshots it alongside the other stages.
    #[cfg(feature = "latency-trace")]
    let mut poll_iter_rec = melin_transport_core::trace::register_stage(
        "dpdk poll: outer iteration (work-iterations only)",
    );
    #[cfg(feature = "latency-trace")]
    let mut poll_iter_start = mono_trace_ns();
    // Paces the recorder flush. The poll thread never idles — it
    // busy-spins — but it stops *recording* as soon as traffic stops,
    // which is enough to strand its samples in the thread-local buffer.
    // Nanoseconds rather than `Instant` so the gate reuses the
    // `mono_trace_ns` read the poll-iteration histogram already takes.
    #[cfg(feature = "latency-trace")]
    let stats_flush_interval_ns =
        melin_transport_core::trace::IDLE_FLUSH_INTERVAL.as_nanos() as u64;
    #[cfg(feature = "latency-trace")]
    let mut last_stats_flush_ns = poll_iter_start;

    loop {
        if shutdown.load(Ordering::Relaxed) {
            // Tell the replicas this primary has gone, as its kernel
            // would on kernel TCP; a DPDK node that just stops says
            // nothing, leaving them to their liveness deadline.
            if let Some(driver) = repl_driver.take() {
                driver.close_all_links(&mut transport);
            }
            break;
        }

        // Set on any bytes received this iteration; gates poll_iter_hist
        // recording so idle iterations don't drown the percentiles.
        #[cfg(feature = "latency-trace")]
        let mut work_done_this_iter = false;

        // Once every TICK_CHECK_INTERVAL poll iterations: journal a
        // waiting `ACK-POLICY` request, and compare the wall clock to the
        // tick deadline, emitting a tick if due.
        tick_check_counter = tick_check_counter.wrapping_add(1);
        if tick_check_counter >= TICK_CHECK_INTERVAL {
            tick_check_counter = 0;
            ack_policy.poll(&mut producer, &halt);
            if tick_enabled {
                // Nothing to re-arm on a published tick: the poll loop
                // never sleeps, so the counter above is its only timer.
                tick_schedule.publish_if_due(Instant::now(), unix_epoch_nanos, &mut producer);
            }
        }

        // TX drain macro — used at multiple points in the loop to move
        // response frames from the SPSC queue into smoltcp socket buffers
        // so the next transport.poll() flushes them to the wire.
        macro_rules! drain_tx {
            () => {
                while let Some((_seq, frame)) = tx_rx.try_consume() {
                    if let Some(&handle) = id_to_handle.get(&frame.connection_id)
                        && !transport.queue_send(handle, frame.as_bytes())
                    {
                        debug!(
                            connection_id = frame.connection_id,
                            "DPDK: TX queue overflow, dropping connection"
                        );
                        transport.close(handle);
                        let _ = control_tx.send(ControlEvent::Disconnected {
                            connection_id: frame.connection_id,
                        });
                        id_to_handle.remove(&frame.connection_id);
                        release_slot(
                            &mut connections,
                            handle.index(),
                            &mut parse_buf_pool,
                            &mut connection_count,
                        );
                    }
                }
            };
        }

        // 1. Drain TX frames from the response stage BEFORE polling.
        // queue_send() stores data in TxQueue; the subsequent poll()
        // calls flush_tx_queues → iface.poll → flush_tx to push it
        // onto the wire in the same iteration. Previously the drain
        // happened after poll, adding a full loop iteration of latency.
        drain_tx!();

        // 2. Poll NIC + smoltcp (flushes TX we just drained + receives RX).
        transport.poll();

        // 3. Accept new connections — dispatch by listen port so the
        //    replication driver gets its own connections, client logic
        //    only sees client-port connections.
        for accepted in transport.take_accepted() {
            if let Some(ref mut driver) = repl_driver
                && accepted.listen_port == repl_listen_port
            {
                driver.accept_connection(accepted.peer, accepted.handle, &mut transport);
                continue;
            }

            // Enforce max_connections limit.
            if crate::connection_limit::connection_cap_reached(
                connection_count as u64,
                max_connections,
            ) {
                warn!(
                    peer = %accepted.peer,
                    "DPDK: connection rejected: max_connections reached"
                );
                transport.close(accepted.handle);
                continue;
            }

            // Encode thread_id in bits 56..63 of connection_id for O(1)
            // response routing. Bits 0..55 are the per-thread sequence.
            let conn_id = ConnectionId((thread_id as u64) << 56 | next_connection_id);
            next_connection_id += 1;

            debug!(
                connection_id = conn_id.0,
                peer = %accepted.peer,
                "DPDK: new connection, starting auth"
            );

            // Fresh random nonce for the challenge, from the CSPRNG above.
            let nonce: [u8; 32] = rng.random();

            // Send the Challenge frame immediately.
            let mut challenge_buf = [0u8; 128];
            let written = control_codec::encode_transport_response(
                &TransportResponse::Challenge { nonce },
                &mut challenge_buf,
            )
            .expect("challenge encodes");
            transport.queue_send(accepted.handle, &challenge_buf[..written]);

            let accepted_idx = accepted.handle.index();
            if accepted_idx + 1 > conn_range_end {
                conn_range_end = accepted_idx + 1;
            }
            connections[accepted_idx] = Some(ConnectionState {
                connection_id: conn_id,
                addr: accepted.peer,
                handle: accepted.handle,
                auth: AuthState::WaitingForResponse {
                    nonce,
                    accepted_at: Instant::now(),
                },
                key_hash: 0,
                // Reuse a pre-allocated buffer from the pool, or allocate
                // if the pool is exhausted (more connections than pre-allocated).
                parse_buf: parse_buf_pool
                    .pop()
                    .unwrap_or_else(|| Vec::with_capacity(MAX_FRAME_SIZE + 4)),
                last_activity: Instant::now(),
            });
            connection_count += 1;
        }

        // 4. Read data from all connections and process.
        // Mid-iteration: drain TX + poll every N connections to keep
        // responses flowing and the NIC busy without waiting for the
        // full connection iteration to complete.
        const POLL_EVERY_N_CONNS: usize = 4;

        // One wall-clock read per outer poll iteration, reused for
        // every request stamped in this pass. Sub-microsecond precision
        // loss at DPDK poll rates; the timestamp drives the application's
        // clock but orders nothing (the pipeline orders by sequence).
        // Deferred until we actually stamp a frame — `clock_gettime`
        // dominates the profile on idle polls with no traffic.
        let mut batch_wall_ns: Option<u64> = None;

        slow_check_counter = slow_check_counter.wrapping_add(1);
        let do_slow_checks = slow_check_counter.is_multiple_of(SLOW_CHECK_INTERVAL);

        // Per-connection batching with a visibility-delay cap happens
        // inside `process_trading_frames`. The previous design opened a
        // single cross-connection batch here and committed it once at
        // the end of every outer poll iteration — fewer cursor stores
        // but unbounded visibility delay for the first frame on any
        // given connection. The kernel-TCP reader hit the same trade-off;
        // see `reader.rs::process_frames` for the measured numbers
        // motivating the cap. Per-connection scope keeps a 10×+ cursor
        // store reduction vs per-event publish, with bounded tail-delay.
        //
        // Counts occupied slots we actually process, to drive the
        // mid-iteration `transport.poll()` cadence.
        let mut active_idx: usize = 0;
        // Indexed loop, not iter_mut: remove paths do `connections[idx].take()`
        // which conflicts with an outer iterator borrow. Bounded by
        // `conn_range_end` (not `connections.len()`) so idle polls don't
        // scan MAX_CONNECTIONS empty slots.
        #[allow(clippy::needless_range_loop)]
        for idx in 0..conn_range_end {
            if connections[idx].is_none() {
                continue;
            }
            if active_idx > 0 && active_idx.is_multiple_of(POLL_EVERY_N_CONNS) {
                // Drain fresh responses before polling so they're flushed
                // to the wire in this poll cycle instead of the next.
                drain_tx!();
                transport.poll();
                // Drive the replication driver at the same cadence as the
                // mid-loop NIC poll. Without this, `tick()` only fires once
                // per outer iteration; under heavy client load the journal
                // stage produces replication batches faster than the driver
                // drains the ring, and the journal evicts both replicas with
                // "ring backpressure timeout".
                if let Some(ref mut driver) = repl_driver {
                    driver.tick(&mut transport, shutdown);
                }
                // drain_tx may have closed the connection at this idx
                if connections[idx].is_none() {
                    continue;
                }
            }
            active_idx += 1;
            let conn = match connections[idx].as_mut() {
                Some(c) => c,
                None => continue,
            };
            // SocketHandle is Copy; capture before any &mut conn borrows
            // so process_auth_frame can take it after the conn borrow.
            let conn_handle = conn.handle;

            // Check auth timeout for pending connections, and for rejected
            // ones whose peer never acknowledged `AuthFailed`. Throttled to
            // avoid a per-poll `Instant::now()` via `elapsed()`.
            if do_slow_checks
                && let AuthState::WaitingForResponse { accepted_at, .. }
                | AuthState::Rejected { accepted_at } = &conn.auth
                && accepted_at.elapsed() > AUTH_TIMEOUT
            {
                debug!(
                    connection_id = conn.connection_id.0,
                    addr = %conn.addr,
                    "DPDK: auth timeout, dropping connection"
                );
                transport.close(conn.handle);
                release_slot(
                    &mut connections,
                    idx,
                    &mut parse_buf_pool,
                    &mut connection_count,
                );
                continue;
            }

            // A rejected connection is closed as soon as the peer has
            // acknowledged `AuthFailed` (see `AuthState::Rejected`).
            if matches!(conn.auth, AuthState::Rejected { .. }) && transport.tx_drained(conn.handle)
            {
                debug!(
                    connection_id = conn.connection_id.0,
                    addr = %conn.addr,
                    "DPDK: AuthFailed delivered, dropping connection"
                );
                transport.close(conn.handle);
                release_slot(
                    &mut connections,
                    idx,
                    &mut parse_buf_pool,
                    &mut connection_count,
                );
                continue;
            }

            // Guard against unbounded buffer growth — check before recv.
            const MAX_PARSE_BUF: usize = 65536;
            if conn.parse_buf.len() >= MAX_PARSE_BUF {
                debug!(
                    connection_id = conn.connection_id.0,
                    buf_len = conn.parse_buf.len(),
                    "parse buffer exceeded limit, dropping connection"
                );
                transport.close(conn.handle);
                let _ = control_tx.send(ControlEvent::Disconnected {
                    connection_id: conn.connection_id.0,
                });
                id_to_handle.remove(&conn.connection_id.0);
                release_slot(
                    &mut connections,
                    idx,
                    &mut parse_buf_pool,
                    &mut connection_count,
                );
                continue;
            }

            // Read directly into parse_buf, skipping the intermediate
            // read_buf stack copy.
            let n = transport.recv_into_vec(conn.handle, &mut conn.parse_buf);
            if n > 0 {
                conn.last_activity = Instant::now();
                #[cfg(feature = "latency-trace")]
                {
                    work_done_this_iter = true;
                }
            }
            if n == 0 {
                // Gone, or half-closed: the client's FIN leaves the socket
                // in close-wait, which the stack still counts as active,
                // and with its data drained the connection is over, as it
                // is on kernel TCP at EOF. Without this the slot was held
                // until the next heartbeat's RST, or for good with
                // heartbeats off.
                if !transport.is_active(conn.handle) || transport.peer_closed(conn.handle) {
                    debug!(
                        connection_id = conn.connection_id.0,
                        addr = %conn.addr,
                        "DPDK: connection closed"
                    );
                    if matches!(conn.auth, AuthState::Authenticated { .. }) {
                        let _ = control_tx.send(ControlEvent::Disconnected {
                            connection_id: conn.connection_id.0,
                        });
                    }
                    transport.close(conn.handle);
                    id_to_handle.remove(&conn.connection_id.0);
                    release_slot(
                        &mut connections,
                        idx,
                        &mut parse_buf_pool,
                        &mut connection_count,
                    );
                }
                // Check idle timeout when no data was received. Throttled
                // to avoid a per-poll `Instant::now()` via `elapsed()`.
                else if do_slow_checks
                    && let Some(timeout) = connection_timeout
                    && matches!(conn.auth, AuthState::Authenticated { .. })
                    && conn.last_activity.elapsed() > timeout
                {
                    debug!(
                        connection_id = conn.connection_id.0,
                        addr = %conn.addr,
                        "DPDK: idle timeout, dropping connection"
                    );
                    transport.close(conn.handle);
                    let _ = control_tx.send(ControlEvent::Disconnected {
                        connection_id: conn.connection_id.0,
                    });
                    id_to_handle.remove(&conn.connection_id.0);
                    release_slot(
                        &mut connections,
                        idx,
                        &mut parse_buf_pool,
                        &mut connection_count,
                    );
                }
                continue;
            }

            // Trace timestamp: the moment `recv_into_vec` handed us these
            // bytes. Captured once per recv — not per frame — and stamped
            // onto every InputSlot parsed below, so the reader-ingest and
            // server-e2e stages measure from true wire receipt (frame decode
            // included) rather than re-sampling after each decode.
            #[allow(clippy::let_unit_value)] // ZST when latency-trace is off
            let recv_ts = melin_transport_core::trace::mono_trace_ns();

            // Process frames based on auth state.
            let was_waiting = matches!(conn.auth, AuthState::WaitingForResponse { .. });
            match &conn.auth {
                AuthState::WaitingForResponse { .. } => {
                    // Try to extract the ChallengeResponse frame.
                    process_auth_frame(
                        conn,
                        &mut transport,
                        &authorized_keys,
                        &control_tx,
                        &mut id_to_handle,
                        conn_handle,
                    );
                }
                // Terminal: whatever the client sends after a failed
                // attempt is discarded unread.
                AuthState::Rejected { .. } => conn.parse_buf.clear(),
                AuthState::Authenticated { role } => {
                    use crate::client_frames::{FrameAction, process_client_frames};
                    let role = *role;
                    let action = process_client_frames::<A>(
                        &mut conn.parse_buf,
                        conn.connection_id.0,
                        conn.key_hash,
                        role,
                        &mut producer,
                        &*decoder,
                        &halt,
                        &mut refusals,
                        *batch_wall_ns.get_or_insert_with(unix_epoch_nanos),
                        recv_ts,
                        #[cfg(feature = "latency-trace")]
                        &mut publish_rec,
                        #[cfg(feature = "tick-to-trade")]
                        &mut ingest_rec,
                    );
                    if matches!(action, FrameAction::Disconnect | FrameAction::PipelineFull) {
                        if matches!(action, FrameAction::PipelineFull) {
                            debug!(
                                connection_id = conn.connection_id.0,
                                "DPDK: pipeline full, dropping connection"
                            );
                        }
                        transport.close(conn.handle);
                        let _ = control_tx.send(ControlEvent::Disconnected {
                            connection_id: conn.connection_id.0,
                        });
                        id_to_handle.remove(&conn.connection_id.0);
                        release_slot(
                            &mut connections,
                            idx,
                            &mut parse_buf_pool,
                            &mut connection_count,
                        );
                        continue;
                    }
                }
            }

            // Track auth → authenticated transition for the connection counter.
            if was_waiting && matches!(conn.auth, AuthState::Authenticated { .. }) {
                active_connections.fetch_add(1, Ordering::Relaxed);
            }
        }

        // Final drain + poll before the replication tick so fresh ACK
        // data from replicas is available (arrived since the last mid-
        // iteration poll) and any remaining client TX is flushed.
        drain_tx!();
        transport.poll();

        // Drive the replication driver's per-iteration work — handshake
        // progression, a joining replica's catch-up (read off the disk by
        // the slot's join worker, sent from here a tick at a time), ack
        // processing, and live data-batch sends. With a single
        // queue + single thread, this is what replaces the previous
        // dedicated replication sender thread; the poll above flushed any TX
        // the driver queued on the prior iteration and received any
        // pending ACKs from replicas.
        if let Some(ref mut driver) = repl_driver {
            driver.tick(&mut transport, shutdown);
        }

        #[cfg(feature = "latency-trace")]
        {
            let now = mono_trace_ns();
            // Skip records once shutdown has been observed: matches the
            // gate on the journal / matching / response stages and keeps
            // diagnostic numbers comparable across runs.
            if work_done_this_iter && !shutdown.load(Ordering::Relaxed) {
                poll_iter_rec.record_elapsed(poll_iter_start, now);
            }
            poll_iter_start = now;

            // Hand buffered samples to the stats registry so a
            // /stats-dump taken after traffic stops still sees them.
            if now.saturating_sub(last_stats_flush_ns) >= stats_flush_interval_ns {
                last_stats_flush_ns = now;
                poll_iter_rec.flush();
                publish_rec.flush();
                #[cfg(feature = "tick-to-trade")]
                ingest_rec.flush();
            }
        }
    }
}

/// What a pending connection's parse buffer amounts to, auth-wise.
#[derive(Debug)]
enum AuthVerdict {
    /// No complete auth frame yet — wait for more bytes.
    Incomplete,
    /// The frame is oversized, undecodable, or carries a signature that
    /// does not admit the client. Answered with `AuthFailed`.
    Rejected,
    /// The client signed the nonce with a key listed for `role`.
    Admitted {
        role: ClientRole<RoleId>,
        public_key: [u8; 32],
    },
}

/// Judge the auth frame at the head of `parse_buf` against `nonce`.
///
/// A complete frame is consumed from the buffer, so on admission the bytes
/// the client pipelined behind it stay in place for the data path. On
/// rejection the caller drops the whole buffer, and an oversized frame is
/// rejected on its prefix without being consumed.
fn evaluate_auth_frame(
    parse_buf: &mut Vec<u8>,
    nonce: &[u8; 32],
    authorized_keys: &AuthorizedKeys,
    connection_id: u64,
) -> AuthVerdict {
    let (decoded, consumed) = match split_frame_limited(parse_buf, MAX_AUTH_FRAME) {
        // Borrow the payload directly — no heap allocation.
        Ok(Some(frame)) => (
            control_codec::decode_challenge_response(frame.payload),
            frame.consumed(),
        ),
        Ok(None) => return AuthVerdict::Incomplete,
        Err(too_large) => {
            debug!(
                connection_id,
                frame_len = too_large.declared,
                "DPDK: auth frame too large"
            );
            return AuthVerdict::Rejected;
        }
    };

    // Compact the parse buffer now that the borrow is released.
    // Single memmove instead of drain()'s per-byte shift.
    let remaining = parse_buf.len() - consumed;
    parse_buf.copy_within(consumed.., 0);
    parse_buf.truncate(remaining);

    let cr = match decoded {
        Ok(cr) => cr,
        Err(e) => {
            debug!(connection_id, error = %e, "DPDK: auth decode error");
            return AuthVerdict::Rejected;
        }
    };

    match crate::client_auth::verify_client(authorized_keys, nonce, &cr.public_key, &cr.signature) {
        Ok(role) => AuthVerdict::Admitted {
            role,
            public_key: cr.public_key,
        },
        Err(e) => {
            debug!(connection_id, error = %e, "DPDK: auth failed");
            AuthVerdict::Rejected
        }
    }
}

/// Process the auth handshake frame from a pending connection.
fn process_auth_frame(
    conn: &mut ConnectionState,
    transport: &mut DpdkTransport,
    authorized_keys: &AuthorizedKeys,
    control_tx: &mpsc::Sender<ControlEvent>,
    id_to_handle: &mut FxHashMap<u64, SocketHandle>,
    handle: SocketHandle,
) {
    // The nonce captured at Challenge-send time, which the client signed.
    let (nonce, accepted_at) = match &conn.auth {
        AuthState::WaitingForResponse { nonce, accepted_at } => (*nonce, *accepted_at),
        _ => unreachable!("process_auth_frame called in wrong state"),
    };

    let (role, public_key_bytes) = match evaluate_auth_frame(
        &mut conn.parse_buf,
        &nonce,
        authorized_keys,
        conn.connection_id.0,
    ) {
        AuthVerdict::Incomplete => return,
        AuthVerdict::Rejected => {
            send_auth_failed(conn, transport);
            // One attempt per connection: the poll loop closes it once
            // the peer has the AuthFailed frame.
            conn.parse_buf.clear();
            conn.auth = AuthState::Rejected { accepted_at };
            return;
        }
        AuthVerdict::Admitted { role, public_key } => (role, public_key),
    };

    // Auth succeeded — send ServerReady.
    let mut buf = [0u8; 16];
    let written =
        control_codec::encode_transport_response(&TransportResponse::ServerReady, &mut buf)
            .expect("ServerReady encodes");
    transport.queue_send(conn.handle, &buf[..written]);

    debug!(
        connection_id = conn.connection_id.0,
        addr = %conn.addr,
        role = authorized_keys.token(KeyRole::Client(role)),
        "DPDK: authenticated"
    );

    // Transition to authenticated state, under the identity the
    // application sees as `ApplyCtx::key_hash`.
    conn.key_hash = melin_app::key_hash(&public_key_bytes);
    conn.auth = AuthState::Authenticated { role };

    // Register with the response stage and ID map — before this thread
    // parses a single request of the connection, and keep it that way:
    // the response stage reads its output ring before it drains this
    // channel, and relies on `Connected` being queued first. See the
    // drain in `dpdk_response::run`.
    id_to_handle.insert(conn.connection_id.0, handle);
    let _ = control_tx.send(ControlEvent::Connected {
        connection_id: conn.connection_id.0,
    });
}

/// Free a closed connection's slot: return its parse buffer to the pool
/// (cleared, capacity kept) and release its `max_connections` count. A
/// no-op on an empty slot, so a path that may already have released it
/// stays balanced.
///
/// Only the slot: closing the socket, telling the response stage, and
/// dropping the `id_to_handle` route stay with the caller, because which
/// of those apply depends on how far the connection got.
#[inline]
fn release_slot(
    connections: &mut [Option<ConnectionState>],
    idx: usize,
    parse_buf_pool: &mut Vec<Vec<u8>>,
    connection_count: &mut usize,
) {
    if let Some(mut removed) = connections[idx].take() {
        removed.parse_buf.clear();
        parse_buf_pool.push(removed.parse_buf);
        *connection_count -= 1;
    }
}

/// Queue an AuthFailed response, best effort. The caller moves the
/// connection to `AuthState::Rejected`, which closes it once the frame
/// has been delivered.
fn send_auth_failed(conn: &ConnectionState, transport: &mut DpdkTransport) {
    let mut buf = [0u8; 16];
    if let Ok(written) =
        control_codec::encode_transport_response(&TransportResponse::AuthFailed, &mut buf)
    {
        // The result is not checked: the queue refuses only past its
        // per-connection limit, and before auth it holds at most the
        // Challenge. Were it refused, only the reason would be lost; the
        // connection is closed on the same terms either way.
        transport.queue_send(conn.handle, &buf[..written]);
    }
}

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

    // The client data path splits frames in `process_client_frames`, shared
    // with the io_uring reader and tested there (`reader.rs`), on the wire
    // protocol's splitter, whose own tests cover the partial, empty,
    // exact-limit and oversized cases. What is left to check here is that
    // an application's encoded events come back out of that splitter, under
    // the cap the DPDK path applies, byte for byte.
    //
    // The in-tree counter example stands in for a production application's
    // event, keeping these tests free of any application crate but the
    // example.
    use counter_server::CounterEvent;
    use melin_app::AppEvent;
    use melin_wire_protocol::framing::Frame;

    /// The first frame of `buf` under the client data path's cap.
    fn split(buf: &[u8]) -> Option<Frame<'_>> {
        split_frame_limited(buf, MAX_FRAME_SIZE).expect("frame within MAX_FRAME_SIZE")
    }

    /// Encode a `CounterEvent` into a length-prefixed wire frame:
    /// `[len:u32][event bytes]` — the framing the DPDK transport extracts.
    fn counter_frame(event: CounterEvent) -> Vec<u8> {
        let mut body = vec![0u8; event.encoded_size()];
        let n = event.encode(&mut body);
        let mut frame = (n as u32).to_le_bytes().to_vec();
        frame.extend_from_slice(&body[..n]);
        frame
    }

    // --- Wire-level round-trip: encode event → extract frame → decode ---

    #[test]
    fn wire_round_trip_increment() {
        let wire = counter_frame(CounterEvent::Increment { amount: 42 });
        let frame = split(&wire).expect("complete frame");
        assert_eq!(frame.consumed(), wire.len());
        let decoded = CounterEvent::decode(frame.payload).unwrap();
        assert!(matches!(decoded, CounterEvent::Increment { amount: 42 }));
    }

    #[test]
    fn wire_round_trip_get_value() {
        let wire = counter_frame(CounterEvent::GetValue);
        let frame = split(&wire).expect("complete frame");
        let decoded = CounterEvent::decode(frame.payload).unwrap();
        assert!(matches!(decoded, CounterEvent::GetValue));
    }

    // --- Incremental accumulation (simulates TCP byte-at-a-time arrival) ---

    #[test]
    fn incremental_frame_accumulation() {
        let wire = counter_frame(CounterEvent::Increment { amount: 7 });
        let written = wire.len();

        // Feed bytes one at a time into a parse buffer.
        let mut parse_buf = Vec::new();
        for (i, &byte) in wire.iter().enumerate() {
            parse_buf.push(byte);
            let result = split(&parse_buf);
            if i < written - 1 {
                assert_eq!(result, None, "byte {i}");
            } else {
                assert!(result.is_some(), "expected a frame at the final byte");
            }
        }
    }

    // --- Multiple frames back-to-back ---

    #[test]
    fn multiple_frames_in_buffer() {
        let frame1 = counter_frame(CounterEvent::Increment { amount: 1 });
        let frame2 = counter_frame(CounterEvent::GetValue);

        // Concatenate two frames.
        let mut combined = Vec::new();
        combined.extend_from_slice(&frame1);
        combined.extend_from_slice(&frame2);

        // First extraction should get frame 1.
        let first = split(&combined).expect("first frame");
        let decoded = CounterEvent::decode(first.payload).unwrap();
        assert!(matches!(decoded, CounterEvent::Increment { amount: 1 }));

        // Second extraction, past the first frame, should get frame 2.
        let second = split(&combined[first.consumed()..]).expect("second frame");
        let decoded = CounterEvent::decode(second.payload).unwrap();
        assert!(matches!(decoded, CounterEvent::GetValue));
        assert_eq!(first.consumed() + second.consumed(), combined.len());
    }

    mod auth_verdict {
        use super::*;

        use base64::Engine;
        use ed25519_dalek::{Signer, SigningKey};
        use melin_wire_protocol::control::ChallengeResponse;

        use crate::test_roles::desk_keys;

        const NONCE: [u8; 32] = [0x5A; 32];

        fn client() -> SigningKey {
            SigningKey::from_bytes(&[0x11; 32])
        }

        fn keys() -> AuthorizedKeys {
            let public = base64::engine::general_purpose::STANDARD
                .encode(client().verifying_key().to_bytes());
            desk_keys("trader", &public)
        }

        /// A length-prefixed ChallengeResponse frame signing `nonce` with `signer`.
        fn auth_frame(signer: &SigningKey, nonce: &[u8; 32]) -> Vec<u8> {
            let response = ChallengeResponse {
                signature: signer.sign(nonce).to_bytes(),
                public_key: client().verifying_key().to_bytes(),
            };
            let mut payload = [0u8; 128];
            let len = control_codec::encode_challenge_response(&response, &mut payload).unwrap();
            let mut frame = (len as u32).to_le_bytes().to_vec();
            frame.extend_from_slice(&payload[..len]);
            frame
        }

        fn evaluate(buf: &mut Vec<u8>) -> AuthVerdict {
            evaluate_auth_frame(buf, &NONCE, &keys(), 1)
        }

        #[test]
        fn a_valid_response_is_admitted_and_pipelined_bytes_are_kept() {
            let mut buf = auth_frame(&client(), &NONCE);
            buf.extend_from_slice(b"next");
            match evaluate(&mut buf) {
                AuthVerdict::Admitted { public_key, .. } => {
                    assert_eq!(public_key, client().verifying_key().to_bytes());
                }
                other => panic!("expected Admitted, got {other:?}"),
            }
            assert_eq!(buf, b"next");
        }

        /// Every prefix of a valid frame waits for more bytes rather than
        /// being judged.
        #[test]
        fn a_partial_frame_is_incomplete_and_left_in_place() {
            let frame = auth_frame(&client(), &NONCE);
            for cut in 0..frame.len() {
                let mut buf = frame[..cut].to_vec();
                assert!(
                    matches!(evaluate(&mut buf), AuthVerdict::Incomplete),
                    "cut {cut}"
                );
                assert_eq!(buf, &frame[..cut]);
            }
        }

        #[test]
        fn a_signature_by_another_key_is_rejected() {
            let impostor = SigningKey::from_bytes(&[0x33; 32]);
            let mut buf = auth_frame(&impostor, &NONCE);
            assert!(matches!(evaluate(&mut buf), AuthVerdict::Rejected));
        }

        /// A signature over another nonce — a replay from an earlier
        /// connection — does not authenticate this one.
        #[test]
        fn a_signature_over_another_nonce_is_rejected() {
            let mut buf = auth_frame(&client(), &[0xA5; 32]);
            assert!(matches!(evaluate(&mut buf), AuthVerdict::Rejected));
        }

        #[test]
        fn an_undecodable_frame_is_rejected() {
            let mut buf = auth_frame(&client(), &NONCE);
            buf[4] ^= 0xFF; // corrupt the tag
            assert!(matches!(evaluate(&mut buf), AuthVerdict::Rejected));
        }

        #[test]
        fn a_zero_length_frame_is_rejected() {
            let mut buf = 0u32.to_le_bytes().to_vec();
            assert!(matches!(evaluate(&mut buf), AuthVerdict::Rejected));
        }

        /// Rejected on the length prefix alone, before the body arrives.
        #[test]
        fn an_oversized_frame_is_rejected_on_its_prefix() {
            let mut buf = ((MAX_AUTH_FRAME + 1) as u32).to_le_bytes().to_vec();
            assert!(matches!(evaluate(&mut buf), AuthVerdict::Rejected));

            let mut at_cap = (MAX_AUTH_FRAME as u32).to_le_bytes().to_vec();
            assert!(matches!(evaluate(&mut at_cap), AuthVerdict::Incomplete));
        }
    }
}