melin-server-runtime 0.18.0

//! Server orchestrator — binds the accept loop, pipeline threads, and reader.
//!
//! On startup:
//! 1. Recovers or creates the `JournaledApp<A>`.
//! 2. Decomposes it into `(A, W)` via `into_parts()`, where `W` is the journal writer.
//! 3. Builds the disruptor pipeline (input ring + output ring).
//! 4. Spawns 3-5 OS threads: journal, matching, response, [repl-accept], [event-publisher].
//! 5. Runs the accept loop, registering connections with the io_uring reader.
//!
//! Fully synchronous — no async runtime needed. Reader threads use io_uring
//! with multishot RECV to multiplex connections, eliminating thread
//! oversubscription. The response thread writes via io_uring SEND.

use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::os::unix::io::AsRawFd;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};

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

use melin_journal::BufferedWriter;
use melin_journal::JournalError;
use melin_journal::JournalWrite;
use melin_transport_core::journaled_app::JournaledApp;
use melin_transport_core::pipeline::{
    InputSlot, OutputSlot as GenericOutputSlot, Pipeline as GenericPipeline,
    build_pipeline_with_replication,
};
/// Internal alias for the disruptor-built pipeline, used only by
/// destructuring `let Pipeline { … }` patterns inside the boot path.
/// Not part of the public API — callers reach the underlying type
/// through `melin_transport_core::pipeline`.
type Pipeline<A> = GenericPipeline<A>;

use crate::StartupEvents;
use crate::reader::RequestDecoderArc;
use crate::response::ResponseEncoderArc;
use melin_app::Application;
use melin_app::auth::AuthorizedKeys;
use melin_app::auth::Permission;
use melin_app::decoder::RequestDecoder;
use melin_app::encoder::ResponseEncoder;
use melin_pipeline::ring::Consumer;
use melin_pipeline::wait::WaitStrategy;

use crate::layout::{DEFAULT_CORES, parse_cores};
/// The layout types live in [`crate::layout`]; re-exported here because
/// `ServerConfig::cores` is one of them and callers reach the config
/// through this module.
pub use crate::layout::{PipelineCores, Placement, ReaderThread};

/// How the orchestrator (main) thread waits on pipeline threads during
/// startup — the seed drain and the promotion epoch bump. It is unpinned
/// and on no latency path, so it yields whatever the layout says: a
/// spinner without a core of its own is the co-location bug with the
/// victim chosen by the scheduler.
const ORCHESTRATOR_WAIT: WaitStrategy = WaitStrategy::SpinThenYield;

/// Output-slot sugar parameterised on the application — saves spelling
/// `<A::Report, A::QueryResponse>` at every pipeline-facing signature
/// that doesn't already destructure them.
type OutputSlot<A> =
    GenericOutputSlot<<A as Application>::Report, <A as Application>::QueryResponse>;

/// Body of the event-publisher thread: a free function with this exact
/// signature (an application with an output feed supplies its
/// publisher; one without supplies nothing). Passing it as a function
/// pointer keeps the runtime decoupled from the application — no
/// application reference inside server.rs — without paying for a
/// boxed closure.
///
/// Threaded into [`run`] (and `run_dpdk` under `feature = "dpdk"`) as `Option<EventPublisherFn>`:
/// `None` disables the publisher unconditionally, `Some(_)` wires it up
/// when `--event-bind` is also set.
pub type EventPublisherFn<A> = fn(
    consumer: Consumer<OutputSlot<A>>,
    bind_addr: SocketAddr,
    authorized_keys: Arc<AuthorizedKeys>,
    shutdown: &AtomicBool,
    wait: WaitStrategy,
);

use melin_wire_protocol::blocking::BlockingFrameWriter;
use melin_wire_protocol::control::ConnectionId;
use melin_wire_protocol::transport::BlockingTransportListener;

/// Default replica pipeline depth (pending ack queue capacity).
///
/// Only bounds persisted-ack *granularity*, never the receive rate: the
/// queue merges instead of blocking when it fills. 256 entries × 16 B is
/// 4 KiB, so the depth is chosen to keep granularity through an fsync
/// hiccup rather than to save memory.
const DEFAULT_REPLICATION_PIPELINE_DEPTH: usize = 256;

/// CLI spelling of [`melin_journal::StagingMode`].
///
/// A separate enum rather than a `clap::ValueEnum` derive on the journal
/// type: `melin-journal` has no clap dependency and should not acquire
/// one for a flag: the crate is a library first, and its callers are not
/// all command-line servers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)]
pub enum JournalStagingMode {
    /// Pre-write every staged segment so appends generate no
    /// extent-conversion metadata. Costs one extra pass over each
    /// segment in device bandwidth.
    #[default]
    ZeroFill,
    /// Allocate staged extents without materialising them. No staging
    /// bandwidth; appends convert extents and `fdatasync` periodically
    /// forces the filesystem log.
    Allocate,
}

impl From<JournalStagingMode> for melin_journal::StagingMode {
    fn from(m: JournalStagingMode) -> Self {
        match m {
            JournalStagingMode::ZeroFill => melin_journal::StagingMode::ZeroFill,
            JournalStagingMode::Allocate => melin_journal::StagingMode::Allocate,
        }
    }
}

/// Server configuration, parsed from CLI arguments via clap.
#[derive(clap::Parser)]
#[command(
    name = "melin-server",
    about = "A node of the Melin replicated sequencer"
)]
pub struct ServerConfig {
    /// Address to bind the TCP listener.
    #[arg(long, default_value = "127.0.0.1:9876")]
    pub bind: SocketAddr,
    /// Path to the journal file for durable event sourcing.
    #[arg(long, default_value = "melin.journal")]
    pub journal: PathBuf,
    /// Path to a snapshot file for faster recovery.
    #[arg(long)]
    pub snapshot: Option<PathBuf>,
    /// Where each pipeline thread runs, as comma-separated `thread=core`
    /// entries in any order. Threads: journal-seq, matching, response, reader,
    /// event-publisher, shadow, repl-handler-0, repl-handler-1, journal-prep,
    /// journal-disk. Every thread must be named: `0` leaves one unpinned,
    /// and `none` unpins every thread. The node warns at boot when
    /// journal-seq, matching, response, reader or journal-disk has no core:
    /// every request passes through them. Core 0 is reserved for OS/IRQ
    /// handling.
    /// reader pins the io_uring reader (TCP) or DPDK poll thread.
    /// event-publisher applies when `--event-bind` is set, shadow when
    /// `--snapshot-interval-ms` > 0. repl-handler-0/1 are for the
    /// per-replica TCP handler threads. journal-disk pins the thread that
    /// writes and syncs the journal; give it a core on the same CCD as
    /// journal-seq, since the two exchange a cache line per batch.
    ///
    /// Each core may carry a suffix saying how that thread waits: `7`
    /// (or `7s`) busy-spins and needs the core to itself, `7y` spins
    /// briefly then yields and may share the core with other `y`
    /// threads. `0` (unpinned) always yields. Two threads on one core
    /// where either busy-spins is refused at startup — they would starve
    /// each other. journal-prep never busy-waits and takes no suffix.
    #[arg(long, default_value = DEFAULT_CORES, value_parser = parse_cores)]
    pub cores: PipelineCores,
    /// Group commit coalescing delay in microseconds. Keep at 0 for TCP.
    #[arg(long, default_value_t = 0)]
    pub group_commit_us: u64,
    /// Heartbeat interval in seconds. The server sends a heartbeat to idle
    /// connections after this many seconds of silence. Set to 0 to disable.
    #[arg(long, default_value_t = 10)]
    pub heartbeat_interval_secs: u64,
    /// Connection timeout in seconds. The server disconnects clients that
    /// have not sent any data within this window. Set to 0 to disable.
    #[arg(long, default_value_t = 30)]
    pub connection_timeout_secs: u64,
    /// Maximum number of concurrent authenticated connections. New
    /// connections are rejected (closed immediately) when this limit is
    /// reached. 0 means unlimited. Prevents fd/memory exhaustion (SEC-02).
    #[arg(long, default_value_t = 1024)]
    pub max_connections: u64,
    /// Path to the authorized keys file for Ed25519 challenge-response
    /// authentication. Every connection must authenticate before sending
    /// requests.
    /// Required for primary mode; ignored in replica mode (--replica-of).
    /// See `AuthorizedKeys` for file format.
    #[arg(long, default_value = "authorized_keys")]
    pub authorized_keys: PathBuf,
    /// Maximum journal size in MiB before automatic rotation at startup.
    /// When the journal exceeds this threshold, the server saves a snapshot
    /// and starts a fresh journal. Set to 0 to disable. Default: 256 MiB.
    #[arg(long, default_value_t = 256)]
    pub max_journal_mib: u64,
    /// How the background preparer materialises the next journal
    /// segment's extents.
    ///
    /// - `zero-fill` (default) physically pre-writes the segment, so
    ///   appends carry no extent-conversion metadata and each flush's
    ///   `fdatasync` never forces the filesystem log. It costs one extra
    ///   pass over every segment in device bandwidth.
    /// - `allocate` reserves the extents and stops. Staging becomes
    ///   free, and the periodic log force returns to the flush path.
    ///
    /// Keep the default on local NVMe, where sequential bandwidth is
    /// cheap and the log forces are what hurts. Consider `allocate` on
    /// network-attached storage (EBS and similar), where staging
    /// bandwidth is metered and drawn from the same budget as the hot
    /// path — measure both on the volume in question rather than
    /// assuming, since which one wins is a property of the device.
    #[arg(long, value_enum, default_value_t = JournalStagingMode::ZeroFill)]
    pub journal_staging_mode: JournalStagingMode,

    /// Address to listen for replica connections (enables synchronous replication).
    /// Mutually exclusive with `--standalone` and `--replica-of`.
    #[arg(long)]
    pub replication_bind: Option<std::net::SocketAddr>,

    /// Disable replication (dev/test mode). The replica quorum cursor stays
    /// at its no-replica sentinel (health reports zero replication lag) and
    /// the ack policy evaluates against the primary's journal alone.
    /// Mutually exclusive with `--replication-bind` and `--replica-of`.
    #[arg(long, default_value_t = false)]
    pub standalone: bool,

    /// Run as a replica connected to the given primary address.
    /// In replica mode, the server does not accept client connections.
    /// Mutually exclusive with `--replication-bind` and `--standalone`.
    #[arg(long)]
    pub replica_of: Option<std::net::SocketAddr>,

    /// Path to the Ed25519 private key for replication authentication.
    /// Required in replica mode (`--replica-of`). The corresponding
    /// public key must be listed in the primary's authorized_keys file
    /// with permission `replication`.
    #[arg(long)]
    pub replication_key: Option<std::path::PathBuf>,

    /// Maximum number of replication ring batches to coalesce into a
    /// single TCP write+flush. Higher values reduce syscall overhead
    /// but increase per-write latency. Default: 128.
    #[arg(long, default_value_t = 128)]
    pub replication_batch_size: usize,

    /// Maximum events per journal fsync batch. Smaller values reduce
    /// tail latency (less work per sync), larger values improve throughput
    /// (fewer fsyncs). Default: 4096.
    #[arg(long, default_value_t = 4096)]
    pub max_journal_batch: usize,

    /// Replication heartbeat interval in seconds. The primary sends a
    /// heartbeat to the replica after this many seconds of idle. Used
    /// for disconnect detection. Default: 5.
    #[arg(long, default_value_t = 5)]
    pub replication_heartbeat_secs: u64,

    /// Number of receive batches the replica tracks individually while
    /// they await local journal fsync. Must be a power of two. Beyond
    /// this depth the receiver keeps receiving and coalesces pending
    /// acks (a later ack, never an early one) — it never applies
    /// backpressure to the primary. Default: 256.
    #[arg(long, default_value_t = DEFAULT_REPLICATION_PIPELINE_DEPTH)]
    pub replication_pipeline_depth: usize,

    /// Number of slots in each replication ring buffer. Must be a power
    /// of two. Each slot holds up to 512 KiB. More slots = more buffering
    /// before eviction. Default: 256 (128 MiB per ring, 256 MiB dual-repl).
    #[arg(long, default_value_t = 256)]
    pub replication_ring_size: usize,

    /// Ack policy: which copies of an event must exist before its
    /// response is released. One of:
    ///
    /// - `disk`               `persisted>=1`. One fsynced copy, on
    ///   whichever node's disk confirms first; required with
    ///   `--standalone`. Dev/staging deployments.
    /// - `ram`                `in_memory>=2`. Two nodes hold the event
    ///   in memory before the ack; every fsync trails off the ack path
    ///   (the journal still syncs every batch). Lowest ack latency;
    ///   survives any single node failure via failover; loses only the
    ///   un-fsynced tail on a whole-cluster power loss. For slow-fsync
    ///   storage (cloud volumes) or RPO-tolerant apps.
    /// - `disk+ram` (default) `persisted>=1 && in_memory>=2`. One
    ///   fsynced copy plus a second copy in another node's memory.
    ///   Single-failure-safe with a brief RAM-only window for the
    ///   second copy. Typical live deployments. Faster than `two-disks`:
    ///   an acknowledgement waits for the second node to receive the
    ///   event, not to fsync it.
    /// - `two-disks`          `persisted>=2`. Two fsynced copies before
    ///   the client ack. Zero RAM-only window; the gate stalls when no
    ///   replica is connected. Compliance-driven deployments.
    ///
    /// `--standalone` requires `disk`. Under every other policy the
    /// gate stalls while no replica is connected — the correct
    /// behaviour for a serious deployment that has lost its replicas.
    /// See `docs/replication.md` for the operational menu.
    #[arg(long, value_enum, default_value_t = crate::ack_policy::AckPolicy::DiskAndRam)]
    pub ack_policy: crate::ack_policy::AckPolicy,

    // --- DPDK configuration (only used with --features dpdk) ---
    /// DPDK EAL arguments (space-separated). Example: --dpdk-eal-args="-l 0-7 --huge-dir /dev/hugepages".
    /// Passed directly to rte_eal_init. Only used when compiled with --features dpdk.
    /// The `=` is required: the value itself starts with a dash, and requiring
    /// the joined form keeps a forgotten value from silently swallowing the
    /// next flag as the EAL string.
    #[arg(long, default_value = "", require_equals = true)]
    pub dpdk_eal_args: String,

    /// DPDK port IDs (comma-separated). For LACP bonds, pass both VF ports
    /// (e.g., "0,1") so traffic arriving on either bond member is received.
    #[arg(long, default_value = "0", value_delimiter = ',')]
    pub dpdk_ports: Vec<u16>,

    /// IPv4 address for the DPDK interface (e.g., "10.0.0.1").
    #[arg(long, default_value = "10.0.0.1")]
    pub dpdk_ip: String,

    /// IPv4 prefix length for the DPDK interface. Default: 24.
    #[arg(long, default_value_t = 24)]
    pub dpdk_prefix_len: u8,

    /// IPv4 gateway for the DPDK interface (optional, needed for cross-subnet traffic).
    #[arg(long)]
    pub dpdk_gateway: Option<String>,

    /// Peer IPv4 used for the bifurcated `rte_flow` steering rule. When
    /// set, the DPDK port opens in isolated mode and only IPv4 packets
    /// sourced from this address are delivered into DPDK queue 0 —
    /// everything else stays with the kernel netdev. Required for L3
    /// setups that share the public NIC with the kernel (SSH, etc.).
    #[arg(long)]
    pub dpdk_peer_ip: Option<String>,

    /// Gateway MAC (aa:bb:cc:dd:ee:ff) seeded into smoltcp for the
    /// `--dpdk-gateway` IP. Required in L3 bifurcated mode because the
    /// gateway's ARP reply would not match the steering rule and would
    /// be eaten by the kernel. Source from `ip neigh` on the host.
    #[arg(long)]
    pub dpdk_gateway_mac: Option<String>,

    /// MAC (aa:bb:cc:dd:ee:ff) of the primary named by `--replica-of`,
    /// seeded into smoltcp so the replica's first frame is addressed
    /// before any ARP. Required on ports that keep a real hardware MAC —
    /// an mlx5 in bifurcated mode shares the kernel netdev's — where the
    /// SR-IOV `02:00:<ip>` fallback is wrong and the replica would spin
    /// on connect with no error. Read it from
    /// `/sys/class/net/<iface>/address` on the primary, or from
    /// `DPDK_MAC` in `/etc/melin-dpdk.conf`. Ignored on a primary, which
    /// learns peer MACs from inbound frames.
    #[arg(long)]
    pub dpdk_peer_mac: Option<String>,

    /// MTU for the DPDK interface. Use 9000 for jumbo frames (6x fewer TCP
    /// segments). Requires switch and PF MTU to be set accordingly.
    #[arg(long, default_value_t = 1500)]
    pub dpdk_mtu: usize,

    /// VLAN ID for hardware VLAN strip/insert. Required in dedicated NIC
    /// mode (dpdk-setup-dedicated.sh) where the kernel doesn't handle VLAN
    /// tags. Not needed for SR-IOV mode (the PF handles VLAN tagging).
    #[arg(long)]
    pub dpdk_vlan: Option<u16>,

    /// Address for the output event publisher. Subscribers connect here
    /// to receive a real-time stream of the application's reports, as the
    /// application's publisher encodes them. Ed25519 auth required
    /// (ReadOnly or above).
    /// Omit to disable (ring has 1 consumer — identical to before).
    #[arg(long)]
    pub event_bind: Option<SocketAddr>,

    /// Address for the health/liveness TCP endpoint. On connect, returns
    /// a one-line status (`OK <conns> <journal_seq> <repl_lag>\n`) and
    /// closes. No auth required. Set to empty string to disable.
    #[arg(long, default_value = "127.0.0.1:9878")]
    pub health_bind: Option<SocketAddr>,

    /// TCP address for the operator admin endpoint. Authenticated with
    /// operator keys (Ed25519 challenge-response, same as all other
    /// admin handshakes). Accepts:
    ///
    /// - `PROMOTE\n` — replica → primary leadership transition (replica
    ///   nodes only; rejected with ERR on a primary).
    /// - `ROTATE\n` — archive the current journal segment at the next
    ///   fsync boundary (any node where runtime rotation is enabled).
    ///
    /// Unset means no admin endpoint is listening; operators can still
    /// rely on the size-driven rotation trigger (`--max-journal-mib`)
    /// without exposing a port.
    #[arg(long)]
    pub admin_bind: Option<SocketAddr>,

    /// Interval in milliseconds between automatic shadow snapshots. The
    /// shadow stage replays journaled events on a cloned `A` and saves a
    /// consistent snapshot at this cadence — no hot-path stall. Set to 0
    /// to disable shadow snapshots entirely. Default: 3 000 000 ms (50 min).
    #[arg(long, default_value_t = 3_000_000)]
    pub snapshot_interval_ms: u64,

    /// Path for shadow snapshots. Defaults to the journal path with a
    /// `.snapshot` extension (same as the startup snapshot path).
    #[arg(long)]
    pub snapshot_path: Option<PathBuf>,

    /// Cadence in milliseconds for the application's clock tick.
    /// The ingress thread (io_uring reader or DPDK poll thread) publishes
    /// a `JournalEvent::Tick { now_ns }` at this interval so the
    /// application's time-driven work (expiries, timeouts, scheduled
    /// transitions) fires in deterministic, journaled lockstep. There is
    /// no separate tick thread on either transport. Set to 0 to disable
    /// tick generation entirely (useful for benchmarks that don't
    /// exercise time-driven features).
    ///
    /// Defaults to 250 ms. Under load the matching stage advances the
    /// application's clock at every-event resolution from `slot.timestamp_ns`
    /// (microsecond precision), so the tick is only the safety net for
    /// quiet periods. 250 ms keeps time-driven work in quiet periods within
    /// a quarter-second of their deadline at a cost of ~4 events/sec of
    /// journal traffic.
    #[arg(long, default_value_t = 250)]
    pub tick_interval_ms: u64,

    /// Disable `mlockall(MCL_CURRENT | MCL_FUTURE)` at startup.
    ///
    /// By default the server locks all current and future pages into
    /// RAM to prevent rare multi-millisecond stalls from page faults
    /// or swap activity on the matching hot path. Locking requires
    /// `CAP_IPC_LOCK` (or running as root) and `RLIMIT_MEMLOCK` raised
    /// to a value larger than the process's RSS — the server raises
    /// the rlimit itself, but only succeeds with the capability.
    ///
    /// Use `--no-mlock` for development / containerised runs where
    /// the privilege isn't available; on a bare-metal production host
    /// leave it on.
    #[arg(long, default_value_t = false)]
    pub no_mlock: bool,

    /// Bind address for the control-plane raft RPC listener. Setting this
    /// enables the control plane on this node: leader election runs and is
    /// observable via the `melin_raft_*` gauges. Requires `--raft-node-id`
    /// and a `--raft-peer` entry for this node; requires
    /// `--replication-key` (peer links authenticate with the replication
    /// key, same trust domain as the data plane). Raft always runs on
    /// kernel TCP, including on DPDK nodes.
    #[arg(long)]
    pub raft_bind: Option<SocketAddr>,

    /// This node's control-plane raft id (non-zero, unique per cluster).
    #[arg(long)]
    pub raft_node_id: Option<u64>,

    /// A control-plane cluster member as `id@host:port#base64-pubkey`
    /// (repeatable). Give **every node the same list, including an entry
    /// for the node itself** — the self entry supplies the dialable
    /// address peers use to reach it, and identical lists keep the
    /// first-boot membership consistent. The pubkey pins the peer's
    /// identity: a connection authenticated with a different key cannot
    /// speak for that id.
    #[arg(long)]
    pub raft_peer: Vec<String>,

    /// Directory for durable raft state (vote, log, membership).
    /// Defaults to the journal path with a `.raft` extension. Must live
    /// on the same durability class as the journal — losing it can
    /// double-grant a vote (see docs/replication.md).
    #[arg(long)]
    pub raft_dir: Option<PathBuf>,

    /// Act on control-plane election wins: a replica elected leader
    /// promotes itself (journaling the election term as its fencing
    /// epoch), and a serving node self-fences when the raft mesh shows
    /// it superseded. Off by default — automatic failover is an explicit
    /// operator policy decision. Requires `--raft-bind` and at least
    /// three configured voters (a two-node cluster cannot elect after
    /// losing either node, so automation would be dead weight with real
    /// misconfiguration risk; see docs/replication.md).
    #[arg(long, default_value_t = false)]
    pub raft_auto_promote: bool,
}

/// Delegates to clap so `#[arg(default_value...)]` is the single source of
/// truth for every default.  Used by the bench crate for struct-literal
/// construction with `..ServerConfig::default()`.
impl Default for ServerConfig {
    fn default() -> Self {
        // On the dpdk branch we spell out every field so that dpdk-specific
        // fields (not known to clap on plain builds) get their defaults.
        // Main uses `Self::parse_from(["melin-server"])` — the values below
        // mirror those clap defaults plus the dpdk extras.
        Self {
            bind: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9876),
            journal: PathBuf::from("melin.journal"),
            snapshot: None,
            cores: PipelineCores {
                journal_seq: Placement::spinning(1),
                matching: Placement::spinning(2),
                response: Placement::spinning(3),
                reader: Placement::spinning(4),
                event_publisher: Placement::spinning(6),
                shadow: Placement::spinning(7),
                repl_handler_0: Placement::spinning(8),
                repl_handler_1: Placement::spinning(9),
                journal_prep: Placement::yielding(10),
                journal_disk: Placement::spinning(11),
            },
            group_commit_us: 0,
            heartbeat_interval_secs: 10,
            connection_timeout_secs: 30,
            max_connections: 1024,
            authorized_keys: PathBuf::from("authorized_keys"),
            max_journal_mib: 256,
            journal_staging_mode: JournalStagingMode::ZeroFill,

            replication_bind: None,
            standalone: false,
            replica_of: None,
            replication_key: None,
            replication_batch_size: 128,
            max_journal_batch: 1024,
            replication_heartbeat_secs: 5,
            replication_pipeline_depth: DEFAULT_REPLICATION_PIPELINE_DEPTH,
            replication_ring_size: 256,
            ack_policy: crate::ack_policy::AckPolicy::DiskAndRam,
            dpdk_eal_args: String::new(),
            dpdk_peer_ip: None,
            dpdk_gateway_mac: None,
            dpdk_peer_mac: None,
            dpdk_ports: vec![0],
            dpdk_ip: "10.0.0.1".into(),
            dpdk_prefix_len: 24,
            dpdk_gateway: None,
            dpdk_mtu: 1500,
            dpdk_vlan: None,
            event_bind: None,
            health_bind: Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9878)),
            admin_bind: None,
            snapshot_interval_ms: 3_000_000,
            snapshot_path: None,
            tick_interval_ms: 250,
            no_mlock: false,
            raft_bind: None,
            raft_node_id: None,
            raft_peer: Vec::new(),
            raft_dir: None,
            raft_auto_promote: false,
        }
    }
}

impl ServerConfig {
    /// Group commit delay as a Duration.
    pub fn group_commit_delay(&self) -> std::time::Duration {
        std::time::Duration::from_micros(self.group_commit_us)
    }

    /// Heartbeat interval as a Duration. Returns `None` if disabled (0).
    pub fn heartbeat_interval(&self) -> Option<std::time::Duration> {
        if self.heartbeat_interval_secs == 0 {
            None
        } else {
            Some(std::time::Duration::from_secs(self.heartbeat_interval_secs))
        }
    }

    /// Connection timeout as a Duration. Returns `None` if disabled (0).
    pub fn connection_timeout(&self) -> Option<std::time::Duration> {
        if self.connection_timeout_secs == 0 {
            None
        } else {
            Some(std::time::Duration::from_secs(self.connection_timeout_secs))
        }
    }

    /// Snapshot path for the shadow stage. Uses the explicit `--snapshot-path`
    /// if set, otherwise derives from the journal path with `.snapshot` extension.
    pub fn shadow_snapshot_path(&self) -> PathBuf {
        self.snapshot_path
            .clone()
            .unwrap_or_else(|| self.journal.with_extension("snapshot"))
    }

    /// Tick generator cadence as a `Duration`. Returns `None` when the tick
    /// thread is disabled (`--tick-interval-ms 0`).
    pub fn tick_interval(&self) -> Option<std::time::Duration> {
        if self.tick_interval_ms == 0 {
            None
        } else {
            Some(std::time::Duration::from_millis(self.tick_interval_ms))
        }
    }

    /// The per-thread layout this node actually runs — see
    /// [`PipelineCores::resolve`]. `reader` says which thread the reader
    /// entry pins on this transport, since one of the two cannot yield.
    ///
    /// An `Err` is a configuration the node refuses to start with.
    pub fn resolved_cores(&self, reader: ReaderThread) -> Result<PipelineCores, String> {
        self.cores.resolve(reader)
    }
}

/// Run the server with automatic transport selection.
///
/// Installs SIGINT/SIGTERM handlers, optionally locks memory, selects
/// the transport (DPDK under the `dpdk` feature, kernel TCP otherwise),
/// and enters the pipeline loop. This is the main entry point for
/// application binaries.
///
/// The application starts from `A::default()` on every node; `startup`
/// is what the node journals on top of that as it becomes primary — see
/// [`StartupEvents`]. `sizing` is what this node reserves memory for,
/// handed to [`Application::prefault`] on every instance the node
/// builds; it is local to the node and never journaled.
///
/// For callers that need a pre-bound listener or an externally
/// controlled shutdown flag (e.g. benchmarks), use
/// [`run_with_listener`] instead.
pub fn run<A>(
    config: ServerConfig,
    startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: impl RequestDecoder<Event = A::Event> + 'static,
    encoder: impl ResponseEncoder<Report = A::Report, Query = A::QueryResponse> + 'static,
    event_publisher: Option<EventPublisherFn<A>>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    // Before anything pins a thread or initialises DPDK, whose EAL can
    // narrow this thread to one lcore: the CPU set captured here is what
    // every unpinned thread runs on.
    melin_app::affinity::capture_home_mask();
    let shutdown = Arc::new(AtomicBool::new(false));
    crate::process::install_shutdown_handler(&shutdown);

    if !config.no_mlock {
        crate::process::try_lock_memory();
    }

    #[cfg(feature = "dpdk")]
    {
        run_dpdk::<A>(
            config,
            startup,
            sizing,
            Arc::new(decoder),
            Arc::new(encoder),
            event_publisher,
            shutdown,
        )
    }

    #[cfg(not(feature = "dpdk"))]
    {
        let listener = melin_wire_protocol::tcp::BlockingTcpListener::bind(config.bind)?;
        run_tcp::<A, _>(
            listener,
            config,
            startup,
            sizing,
            Arc::new(decoder),
            Arc::new(encoder),
            event_publisher,
            shutdown,
        )
    }
}

/// Run the server with a caller-supplied listener.
///
/// Use this when the listener must be pre-bound before the server
/// starts (e.g. benchmarks that need the kernel backlog to queue
/// connections immediately).
///
/// Set `shutdown` to `true` to trigger a clean shutdown of all
/// pipeline threads.
pub fn run_with_listener<A>(
    listener: impl BlockingTransportListener,
    config: ServerConfig,
    startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: impl RequestDecoder<Event = A::Event> + 'static,
    encoder: impl ResponseEncoder<Report = A::Report, Query = A::QueryResponse> + 'static,
    event_publisher: Option<EventPublisherFn<A>>,
    shutdown: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    // As in `run`: the CPU set unpinned threads run on, captured before
    // anything pins a thread.
    melin_app::affinity::capture_home_mask();
    run_tcp::<A, _>(
        listener,
        config,
        startup,
        sizing,
        Arc::new(decoder),
        Arc::new(encoder),
        event_publisher,
        shutdown,
    )
}

fn run_tcp<A, L>(
    listener: L,
    config: ServerConfig,
    startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: RequestDecoderArc<A>,
    encoder: ResponseEncoderArc<A>,
    event_publisher: Option<EventPublisherFn<A>>,
    shutdown: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
    L: BlockingTransportListener,
{
    run_impl::<A, L>(
        listener,
        config,
        startup,
        sizing,
        decoder,
        encoder,
        event_publisher,
        shutdown,
    )
}

/// Announce at boot that this binary was built without the
/// tamper-evident journal chain.
///
/// A chain-less build is otherwise silent: the journal still writes and
/// replication still streams, but segment-continuity verification at
/// recovery, the snapshot/journal cross-check, and cross-node divergence
/// detection are all absent — a rejoining ex-primary carrying a
/// journaled-but-unreplicated suffix gets streamed to rather than
/// resynced. A deployment can arrive here without meaning to, because
/// `--no-default-features` anywhere in the dependency graph turns the
/// chain off wholesale, so boot says so out loud rather than leaving the
/// absence to be discovered during an incident.
///
/// `warn!` rather than `error!`: it is a deliberate build configuration,
/// not a malfunction.
fn warn_if_chain_disabled() {
    #[cfg(not(feature = "hash-chain"))]
    warn!(
        "built without the `hash-chain` feature: journal tamper evidence, \
         segment-continuity and snapshot cross-checks at recovery, and \
         cross-node divergence detection are all disabled"
    );
}

/// Log the layout the node runs, and warn when it leaves a mandatory
/// thread without a core. Every request passes through those threads, so
/// a core shared with whatever else the scheduler puts there is paid for
/// on every acknowledgement; the auxiliary threads' placement is the
/// operator's documented trade and draws no warning. `none` is called
/// out on its own: nothing pinned is the development layout, and figures
/// measured under it say nothing about the node. One function for both
/// transports, so the two paths cannot drift on what they warn about.
///
/// `warn!` rather than `error!`: a layout the operator chose, not a
/// malfunction.
fn log_layout(cores: &PipelineCores) {
    info!(cores = %cores, "pipeline layout");
    if cores.pins_nothing() {
        warn!(
            "--cores none: no pipeline thread is pinned. Every thread runs wherever the \
             scheduler puts it and shares that core with everything else on the host. A \
             development layout: latency measured under it is not representative. Name a \
             core per thread in --cores"
        );
        return;
    }
    let unpinned = cores.unpinned_mandatory();
    if !unpinned.is_empty() {
        warn!(
            threads = %unpinned.join(", "),
            "--cores: mandatory pipeline threads have no core of their own. Every request \
             passes through them, and each runs wherever the scheduler puts it, sharing \
             that core with whatever else is there. Give them cores"
        );
    }
}

fn run_impl<A, L>(
    listener: L,
    config: ServerConfig,
    mut startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: RequestDecoderArc<A>,
    encoder: ResponseEncoderArc<A>,
    event_publisher: Option<EventPublisherFn<A>>,
    shutdown: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
    L: BlockingTransportListener,
{
    warn_if_chain_disabled();

    // The layout every spawn site below reads, refused outright if two
    // threads would starve each other.
    let config = ServerConfig {
        cores: config.resolved_cores(ReaderThread::IoUring)?,
        ..config
    };
    log_layout(&config.cores);

    // Shared ack-policy atomic, constructed once per process and
    // threaded through both roles. Wiring it on the replica path
    // (where the live node has no response stage) lets an operator
    // pre-stage the post-promotion policy with `ACK-POLICY <policy>`
    // before issuing `PROMOTE`; the same `Arc` becomes the response
    // stage's source of truth after the replica → primary transition.
    let ack_policy_atomic = Arc::new(AtomicU8::new(config.ack_policy.as_u8()));

    // Validate before the bind below so an invalid config has no side
    // effects. `run_as_primary` re-checks (it is also reached from the
    // DPDK promotion fallback, which does not pass through here).
    if config.replication_bind.is_some() && config.standalone {
        return Err("--replication-bind and --standalone are mutually exclusive".into());
    }

    // Bind the replication listener up front, before any pipeline thread
    // exists and regardless of role: a bad or stolen --replication-bind
    // fails the boot cleanly (nothing spawned yet to leak), and a
    // replica holds its configured port from boot, so a later promotion
    // can neither lose the port to another process nor die on a
    // mid-failover bind. Pending connections just sit in the backlog
    // until a promotion starts the sender's accept loop — replicas dial
    // the configured primary, not each other, so nothing queues there
    // in normal operation.
    let repl_listener = config
        .replication_bind
        .map(bind_replication_listener)
        .transpose()?;

    // Replica mode: connect to primary, receive journal stream, replay.
    // Must run before init_engine — the replica's journal is created from
    // the primary's segment lineage during the replication handshake.
    if let Some(primary_addr) = config.replica_of {
        info!(primary = %primary_addr, "starting in replica mode");
        // A replica receives the genesis events in the primary's history
        // and never journals its own, promoted or not — nothing to hold
        // them for, and a large genesis is worth freeing.
        startup.genesis = Vec::new();

        // Load replication signing key.
        let replication_key_path = config.replication_key.as_ref().ok_or_else(|| {
            std::io::Error::other("--replication-key is required in replica mode (--replica-of)")
        })?;
        let signing_key = {
            let seed = std::fs::read(replication_key_path).map_err(|e| {
                std::io::Error::other(format!(
                    "failed to read replication key {}: {e}",
                    replication_key_path.display()
                ))
            })?;
            if seed.len() != 32 {
                return Err(format!(
                    "replication key must be 32 bytes, got {} ({})",
                    seed.len(),
                    replication_key_path.display()
                )
                .into());
            }
            let mut bytes = [0u8; 32];
            bytes.copy_from_slice(&seed);
            ed25519_dalek::SigningKey::from_bytes(&bytes)
        };

        // Load authorized keys early — the admin listener needs them for
        // Ed25519 challenge-response auth (operator keys only).
        let authorized_keys = Arc::new(AuthorizedKeys::load(&config.authorized_keys)?);
        info!(
            keys = authorized_keys.len(),
            path = %config.authorized_keys.display(),
            "loaded authorized keys (replica mode)"
        );

        // The replica's control-plane bundle: promotion request (admin
        // PROMOTE / raft auto-promotion), tip readiness + advertised
        // journal tip (vote recency), primary link state, and the
        // primary's advertised ack policy. Constructed before
        // mode-detection so it survives a replica → primary transition.
        // The rotate flag is re-wired into the new primary's journal
        // stage by `run_as_primary`.
        let control = crate::replication::ReplicaControlPlane::new();
        let promotion_request = control.promote.clone();
        let rotate_flag = config.admin_bind.map(|_| Arc::new(AtomicBool::new(false)));
        let _admin_handle = config
            .admin_bind
            .map(|addr| {
                crate::admin::spawn(
                    addr,
                    Some(promotion_request.clone()),
                    rotate_flag.clone(),
                    Some(Arc::clone(&ack_policy_atomic)),
                    Arc::clone(&shutdown),
                    Arc::clone(&authorized_keys),
                )
            })
            .transpose()?;

        // Shared fencing state, created before mode detection so it
        // survives a replica → primary transition. Seeded at epoch 0; the
        // receiver raises it from the replica's recovered journal and the
        // replication stream, and a promotion bumps it by injecting an
        // `EpochBump`. Carried into `run_as_primary` post-promotion.
        let fence_state = Arc::new(melin_transport_core::fence::FenceState::new(0));

        // Control-plane raft (observational election). Spawned before the
        // receive loop so the driver — like the fence state and admin
        // flags — survives a replica → primary promotion untouched: it
        // shares the process shutdown flag and runs on its own thread.
        // The advertised tip isn't trustworthy until journal recovery
        // seeds the fence epoch and sequence, so `control.tip_ready`
        // starts false and the receiver flips it — until then the driver
        // refuses to grant votes. The receiver owns the sequence half
        // until a promotion hands it to the new primary's journal stage.
        let journal_tip = control.journal_tip.clone();
        let mut raft = match crate::raft::build_raft_config(&config)? {
            None => crate::raft::RaftDriverGuard::disabled(&shutdown),
            Some(cfg) => crate::raft::spawn_raft_driver(
                cfg,
                config.raft_auto_promote,
                &signing_key,
                &authorized_keys,
                &fence_state,
                journal_tip.clone(),
                Arc::clone(&control.tip_ready),
                // A replica claims to be serving once a promotion is in
                // flight (see SupersessionPolicy).
                {
                    let promote = control.promote.clone();
                    Arc::new(move || promote.is_requested())
                },
                &shutdown,
            )?,
        };
        // Act on election wins when the operator opted in. Genesis
        // primaries have nothing to promote — replica paths only. The
        // guard joins the thread on every exit path.
        if config.raft_auto_promote
            && let Some(wiring) = raft.promotion_wiring()
        {
            raft.arm_promotion(crate::raft_promotion::spawn_auto_promotion(
                wiring.status,
                control.clone(),
                Arc::clone(&fence_state),
                Arc::clone(&ack_policy_atomic),
                wiring.peer_tips,
                wiring.peer_ids,
                wiring.elect_requested,
                wiring.elect_enabled,
                Arc::clone(&shutdown),
            ));
        }
        let raft_status = raft.status();
        // A raft-enabled replica exposes election gauges on --health-bind;
        // stopped explicitly before promotion so run_as_primary can rebind
        // the port, and by its guard on every other exit path.
        let mut replica_health = crate::raft::spawn_replica_health(
            &config,
            &fence_state,
            raft_status.as_ref(),
            Arc::clone(&control.pipeline_healthy),
        )?;

        // No local rotation triggers on the replica side: segment
        // rotation is primary-driven (the replica adopts the boundaries
        // announced over the replication stream), so replica journals
        // stay bitwise mirrors of the primary's. `--max-journal-mib`
        // and the admin `ROTATE` command only act on primaries.

        match crate::replication::run_receiver::<A>(
            primary_addr,
            &config.journal,
            &signing_key,
            &shutdown,
            &control,
            config.snapshot_interval_ms,
            config.shadow_snapshot_path(),
            config.cores,
            config.journal_staging_mode.into(),
            config.group_commit_delay(),
            config.replication_pipeline_depth,
            Arc::clone(&fence_state),
            &sizing,
        )? {
            // Clean shutdown — the raft and health guards tear down on drop.
            None => return Ok(()),
            Some((mut app, writer)) => {
                // Promotion! Transition to primary mode. Bump the epoch so a
                // paused/partitioned ex-primary is fenced when it reconnects.
                info!("replica promoted — transitioning to primary");
                // Release --health-bind before run_as_primary rebinds it
                // with the full primary health state.
                replica_health.stop();
                // A replica that ran a pipeline sized this instance
                // already; one promoted before its first session did
                // not (recovered from disk, never streamed). Sizing is
                // idempotent, so size here either way.
                <A as Application>::prefault(&mut app, &sizing);

                // A ROTATE received while this node was a replica latched
                // the flag but rotated nothing (rotation is primary-driven
                // and replicas follow the primary's boundaries). Clear it
                // so the stale latch can't fire a surprise rotation on the
                // first post-promotion fsync.
                if let Some(ref flag) = rotate_flag {
                    flag.store(false, Ordering::Release);
                }

                // The raft guard drops — stopping the driver and joining
                // the (already exited) promotion thread — after this
                // returns, so the driver serves elections and the fencing
                // channel for the whole primary tenure.
                return run_as_primary::<A, L>(
                    app,
                    writer,
                    listener,
                    repl_listener,
                    &config,
                    startup,
                    decoder,
                    encoder,
                    event_publisher,
                    Arc::clone(&shutdown),
                    authorized_keys,
                    false, // no seeding needed — state comes from replication
                    rotate_flag,
                    ack_policy_atomic,
                    fence_state,
                    promotion_request.pending(), // promoted — EpochBump with the request's epoch floor
                    raft_status,
                    journal_tip,
                );
            }
        }
    }
    // Load authorized keys for challenge-response authentication.
    let authorized_keys = Arc::new(AuthorizedKeys::load(&config.authorized_keys)?);
    info!(
        keys = authorized_keys.len(),
        path = %config.authorized_keys.display(),
        "loaded authorized keys"
    );

    // Spawn the admin listener once if configured. PROMOTE is rejected
    // (with ERR) on a primary because no promote flag is wired here —
    // promotion is meaningful only on a replica. ROTATE is wired
    // whenever the admin endpoint is configured.
    let rotate_flag = config.admin_bind.map(|_| Arc::new(AtomicBool::new(false)));
    let _admin_handle = config
        .admin_bind
        .map(|addr| {
            crate::admin::spawn(
                addr,
                None,
                rotate_flag.clone(),
                Some(Arc::clone(&ack_policy_atomic)),
                Arc::clone(&shutdown),
                Arc::clone(&authorized_keys),
            )
        })
        .transpose()?;

    // Initialize or recover the app. `needs_seeding` is true on first
    // startup — the genesis events will flow through the pipeline later.
    let (mut app, writer, needs_seeding, recovered_epoch) =
        init_engine::<A, BufferedWriter<A::Event>>(&config, &sizing)?;

    // Size and pre-fault application-owned memory (slabs, indices) so
    // growth and page faults happen now, not on the hot path. Runs on the
    // recovered state too: a snapshot restores contents, not capacity,
    // and `init_engine` sizes a genesis instance before replay only.
    <A as Application>::prefault(&mut app, &sizing);

    // A primary booting directly (not via promotion) keeps whatever epoch
    // its journal recovered; no bump.
    let fence_state = Arc::new(melin_transport_core::fence::FenceState::new(
        recovered_epoch,
    ));

    // Control-plane raft (observational election). The primary needs the
    // replication signing key only when raft is on — peer links
    // authenticate with it (build_raft_config enforces the flag).
    // A primary's fence epoch is already recovered here, so its tip is
    // trustworthy from the moment the driver starts. Seed the advertised
    // sequence from the recovered journal for the same reason — the
    // pipeline's journal stage takes the handle over once it runs.
    let journal_tip = melin_transport_core::AdvertisedJournalTip::new(
        melin_transport_core::WireSeq::new(writer.next_sequence().saturating_sub(1)),
    );
    let raft = match crate::raft::build_raft_config(&config)? {
        None => crate::raft::RaftDriverGuard::disabled(&shutdown),
        Some(cfg) => {
            let signing_key = load_replication_key(&config)?;
            crate::raft::spawn_raft_driver(
                cfg,
                config.raft_auto_promote,
                &signing_key,
                &authorized_keys,
                &fence_state,
                journal_tip.clone(),
                Arc::new(AtomicBool::new(true)),
                // A primary always claims to be serving.
                Arc::new(|| true),
                &shutdown,
            )?
        }
    };
    let raft_status = raft.status();

    // The raft guard drops — stopping the driver — after this returns.
    run_as_primary::<A, L>(
        app,
        writer,
        listener,
        repl_listener,
        &config,
        startup,
        decoder,
        encoder,
        event_publisher,
        Arc::clone(&shutdown),
        authorized_keys,
        needs_seeding,
        rotate_flag,
        ack_policy_atomic,
        fence_state,
        None, // not promoted — no EpochBump injection
        raft_status,
        journal_tip,
    )
}

/// Bind the kernel-TCP replication listener, non-blocking (the sender's
/// accept loop polls the shutdown flag between accepts). The kernel path
/// binds at boot (`run_impl`) so a failure cannot leak pipeline threads
/// and a promotion cannot fail on it; the DPDK promotion fallback binds
/// at promotion, where its kernel listeners come up.
fn bind_replication_listener(
    addr: std::net::SocketAddr,
) -> Result<crate::replication::ReplicationListener, Box<dyn std::error::Error>> {
    let listener = std::net::TcpListener::bind(addr)
        .map_err(|e| format!("failed to bind replication listener on {addr}: {e}"))?;
    // The newtype sets non-blocking — the invariant the sender's accept
    // loop needs is enforced by construction, not by this call site.
    let listener = crate::replication::ReplicationListener::new(listener)
        .map_err(|e| format!("failed to set non-blocking on replication listener: {e}"))?;
    info!(%addr, "replication listener bound");
    Ok(listener)
}

/// Load the Ed25519 replication signing key from `--replication-key` —
/// shared by the replica connect path and raft-enabled primaries.
fn load_replication_key(
    config: &ServerConfig,
) -> Result<ed25519_dalek::SigningKey, Box<dyn std::error::Error>> {
    let path = config.replication_key.as_ref().ok_or_else(|| {
        std::io::Error::other("--replication-key is required (replica mode or --raft-bind)")
    })?;
    let seed = std::fs::read(path).map_err(|e| {
        std::io::Error::other(format!(
            "failed to read replication key {}: {e}",
            path.display()
        ))
    })?;
    if seed.len() != 32 {
        return Err(format!(
            "replication key must be 32 bytes, got {} ({})",
            seed.len(),
            path.display()
        )
        .into());
    }
    let mut bytes = [0u8; 32];
    bytes.copy_from_slice(&seed);
    Ok(ed25519_dalek::SigningKey::from_bytes(&bytes))
}

/// Run the server as a primary: build the disruptor pipeline, spawn
/// pipeline threads, journal the application's startup events, then
/// accept client connections.
///
/// Control event for the response stage. The io_uring response path reads
/// `fd` for I/O; the `writer` keeps the fd alive via ownership.
pub use crate::ControlEvent;

/// Joinable handles for every long-lived thread spawned by a primary.
/// Optional handles are `None` when their feature is disabled (e.g.,
/// no replication, no health endpoint) or unsupported on a transport
/// (e.g., DPDK runs without an event publisher or shadow snapshotter).
struct PipelineHandles<A: Send + 'static, W: Send + 'static> {
    journal: std::thread::JoinHandle<Result<W, JournalError>>,
    matching: std::thread::JoinHandle<A>,
    response: std::thread::JoinHandle<()>,
    replication: Option<std::thread::JoinHandle<()>>,
    event_publisher: Option<std::thread::JoinHandle<()>>,
    shadow: Option<std::thread::JoinHandle<()>>,
    health: Option<std::thread::JoinHandle<()>>,
    // No standalone tick thread on either transport: io_uring emits ticks
    // from inside the reader (via IORING_OP_TIMEOUT) and DPDK emits them
    // from inside the poll thread (via a wall-clock check between bursts).
}

/// Drain the pipeline and join every worker thread, surfacing panics
/// and journal-stage errors as a single `pipeline failure` return.
///
/// `extras` is a list of pre-joined results (used by the DPDK path,
/// which joins its poll threads before draining the pipeline).
fn shutdown_pipeline_stages<A: Send + 'static, W: Send + 'static>(
    handles: PipelineHandles<A, W>,
    extras: Vec<(String, std::thread::Result<()>)>,
    pipeline_healthy: &AtomicBool,
    shutdown: &AtomicBool,
) -> Result<(), Box<dyn std::error::Error>> {
    info!("shutdown: draining pipeline");
    pipeline_healthy.store(false, Ordering::Relaxed);
    shutdown.store(true, Ordering::Relaxed);

    let mut thread_panicked = false;
    let mut check_join = |name: &str, result: std::thread::Result<()>| {
        if let Err(panic) = result {
            let msg = panic
                .downcast_ref::<&str>()
                .copied()
                .or_else(|| panic.downcast_ref::<String>().map(|s| s.as_str()))
                .unwrap_or("<non-string panic>");
            error!(thread = name, message = msg, "pipeline thread panicked");
            thread_panicked = true;
        }
    };

    let journal_result = handles.journal.join();
    let journal_failed = matches!(&journal_result, Ok(Err(_)));
    if let Ok(Err(ref e)) = journal_result {
        error!(thread = "journal-seq", error = %e, "journal stage returned error");
    }
    check_join("journal-seq", journal_result.map(|_| ()));
    check_join("matching", handles.matching.join().map(|_| ()));
    check_join("response", handles.response.join());
    for (name, r) in extras {
        check_join(&name, r);
    }
    if let Some(h) = handles.replication {
        check_join("replication-sender", h.join());
    }
    if let Some(h) = handles.event_publisher {
        check_join("event-publisher", h.join());
    }
    if let Some(h) = handles.shadow {
        check_join("shadow", h.join());
    }
    if let Some(h) = handles.health {
        check_join("health", h.join());
    }

    // After every stage thread has joined, dump the per-stage latency
    // histograms to stderr. No-op when `latency-trace` is disabled.
    melin_transport_core::trace::print_report_all();

    if thread_panicked || journal_failed {
        error!("shutdown complete (with pipeline failure)");
        return Err("pipeline failure".into());
    }

    info!("shutdown complete");
    Ok(())
}

/// Used by both the normal primary startup path and the promotion path
/// (replica → primary transition).
///
/// `rotate_flag` is the shared `AtomicBool` toggled by the admin
/// endpoint (or `None` if no admin endpoint was configured). On the
/// promotion path the replica's journal stage is torn down and a new
/// primary stage is built here; passing the flag through means the
/// admin endpoint, which was spawned once at process start, keeps
/// driving the new stage's rotation.
#[allow(clippy::too_many_arguments)]
fn run_as_primary<A, L>(
    app: A,
    writer: BufferedWriter<A::Event>,
    mut listener: L,
    // Pre-bound replication listener (non-blocking by construction),
    // `Some` iff `--replication-bind` is set. Bound in `run_impl` before
    // any pipeline thread exists so a bind failure cannot leak threads,
    // and held from boot on replicas so a promotion cannot fail on it.
    repl_listener: Option<crate::replication::ReplicationListener>,
    config: &ServerConfig,
    startup: StartupEvents<A::Event>,
    decoder: RequestDecoderArc<A>,
    encoder: ResponseEncoderArc<A>,
    event_publisher: Option<EventPublisherFn<A>>,
    shutdown: Arc<AtomicBool>,
    authorized_keys: Arc<AuthorizedKeys>,
    needs_seeding: bool,
    rotate_flag: Option<Arc<AtomicBool>>,
    ack_policy_atomic: Arc<AtomicU8>,
    fence_state: Arc<melin_transport_core::fence::FenceState>,
    promotion: Option<u64>,
    raft_status: Option<Arc<melin_transport_core::health::RaftStatus>>,
    // Control-plane advertised tip. This primary's journal stage becomes
    // its writer (the raft driver keeps reading the same handle across a
    // promotion) — see `AdvertisedJournalTip`.
    journal_tip: melin_transport_core::AdvertisedJournalTip,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
    L: BlockingTransportListener,
{
    // Active connection counter shared between accept loop, response
    // stage, and matching stage (for stats queries).
    // Incremented on successful auth, decremented on disconnect.
    // Used to enforce max_connections (SEC-02).
    let active_connections = Arc::new(AtomicU64::new(0));

    // Determine replication mode. Every application's binary ships the
    // full durable transport (journal + replication + shadow), so the
    // same config knob drives any of them.
    let enable_replication = config.replication_bind.is_some();
    if enable_replication && config.standalone {
        return Err("--replication-bind and --standalone are mutually exclusive".into());
    }
    // `--standalone` declares "no replicas ever" — only `disk` can be
    // satisfied. Every other policy requires a second node and would
    // stall the gate forever; reject loudly at startup with the fix in
    // the message.
    if config.standalone && config.ack_policy != crate::ack_policy::AckPolicy::Disk {
        return Err(format!(
            "--standalone requires --ack-policy disk; got `{}` (this policy needs at least one connected replica)",
            config.ack_policy,
        )
        .into());
    }
    // Clone the application for the shadow snapshot stage before the pipeline
    // consumes it. `Application` does not require `Clone`, so this goes
    // through `clone_via_snapshot` (a snapshot round-trip unless the
    // application overrides it with something cheaper).
    let enable_shadow = config.snapshot_interval_ms > 0;
    let shadow_app = if enable_shadow {
        Some(<A as Application>::clone_via_snapshot(&app)?)
    } else {
        None
    };

    // Build the disruptor pipeline with optional replication consumer.
    // An application without an event publisher silently ignores
    // `--event-bind`, so the same invocation works against any binary.
    // The caller (binary) decides whether an event-publisher fn is
    // available; we only allocate the consumer slot when both the fn is
    // wired AND `--event-bind` is set.
    let enable_event_publisher = event_publisher.is_some() && config.event_bind.is_some();
    let Pipeline {
        input_producer,
        journal_stage,
        matching_stage,
        mut output_consumers,
        events_processed,
        input_cursor,
        replication_consumers,
        replicas_connected,
        shadow_consumer,
        chain_hash_lock,
        replication_ring_progress,
        cursors,
    } = build_pipeline_with_replication(
        app,
        writer,
        config.group_commit_delay(),
        Arc::clone(&active_connections),
        enable_replication,
        config.max_journal_batch,
        config.replication_ring_size,
        config.cores.stage_waits(),
        enable_event_publisher,
        enable_shadow,
        Arc::clone(&fence_state),
    );
    // Ring-position cursors for the seed-drain gate below (Acquire loads,
    // stronger than the bundle's monitoring reads). The ack gate and
    // the replication sender pull their typed handles straight from
    // `cursors`; the health endpoint takes `cursors` itself — the quorum
    // and fastest-replica gauges are derived from the bundle's per-slot
    // cursors at read time.
    let journal_cursor = cursors.journal_ring_arc();
    let matching_cursor = cursors.matching_ring_arc();

    // Consumer 0 is always the response stage. Consumer 1 (if present)
    // is the event publisher — only created when --event-bind is set.
    let output_consumer = output_consumers.remove(0);
    let event_publisher_consumer = if enable_event_publisher {
        Some(output_consumers.remove(0))
    } else {
        None
    };

    // Control channel for connect/disconnect events → response stage.
    let (control_tx, control_rx) = std::sync::mpsc::channel();

    // Client writes a halted node refuses at ingress, reader → response
    // stage, counted for the health endpoint. See `crate::halt`.
    let refused_writes = Arc::new(AtomicU64::new(0));
    let halt_gate = crate::halt::HaltGate::new(
        replicas_connected.clone(),
        Arc::clone(&fence_state),
        Arc::clone(&refused_writes),
    );
    let (refusal_tx, refusal_rx) =
        crate::halt::refusal_channel::<A::Report>(Arc::clone(&matching_cursor));

    // Spawn the io_uring reader thread. A single reader uses multishot RECV
    // to multiplex every TCP connection on the server. Pinned to
    // cores.reader. The matching stage is the throughput limit, so a
    // second reader would not raise throughput — only re-introduce the
    // multi-producer ordering race on the input ring.
    let connection_timeout = config.connection_timeout();
    let heartbeat_interval = config.heartbeat_interval();

    // Startup events flow through the disruptor like regular events so
    // they're journaled, replicated, and processed by the matching stage via
    // the normal pipeline. The input ring is single-producer: main publishes
    // them through `input_producer`, then moves it into the reader thread
    // which becomes the sole steady-state producer. No cloning required.
    let mut input_producer = input_producer;

    // Spawn pipeline OS threads.
    let cores = config.cores;

    // Wire runtime journal rotation into the journal stage. The flag is
    // shared with the admin endpoint (spawned once at process start in
    // `run`) — passing it through here means PROMOTE-then-
    // ROTATE on a freshly-promoted node still drives rotation against
    // the new primary's stage. The size threshold uses the same
    // `max_journal_mib` knob that drives startup rotation.
    let mut journal_stage = journal_stage;
    let max_journal_bytes = config.max_journal_mib.saturating_mul(1024 * 1024);
    journal_stage.set_rotation(max_journal_bytes, rotate_flag.clone());
    config.cores.place_journal_children(&mut journal_stage);
    journal_stage.set_staging_mode(config.journal_staging_mode.into());
    // On a primary the journal stage owns the control-plane advertised
    // tip (durable cursor after each fsync batch). On the promotion path
    // this takes over the handle the receiver was advancing.
    journal_stage.set_advertised_tip_publisher(journal_tip.clone());
    if config.max_journal_mib > 0 {
        info!(
            max_journal_mib = config.max_journal_mib,
            "runtime journal rotation enabled (size threshold)"
        );
    }

    // Extract utilization handles before stages are moved into threads.
    let journal_utilization = journal_stage.utilization();
    let matching_utilization = matching_stage.utilization();
    let response_utilization = Arc::new(melin_transport_core::pipeline::StageUtilization::new());

    // Wrap each spawn closure with a tail log so a thread that returns
    // (cleanly OR via `Err`) is visible in the trace stream — the
    // accept loop's `is_finished` check fires error!("pipeline thread
    // died") only on its next tick, which can lag the actual exit by
    // up to 100ms and races with the listener teardown.
    let s1 = Arc::clone(&shutdown);
    let shutdown_for_journal = Arc::clone(&shutdown);
    // Start the journal's disk and preparer threads here rather than on
    // journal-seq, which is pinned by the time it runs: a thread inherits
    // its creator's placement. See `JournalStage::start`.
    let sequencer = journal_stage
        .start()
        .map_err(|e| format!("start the journal stage: {e}"))?;
    let journal_handle = std::thread::Builder::new()
        .name("journal-seq".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("journal-seq", cores.journal_seq.core);
            let result = sequencer.run(&s1);
            let was_shutdown = shutdown_for_journal.load(Ordering::Relaxed);
            match &result {
                Ok(_) if was_shutdown => info!("journal-seq thread exited cleanly on shutdown"),
                Ok(_) => error!("journal-seq thread returned without shutdown signal"),
                Err(e) => error!(error = %e, "journal-seq thread returned Err"),
            }
            result
        })
        .map_err(|e| format!("spawn journal-seq thread: {e}"))?;

    let s2 = Arc::clone(&shutdown);
    let shutdown_for_matching = Arc::clone(&shutdown);
    let matching_handle = std::thread::Builder::new()
        .name("matching".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("matching", cores.matching.core);
            let app = matching_stage.run(&s2);
            let was_shutdown = shutdown_for_matching.load(Ordering::Relaxed);
            if was_shutdown {
                info!("matching thread exited cleanly on shutdown");
            } else {
                error!("matching thread returned without shutdown signal");
            }
            app
        })
        .map_err(|e| format!("spawn matching thread: {e}"))?;

    let replication_metrics = build_replication_metrics(
        replication_consumers.is_some(),
        &ack_policy_atomic,
        config.ack_policy,
    );

    // Per-slot active flags exposed by the journal stage's replication
    // ring; the response gate filters disconnected slots out of the
    // policy's cursor view via these.
    let replica_active: Option<[Arc<AtomicBool>; 2]> =
        replication_ring_progress.as_ref().map(|rp| {
            [
                Arc::clone(&rp.active_flags[0]),
                Arc::clone(&rp.active_flags[1]),
            ]
        });

    // Typed durable-cursor handle for the response thread's gate.
    let journal_persisted_wire_seq_response = cursors.durable_wire_seq();
    let replication_metrics_response = replication_metrics.as_ref().map(Arc::clone);
    let replica_active_response = replica_active.clone();
    let ack_policy_response = Arc::clone(&ack_policy_atomic);
    let s3 = Arc::clone(&shutdown);
    let shutdown_for_response = Arc::clone(&shutdown);
    let response_utilization_thread = Arc::clone(&response_utilization);
    let response_fence = Arc::clone(&fence_state);
    let active_connections_response = Arc::clone(&active_connections);
    let response_handle = std::thread::Builder::new()
        .name("response".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("response", cores.response.core);
            crate::response::run::<A>(
                output_consumer,
                control_rx,
                crate::response::Response::<A> {
                    journal_persisted_wire_seq: journal_persisted_wire_seq_response,
                    ack_policy: ack_policy_response,
                    replication_metrics: replication_metrics_response,
                    replica_active: replica_active_response,
                    heartbeat_interval,
                    wait: cores.response.wait,
                    utilization: response_utilization_thread,
                    encoder,
                    fence_state: response_fence,
                    active_connections: active_connections_response,
                    refusals: refusal_rx,
                    #[cfg(test)]
                    pause_after_control_drain: None,
                },
                &s3,
            );
            let was_shutdown = shutdown_for_response.load(Ordering::Relaxed);
            if was_shutdown {
                info!("response thread exited cleanly on shutdown");
            } else {
                error!("response thread returned without shutdown signal");
            }
        })
        .map_err(|e| format!("spawn response thread: {e}"))?;

    // Spawn replication sender thread if enabled. The journal stage publishes
    // encoded batches to a pre-allocated ring; the sender thread consumes them.
    // `replica_ready` is set when the first replica connects — seeding waits
    // on this to ensure seed events aren't drained before the replica arrives.
    let replica_ready = Arc::new(AtomicBool::new(false));
    // Ring depth monitoring: the producer cursors are in ReplicationRingProgress
    // (owned by this function), so we compute depth via ring_progress rather
    // than storing Box<dyn QueueCursor> in ReplicationMetrics. The health
    // snapshot reads consumer cursors from ReplicationMetrics and producer
    // cursors are not needed — ring depth is a secondary metric.

    let replication_handle = if let Some((repl_consumer_1, repl_consumer_2)) = replication_consumers
    {
        let s_repl = Arc::clone(&shutdown);
        let repl_slots = cursors.replica_slot_cursors();
        let ready_flag = Arc::clone(&replica_ready);
        let connected_counter = replicas_connected
            .clone()
            .ok_or("replicas_connected must be Some when replication is enabled")?;

        let batch_size = config.replication_batch_size;
        let heartbeat_secs = config.replication_heartbeat_secs;
        let journal_path = config.journal.clone();
        let repl_auth_keys = Arc::clone(&authorized_keys);
        let evict_flags = replication_ring_progress
            .as_ref()
            .map(|rp| {
                [
                    Arc::clone(&rp.evict_flags[0]),
                    Arc::clone(&rp.evict_flags[1]),
                ]
            })
            .unwrap_or_else(|| {
                [
                    Arc::new(AtomicBool::new(false)),
                    Arc::new(AtomicBool::new(false)),
                ]
            });
        let active_flags = replication_ring_progress
            .as_ref()
            .map(|rp| {
                [
                    Arc::clone(&rp.active_flags[0]),
                    Arc::clone(&rp.active_flags[1]),
                ]
            })
            .unwrap_or_else(|| {
                [
                    Arc::new(AtomicBool::new(false)),
                    Arc::new(AtomicBool::new(false)),
                ]
            });
        let repl_metrics = replication_metrics
            .clone()
            .ok_or("replication_metrics must be Some when replication is enabled")?;
        let handlers = [cores.repl_handler_0, cores.repl_handler_1];
        let sender_fence = Arc::clone(&fence_state);
        let sender_ack_policy = Arc::clone(&ack_policy_atomic);
        // Bound in `run_impl` before any pipeline thread was spawned —
        // see the `repl_listener` parameter doc.
        let repl_listener = repl_listener
            .ok_or("replication listener must be pre-bound when replication is enabled")?;
        let repl_accept_handle = std::thread::Builder::new()
            // `repl-accept`, not `repl-sender`: this thread sends nothing.
            // The ring consumers are moved into the per-replica handler
            // threads, which do all the streaming; this one accepts
            // connections, reaps finished handlers, and sleeps 50 ms.
            .name("repl-accept".into())
            .spawn(move || {
                // Deliberately NOT pinned. Reserving a core — at
                // SCHED_FIFO, on an isolated one — for a thread that idles
                // 99.9% of the time bought nothing.
                //
                // Staying unpinned is now load-bearing, not just thrifty.
                // The handler threads are spawned from here and inherit
                // this thread's affinity and scheduling policy at
                // creation; a child of a pinned real-time parent cannot
                // move itself off that core, because moving itself
                // requires running. That is the DPDK handshake hang, and
                // this path escaped it only because the 50 ms sleep
                // happened to yield the core. See
                // `replication::validation_worker`.
                //
                // `PipelineCores` consequently has no field for it, and
                // `--cores` no name.
                crate::replication::run_sender::<A>(
                    crate::replication::Sender {
                        listener: repl_listener,
                        repl_consumer_1,
                        repl_consumer_2,
                        replica_slots: repl_slots,
                        journal_path,
                        authorized_keys: repl_auth_keys,
                        evict_flags,
                        active_flags,
                        metrics: repl_metrics,
                        handlers,
                        batch_size,
                        heartbeat_secs,
                        fence_state: sender_fence,
                        ack_policy: sender_ack_policy,
                    },
                    &s_repl,
                    &ready_flag,
                    &connected_counter,
                );
            })
            .map_err(|e| format!("spawn replication sender thread: {e}"))?;

        Some(repl_accept_handle)
    } else {
        if !config.standalone && config.replica_of.is_none() {
            info!("running in standalone mode (no replication)");
        }
        None
    };

    // Spawn event publisher thread if enabled. Consumes from output ring
    // consumer 1 and broadcasts the application's reports to TCP
    // subscribers. Caller-supplied (an application with an output feed
    // wires its publisher; one without passes `None`), so the runtime
    // carries no application-specific reference.
    let event_publisher_handle = spawn_event_publisher::<A>(
        event_publisher_consumer,
        event_publisher,
        config,
        &cores,
        &authorized_keys,
        &shutdown,
    )?;

    let shadow_handle = spawn_shadow_stage::<A>(
        shadow_consumer,
        shadow_app,
        chain_hash_lock,
        config,
        &cores,
        &shutdown,
        fence_state.epoch(),
    )?;

    // Spawn the health endpoint BEFORE the wait-for-replica gate so
    // operators (and the failover test harness) can probe `/healthz`
    // to confirm the server has bound its sockets and is ready to
    // accept replica connections — even though seeding hasn't started
    // yet. The health snapshot legitimately reports `halted` while
    // waiting for the first replica; that's accurate, not a
    // misreport. Replicas connect to the *replication* endpoint, not
    // the client port, so spawning health here doesn't change accept
    // semantics.
    let pipeline_healthy = Arc::new(AtomicBool::new(true));
    let health_handle = spawn_health_endpoint(
        config,
        &active_connections,
        &events_processed,
        &refused_writes,
        &cursors,
        input_cursor,
        &pipeline_healthy,
        &replicas_connected,
        &fence_state,
        &replication_metrics,
        &replica_active,
        &replication_ring_progress,
        &journal_utilization,
        &matching_utilization,
        &response_utilization,
        &shutdown,
        &raft_status,
    )?;

    // Promotion fencing: a node that reached primary via promotion injects
    // an `EpochBump` as the first journaled entry of its tenure, raising the
    // cluster epoch so a paused/partitioned ex-primary self-demotes when a
    // handshake crosses (see `melin_transport_core::fence`). Genesis primaries
    // skip this — their epoch stays at whatever the journal recovered. The
    // bump rides the input ring like a seed event (connection_id == 0), so it
    // flows through journal + replication to every replica. We wait for the
    // matching stage to apply it (epoch advanced) before serving so the first
    // handshake already advertises the new epoch.
    //
    // The new epoch honours the promotion request's floor: a manual
    // `PROMOTE` carries `MANUAL` (= 1) and resolves to the classic
    // `epoch + 1`; a raft auto-promotion carries its election term
    // (strictly above the old epoch by the driver's request rule), so
    // tenure epochs align with raft terms and two overlapping
    // promotions from different elections always allocate distinct
    // epochs — the newer one fences the older.
    if let Some(requested_epoch) = promotion {
        use melin_app::unix_epoch_nanos;
        use melin_journal::JournalEvent;
        use melin_transport_core::trace::mono_trace_ns;

        let new_epoch = fence_state.epoch().saturating_add(1).max(requested_epoch);
        // Re-validate the term↔epoch alignment at the moment the epoch is
        // minted, not just at the driver's request-time check: a streamed
        // `EpochBump` from a concurrent promotion elsewhere can raise the
        // fence during the drain, in which case `max` allocates `epoch+1`
        // instead of the election term. Fencing still converges (the
        // epochs stay distinct and the higher one wins), but the skew
        // makes later "epochs outran raft terms" refusals — so say what
        // actually happened while the evidence exists. Manual promotions
        // (requested == MANUAL) are exempt: they never claimed alignment.
        if requested_epoch > crate::promotion::PromotionRequest::MANUAL
            && new_epoch != requested_epoch
        {
            warn!(
                new_epoch,
                requested_epoch,
                "promotion epoch does not match its election term — a concurrent \
                 promotion advanced the fencing epoch mid-drain; auto-promotion \
                 refusals may report term/epoch misalignment until a newer election"
            );
        }
        info!(
            new_epoch,
            requested_epoch, "promotion: injecting epoch bump"
        );
        input_producer.publish(InputSlot {
            connection_id: 0,
            key_hash: 0,
            sequence: 0,
            timestamp_ns: unix_epoch_nanos(),
            event: JournalEvent::EpochBump { epoch: new_epoch },
            publish_ts: mono_trace_ns(),
            recv_ts: mono_trace_ns(),
        });
        // Wait until the matching stage observes the bump (epoch raised) so
        // the node advertises `new_epoch` on the very first handshake. Bounded
        // by the shutdown flag so a stuck pipeline can't wedge startup.
        ORCHESTRATOR_WAIT
            .wait_until(|| fence_state.epoch() >= new_epoch || shutdown.load(Ordering::Relaxed));
    }

    // Startup events: genesis on a new journal, then `on_primary` — after
    // the epoch bump on promotion, so the bump stays the first entry of
    // the tenure. (Genesis and promotion never coincide: a promoted node
    // has its history from the stream.)
    //
    // When replication is enabled, a new journal waits for the first
    // replica to connect before publishing. replica_ready is set by the
    // replica handler thread after catch-up completes and it enters the
    // live streaming loop — this ensures the ring consumer is actively
    // draining before a large genesis starts flowing. The health endpoint
    // was spawned above so `/healthz` answers during the wait.
    if enable_replication && needs_seeding {
        info!("waiting for replica to connect before seeding...");
        while !replica_ready.load(Ordering::Acquire) {
            if shutdown.load(Ordering::Relaxed) {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
    }
    let genesis = if needs_seeding {
        startup.genesis
    } else {
        Vec::new()
    };
    journal_startup_events(
        genesis,
        startup.on_primary,
        &mut input_producer,
        &journal_cursor,
        &matching_cursor,
        &replication_ring_progress,
        &shutdown,
    );

    // Now that seeding is fully drained, spawn the reader thread. From here
    // on the reader is the sole producer on the input ring (the seed loop
    // has finished and its clone of `input_producer` was dropped at the end
    // of the block above). Any subsequent ticks the reader emits cannot
    // race with seed events: there are none in flight.
    //
    // If shutdown was requested while seed was draining we still spawn the
    // reader so the unified shutdown sequence below joins every thread.
    let reader_shutdown = Arc::new(AtomicBool::new(false));
    let mut reader_handle = crate::reader::spawn_reader::<A, _>(
        input_producer,
        decoder,
        halt_gate,
        refusal_tx,
        control_tx.clone(),
        config.cores.reader.core,
        connection_timeout,
        config.tick_interval(),
        Arc::clone(&reader_shutdown),
    );

    // Health endpoint and `pipeline_healthy` were already spawned/created
    // earlier (before the wait-for-replica gate) so probes can succeed
    // before seeding completes.

    // Set the listener to non-blocking so accept() returns immediately
    // with WouldBlock when no connection is pending. This lets the accept
    // loop check the shutdown flag without blocking indefinitely.
    // Rust's std TcpListener retries on EINTR, so signals alone can't
    // interrupt a blocking accept().
    listener.set_nonblocking(true);

    info!(addr = %config.bind, "listening");

    // Monotonically increasing connection ID counter. AtomicU64 because
    // the accept loop is the only writer, but using atomic for future
    // flexibility (e.g., multiple listeners).
    let next_connection_id = AtomicU64::new(1);

    // Accept loop — non-blocking with 100ms sleep on WouldBlock. Each
    // accepted connection is registered with the reader thread (no
    // per-connection threads).
    loop {
        if shutdown.load(Ordering::Relaxed) {
            info!("shutdown signal received");
            break;
        }

        // Detect pipeline thread death early so we don't keep accepting
        // connections into a broken pipeline. These threads only exit on
        // shutdown or panic — if one is finished while shutdown is false,
        // it panicked. Re-check shutdown to avoid a TOCTOU race where a
        // clean shutdown signal arrives between the two checks.
        let event_pub_died = event_publisher_handle
            .as_ref()
            .is_some_and(|h| h.is_finished());
        let shadow_died = shadow_handle.as_ref().is_some_and(|h| h.is_finished());
        if (journal_handle.is_finished()
            || matching_handle.is_finished()
            || response_handle.is_finished()
            || event_pub_died
            || shadow_died)
            && !shutdown.load(Ordering::Relaxed)
        {
            error!("pipeline thread died, initiating shutdown");
            pipeline_healthy.store(false, Ordering::Relaxed);
            break;
        }

        let (mut std_read, mut std_write, addr) = match listener.accept() {
            Ok(conn) => conn,
            Err(e) => {
                if e.kind() == std::io::ErrorKind::WouldBlock {
                    // No pending connection — sleep briefly then retry.
                    std::thread::sleep(std::time::Duration::from_millis(100));
                    continue;
                }
                error!(error = %e, "accept error");
                continue;
            }
        };

        // Enforce max_connections limit (SEC-02). Reject early before
        // spending time on auth. The counter is decremented by the response
        // stage on disconnect or write error.
        if config.max_connections > 0
            && active_connections.load(Ordering::Relaxed) >= config.max_connections
        {
            warn!(addr = %addr, "connection rejected: max_connections reached");
            drop(std_read);
            drop(std_write);
            continue;
        }

        let connection_id = ConnectionId(next_connection_id.fetch_add(1, Ordering::Relaxed));

        debug!(connection_id = connection_id.0, addr = %addr, "new connection");

        // Set a 5-second read timeout for the auth handshake to prevent
        // slow/malicious clients from blocking the accept loop.
        if let Err(e) = set_read_timeout(&std_read, Some(std::time::Duration::from_secs(5))) {
            debug!(connection_id = connection_id.0, error = %e, "failed to set auth timeout");
        }

        // Challenge-response authentication handshake (cold path).
        // 1. Send Challenge with random nonce
        // 2. Read ChallengeResponse (signature + public key)
        // 3. Verify signature and look up key in authorized_keys
        // 4. Send ServerReady on success, AuthFailed on failure
        let (permission, public_key_bytes) = match authenticate_connection(
            connection_id,
            addr,
            &mut std_read,
            &mut std_write,
            &authorized_keys,
        ) {
            Ok(pair) => pair,
            Err(e) => {
                debug!(connection_id = connection_id.0, addr = %addr, error = %e, "auth failed, dropping");
                continue;
            }
        };

        // The identity the application sees as `ApplyCtx::key_hash`.
        let key_hash = melin_app::key_hash(&public_key_bytes);

        active_connections.fetch_add(1, Ordering::Relaxed);

        // Clear the read timeout before handing to the io_uring reader.
        // io_uring uses kernel-managed I/O, so the timeout is irrelevant,
        // but clearing it avoids surprising behavior if the fd is ever
        // used in blocking mode again.
        if let Err(e) = set_read_timeout(&std_read, None) {
            debug!(connection_id = connection_id.0, error = %e, "failed to clear auth timeout");
        }

        // Enable SO_BUSY_POLL on the client data socket. The reader
        // thread already busy-spins on io_uring CQEs, so the kernel's
        // NIC busy-poll happens during cycles that would have been
        // spent spinning anyway — net cost is zero, and we avoid the
        // softirq → wakeup handoff for every recv on this connection.
        if let Err(e) = set_busy_poll(&std_read, BUSY_POLL_US) {
            // Best-effort: a kernel without CAP_NET_ADMIN or with
            // SO_BUSY_POLL disabled by sysctl will reject this. Logged
            // as debug — it's a per-connection event and only affects
            // latency, not correctness.
            debug!(
                connection_id = connection_id.0,
                error = %e,
                "failed to set SO_BUSY_POLL on client socket"
            );
        }

        // Belt-and-braces write timeout (SEC-01). The real slow-client
        // protection is in the response stage — MSG_DONTWAIT sends,
        // paced retries, and the blocked-send timeout — since io_uring
        // SEND does not honor SO_SNDTIMEO. This bounds any residual
        // blocking write(2) on this fd (none exist today; the response
        // stage is the sole egress writer).
        if let Err(e) = set_write_timeout(&std_write, Some(std::time::Duration::from_secs(5))) {
            debug!(connection_id = connection_id.0, error = %e, "failed to set write timeout");
        }

        // Register the writer with the response thread before the
        // reader, and keep it that way: the response stage reads its
        // output ring before it drains this channel, and relies on
        // `Connected` being queued before any request of the connection
        // can be read — see the drain in `response::run`. Queued, not
        // yet applied: the stage picks the event up on its next
        // iteration, which that order makes early enough.
        let fd = std_write.as_raw_fd();
        let boxed_writer: Box<dyn std::io::Write + Send> = Box::new(std_write);
        let control_event = ControlEvent::Connected {
            connection_id: connection_id.0,
            fd,
            writer: BlockingFrameWriter::new(boxed_writer),
        };
        if control_tx.send(control_event).is_err() {
            info!("response thread gone, shutting down");
            break;
        }

        // Register the reader fd with the io_uring reader thread.
        reader_handle.register(crate::reader::ReaderRegistration {
            connection_id,
            reader: std_read,
            addr,
            permission,
            key_hash,
        });
    }

    // --- Ordered shutdown sequence ---
    // 1. Stop readers first so no new events enter the disruptor.
    info!("shutdown: stopping reader thread");
    reader_handle.shutdown();
    reader_handle.join();

    // 2. Now drain the pipeline and join every worker thread.
    shutdown_pipeline_stages(
        PipelineHandles {
            journal: journal_handle,
            matching: matching_handle,
            response: response_handle,
            replication: replication_handle,
            event_publisher: event_publisher_handle,
            shadow: shadow_handle,
            health: health_handle,
        },
        Vec::new(),
        &pipeline_healthy,
        &shutdown,
    )
}

/// Run the server with DPDK kernel-bypass networking.
///
/// Replaces the kernel TCP stack entirely. The DPDK poll thread handles
/// all NIC I/O and TCP processing via smoltcp. The response stage encodes
/// frames and pushes them through an mpsc channel to the poll thread.
///
/// Thread layout:
/// - Core N:   DPDK poll thread (rx_burst, smoltcp, frame decode, tx_burst)
/// - Core 1:   Journal stage
/// - Core 2:   Matching stage
/// - Core 3:   Response stage (encodes to TX channel)
///
/// See [`run`] for the role of `startup`.
#[cfg(feature = "dpdk")]
fn run_dpdk<A>(
    config: ServerConfig,
    startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: RequestDecoderArc<A>,
    encoder: ResponseEncoderArc<A>,
    event_publisher: Option<EventPublisherFn<A>>,
    shutdown: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    let dpdk_config = dpdk_config_from(&config)?;

    run_dpdk_impl::<A>(
        config,
        startup,
        sizing,
        decoder,
        encoder,
        event_publisher,
        dpdk_config,
        shutdown,
    )
}

/// Translate the CLI surface into a [`melin_dpdk::DpdkConfig`].
///
/// Every operator-supplied string is validated here, before EAL init and
/// before any thread is spawned, and a bad value is returned as an error
/// rather than a panic. That ordering is the point: the peer MAC is only
/// consumed inside the replica's reconnect loop, so parsing it there
/// would turn a typo into a crash of an already-running node.
#[cfg(feature = "dpdk")]
fn dpdk_config_from(cfg: &ServerConfig) -> Result<melin_dpdk::DpdkConfig, String> {
    let parse_ip = |value: &str, flag: &str| -> Result<std::net::Ipv4Addr, String> {
        value
            .parse()
            .map_err(|e| format!("invalid {flag} address '{value}': {e}"))
    };
    let parse_mac = |value: &str, flag: &str| -> Result<[u8; 6], String> {
        melin_dpdk::try_parse_mac(value).map_err(|e| format!("invalid {flag}: {e}"))
    };

    Ok(melin_dpdk::DpdkConfig {
        eal_args: cfg
            .dpdk_eal_args
            .split_whitespace()
            .map(String::from)
            .collect(),
        port_ids: cfg.dpdk_ports.clone(),
        ip_addr: parse_ip(&cfg.dpdk_ip, "--dpdk-ip")?,
        prefix_len: cfg.dpdk_prefix_len,
        gateway: cfg
            .dpdk_gateway
            .as_deref()
            .map(|s| parse_ip(s, "--dpdk-gateway"))
            .transpose()?,
        listen_port: cfg.bind.port(),
        mtu: cfg.dpdk_mtu,
        vlan_id: cfg.dpdk_vlan,
        peer_ip: cfg
            .dpdk_peer_ip
            .as_deref()
            .map(|s| parse_ip(s, "--dpdk-peer-ip"))
            .transpose()?,
        gateway_mac: cfg
            .dpdk_gateway_mac
            .as_deref()
            .map(|s| parse_mac(s, "--dpdk-gateway-mac"))
            .transpose()?,
        peer_mac: cfg
            .dpdk_peer_mac
            .as_deref()
            .map(|s| parse_mac(s, "--dpdk-peer-mac"))
            .transpose()?,
        num_queues: 1,
    })
}

#[cfg(feature = "dpdk")]
fn run_dpdk_impl<A>(
    config: ServerConfig,
    mut startup: StartupEvents<A::Event>,
    sizing: A::Sizing,
    decoder: RequestDecoderArc<A>,
    encoder: ResponseEncoderArc<A>,
    event_publisher: Option<EventPublisherFn<A>>,
    dpdk_config: melin_dpdk::DpdkConfig,
    shutdown: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error>>
where
    A: Application + Send + 'static,
    A::Event: Send + Sync + 'static,
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    warn_if_chain_disabled();

    // As on the kernel-TCP path: resolve the layout once, up front. The
    // reader entry pins the NIC poll thread here, which never yields.
    let config = ServerConfig {
        cores: config.resolved_cores(ReaderThread::DpdkPoll)?,
        ..config
    };
    log_layout(&config.cores);
    // The one placement the layout check cannot make honest: the poll
    // thread polls flat out whatever its entry says, and unpinned it
    // does so wherever the scheduler puts it — next to whatever else is
    // there. Accepted (test hosts run this way), but said out loud.
    if !config.cores.reader.is_pinned() {
        warn!(
            "--cores: the reader is unpinned, but on DPDK it is the NIC poll thread and \
             busy-polls regardless; it will take a full core wherever the scheduler places \
             it. Give it a core of its own"
        );
    }

    // Mirrors the kernel-TCP `run` path: one atomic per
    // process, threaded into both replica (pre-staging for promotion)
    // and primary admin listeners.
    let ack_policy_atomic = Arc::new(AtomicU8::new(config.ack_policy.as_u8()));
    // Initialize shared DPDK resources (EAL, mempool, ports with N queues).
    let shared = melin_dpdk::DpdkShared::init(&dpdk_config)?;
    // Actual queue count may be less than requested (TAP only supports 1).
    let num_dpdk_threads = shared.num_queues as usize;

    // --- Replica mode (DPDK) ---
    // If --replica-of is set, run the DPDK replication receiver instead
    // of the primary path. The receiver uses one queue pair for the
    // outbound connection to the primary.
    if let Some(primary_addr) = config.replica_of {
        info!(primary = %primary_addr, "starting in replica mode (DPDK)");
        // As on the kernel-TCP path: a replica never journals genesis.
        startup.genesis = Vec::new();

        // Load authorized keys early — the admin listener needs them for
        // Ed25519 challenge-response auth (operator keys only).
        let authorized_keys = Arc::new(AuthorizedKeys::load(&config.authorized_keys)?);
        info!(
            keys = authorized_keys.len(),
            path = %config.authorized_keys.display(),
            "loaded authorized keys (DPDK replica mode)"
        );

        // Load the replication signing key — the replica signs the primary's
        // challenge with it. Mirrors the kernel-TCP replica path.
        let replication_key_path = config.replication_key.as_ref().ok_or_else(|| {
            std::io::Error::other("--replication-key is required in replica mode (--replica-of)")
        })?;
        let signing_key = {
            let seed = std::fs::read(replication_key_path).map_err(|e| {
                std::io::Error::other(format!(
                    "failed to read replication key {}: {e}",
                    replication_key_path.display()
                ))
            })?;
            if seed.len() != 32 {
                return Err(format!(
                    "replication key must be 32 bytes, got {} ({})",
                    seed.len(),
                    replication_key_path.display()
                )
                .into());
            }
            let mut bytes = [0u8; 32];
            bytes.copy_from_slice(&seed);
            ed25519_dalek::SigningKey::from_bytes(&bytes)
        };

        // The replica's control-plane bundle, same shape as the kernel
        // TCP replica path — see `run` for the rationale on lifetime.
        let control = crate::replication::ReplicaControlPlane::new();
        let promotion_request = control.promote.clone();
        let rotate_flag = config.admin_bind.map(|_| Arc::new(AtomicBool::new(false)));
        let _admin_handle = config
            .admin_bind
            .map(|addr| {
                crate::admin::spawn(
                    addr,
                    Some(promotion_request.clone()),
                    rotate_flag.clone(),
                    Some(Arc::clone(&ack_policy_atomic)),
                    Arc::clone(&shutdown),
                    Arc::clone(&authorized_keys),
                )
            })
            .transpose()?;

        // Shared fencing state, created before mode detection so it
        // survives a replica → primary transition. Seeded at epoch 0; the
        // receiver raises it from the replica's recovered journal and the
        // replication stream, and a promotion bumps it by injecting an
        // `EpochBump`. Carried into `run_as_primary` post-promotion.
        let fence_state = Arc::new(melin_transport_core::fence::FenceState::new(0));

        // Control-plane raft runs on kernel TCP even on DPDK nodes — the
        // control plane is off the hot path and must not consume a DPDK
        // queue. Same promotion-surviving placement and tip ownership as
        // the kernel replica path.
        let journal_tip = control.journal_tip.clone();
        let mut raft = match crate::raft::build_raft_config(&config)? {
            None => crate::raft::RaftDriverGuard::disabled(&shutdown),
            Some(cfg) => crate::raft::spawn_raft_driver(
                cfg,
                config.raft_auto_promote,
                &signing_key,
                &authorized_keys,
                &fence_state,
                journal_tip.clone(),
                Arc::clone(&control.tip_ready),
                // A replica claims to be serving once a promotion is in
                // flight (see SupersessionPolicy).
                {
                    let promote = control.promote.clone();
                    Arc::new(move || promote.is_requested())
                },
                &shutdown,
            )?,
        };
        // Act on election wins when the operator opted in. Genesis
        // primaries have nothing to promote — replica paths only. The
        // guard joins the thread on every exit path.
        if config.raft_auto_promote
            && let Some(wiring) = raft.promotion_wiring()
        {
            raft.arm_promotion(crate::raft_promotion::spawn_auto_promotion(
                wiring.status,
                control.clone(),
                Arc::clone(&fence_state),
                Arc::clone(&ack_policy_atomic),
                wiring.peer_tips,
                wiring.peer_ids,
                wiring.elect_requested,
                wiring.elect_enabled,
                Arc::clone(&shutdown),
            ));
        }
        let raft_status = raft.status();
        let mut replica_health = crate::raft::spawn_replica_health(
            &config,
            &fence_state,
            raft_status.as_ref(),
            Arc::clone(&control.pipeline_healthy),
        )?;

        // No local rotation triggers on the replica side — rotation is
        // primary-driven (see the kernel-TCP receiver path).

        // Use queue 0 for the replication receiver's smoltcp connection.
        // Listen port is unused (receiver connects outbound, doesn't listen),
        // but DpdkTransport requires one — use an ephemeral port.
        let mut repl_transport = melin_dpdk::DpdkTransport::from_shared_with_port(
            &shared,
            &dpdk_config,
            0,
            39999, // Ephemeral — receiver connects outbound, doesn't accept.
        )?;
        repl_transport.send_gratuitous_arp();

        let primary_ipv4 = match primary_addr.ip() {
            std::net::IpAddr::V4(ip) => ip,
            std::net::IpAddr::V6(_) => {
                return Err("DPDK replication requires IPv4".into());
            }
        };

        match crate::replication::run_receiver_dpdk::<A>(
            repl_transport,
            primary_ipv4,
            primary_addr.port(),
            &signing_key,
            &config.journal,
            &shutdown,
            &control,
            config.snapshot_interval_ms,
            config.shadow_snapshot_path(),
            config.cores,
            config.journal_staging_mode.into(),
            config.group_commit_delay(),
            config.replication_pipeline_depth,
            Arc::clone(&fence_state),
            &sizing,
        )? {
            // Clean shutdown — the raft and health guards tear down on drop.
            None => return Ok(()),
            Some((mut app, writer)) => {
                // Promotion! Transition to primary mode (DPDK).
                info!("replica promoted (DPDK) — transitioning to primary");
                // Release --health-bind before run_as_primary rebinds it.
                replica_health.stop();
                // Idempotent; see the kernel-TCP promotion path.
                <A as Application>::prefault(&mut app, &sizing);

                // Clear a ROTATE latched while this node was a replica —
                // see the kernel-TCP promotion path.
                if let Some(ref flag) = rotate_flag {
                    flag.store(false, Ordering::Release);
                }

                // TODO: run_as_primary_dpdk — for now, fall back to
                // kernel TCP primary after promotion.
                warn!("DPDK primary promotion not yet implemented — falling back to kernel TCP");
                let listener = melin_wire_protocol::tcp::BlockingTcpListener::bind(config.bind)?;
                // Unlike the kernel path (bound at boot in `run_impl`),
                // the fallback's kernel listeners only come up here at
                // promotion — the DPDK stack carried replication until
                // now. See `bind_replication_listener`.
                let repl_listener = config
                    .replication_bind
                    .map(bind_replication_listener)
                    .transpose()?;
                // The raft guard drops — stopping the driver and joining
                // the (already exited) promotion thread — after this
                // returns; see the kernel-TCP promotion path.
                return run_as_primary::<A, _>(
                    app,
                    writer,
                    listener,
                    repl_listener,
                    &config,
                    startup,
                    decoder,
                    encoder,
                    event_publisher,
                    Arc::clone(&shutdown),
                    authorized_keys,
                    false,
                    rotate_flag,
                    ack_policy_atomic,
                    fence_state,
                    promotion_request.pending(), // promoted — EpochBump with the request's epoch floor
                    raft_status,
                    journal_tip,
                );
            }
        }
    }

    // Create per-thread transports (each gets its own queue pair + smoltcp stack).
    // The single-queue refactor (`feat/dpdk-single-queue`) collapsed
    // replication onto the same queue/thread as client traffic — no
    // separate queue is reserved any more, so `num_client_queues`
    // simply tracks `num_dpdk_threads`.
    let num_client_queues = num_dpdk_threads;
    let mut transports = Vec::with_capacity(num_client_queues);
    for q in 0..num_client_queues {
        let mut transport =
            melin_dpdk::DpdkTransport::from_shared(&shared, &dpdk_config, q as u16)?;
        // Send a gratuitous ARP from the first queue so the switch learns
        // our VF's MAC address. Without this, SR-IOV VFs that can't enable
        // promiscuous mode are unreachable — the switch has no forwarding
        // entry and drops unicast frames to our MAC.
        if q == 0 {
            transport.send_gratuitous_arp();
        }
        transports.push(transport);
    }

    // Load authorized keys for challenge-response authentication.
    let authorized_keys = Arc::new(AuthorizedKeys::load(&config.authorized_keys)?);
    info!(
        keys = authorized_keys.len(),
        path = %config.authorized_keys.display(),
        "loaded authorized keys"
    );

    // Initialize or recover the application, then size it — see the
    // kernel-TCP primary path.
    let (mut app, writer, needs_seeding, recovered_epoch) =
        init_engine::<A, BufferedWriter<A::Event>>(&config, &sizing)?;
    <A as Application>::prefault(&mut app, &sizing);

    // Fencing state for this DPDK primary, seeded with the recovered epoch.
    let fence_state = Arc::new(melin_transport_core::fence::FenceState::new(
        recovered_epoch,
    ));

    // Control-plane raft (kernel TCP even on DPDK — off the hot path, no
    // DPDK queue consumed). Signing key loaded only when raft is on.
    // A DPDK primary's fence epoch is recovered here too; seed the
    // advertised sequence from the recovered journal (see the kernel
    // primary path). The pipeline's journal stage owns the handle below.
    let journal_tip = melin_transport_core::AdvertisedJournalTip::new(
        melin_transport_core::WireSeq::new(writer.next_sequence().saturating_sub(1)),
    );
    let raft = match crate::raft::build_raft_config(&config)? {
        None => crate::raft::RaftDriverGuard::disabled(&shutdown),
        Some(cfg) => {
            let signing_key = load_replication_key(&config)?;
            crate::raft::spawn_raft_driver(
                cfg,
                config.raft_auto_promote,
                &signing_key,
                &authorized_keys,
                &fence_state,
                journal_tip.clone(),
                Arc::new(AtomicBool::new(true)),
                // A primary always claims to be serving.
                Arc::new(|| true),
                &shutdown,
            )?
        }
    };
    let raft_status = raft.status();

    // Clone the application's state for the shadow snapshot stage before
    // moving it into the pipeline (same as the kernel TCP path).
    let enable_shadow = config.snapshot_interval_ms > 0;
    let shadow_app = if enable_shadow {
        Some(<A as Application>::clone_via_snapshot(&app)?)
    } else {
        None
    };

    let active_connections = Arc::new(AtomicU64::new(0));

    // Replication setup (same as TCP path).
    let enable_replication = config.replication_bind.is_some();
    if enable_replication && config.standalone {
        return Err("--replication-bind and --standalone are mutually exclusive".into());
    }
    // `--standalone` declares "no replicas ever" — only `disk` can be
    // satisfied. Every other policy requires a second node and would
    // stall the gate forever; reject loudly at startup with the fix in
    // the message.
    if config.standalone && config.ack_policy != crate::ack_policy::AckPolicy::Disk {
        return Err(format!(
            "--standalone requires --ack-policy disk; got `{}` (this policy needs at least one connected replica)",
            config.ack_policy,
        )
        .into());
    }

    // Build disruptor pipeline (same flags as the kernel TCP path).
    // DPDK doesn't currently spawn the publisher thread (see
    // `event_publisher: None` in the handles bundle further down) but
    // gating consumer allocation on the same `Some`/`event_bind` pair
    // the kernel-TCP path uses means passing `None` from the binary is
    // honoured here too — no orphan consumer slot is wired up.
    let enable_event_publisher = event_publisher.is_some() && config.event_bind.is_some();
    let enable_shadow = config.snapshot_interval_ms > 0;
    let Pipeline {
        input_producer,
        journal_stage,
        matching_stage,
        mut output_consumers,
        events_processed,
        input_cursor,
        replication_consumers,
        replicas_connected,
        shadow_consumer,
        chain_hash_lock,
        replication_ring_progress,
        cursors,
    } = build_pipeline_with_replication(
        app,
        writer,
        config.group_commit_delay(),
        Arc::clone(&active_connections),
        enable_replication,
        config.max_journal_batch,
        config.replication_ring_size,
        config.cores.stage_waits(),
        enable_event_publisher,
        enable_shadow,
        Arc::clone(&fence_state),
    );
    // Ring-position cursors for the seed-drain gate below (Acquire loads,
    // stronger than the bundle's monitoring reads). The ack gate and
    // the replication sender pull their typed handles straight from
    // `cursors`; the health endpoint takes `cursors` itself.
    let journal_cursor = cursors.journal_ring_arc();
    let matching_cursor = cursors.matching_ring_arc();

    let heartbeat_interval = config.heartbeat_interval();

    // Seed events flow through the disruptor like regular events. The input
    // ring is single-producer: main publishes seeds, then moves the
    // producer into the DPDK poll thread. No cloning required.
    let mut input_producer = input_producer;

    // The DPDK poll thread also generates the application's clock ticks via
    // a wall-clock comparison between NIC bursts (see `run_dpdk_poll`). The
    // input ring is therefore single-producer in steady state alongside the
    // one-shot seed loop — same property as the io_uring transport.
    let tick_cadence = config.tick_interval();

    // Control channel: DPDK poll thread → response stage (connect/disconnect).
    let (control_tx, control_rx) = std::sync::mpsc::channel();

    // Client writes a halted node refuses at ingress, poll thread →
    // response stage, counted for the health endpoint. See `crate::halt`.
    let refused_writes = Arc::new(AtomicU64::new(0));
    let halt_gate = crate::halt::HaltGate::new(
        replicas_connected.clone(),
        Arc::clone(&fence_state),
        Arc::clone(&refused_writes),
    );
    let (refusal_tx, refusal_rx) =
        crate::halt::refusal_channel::<A::Report>(Arc::clone(&matching_cursor));

    // TX SPSC: response stage → DPDK poll thread (encoded frames).
    // Lock-free, fixed-size slots — no heap allocation per frame.
    // 4096 slots × ~140 bytes = ~560 KiB. Enough to buffer a burst
    // without backpressuring the response stage.
    // One SPSC channel per DPDK poll thread. The response stage routes
    // frames to the correct thread based on thread_id encoded in
    // connection_id bits 56..63.
    let mut tx_producers = Vec::with_capacity(num_dpdk_threads);
    let mut tx_consumers = Vec::with_capacity(num_dpdk_threads);
    // The response thread produces into these, so they wait its way.
    let wait = config.cores.response.wait;
    for _ in 0..num_dpdk_threads {
        let (tx_out, tx_rx) =
            melin_pipeline::spsc::channel::<crate::dpdk_response::TxFrame>(4096, wait);
        tx_producers.push(tx_out);
        tx_consumers.push(tx_rx);
    }

    // Spawn pipeline threads (journal, matching — identical to TCP path).
    let cores = config.cores;

    // Wire runtime rotation into the journal stage (DPDK primary path).
    // PROMOTE is rejected on a primary (no flag wired); ROTATE shares the
    // same flag the journal stage observes.
    let mut journal_stage = journal_stage;
    let rotate_flag = config.admin_bind.map(|_| Arc::new(AtomicBool::new(false)));
    let _admin_handle = config
        .admin_bind
        .map(|addr| {
            crate::admin::spawn(
                addr,
                None,
                rotate_flag.clone(),
                Some(Arc::clone(&ack_policy_atomic)),
                Arc::clone(&shutdown),
                Arc::clone(&authorized_keys),
            )
        })
        .transpose()?;
    let max_journal_bytes = config.max_journal_mib.saturating_mul(1024 * 1024);
    journal_stage.set_rotation(max_journal_bytes, rotate_flag.clone());
    config.cores.place_journal_children(&mut journal_stage);
    journal_stage.set_staging_mode(config.journal_staging_mode.into());
    // On a primary the journal stage owns the control-plane advertised
    // tip (durable cursor after each fsync batch). On the promotion path
    // this takes over the handle the receiver was advancing.
    journal_stage.set_advertised_tip_publisher(journal_tip.clone());
    if config.max_journal_mib > 0 {
        info!(
            max_journal_mib = config.max_journal_mib,
            "runtime journal rotation enabled (size threshold, DPDK)"
        );
    }

    // Extract utilization handles before stages are moved into threads.
    let journal_utilization = journal_stage.utilization();
    let matching_utilization = matching_stage.utilization();
    let response_utilization = Arc::new(melin_transport_core::pipeline::StageUtilization::new());

    let s1 = Arc::clone(&shutdown);
    // As on kernel TCP: start the journal's helper threads here, not on
    // journal-seq. See `JournalStage::start`.
    let sequencer = journal_stage
        .start()
        .map_err(|e| format!("start the journal stage: {e}"))?;
    let journal_handle = std::thread::Builder::new()
        .name("journal-seq".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("journal-seq", cores.journal_seq.core);
            sequencer.run(&s1)
        })
        .map_err(|e| format!("spawn journal-seq thread: {e}"))?;

    let s2 = Arc::clone(&shutdown);
    let matching_handle = std::thread::Builder::new()
        .name("matching".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("matching", cores.matching.core);
            matching_stage.run(&s2)
        })
        .map_err(|e| format!("spawn matching thread: {e}"))?;

    let replication_metrics = build_replication_metrics(
        replication_consumers.is_some(),
        &ack_policy_atomic,
        config.ack_policy,
    );

    let replica_active: Option<[Arc<AtomicBool>; 2]> =
        replication_ring_progress.as_ref().map(|rp| {
            [
                Arc::clone(&rp.active_flags[0]),
                Arc::clone(&rp.active_flags[1]),
            ]
        });

    // Spawn DPDK response stage (encodes to TX channel instead of kernel sockets).
    let output_consumer = output_consumers.remove(0);
    // Typed durable-cursor handle for the response thread's gate.
    let journal_persisted_wire_seq_response = cursors.durable_wire_seq();
    let replication_metrics_response = replication_metrics.as_ref().map(Arc::clone);
    let replica_active_response = replica_active.clone();
    let ack_policy_response = Arc::clone(&ack_policy_atomic);
    let active_connections_response = Arc::clone(&active_connections);
    let s3 = Arc::clone(&shutdown);
    let response_utilization_thread = Arc::clone(&response_utilization);
    let response_handle = std::thread::Builder::new()
        .name("response".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("response", cores.response.core);
            crate::dpdk_response::run::<A>(
                output_consumer,
                control_rx,
                journal_persisted_wire_seq_response,
                ack_policy_response,
                replication_metrics_response,
                replica_active_response,
                &s3,
                heartbeat_interval,
                active_connections_response,
                tx_producers,
                response_utilization_thread,
                wait,
                encoder,
                refusal_rx,
            );
        })
        .map_err(|e| format!("spawn response thread: {e}"))?;

    let shadow_handle = spawn_shadow_stage::<A>(
        shadow_consumer,
        shadow_app,
        chain_hash_lock,
        &config,
        &cores,
        &shutdown,
        fence_state.epoch(),
    )?;

    // Spawn DPDK replication sender if enabled. Uses its own DPDK queue pair
    // and smoltcp stack so the replication channel goes through kernel bypass.
    // `replication_metrics` was constructed above so the response gate can
    // read per-slot cursors; the sender thread shares the same instance.
    let replica_ready = Arc::new(AtomicBool::new(false));
    // Replication, if enabled: build a `DpdkReplicationDriver` for the
    // single client poll thread to drive. The driver's accept dispatch
    // hangs off the second listener we added on the client transport
    // earlier (port == repl_bind.port()).
    let (repl_driver, repl_listen_port) = if let Some((repl_consumer_1, repl_consumer_2)) =
        replication_consumers
    {
        let repl_bind = config
            .replication_bind
            .ok_or("replication_bind must be set when replication is enabled")?;
        let repl_port = repl_bind.port();

        let repl_slots = cursors.replica_slot_cursors();
        let ready_flag = Arc::clone(&replica_ready);
        let batch_size = config.replication_batch_size;
        let heartbeat_secs = config.replication_heartbeat_secs;
        let repl_metrics = replication_metrics
            .clone()
            .ok_or("replication_metrics must be Some when replication is enabled")?;
        let connected_counter = replicas_connected
            .clone()
            .ok_or("replicas_connected must be Some when replication is enabled")?;
        let dpdk_active_flags: [Arc<AtomicBool>; 2] = replication_ring_progress
            .as_ref()
            .map(|rp| {
                [
                    Arc::clone(&rp.active_flags[0]),
                    Arc::clone(&rp.active_flags[1]),
                ]
            })
            .unwrap_or_else(|| {
                [
                    Arc::new(AtomicBool::new(false)),
                    Arc::new(AtomicBool::new(false)),
                ]
            });
        let dpdk_evict_flags: [Arc<AtomicBool>; 2] = replication_ring_progress
            .as_ref()
            .map(|rp| {
                [
                    Arc::clone(&rp.evict_flags[0]),
                    Arc::clone(&rp.evict_flags[1]),
                ]
            })
            .unwrap_or_else(|| {
                [
                    Arc::new(AtomicBool::new(false)),
                    Arc::new(AtomicBool::new(false)),
                ]
            });
        let journal_path = config.journal.clone();

        // Add the replication listener to the client transport so the
        // poll thread accepts both client and replication connections
        // off the same queue.
        // Replication needs larger TX buffers than clients: journal batches
        // can be 100-200 KiB, far exceeding the 16 KiB client TX buffer.
        const REPL_TX_BUF: usize = 512 * 1024;
        const REPL_TX_QUEUE: usize = 512 * 1024;
        const REPL_RX_BUF: usize = 64 * 1024;
        transports[0]
            .add_listener_with_buffers(
                repl_port,
                REPL_RX_BUF,
                REPL_TX_BUF,
                REPL_TX_QUEUE,
                crate::replication::REPL_DISPATCH_BURST,
            )
            .map_err(|e| format!("add replication listener: {e}"))?;

        let driver = crate::replication::DpdkReplicationDriver::new(
            [repl_consumer_1, repl_consumer_2],
            repl_slots,
            journal_path,
            ready_flag,
            connected_counter,
            dpdk_evict_flags,
            dpdk_active_flags,
            repl_metrics,
            batch_size,
            heartbeat_secs,
            Arc::clone(&fence_state),
            Arc::clone(&ack_policy_atomic),
            Arc::clone(&authorized_keys),
        )?;
        // Legacy text match — `lan-bench-suite.sh` `wait_for_log` keys
        // off "DPDK replication sender started" to know the primary
        // is ready to accept replicas. Kept verbatim for backward
        // compatibility with existing bench tooling.
        info!(addr = %repl_bind, "DPDK replication sender started (single-queue, in client poll thread)");
        (Some(driver), repl_port)
    } else {
        if !config.standalone && config.replica_of.is_none() {
            info!("running in standalone mode (no replication)");
        }
        (None, 0)
    };
    // The DPDK replication-sender thread used to be spawned here; with
    // the single-queue / single-thread refactor the driver lives inside
    // the client poll thread instead. Nothing to join.
    let replication_handle: Option<std::thread::JoinHandle<()>> = None;

    // Seed through the pipeline if needed.
    //
    // The previous DPDK path waited here for at least one replica to
    // connect before seeding so the replica picked up the seed events
    // live (skipping a journal-catch-up step). That gate is incompatible
    // with the single-queue / single-thread design: the main thread IS
    // the poll thread that accepts replica connections, and main is
    // currently blocked here pre-seed → replicas have nowhere to land →
    // deadlock until shutdown.
    //
    // Drop the gate. Seeding proceeds unconditionally; replicas that
    // connect after seeding catch up via the journal protocol (a few
    // hundred batches at startup — measured in tens of milliseconds, not
    // a meaningful operational cost). The kernel-TCP path still has its
    // own gate via the spawned `repl-accept` thread which can accept
    // replicas independently of seeding; this only changes DPDK behavior.
    //
    // A DPDK primary is never a promoted one (promotion falls back to the
    // kernel-TCP primary), so there is no epoch bump to order against.
    let _ = (&enable_replication, &replica_ready); // suppress unused warnings on non-DPDK paths
    let genesis = if needs_seeding {
        startup.genesis
    } else {
        Vec::new()
    };
    journal_startup_events(
        genesis,
        startup.on_primary,
        &mut input_producer,
        &journal_cursor,
        &matching_cursor,
        &replication_ring_progress,
        &shutdown,
    );

    // Note: the DPDK poll threads are spawned BELOW, after this seed-drain
    // block. That ordering is what keeps the input ring single-producer
    // during seeding — the same property the TCP path achieves by deferring
    // its `spawn_reader` call to after seed-drain.

    let pipeline_healthy = Arc::new(AtomicBool::new(true));
    let health_handle = spawn_health_endpoint(
        &config,
        &active_connections,
        &events_processed,
        &refused_writes,
        &cursors,
        input_cursor,
        &pipeline_healthy,
        &replicas_connected,
        &fence_state,
        &replication_metrics,
        &replica_active,
        &replication_ring_progress,
        &journal_utilization,
        &matching_utilization,
        &response_utilization,
        &shutdown,
        &raft_status,
    )?;

    info!(
        ip = %dpdk_config.ip_addr,
        port = dpdk_config.listen_port,
        num_dpdk_threads,
        "DPDK transport listening"
    );

    let connection_timeout = config.connection_timeout();
    let max_conns = config.max_connections;
    let reader_core = config.cores.reader.core;

    // Exactly one client poll queue (LMAX: single reader → single matcher).
    // Additional DPDK queues, if any, are dedicated to the replication sender
    // on its own thread and don't touch the input ring.
    assert_eq!(
        transports.len(),
        1,
        "expected exactly one client DPDK transport"
    );
    let transport_0 = transports.pop().expect("one client transport");
    let tx_rx_0 = tx_consumers.remove(0);
    melin_app::affinity::pin_thread("dpdk-poll-0", reader_core);
    crate::dpdk_transport::run_dpdk_poll::<A>(
        transport_0,
        input_producer,
        decoder,
        halt_gate,
        refusal_tx,
        control_tx,
        tx_rx_0,
        &shutdown,
        authorized_keys,
        connection_timeout,
        tick_cadence,
        max_conns,
        Arc::clone(&active_connections),
        0,
        repl_driver,
        repl_listen_port,
    );

    // The client poll runs on the main thread, so no extra DPDK threads
    // to join here — replication sender (if enabled) is joined below.
    let dpdk_extras: Vec<(String, std::thread::Result<()>)> = Vec::new();

    // The raft guard drops — stopping the driver — after this returns.
    shutdown_pipeline_stages(
        PipelineHandles {
            journal: journal_handle,
            matching: matching_handle,
            response: response_handle,
            replication: replication_handle,
            event_publisher: None,
            shadow: shadow_handle,
            health: health_handle,
        },
        dpdk_extras,
        &pipeline_healthy,
        &shutdown,
    )
}

/// Journal a primary's [`StartupEvents`] — `genesis` (empty unless the
/// journal is new), then `on_primary` — and return once they are applied.
///
/// The events ride the input ring as the node's own (connection 0, key
/// hash 0: no client identity), so the normal pipeline
/// journals, replicates and applies them. The caller must still be the
/// ring's only producer, and must not serve a client until this returns:
/// a client request would otherwise run before the configuration in
/// `on_primary` is in force, and would compete with a large genesis for
/// pipeline time.
///
/// Gates on the journal + matching cursors (disruptor sequence space),
/// then on every active replication ring being consumed — the sender
/// threads have read every batch (sent or being sent to replicas).
/// Stronger than no gate, faster than waiting for replica acks, and
/// deadlock-free because the ring backpressures instead of dropping
/// batches. Every wait yields to the shutdown flag, and goes through the
/// orchestrator wait strategy, never a bare spin: the stage threads may
/// share this thread's core on a small box.
fn journal_startup_events<E: melin_app::AppEvent>(
    genesis: Vec<E>,
    on_primary: Vec<E>,
    input_producer: &mut melin_pipeline::ring::Producer<InputSlot<E>>,
    journal_cursor: &melin_pipeline::padding::Sequence,
    matching_cursor: &melin_pipeline::padding::Sequence,
    replication_ring_progress: &Option<melin_transport_core::pipeline::ReplicationRingProgress>,
    shutdown: &AtomicBool,
) {
    use melin_app::unix_epoch_nanos;
    use melin_journal::JournalEvent;
    use melin_transport_core::trace::mono_trace_ns;

    let (genesis_count, on_primary_count) = (genesis.len(), on_primary.len());
    // Nothing published: the cursors never advance past a target, so
    // waiting for one would hang the boot.
    if genesis_count + on_primary_count == 0 {
        return;
    }
    let start = std::time::Instant::now();

    // `sequence: 0` — the journal stage allocates sequences in disruptor
    // cursor order at encode time. The runtime wraps each application
    // event as `JournalEvent::App` and stamps transport-level metadata.
    let mut last_published_seq = 0u64;
    for event in genesis.into_iter().chain(on_primary) {
        last_published_seq = input_producer.publish(InputSlot {
            connection_id: 0,
            key_hash: 0,
            sequence: 0,
            timestamp_ns: unix_epoch_nanos(),
            event: JournalEvent::App(event),
            publish_ts: mono_trace_ns(),
            recv_ts: mono_trace_ns(),
        });
    }
    let publish_elapsed = start.elapsed();

    let target = last_published_seq + 1; // cursor = next-to-consume
    info!(
        target,
        journal = journal_cursor.get().load(Ordering::Relaxed),
        matching = matching_cursor.get().load(Ordering::Relaxed),
        "startup events: waiting for pipeline cursors"
    );
    ORCHESTRATOR_WAIT.wait_until(|| {
        shutdown.load(Ordering::Relaxed)
            || (journal_cursor.get().load(Ordering::Acquire) >= target
                && matching_cursor.get().load(Ordering::Acquire) >= target)
    });

    // Inactive rings (no connected replica) were never published to, so
    // their producer cursor is 0 — no wait needed.
    if let Some(ring_progress) = replication_ring_progress {
        for i in 0..ring_progress.producer_cursors.len() {
            if !ring_progress.active_flags[i].load(Ordering::Relaxed) {
                continue;
            }
            let ring_target = ring_progress.producer_cursors[i].load();
            ORCHESTRATOR_WAIT.wait_until(|| {
                shutdown.load(Ordering::Relaxed)
                    || ring_progress.consumer_cursors[i]
                        .get()
                        .load(Ordering::Acquire)
                        >= ring_target
            });
        }
    }

    info!(
        genesis_events = genesis_count,
        on_primary_events = on_primary_count,
        publish_ms = publish_elapsed.as_millis(),
        total_ms = start.elapsed().as_millis(),
        "journaled startup events through pipeline"
    );
}

/// Bootstrap source chosen by [`init_engine`] from the on-disk layout.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BootstrapSource {
    /// Snapshot plus at least one journal segment (live or archived):
    /// restore the snapshot, replay the post-snapshot delta.
    SnapshotAndJournal,
    /// Snapshot with no journal segment at all: restore the snapshot and
    /// open a fresh segment continuing its sequence + chain.
    SnapshotOnly,
    /// Journal segments but no snapshot: full replay from genesis.
    JournalOnly,
    /// Nothing on disk: first-ever startup.
    Fresh,
}

/// Pure bootstrap decision, extracted so the matrix is unit-testable.
///
/// The critical cell: `(snapshot, no live, archives)` must route through
/// snapshot+journal recovery — the archives may hold acked events past
/// the snapshot (post-rotation crash window), and only the recovery walk
/// replays them, verifies the lineage, and synthesizes the missing live
/// segment. Likewise `(no snapshot, no live, archives)` is journal
/// recovery, not a fresh start.
fn choose_bootstrap(
    snapshot_exists: bool,
    live_exists: bool,
    archives_exist: bool,
) -> BootstrapSource {
    let lineage_exists = live_exists || archives_exist;
    match (snapshot_exists, lineage_exists) {
        (true, true) => BootstrapSource::SnapshotAndJournal,
        (true, false) => BootstrapSource::SnapshotOnly,
        (false, true) => BootstrapSource::JournalOnly,
        (false, false) => BootstrapSource::Fresh,
    }
}

/// Initialize or recover the journaled application from disk.
///
/// Returns `(app, writer, needs_seeding, recovered_epoch)`. `needs_seeding`
/// is true on first startup (fresh journal) — the caller must publish seed
/// events through the pipeline. `recovered_epoch` is the fencing epoch
/// recovered from the snapshot + journal (0 for a genesis node); the caller
/// seeds the node's `FenceState` with it. The recovery paths
/// (snapshot+journal, snapshot only, journal only, fresh) are
/// transport-level concerns and work uniformly for any `A: Application`
/// via `JournaledApp<A>`. Same engine initialization the TCP / DPDK paths use.
///
/// `sizing` is applied to a genesis instance before a journal is replayed
/// into it, so the history lands in reserved collections; the caller
/// sizes the result again afterwards, which covers the snapshot paths.
pub(crate) fn init_engine<A, W>(
    config: &ServerConfig,
    sizing: &A::Sizing,
) -> Result<(A, W, bool, u64), Box<dyn std::error::Error>>
where
    A: Application,
    W: JournalWrite<A::Event>,
{
    // Check for a snapshot: either the explicit --snapshot path, or the
    // default derived path (used by auto-rotation when --snapshot is not set).
    let derived_snap = config.journal.with_extension("snapshot");
    let snap_path = config.snapshot.as_deref().or_else(|| {
        if derived_snap.exists() {
            Some(derived_snap.as_path())
        } else {
            None
        }
    });

    // The journal *lineage* exists if either the live segment or any
    // archived segment is on disk. Deciding on the live file alone is
    // how acked events used to vanish: a crash between `archive_live`
    // and `create_continuing` leaves archives (holding events past the
    // snapshot) with no live file — bootstrapping "snapshot only" from
    // that layout would silently rewind every event in the archives.
    // `JournaledApp::recover*` handles the missing-live case itself
    // (replays archives, synthesizes a fresh live continuing the chain).
    let journal_exists = config.journal.exists();
    let archives_exist = !melin_journal::segment::list_archives(&config.journal)?.is_empty();
    let snapshot_exists = snap_path.is_some_and(|p| p.exists());
    let mut engine: JournaledApp<A, W> =
        match choose_bootstrap(snapshot_exists, journal_exists, archives_exist) {
            BootstrapSource::SnapshotAndJournal => {
                let snap_path = snap_path.expect("snapshot_exists implies snap_path");
                info!(snapshot = %snap_path.display(), "recovering from snapshot + journal");
                JournaledApp::<A, W>::recover_from_snapshot(snap_path, &config.journal)?
            }
            BootstrapSource::SnapshotOnly => {
                // Snapshot exists but no journal segment survives at all —
                // recover from the snapshot alone and start a fresh segment
                // continuing its sequence and chain.
                let snap_path = snap_path.expect("snapshot_exists implies snap_path");
                info!(
                    snapshot = %snap_path.display(),
                    "recovering from snapshot only (no journal segments on disk)"
                );
                let (app, snap_sequence, snap_chain_hash, snap_epoch) =
                    melin_transport_core::snapshot::load::<A>(snap_path)?;
                let writer =
                    W::create_continuing(&config.journal, snap_sequence + 1, snap_chain_hash)?;
                JournaledApp::<A, W>::from_parts(app, writer, snap_epoch)
            }
            BootstrapSource::JournalOnly => {
                info!("recovering from journal");
                // Sized before the history is applied to it, as a replica
                // is before it applies the stream: replay must not grow
                // the collections the sizing would have reserved.
                let mut app = A::default();
                <A as Application>::prefault(&mut app, sizing);
                JournaledApp::<A, W>::recover(app, &config.journal)?
            }
            BootstrapSource::Fresh => {
                info!("creating new journal");
                JournaledApp::<A, W>::create(A::default(), &config.journal)?
            }
        };

    // Genesis only on a genuinely fresh start — any surviving lineage
    // (live or archived) already contains the genesis events.
    let needs_seeding = !journal_exists && !archives_exist;

    // Archive the live journal segment if it exceeds the configured
    // size threshold. The shadow stage owns snapshot writes; here we
    // only rotate the segment so disk usage stays bounded across
    // restarts. Recovery walks the archive chain forward from the
    // latest shadow snapshot.
    if config.max_journal_mib > 0 {
        let threshold = config.max_journal_mib * 1024 * 1024;
        let current_size = engine.journal_size();
        if current_size > threshold {
            info!(
                current_mib = current_size / (1024 * 1024),
                threshold_mib = config.max_journal_mib,
                "journal exceeds threshold, rotating segment"
            );
            engine.rotate_segment()?;
            info!("journal segment rotated successfully");
        }
    }

    let recovered_epoch = engine.recovered_epoch();
    let (app, writer) = engine.into_parts();
    Ok((app, writer, needs_seeding, recovered_epoch))
}

// ---------------------------------------------------------------------------
// Shared helpers — used identically by both TCP and DPDK boot paths.
// ---------------------------------------------------------------------------

fn build_replication_metrics(
    has_replication: bool,
    ack_policy_atomic: &AtomicU8,
    config_policy: crate::ack_policy::AckPolicy,
) -> Option<Arc<crate::replication::ReplicationMetrics>> {
    let metrics = if has_replication {
        Some(Arc::new(crate::replication::ReplicationMetrics::default()))
    } else {
        None
    };

    let active_policy =
        crate::ack_policy::AckPolicy::from_u8(ack_policy_atomic.load(Ordering::Relaxed))
            .unwrap_or(config_policy);
    info!(
        ack_policy = %active_policy,
        clauses = %active_policy.to_policy(),
        "ack policy active"
    );

    metrics
}

fn spawn_shadow_stage<A: Application + Send + 'static>(
    shadow_consumer: Option<Consumer<InputSlot<A::Event>>>,
    shadow_app: Option<A>,
    chain_hash_lock: Option<
        melin_pipeline::seqlock::SeqLockReader<melin_transport_core::pipeline::FsyncState>,
    >,
    config: &ServerConfig,
    cores: &PipelineCores,
    shutdown: &Arc<AtomicBool>,
    initial_epoch: u64,
) -> Result<Option<std::thread::JoinHandle<()>>, Box<dyn std::error::Error>>
where
    A::Event: Send + Sync + 'static,
{
    let Some(shadow_cons) = shadow_consumer else {
        return Ok(None);
    };
    let snap_path = config.shadow_snapshot_path();
    let interval = std::time::Duration::from_millis(config.snapshot_interval_ms);
    let chain_hash =
        chain_hash_lock.ok_or("chain hash lock must be Some when shadow is enabled")?;
    let shadow_ex = shadow_app.ok_or("shadow application must be Some when shadow is enabled")?;
    let s_shadow = Arc::clone(shutdown);
    let shadow = cores.shadow;
    let shadow_initial_epoch = initial_epoch;
    let handle = std::thread::Builder::new()
        .name("shadow".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("shadow", shadow.core);
            melin_transport_core::shadow::run(
                shadow_cons,
                shadow_ex,
                snap_path,
                interval,
                chain_hash,
                &s_shadow,
                shadow.wait,
                shadow_initial_epoch,
            );
        })
        .map_err(|e| format!("spawn shadow thread: {e}"))?;

    info!(
        interval_ms = config.snapshot_interval_ms,
        path = %config.shadow_snapshot_path().display(),
        "shadow snapshot stage started"
    );
    Ok(Some(handle))
}

#[allow(clippy::too_many_arguments)]
fn spawn_health_endpoint(
    config: &ServerConfig,
    active_connections: &Arc<AtomicU64>,
    events_processed: &Arc<AtomicU64>,
    refused_writes: &Arc<AtomicU64>,
    cursors: &melin_transport_core::PipelineCursors,
    input_cursor: Box<dyn melin_pipeline::ring::QueueCursor>,
    pipeline_healthy: &Arc<AtomicBool>,
    replicas_connected: &Option<Arc<std::sync::atomic::AtomicU32>>,
    fence_state: &Arc<melin_transport_core::fence::FenceState>,
    replication_metrics: &Option<Arc<crate::replication::ReplicationMetrics>>,
    replica_active: &Option<[Arc<AtomicBool>; 2]>,
    replication_ring_progress: &Option<melin_transport_core::pipeline::ReplicationRingProgress>,
    journal_utilization: &Arc<melin_transport_core::pipeline::StageUtilization>,
    matching_utilization: &Arc<melin_transport_core::pipeline::StageUtilization>,
    response_utilization: &Arc<melin_transport_core::pipeline::StageUtilization>,
    shutdown: &Arc<AtomicBool>,
    raft_status: &Option<Arc<melin_transport_core::health::RaftStatus>>,
) -> Result<Option<std::thread::JoinHandle<()>>, Box<dyn std::error::Error>> {
    let Some(health_addr) = config.health_bind else {
        return Ok(None);
    };

    let (repl_ring_producers, repl_ring_consumers) = replication_ring_progress
        .as_ref()
        .map(|rp| {
            (
                Some([
                    Arc::clone(&rp.producer_cursors[0]),
                    Arc::clone(&rp.producer_cursors[1]),
                ]),
                Some([
                    Arc::clone(&rp.consumer_cursors[0]),
                    Arc::clone(&rp.consumer_cursors[1]),
                ]),
            )
        })
        .unwrap_or((None, None));
    Ok(Some(melin_transport_core::health::spawn(
        health_addr,
        melin_transport_core::health::HealthState {
            active_connections: Arc::clone(active_connections),
            events_processed: Arc::clone(events_processed),
            refused_writes: Arc::clone(refused_writes),
            cursors: cursors.clone(),
            input_cursor,
            pipeline_healthy: Arc::clone(pipeline_healthy),
            replicas_connected: replicas_connected.clone(),
            fence_state: Some(Arc::clone(fence_state)),
            replication_metrics: replication_metrics.clone(),
            replica_active: replica_active.clone(),
            replication_ring_producer_cursors: repl_ring_producers,
            replication_ring_consumer_cursors: repl_ring_consumers,
            journal_utilization: Arc::clone(journal_utilization),
            matching_utilization: Arc::clone(matching_utilization),
            response_utilization: Arc::clone(response_utilization),
            raft: raft_status.clone(),
        },
        Arc::clone(shutdown),
    )?))
}

/// Spawn the event-publisher thread on the affinity core configured by
/// `cores.event_publisher`, delegating the loop body to the
/// caller-supplied [`EventPublisherFn`]. Returns `Ok(None)` when either
/// the consumer slot wasn't wired (the pipeline build saw
/// `enable_event_publisher == false`) or the binary passed no
/// publisher fn. The runtime owns thread lifecycle and CPU pinning so
/// the caller's fn stays domain-only.
fn spawn_event_publisher<A: Application>(
    consumer: Option<Consumer<OutputSlot<A>>>,
    run_fn: Option<EventPublisherFn<A>>,
    config: &ServerConfig,
    cores: &PipelineCores,
    authorized_keys: &Arc<AuthorizedKeys>,
    shutdown: &Arc<AtomicBool>,
) -> Result<Option<std::thread::JoinHandle<()>>, Box<dyn std::error::Error>>
where
    A::Report: Send + 'static,
    A::QueryResponse: Send + 'static,
{
    let (Some(event_consumer), Some(run_fn)) = (consumer, run_fn) else {
        return Ok(None);
    };
    let event_bind = config
        .event_bind
        .ok_or("event_bind must be set when event publisher is enabled")?;
    let s_event = Arc::clone(shutdown);
    let event_keys = Arc::clone(authorized_keys);
    let publisher = cores.event_publisher;
    let event_handle = std::thread::Builder::new()
        .name("event-publisher".into())
        .spawn(move || {
            melin_app::affinity::pin_thread("event-publisher", publisher.core);
            run_fn(
                event_consumer,
                event_bind,
                event_keys,
                &s_event,
                publisher.wait,
            );
        })
        .map_err(|e| format!("spawn event publisher thread: {e}"))?;
    info!(addr = %event_bind, "event publisher started");
    Ok(Some(event_handle))
}

/// Perform challenge-response authentication on a new connection.
///
/// Runs on the accept thread (cold path, blocking). The caller must set
/// a read timeout on the stream before calling to prevent slow clients
/// from stalling the accept loop.
///
/// Uses raw `read_exact` instead of `BufReader` to avoid over-reading
/// bytes that belong to the first post-auth request.
///
/// Returns `(Permission, public_key_bytes)` on success.
fn authenticate_connection<R: std::io::Read, W: std::io::Write>(
    connection_id: ConnectionId,
    addr: SocketAddr,
    reader: &mut R,
    writer: &mut W,
    authorized_keys: &AuthorizedKeys,
) -> Result<(Permission, [u8; 32]), Box<dyn std::error::Error>> {
    use std::io;

    use melin_wire_protocol::control::TransportResponse;
    use melin_wire_protocol::control_codec;

    // Generate a 32-byte random nonce for this connection.
    // Explicit OsRng for cryptographic material (SEC-10).
    let mut nonce = [0u8; 32];
    getrandom::fill(&mut nonce).map_err(|e| io::Error::other(format!("getrandom failed: {e}")))?;

    let mut buf = [0u8; 128];
    let written =
        control_codec::encode_transport_response(&TransportResponse::Challenge { nonce }, &mut buf)
            .map_err(|e| io::Error::other(format!("encode Challenge: {e}")))?;
    writer.write_all(&buf[..written])?;
    writer.flush()?;

    // Read ChallengeResponse frame directly (no BufReader). Using raw
    // read_exact avoids BufReader over-reading bytes that belong to the
    // first post-auth request — those bytes would be lost when the
    // BufReader is dropped and the fd moves to the io_uring reader.
    let mut len_buf = [0u8; 4];
    reader
        .read_exact(&mut len_buf)
        .map_err(|e| io::Error::other(format!("read auth frame length: {e}")))?;
    let frame_len = u32::from_le_bytes(len_buf) as usize;
    // ChallengeResponse is 1 (tag) + 64 (signature) + 32 (public key) = 97 bytes.
    if frame_len > 256 {
        send_auth_failed(writer);
        return Err(io::Error::other(format!("auth frame too large: {frame_len}")).into());
    }
    let mut frame_buf = [0u8; 256];
    reader
        .read_exact(&mut frame_buf[..frame_len])
        .map_err(|e| io::Error::other(format!("read auth frame payload: {e}")))?;

    let cr = match control_codec::decode_challenge_response(&frame_buf[..frame_len]) {
        Ok(cr) => cr,
        Err(e) => {
            send_auth_failed(writer);
            return Err(io::Error::other(format!("decode ChallengeResponse: {e}")).into());
        }
    };

    let public_key_bytes = cr.public_key;
    let permission = crate::client_auth::verify_client(
        authorized_keys,
        &nonce,
        &public_key_bytes,
        &cr.signature,
    )
    .inspect_err(|_| send_auth_failed(writer))?;

    // Auth succeeded — send ServerReady.
    let written =
        control_codec::encode_transport_response(&TransportResponse::ServerReady, &mut buf)
            .map_err(|e| io::Error::other(format!("encode ServerReady: {e}")))?;
    writer.write_all(&buf[..written])?;
    writer.flush()?;

    debug!(
        connection_id = connection_id.0,
        addr = %addr,
        permission = ?permission,
        "authenticated"
    );

    Ok((permission, public_key_bytes))
}

/// Set a read timeout on a raw fd via `setsockopt(SO_RCVTIMEO)`.
///
/// Works for both TCP and UDS since both are sockets. Uses `AsRawFd`
/// to avoid requiring a concrete stream type.
fn set_read_timeout<F: std::os::unix::io::AsRawFd>(
    fd: &F,
    timeout: Option<std::time::Duration>,
) -> std::io::Result<()> {
    let tv = match timeout {
        Some(d) => libc::timeval {
            tv_sec: d.as_secs() as libc::time_t,
            tv_usec: d.subsec_micros() as libc::suseconds_t,
        },
        None => libc::timeval {
            tv_sec: 0,
            tv_usec: 0,
        },
    };
    let ret = unsafe {
        libc::setsockopt(
            fd.as_raw_fd(),
            libc::SOL_SOCKET,
            libc::SO_RCVTIMEO,
            &tv as *const libc::timeval as *const libc::c_void,
            std::mem::size_of::<libc::timeval>() as libc::socklen_t,
        )
    };
    if ret < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

/// Enable `SO_BUSY_POLL` on a TCP data socket so the kernel busy-polls
/// the NIC for incoming data instead of going to sleep on the softirq
/// → wakeup handoff. Removes scheduler-wakeup latency from the recv
/// path on hardware where IRQ delivery is the dominant per-packet cost
/// (e.g. ixgbe-class NICs). 50 µs covers a typical LAN ack RTT.
///
/// Best-effort: requires `CAP_NET_ADMIN` (or unprivileged operation
/// permitted via sysctl). Failures are surfaced as warnings by the
/// caller — they only cost latency, not correctness, and we don't want
/// a misconfigured kernel to halt connection acceptance.
///
/// Only beneficial when the receiving thread is already busy-spinning
/// (otherwise the spin cost is wasted on idle connections). All
/// callers in Melin meet that condition: client reader threads spin on
/// io_uring CQEs, the replication sender's ack-recv thread spins, and
/// the bench client thread spins.
pub(crate) fn set_busy_poll<F: std::os::unix::io::AsRawFd>(
    fd: &F,
    micros: i32,
) -> std::io::Result<()> {
    // SAFETY: fd is a live socket fd owned by the caller for the
    // duration of the call; the option pointer is to a stack-local i32
    // with the right size for SO_BUSY_POLL.
    let val: libc::c_int = micros;
    let ret = unsafe {
        libc::setsockopt(
            fd.as_raw_fd(),
            libc::SOL_SOCKET,
            libc::SO_BUSY_POLL,
            &val as *const libc::c_int as *const libc::c_void,
            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
        )
    };
    if ret < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

/// Default `SO_BUSY_POLL` window in microseconds. Matches the value
/// already used on the replica receive socket; chosen to cover a
/// typical LAN round-trip without burning excessive CPU on quiet
/// connections.
pub(crate) const BUSY_POLL_US: i32 = 50;

/// Set `SO_SNDTIMEO` on a socket. Prevents blocking writes from stalling
/// the response thread when a client stops reading (SEC-01).
fn set_write_timeout<F: std::os::unix::io::AsRawFd>(
    fd: &F,
    timeout: Option<std::time::Duration>,
) -> std::io::Result<()> {
    let tv = match timeout {
        Some(d) => libc::timeval {
            tv_sec: d.as_secs() as libc::time_t,
            tv_usec: d.subsec_micros() as libc::suseconds_t,
        },
        None => libc::timeval {
            tv_sec: 0,
            tv_usec: 0,
        },
    };
    let ret = unsafe {
        libc::setsockopt(
            fd.as_raw_fd(),
            libc::SOL_SOCKET,
            libc::SO_SNDTIMEO,
            &tv as *const libc::timeval as *const libc::c_void,
            std::mem::size_of::<libc::timeval>() as libc::socklen_t,
        )
    };
    if ret < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

/// Best-effort send of AuthFailed before dropping a connection.
fn send_auth_failed(writer: &mut impl std::io::Write) {
    use melin_wire_protocol::control::TransportResponse;
    use melin_wire_protocol::control_codec;

    let mut buf = [0u8; 8];
    if let Ok(written) =
        control_codec::encode_transport_response(&TransportResponse::AuthFailed, &mut buf)
    {
        let _ = writer.write_all(&buf[..written]);
        let _ = writer.flush();
    }
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};
    use std::os::unix::net::UnixStream;

    use ed25519_dalek::{Signer, SigningKey};
    use melin_app::auth::AuthorizedKeys;
    use melin_app::auth::Permission;
    use melin_wire_protocol::control::ConnectionId;
    use melin_wire_protocol::control_codec::{
        TAG_AUTH_FAILED, TAG_CHALLENGE, TAG_CHALLENGE_RESPONSE, TAG_RESPONSE_HEARTBEAT,
        TAG_SERVER_READY,
    };

    use super::authenticate_connection;
    use super::{BootstrapSource, choose_bootstrap};

    /// `--dpdk-eal-args` takes its value only in the joined form. The
    /// value itself starts with a dash, and the space form let a
    /// forgotten value silently swallow the next flag as the EAL
    /// string; both mistakes must be startup errors instead.
    #[test]
    fn dpdk_eal_args_requires_the_joined_form() {
        use clap::Parser;
        let cfg = super::ServerConfig::try_parse_from([
            "melin-server",
            "--dpdk-eal-args=-l 0-7 --huge-dir /dev/hugepages",
        ])
        .expect("joined form must parse");
        assert_eq!(cfg.dpdk_eal_args, "-l 0-7 --huge-dir /dev/hugepages");

        let space_form =
            super::ServerConfig::try_parse_from(["melin-server", "--dpdk-eal-args", "-l 0-7"]);
        assert!(space_form.is_err(), "space-separated form must be rejected");

        let forgotten_value =
            super::ServerConfig::try_parse_from(["melin-server", "--dpdk-eal-args", "--no-mlock"]);
        assert!(
            forgotten_value.is_err(),
            "a forgotten value must error, not swallow the next flag"
        );
    }

    /// Full bootstrap decision matrix. The two archive-only cells are
    /// the regression guard: a post-rotation crash leaves archives with
    /// no live segment, and bootstrapping "snapshot only" (or "fresh")
    /// from that layout silently rewinds every acked event held in the
    /// archives. Recovery must own any layout where a lineage survives.
    #[test]
    fn bootstrap_decision_routes_archives_through_recovery() {
        use BootstrapSource::*;
        // (snapshot, live, archives) → source
        let matrix = [
            ((false, false, false), Fresh),
            ((false, false, true), JournalOnly), // post-rotation crash, no snapshot
            ((false, true, false), JournalOnly),
            ((false, true, true), JournalOnly),
            ((true, false, false), SnapshotOnly),
            ((true, false, true), SnapshotAndJournal), // post-rotation crash
            ((true, true, false), SnapshotAndJournal),
            ((true, true, true), SnapshotAndJournal),
        ];
        for ((snap, live, arch), expected) in matrix {
            assert_eq!(
                choose_bootstrap(snap, live, arch),
                expected,
                "snapshot={snap} live={live} archives={arch}"
            );
        }
    }

    /// Deterministic test key.
    fn test_key() -> SigningKey {
        SigningKey::from_bytes(&[0xAA; 32])
    }

    /// Build an `AuthorizedKeys` containing the test key with the given permission.
    fn keys_with_test_key(perm: &str) -> AuthorizedKeys {
        // Encode the public key bytes as base64 with the local helper so
        // the test has no external codec dependency (all test keys produce
        // valid base64), then feed it through AuthorizedKeys::parse.
        let pub_bytes = test_key().verifying_key().to_bytes();
        let pub_b64 = base64_encode(&pub_bytes);
        AuthorizedKeys::parse(&format!("{perm} {pub_b64} test\n")).unwrap()
    }

    /// Minimal base64 encoder for test use only. Avoids adding base64
    /// as a dev-dependency to the server crate.
    fn base64_encode(bytes: &[u8]) -> String {
        const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
        let mut out = String::new();
        for chunk in bytes.chunks(3) {
            let b0 = chunk[0] as u32;
            let b1 = if chunk.len() > 1 { chunk[1] as u32 } else { 0 };
            let b2 = if chunk.len() > 2 { chunk[2] as u32 } else { 0 };
            let n = (b0 << 16) | (b1 << 8) | b2;
            out.push(ALPHABET[(n >> 18 & 0x3F) as usize] as char);
            out.push(ALPHABET[(n >> 12 & 0x3F) as usize] as char);
            if chunk.len() > 1 {
                out.push(ALPHABET[(n >> 6 & 0x3F) as usize] as char);
            } else {
                out.push('=');
            }
            if chunk.len() > 2 {
                out.push(ALPHABET[(n & 0x3F) as usize] as char);
            } else {
                out.push('=');
            }
        }
        out
    }

    /// Run `authenticate_connection` on one end of a `UnixStream::pair()`,
    /// returning the result. Maps the error to `String` so it's `Send`.
    fn run_server_auth(
        mut stream: UnixStream,
        keys: AuthorizedKeys,
    ) -> std::thread::JoinHandle<Result<Permission, String>> {
        std::thread::spawn(move || {
            // Clone the stream so we have independent read/write halves.
            let mut writer = stream.try_clone().unwrap();
            authenticate_connection(
                ConnectionId(1),
                "127.0.0.1:0".parse().unwrap(),
                &mut stream,
                &mut writer,
                &keys,
            )
            .map(|(perm, _pk)| perm)
            .map_err(|e| e.to_string())
        })
    }

    /// Read a length-prefixed Challenge frame and return its 32-byte
    /// nonce. Frame layout: `[len:u32][TAG_CHALLENGE][nonce:32]`.
    fn read_challenge(stream: &mut UnixStream) -> [u8; 32] {
        let mut len_buf = [0u8; 4];
        stream.read_exact(&mut len_buf).unwrap();
        let len = u32::from_le_bytes(len_buf) as usize;
        let mut payload = vec![0u8; len];
        stream.read_exact(&mut payload).unwrap();
        assert_eq!(payload[0], TAG_CHALLENGE, "expected Challenge");
        payload[1..33].try_into().unwrap()
    }

    /// Write a length-prefixed ChallengeResponse frame, matching the
    /// layout the runtime's `control_codec::decode_challenge_response`
    /// expects: `[len:u32][TAG_CHALLENGE_RESPONSE][sig:64][pubkey:32]`.
    fn write_challenge_response(
        stream: &mut UnixStream,
        signature: [u8; 64],
        public_key: [u8; 32],
    ) {
        let mut frame = Vec::with_capacity(97);
        frame.push(TAG_CHALLENGE_RESPONSE);
        frame.extend_from_slice(&signature);
        frame.extend_from_slice(&public_key);
        stream
            .write_all(&(frame.len() as u32).to_le_bytes())
            .unwrap();
        stream.write_all(&frame).unwrap();
        stream.flush().unwrap();
    }

    /// Read a Challenge frame from the client end, sign the nonce, and
    /// write a valid ChallengeResponse back.
    fn client_sign_challenge(stream: &mut UnixStream, key: &SigningKey) {
        let nonce = read_challenge(stream);
        let sig = key.sign(&nonce);
        write_challenge_response(stream, sig.to_bytes(), key.verifying_key().to_bytes());
    }

    /// Like `client_sign_challenge` but corrupts the signature.
    fn client_sign_challenge_bad(stream: &mut UnixStream, key: &SigningKey) {
        let nonce = read_challenge(stream);
        let mut sig_bytes = key.sign(&nonce).to_bytes();
        sig_bytes[0] ^= 0xFF;
        write_challenge_response(stream, sig_bytes, key.verifying_key().to_bytes());
    }

    /// Read one length-prefixed control frame and return its tag byte.
    fn read_response_tag(stream: &mut UnixStream) -> u8 {
        let mut len_buf = [0u8; 4];
        stream.read_exact(&mut len_buf).unwrap();
        let len = u32::from_le_bytes(len_buf) as usize;
        let mut buf = vec![0u8; len];
        stream.read_exact(&mut buf).unwrap();
        buf[0]
    }

    #[test]
    fn auth_success_returns_permission() {
        let keys = keys_with_test_key("trader");
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_SERVER_READY);

        let result = handle.join().unwrap();
        assert_eq!(result.unwrap(), Permission::Trader);
    }

    #[test]
    fn auth_admin_permission() {
        let keys = keys_with_test_key("operator");
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_SERVER_READY);

        assert_eq!(handle.join().unwrap().unwrap(), Permission::Operator);
    }

    #[test]
    fn auth_unknown_key_sends_auth_failed() {
        let keys = AuthorizedKeys::parse("").unwrap();
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_bad_signature_sends_auth_failed() {
        let keys = keys_with_test_key("operator");
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge_bad(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_wrong_message_type_sends_auth_failed() {
        let keys = keys_with_test_key("trader");
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        // Read and discard the Challenge.
        read_challenge(&mut s2);

        // Send a frame carrying a transport-heartbeat tag where a
        // ChallengeResponse is expected. It is the right length (97) so
        // the auth decoder reaches the tag check and rejects the
        // unexpected tag, replying AuthFailed.
        let mut frame = vec![0u8; 97];
        frame[0] = TAG_RESPONSE_HEARTBEAT;
        s2.write_all(&(frame.len() as u32).to_le_bytes()).unwrap();
        s2.write_all(&frame).unwrap();
        s2.flush().unwrap();

        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_client_disconnects_is_error() {
        let keys = keys_with_test_key("trader");
        let (s1, s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        // Drop immediately — server fails reading the ChallengeResponse.
        drop(s2);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_different_key_than_authorized_is_rejected() {
        // Authorize the test key, but connect with a different key.
        let keys = keys_with_test_key("trader");
        let wrong_key = SigningKey::from_bytes(&[0xCC; 32]);
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &wrong_key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_oversized_frame_sends_auth_failed() {
        let keys = keys_with_test_key("trader");
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        // Read and discard Challenge.
        read_challenge(&mut s2);

        // Send a frame claiming to be 1000 bytes (way over the 256 limit).
        let fake_len: u32 = 1000;
        s2.write_all(&fake_len.to_le_bytes()).unwrap();
        s2.flush().unwrap();

        // Server should send AuthFailed before dropping.
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_zero_length_frame_sends_auth_failed() {
        let keys = keys_with_test_key("trader");
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        // Read and discard Challenge.
        read_challenge(&mut s2);

        // Send a zero-length frame — decode_request will fail on empty input.
        let zero_len: u32 = 0;
        s2.write_all(&zero_len.to_le_bytes()).unwrap();
        s2.flush().unwrap();

        // Server should send AuthFailed before dropping.
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        assert!(handle.join().unwrap().is_err());
    }

    #[test]
    fn auth_readonly_permission() {
        let keys = keys_with_test_key("readonly");
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_SERVER_READY);

        let perm = handle.join().unwrap().unwrap();
        assert_eq!(perm, Permission::ReadOnly);
        assert!(!perm.can_trade());
    }

    /// A replication key signs correctly and is in the keys file, but it
    /// authorizes node-to-node streaming only: the client listener refuses
    /// it at the handshake, before any request could reach the decoder.
    #[test]
    fn auth_replication_key_refused_on_client_listener() {
        let keys = keys_with_test_key("replication");
        let key = test_key();
        let (s1, mut s2) = UnixStream::pair().unwrap();

        let handle = run_server_auth(s1, keys);

        client_sign_challenge(&mut s2, &key);
        let resp = read_response_tag(&mut s2);
        assert_eq!(resp, TAG_AUTH_FAILED);

        let err = handle.join().unwrap().unwrap_err();
        assert!(err.contains("client listener"), "unexpected error: {err}");
    }
}

/// Startup-time validation of the DPDK CLI surface.
///
/// The value these pin is *when* a bad flag is caught. `--dpdk-peer-mac`
/// is consumed inside the replica's reconnect loop, so the failure has to
/// surface here — during config translation, before EAL init and before
/// any thread exists — rather than as a panic in a node that is already
/// serving.
#[cfg(all(test, feature = "dpdk"))]
mod dpdk_config_tests {
    use super::{ServerConfig, dpdk_config_from};

    #[test]
    fn a_supplied_peer_mac_reaches_the_transport_config() {
        let cfg = ServerConfig {
            dpdk_peer_mac: Some("0c:42:a1:5b:2e:80".into()),
            ..Default::default()
        };
        let dpdk = dpdk_config_from(&cfg).expect("valid config");
        assert_eq!(dpdk.peer_mac, Some([0x0c, 0x42, 0xa1, 0x5b, 0x2e, 0x80]));
    }

    #[test]
    fn an_absent_peer_mac_leaves_the_transport_to_derive_one() {
        let cfg = ServerConfig::default();
        let dpdk = dpdk_config_from(&cfg).expect("valid config");
        assert_eq!(dpdk.peer_mac, None);
    }

    #[test]
    fn a_malformed_peer_mac_fails_config_translation() {
        let cfg = ServerConfig {
            dpdk_peer_mac: Some("0c:42:a1:5b:2e".into()),
            ..Default::default()
        };
        let err = dpdk_config_from(&cfg).expect_err("a 5-octet MAC must be rejected");
        assert!(err.contains("--dpdk-peer-mac"), "{err}");
    }

    #[test]
    fn a_malformed_gateway_mac_fails_config_translation() {
        // The same guarantee for the pre-existing flag, which used to
        // panic out of this function.
        let cfg = ServerConfig {
            dpdk_gateway_mac: Some("not-a-mac".into()),
            ..Default::default()
        };
        let err = dpdk_config_from(&cfg).expect_err("a non-MAC must be rejected");
        assert!(err.contains("--dpdk-gateway-mac"), "{err}");
    }

    #[test]
    fn a_malformed_ip_names_the_flag_it_came_from() {
        let cfg = ServerConfig {
            dpdk_peer_ip: Some("10.0.0".into()),
            ..Default::default()
        };
        let err = dpdk_config_from(&cfg).expect_err("a truncated IPv4 must be rejected");
        assert!(err.contains("--dpdk-peer-ip"), "{err}");
    }

    #[test]
    fn a_valid_configuration_translates() {
        let cfg = ServerConfig {
            dpdk_gateway: Some("10.0.0.254".into()),
            dpdk_gateway_mac: Some("0c:42:a1:5b:2e:80".into()),
            ..Default::default()
        };
        let dpdk = dpdk_config_from(&cfg).expect("valid config");
        assert_eq!(dpdk.gateway_mac, Some([0x0c, 0x42, 0xa1, 0x5b, 0x2e, 0x80]));
    }
}