rns_net/interface/

backbone.rs

//! Backbone TCP mesh interface using cross-platform polling.
//!
//! Server mode: listens on a TCP port, accepts peer connections, spawns
//! dynamic per-peer interfaces. Uses a single poll thread to multiplex
//! all client sockets. HDLC framing for packet boundaries.
//!
//! Client mode: connects to a remote backbone server, single TCP connection
//! with HDLC framing. Reconnects on disconnect.
//!
//! Matches Python `BackboneInterface.py`.

use std::collections::HashMap;
use std::hash::{BuildHasher, Hasher};
use std::io;
#[cfg(test)]
use std::io::Write;
use std::net::{IpAddr, Shutdown, TcpListener, TcpStream, ToSocketAddrs};
use std::os::unix::io::AsRawFd;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::thread;
use std::time::{Duration, Instant};

use socket2::{SockRef, TcpKeepalive};

use rns_core::constants;
use rns_core::transport::types::{IngressControlConfig, InterfaceId, InterfaceInfo};

use crate::event::{Event, EventSender};
use crate::event::{LoopContext, LoopSource, SourceStatus};
use crate::hdlc;
use crate::interface::loop_writer::LoopEgress;
use crate::interface::tcp::{read_stream_frames, StreamRead};
use crate::interface::transmit_buffer::{
    EgressController, TransmitBuffer, EGRESS_DEAD_TIME, EGRESS_HIGH_WATERMARK,
};
use crate::interface::{
    auto_mtu_for_bitrate, lock_or_recover, wait_while_active, DynamicInterfaceTemplate,
    EgressControl, InterfaceConfigData, InterfaceFactory, ListenerControl, StartContext,
    StartResult, Writer,
};
use crate::BackbonePeerStateEntry;

/// Upstream listener bitrate estimate used for accepted peers.
pub(crate) const SERVER_BITRATE_GUESS: u64 = 100_000_000;
/// Standalone Backbone client bitrate estimate.
pub(crate) const CLIENT_BITRATE_GUESS: u64 = 100_000_000;

/// Effective frame size selected by upstream autoconfiguration at 100 Mbps.
pub(crate) const HW_MTU: usize = match auto_mtu_for_bitrate(CLIENT_BITRATE_GUESS) {
    Some(mtu) => mtu as usize,
    None => 1_048_576,
};
/// MTU selected for listener-accepted peers at the listener bitrate estimate.
pub(crate) const SERVER_PEER_MTU: u32 = match auto_mtu_for_bitrate(SERVER_BITRATE_GUESS) {
    Some(mtu) => mtu,
    None => 1_048_576,
};

const DP_IC_HIGH_WM_PCT: usize = 90;
const DP_IC_MID_WM_PCT: usize = 68;
const DP_IC_LOW_WM_PCT: usize = 10;
const DP_IC_INTERVAL: Duration = Duration::from_millis(250);
const DP_IC_RCVBUF: usize = 32_768;
const DP_IC_IF_HEADROOM: usize = 32;
const DP_IC_PENALTY: f64 = 1.5;

/// Configuration for a backbone interface.
#[derive(Debug, Clone)]
pub struct BackboneConfig {
    pub name: String,
    pub listen_ip: String,
    pub listen_port: u16,
    pub interface_id: InterfaceId,
    pub mode: u8,
    pub gravity: i64,
    pub recursive_prs: bool,
    pub announces_from_internal: bool,
    pub announces_to_internal: Option<bool>,
    pub max_connections: Option<usize>,
    pub idle_timeout: Option<Duration>,
    pub write_stall_timeout: Option<Duration>,
    pub abuse: BackboneAbuseConfig,
    pub fast_flap: BackboneFastFlapConfig,
    pub fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
    pub ingress_control: IngressControlConfig,
    pub runtime: Arc<Mutex<BackboneServerRuntime>>,
    pub peer_state: Arc<Mutex<BackbonePeerMonitor>>,
}

/// Configurable behavior-based abuse detection for inbound peers.
#[derive(Debug, Clone, Default)]
pub struct BackboneAbuseConfig {
    pub max_penalty_duration: Option<Duration>,
}

/// Upstream-compatible automatic blocking for repeatedly short-lived inbound
/// Backbone connections.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct BackboneFastFlapConfig {
    pub enabled: bool,
    pub connection_threshold: Duration,
    pub grace: u64,
    pub block_duration: Duration,
}

impl Default for BackboneFastFlapConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            connection_threshold: Duration::from_secs(20),
            grace: 5,
            block_duration: Duration::from_secs(12 * 60 * 60),
        }
    }
}

#[derive(Debug, Clone)]
struct FastFlapPeerState {
    first_flap_at: Instant,
    last_flap_at: Instant,
    flap_count: u64,
}

/// Result of recording one short-lived connection.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct FastFlapRecord {
    pub flap_count: u64,
    pub blocked: bool,
    pub newly_blocked: bool,
    pub blocked_until: Option<Instant>,
    pub elapsed_since_first: Duration,
}

/// Process-wide fast-flap history. Policy remains per listener, while the IP
/// history is shared like upstream's class-level `fast_flapping` table.
#[derive(Debug, Default)]
pub struct BackboneFastFlapMonitor {
    peers: HashMap<IpAddr, FastFlapPeerState>,
}

impl BackboneFastFlapMonitor {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn record_disconnect(
        &mut self,
        peer_ip: IpAddr,
        connected_for: Duration,
        config: &BackboneFastFlapConfig,
        now: Instant,
    ) -> Option<FastFlapRecord> {
        if !config.enabled || connected_for >= config.connection_threshold {
            return None;
        }

        self.expire_peer(peer_ip, config, now);
        let state = self.peers.entry(peer_ip).or_insert(FastFlapPeerState {
            first_flap_at: now,
            last_flap_at: now,
            flap_count: 0,
        });
        let was_blocked = state.flap_count > config.grace;
        state.last_flap_at = now;
        state.flap_count = state.flap_count.saturating_add(1);
        let blocked = state.flap_count > config.grace;

        Some(FastFlapRecord {
            flap_count: state.flap_count,
            blocked,
            newly_blocked: blocked && !was_blocked,
            blocked_until: blocked
                .then(|| now.checked_add(config.block_duration))
                .flatten(),
            elapsed_since_first: now.saturating_duration_since(state.first_flap_at),
        })
    }

    pub fn is_blocked(
        &mut self,
        peer_ip: IpAddr,
        config: &BackboneFastFlapConfig,
        now: Instant,
    ) -> bool {
        self.blocked_until(peer_ip, config, now).is_some()
    }

    pub fn blocked_until(
        &mut self,
        peer_ip: IpAddr,
        config: &BackboneFastFlapConfig,
        now: Instant,
    ) -> Option<Instant> {
        if !config.enabled {
            return None;
        }
        self.expire_peer(peer_ip, config, now);
        self.peers
            .get(&peer_ip)
            .filter(|state| state.flap_count > config.grace)
            .and_then(|state| state.last_flap_at.checked_add(config.block_duration))
    }

    pub fn blocked_ip_count(&mut self, config: &BackboneFastFlapConfig, now: Instant) -> usize {
        if !config.enabled {
            return 0;
        }
        self.expire_all(config, now);
        self.peers
            .values()
            .filter(|state| state.flap_count > config.grace)
            .count()
    }

    pub fn blocked_ip_list(
        &mut self,
        config: &BackboneFastFlapConfig,
        now: Instant,
    ) -> Vec<IpAddr> {
        if !config.enabled {
            return Vec::new();
        }
        self.expire_all(config, now);
        let mut blocked: Vec<_> = self
            .peers
            .iter()
            .filter(|(_, state)| state.flap_count > config.grace)
            .map(|(peer_ip, _)| *peer_ip)
            .collect();
        blocked.sort_unstable();
        blocked
    }

    pub fn flap_count(&self, peer_ip: IpAddr) -> u64 {
        self.peers
            .get(&peer_ip)
            .map(|state| state.flap_count)
            .unwrap_or(0)
    }

    pub fn clear(&mut self, peer_ip: IpAddr) -> bool {
        self.peers.remove(&peer_ip).is_some()
    }

    fn expire_peer(&mut self, peer_ip: IpAddr, config: &BackboneFastFlapConfig, now: Instant) {
        let expired = self.peers.get(&peer_ip).is_some_and(|state| {
            now.saturating_duration_since(state.last_flap_at) > config.block_duration
        });
        if expired {
            self.peers.remove(&peer_ip);
        }
    }

    fn expire_all(&mut self, config: &BackboneFastFlapConfig, now: Instant) {
        self.peers.retain(|_, state| {
            now.saturating_duration_since(state.last_flap_at) <= config.block_duration
        });
    }
}

fn shared_fast_flap_monitor() -> Arc<Mutex<BackboneFastFlapMonitor>> {
    static MONITOR: OnceLock<Arc<Mutex<BackboneFastFlapMonitor>>> = OnceLock::new();
    Arc::clone(MONITOR.get_or_init(|| Arc::new(Mutex::new(BackboneFastFlapMonitor::new()))))
}

/// Live runtime state for a backbone server interface.
#[derive(Debug, Clone)]
pub struct BackboneServerRuntime {
    pub max_connections: Option<usize>,
    pub idle_timeout: Option<Duration>,
    pub write_stall_timeout: Option<Duration>,
    pub abuse: BackboneAbuseConfig,
}

impl BackboneServerRuntime {
    pub fn from_config(config: &BackboneConfig) -> Self {
        Self {
            max_connections: config.max_connections,
            idle_timeout: config.idle_timeout,
            write_stall_timeout: config.write_stall_timeout,
            abuse: config.abuse.clone(),
        }
    }
}

#[derive(Debug, Clone)]
pub struct BackboneRuntimeConfigHandle {
    pub interface_name: String,
    pub runtime: Arc<Mutex<BackboneServerRuntime>>,
    pub startup: BackboneServerRuntime,
}

#[derive(Debug, Clone)]
pub struct BackbonePeerStateHandle {
    pub interface_id: InterfaceId,
    pub interface_name: String,
    pub mode: u8,
    pub gravity: i64,
    pub announces_to_internal: Option<bool>,
    pub peer_state: Arc<Mutex<BackbonePeerMonitor>>,
    pub fast_flap: BackboneFastFlapConfig,
    pub fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
}

impl Default for BackboneConfig {
    fn default() -> Self {
        let mut config = BackboneConfig {
            name: String::new(),
            listen_ip: "0.0.0.0".into(),
            listen_port: 0,
            interface_id: InterfaceId(0),
            mode: constants::MODE_FULL,
            gravity: 0,
            recursive_prs: false,
            announces_from_internal: true,
            announces_to_internal: None,
            max_connections: None,
            idle_timeout: None,
            write_stall_timeout: None,
            abuse: BackboneAbuseConfig::default(),
            fast_flap: BackboneFastFlapConfig::default(),
            fast_flap_state: shared_fast_flap_monitor(),
            ingress_control: IngressControlConfig::enabled(),
            runtime: Arc::new(Mutex::new(BackboneServerRuntime {
                max_connections: None,
                idle_timeout: None,
                write_stall_timeout: None,
                abuse: BackboneAbuseConfig::default(),
            })),
            peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
        };
        let startup = BackboneServerRuntime::from_config(&config);
        config.runtime = Arc::new(Mutex::new(startup));
        config
    }
}

/// Writer that sends HDLC-framed data over a cloned TCP stream (server mode).
///
/// The server poller owns only readable interest on the original stream. This
/// independent writer therefore never enables or re-arms writable poll events;
/// an already-pending buffer is retried directly before accepting another
/// frame.
struct BackboneWriter {
    stream: TcpStream,
    runtime: Arc<Mutex<BackboneServerRuntime>>,
    interface_name: String,
    interface_id: InterfaceId,
    event_tx: EventSender,
    transmit_buffer: TransmitBuffer,
    egress_controller: EgressController,
    egress_control: EgressControl,
    egress_started: Instant,
    stall_started: Option<Instant>,
    disconnect_notified: bool,
    write_stall_flag: Arc<AtomicBool>,
}

impl Writer for BackboneWriter {
    fn send_frame_confirmed(&mut self, data: &[u8]) -> io::Result<()> {
        // The async writer has already reserved this frame's byte budget.
        // Once admitted, retain it even if the egress gate subsequently closes.
        if !self
            .transmit_buffer
            .append_with_limit(hdlc::frame(data), Some(EGRESS_HIGH_WATERMARK))
        {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "confirmed frame exceeds backbone buffer",
            ));
        }
        self.transmit_buffer.flush();
        let timeout = lock_or_recover(&self.runtime, "backbone runtime").write_stall_timeout;
        self.drain_transmit(timeout)
    }
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        self.send_frames(&[crate::interface::FrameBytes::Owned(data.to_vec())])
    }

    fn send_frames(&mut self, frames: &[crate::interface::FrameBytes]) -> io::Result<()> {
        let write_stall_timeout =
            lock_or_recover(&self.runtime, "backbone runtime").write_stall_timeout;
        if self.transmit_buffer.buffered_bytes() > 0 {
            self.drain_transmit(write_stall_timeout)?;
            if self.transmit_buffer.buffered_bytes() > 0 {
                return Err(io::Error::new(
                    io::ErrorKind::WouldBlock,
                    "backbone writer still stalled",
                ));
            }
        }

        let elapsed = self.egress_started.elapsed();
        let _ = self.egress_controller.evaluate(
            elapsed,
            self.transmit_buffer.buffered_bytes(),
            self.transmit_buffer.sendable_bytes(),
            self.transmit_buffer.total_sent_bytes(),
        );
        self.egress_control
            .set_stalled(self.egress_controller.stalled());
        for frame in frames {
            let encoded = hdlc::frame(frame);
            let encoded_len = encoded.len();
            if self.egress_control.stalled()
                || !self
                    .transmit_buffer
                    .append_with_limit(encoded, Some(EGRESS_HIGH_WATERMARK))
            {
                self.egress_control.record_drop(encoded_len);
                log::debug!(
                    "[{}:{}] egress control dropped outbound frame of {} bytes",
                    self.interface_name,
                    self.interface_id.0,
                    encoded_len
                );
            }
        }
        self.transmit_buffer.flush();
        self.drain_transmit(write_stall_timeout)
    }

    fn flush(&mut self) -> io::Result<()> {
        let timeout = lock_or_recover(&self.runtime, "backbone runtime").write_stall_timeout;
        self.drain_transmit(timeout)
    }

    fn egress_control(&self) -> Option<EgressControl> {
        Some(self.egress_control.clone())
    }

    fn loop_socket(&self) -> Option<std::os::fd::RawFd> {
        Some(self.stream.as_raw_fd())
    }

    fn loop_egress(&mut self) -> Option<LoopEgress> {
        let runtime = Arc::clone(&self.runtime);
        Some(
            LoopEgress::budget(self.egress_control.clone()).with_stall_watch(
                move || lock_or_recover(&runtime, "backbone runtime").write_stall_timeout,
                Arc::clone(&self.write_stall_flag),
            ),
        )
    }

    fn shutdown(&mut self) {
        let _ = self.stream.shutdown(Shutdown::Both);
    }
}

impl BackboneWriter {
    fn drain_transmit(&mut self, write_stall_timeout: Option<Duration>) -> io::Result<()> {
        if self.transmit_buffer.buffered_bytes() == 0 {
            return Ok(());
        }

        let written = self.transmit_buffer.drain_to(&mut self.stream)?;
        if written > 0 {
            self.stall_started = None;
        }
        let elapsed = self.egress_started.elapsed();
        let dead = self.egress_controller.evaluate(
            elapsed,
            self.transmit_buffer.buffered_bytes(),
            self.transmit_buffer.sendable_bytes(),
            self.transmit_buffer.total_sent_bytes(),
        );
        self.egress_control
            .set_stalled(self.egress_controller.stalled());
        if dead {
            return Err(self.disconnect_for_write_stall(EGRESS_DEAD_TIME));
        }
        if self.transmit_buffer.buffered_bytes() == 0 {
            return Ok(());
        }

        let now = Instant::now();
        let started = self.stall_started.get_or_insert(now);
        let timeout = write_stall_timeout.unwrap_or(EGRESS_DEAD_TIME);
        if now.duration_since(*started) >= timeout {
            return Err(self.disconnect_for_write_stall(timeout));
        }
        Err(io::Error::new(
            io::ErrorKind::WouldBlock,
            "backbone writer would block",
        ))
    }

    fn disconnect_for_write_stall(&mut self, timeout: Duration) -> io::Error {
        if !self.disconnect_notified {
            log::warn!(
                "[{}] backbone client {} disconnected due to write stall timeout ({:?})",
                self.interface_name,
                self.interface_id.0,
                timeout
            );
            self.write_stall_flag.store(true, Ordering::Relaxed);
            let _ = self.stream.shutdown(Shutdown::Both);
            let _ = self.event_tx.send(Event::InterfaceDown(self.interface_id));
            self.disconnect_notified = true;
        }
        io::Error::new(
            io::ErrorKind::TimedOut,
            format!("backbone writer stalled for {:?}", timeout),
        )
    }
}

/// Start a backbone interface. Binds the TCP listener, which is served
/// with its peers on the receiving (driver) thread.
pub fn start(config: BackboneConfig, tx: EventSender, next_id: Arc<AtomicU64>) -> io::Result<()> {
    start_with_template(config, tx, next_id, None, None)
}

fn start_with_template(
    config: BackboneConfig,
    tx: EventSender,
    next_id: Arc<AtomicU64>,
    dynamic_template: Option<DynamicInterfaceTemplate>,
    underlay_mark: Option<u32>,
) -> io::Result<()> {
    let addr = format!("{}:{}", config.listen_ip, config.listen_port);
    let listener = TcpListener::bind(&addr)?;
    super::apply_underlay_mark(listener.as_raw_fd(), underlay_mark)?;
    listener.set_nonblocking(true)?;

    log::info!(
        "[{}] backbone server listening on {}",
        config.name,
        listener
            .local_addr()
            .unwrap_or_else(|_| std::net::SocketAddr::from(([0, 0, 0, 0], config.listen_port)))
    );

    let ifac_size = dynamic_template
        .as_ref()
        .and_then(|template| template.ifac.as_ref())
        .map(|ifac| ifac.size)
        .unwrap_or(0);
    // Queue capacities are finalized before interfaces start. Derive the
    // ingress controller's watermarks once, before any listener can read.
    let ingress_thresholds = ingress_thresholds(tx.inbound_queue_snapshot().capacities[0]);
    let server = Arc::new(Server {
        name: config.name.clone(),
        server_interface_id: config.interface_id,
        tx: tx.clone(),
        next_id,
        runtime: Arc::clone(&config.runtime),
        peer_state: Arc::clone(&config.peer_state),
        fast_flap: config.fast_flap,
        fast_flap_state: Arc::clone(&config.fast_flap_state),
        ingress_control: config.ingress_control,
        accepted_peer_mode: config.mode,
        accepted_peer_gravity: config.gravity,
        accepted_peer_recursive_prs: config.recursive_prs,
        accepted_peer_announces_from_internal: config.announces_from_internal,
        accepted_peer_announces_to_internal: config.announces_to_internal,
        dynamic_template,
        ifac_size,
        underlay_mark,
        ingress_thresholds,
        state: Mutex::new(ServerState {
            clients: HashMap::new(),
            peers: HashMap::new(),
            ingress_snapshot_at: Instant::now(),
        }),
    });
    tx.register_source(
        crate::interface::loop_writer::read_key(config.interface_id),
        Box::new(ListenerSource {
            listener,
            server,
            next_interval: Instant::now() + DP_IC_INTERVAL,
        }),
    )
}

/// Connections accepted per readiness before other sources get a turn.
const ACCEPT_BURST: usize = 64;

/// A listener's configuration and the state its peers share. The listener
/// and all its peers are served on the same thread, so `state` is never
/// contended.
struct Server {
    name: String,
    server_interface_id: InterfaceId,
    tx: EventSender,
    next_id: Arc<AtomicU64>,
    runtime: Arc<Mutex<BackboneServerRuntime>>,
    peer_state: Arc<Mutex<BackbonePeerMonitor>>,
    fast_flap: BackboneFastFlapConfig,
    fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
    ingress_control: IngressControlConfig,
    accepted_peer_mode: u8,
    accepted_peer_gravity: i64,
    accepted_peer_recursive_prs: bool,
    accepted_peer_announces_from_internal: bool,
    accepted_peer_announces_to_internal: Option<bool>,
    dynamic_template: Option<DynamicInterfaceTemplate>,
    ifac_size: usize,
    underlay_mark: Option<u32>,
    ingress_thresholds: IngressThresholds,
    state: Mutex<ServerState>,
}

struct ServerState {
    /// Connected peers by poller key.
    clients: HashMap<usize, ClientState>,
    peers: HashMap<IpAddr, PeerBehaviorState>,
    ingress_snapshot_at: Instant,
}

/// Per-client tracking state.
struct ClientState {
    id: InterfaceId,
    peer_ip: IpAddr,
    peer_port: u16,
    /// The peer's socket, owned by its source, which removes this entry
    /// before the socket is closed.
    fd: std::os::fd::RawFd,
    connected_at: Instant,
    has_received_data: bool,
    write_stall_flag: Arc<AtomicBool>,
    /// The idle timeout shut the socket down; the peer source reports it
    /// once it reads the end of the stream.
    idle_timed_out: bool,
    dp_ingress_bytes: u64,
    dp_ingress_packets: u64,
    dp_ingress_hold: Option<Instant>,
    dp_ingress_gated: bool,
    dp_ingress_throttle_count: u64,
}

impl Server {
    fn lock(&self) -> std::sync::MutexGuard<'_, ServerState> {
        lock_or_recover(&self.state, "backbone server state")
    }

    /// Take blacklist changes made through management and expire what is
    /// due, then publish the merged view.
    fn sync_peer_state(&self, state: &mut ServerState) {
        let mut monitor = lock_or_recover(&self.peer_state, "backbone peer state");
        monitor.sync_into(&mut state.peers);
        cleanup_peer_state(&mut state.peers);
        monitor.upsert_snapshot(&state.peers);
    }

    /// Watch a peer for readability again, or stop watching it.
    fn set_readable(&self, key: usize, fd: std::os::fd::RawFd, readable: bool) {
        // SAFETY: `fd` belongs to a registered peer source (see `ClientState`).
        let source = unsafe { std::os::fd::BorrowedFd::borrow_raw(fd) };
        if let Err(error) = self.tx.set_source_readable(source, key, readable) {
            log::warn!(
                "[{}] cannot update peer read interest: {}",
                self.name,
                error
            );
        }
    }

    /// Stop reading the peer that produced the most packets this sample
    /// for a hold proportional to the sample span. Returns its key.
    fn throttle_busiest_ingress(&self, state: &mut ServerState) -> Option<usize> {
        let key = busiest_ingress_key(
            state
                .clients
                .iter()
                .map(|(key, client)| (*key, client.dp_ingress_packets)),
        )?;
        let allocation_divisor = state.clients.len().max(DP_IC_IF_HEADROOM) as f64 * DP_IC_PENALTY;
        let sample_span = state.ingress_snapshot_at.elapsed();
        let client = state.clients.get_mut(&key)?;
        if client.dp_ingress_gated {
            return None;
        }
        let hold = Duration::from_secs_f64(
            (sample_span.as_secs_f64().max(0.001) * allocation_divisor).max(0.001),
        );
        client.dp_ingress_gated = true;
        client.dp_ingress_hold = Some(Instant::now() + hold);
        client.dp_ingress_throttle_count = client.dp_ingress_throttle_count.saturating_add(1);
        log::info!(
            "ingress throttled on BackboneInterface/{} for {:.3}s",
            client.id.0,
            hold.as_secs_f64(),
        );
        Some(key)
    }

    /// Resume reading one gated peer whose hold has passed.
    fn release_one_ingress(&self, state: &mut ServerState, now: Instant) {
        let Some((key, client)) = state.clients.iter_mut().find(|(_, client)| {
            client.dp_ingress_gated
                && client
                    .dp_ingress_hold
                    .is_none_or(|hold_until| now >= hold_until)
        }) else {
            return;
        };
        client.dp_ingress_gated = false;
        client.dp_ingress_hold = None;
        log::info!(
            "released ingress throttle on BackboneInterface/{}",
            client.id.0
        );
        self.set_readable(*key, client.fd, true);
    }

    /// The listener's periodic work: ingress throttling and idle timeouts.
    fn interval(&self) {
        let mut state = self.lock();
        let state = &mut *state;
        self.sync_peer_state(state);
        let now = Instant::now();
        if now.duration_since(state.ingress_snapshot_at) >= DP_IC_INTERVAL {
            let height = self.tx.inbound_queue_snapshot().heights[0];
            if height > self.ingress_thresholds.mid {
                if let Some(key) = self.throttle_busiest_ingress(state) {
                    self.set_readable(key, state.clients[&key].fd, false);
                }
            } else if height < self.ingress_thresholds.low {
                self.release_one_ingress(state, now);
            }
            for client in state.clients.values_mut() {
                client.dp_ingress_bytes = 0;
                client.dp_ingress_packets = 0;
            }
            state.ingress_snapshot_at = now;
        }

        let idle_timeout = lock_or_recover(&self.runtime, "backbone runtime").idle_timeout;
        if let Some(timeout) = idle_timeout {
            for (key, client) in state.clients.iter_mut() {
                if client.has_received_data
                    || client.idle_timed_out
                    || now.duration_since(client.connected_at) < timeout
                {
                    continue;
                }
                // The peer source reports the disconnect when it reads the
                // end of the stream, gated or not.
                client.idle_timed_out = true;
                // SAFETY: the descriptor is open (see `ClientState`).
                unsafe { libc::shutdown(client.fd, libc::SHUT_RDWR) };
                self.set_readable(*key, client.fd, true);
            }
        }
    }

    /// Accept one connection; returns `false` once the driver is gone.
    fn admit(
        self: &Arc<Self>,
        state: &mut ServerState,
        stream: TcpStream,
        peer_addr: std::net::SocketAddr,
        cx: &mut LoopContext<'_>,
    ) -> bool {
        let name = &self.name;
        if let Err(error) = super::apply_underlay_mark(stream.as_raw_fd(), self.underlay_mark) {
            log::error!(
                "[{}] failed to mark accepted underlay socket: {}",
                name,
                error
            );
            return true;
        }
        let peer_ip = peer_addr.ip();
        let peer_port = peer_addr.port();

        let fast_flap_blocked_until = lock_or_recover(
            &self.fast_flap_state,
            "backbone fast-flap state",
        )
        .blocked_until(peer_ip, &self.fast_flap, Instant::now());
        if is_ip_blacklisted(&mut state.peers, peer_ip) || fast_flap_blocked_until.is_some() {
            let peer = state
                .peers
                .entry(peer_ip)
                .or_insert_with(PeerBehaviorState::new);
            if let Some(blocked_until) = fast_flap_blocked_until {
                peer.blacklisted_until = Some(blocked_until);
                peer.blacklist_reason = Some("fast flapping".into());
            }
            peer.reject_count = peer.reject_count.saturating_add(1);
            lock_or_recover(&self.peer_state, "backbone peer state").upsert_snapshot(&state.peers);
            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
                "[{}] rejecting blocked peer {}",
                name,
                peer_addr,
            );
            return true;
        }

        let max_connections = lock_or_recover(&self.runtime, "backbone runtime").max_connections;
        if let Some(max) = max_connections {
            if state.clients.len() >= max {
                log::warn!(
                    "[{}] max connections ({}) reached, rejecting {}",
                    name,
                    max,
                    peer_addr
                );
                return true;
            }
        }

        stream.set_nonblocking(true).ok();
        stream.set_nodelay(true).ok();
        set_tcp_keepalive(&stream).ok();
        SockRef::from(&stream)
            .set_recv_buffer_size(DP_IC_RCVBUF)
            .ok();

        // Prevent SIGPIPE on macOS when writing to broken pipes
        #[cfg(target_os = "macos")]
        {
            let sock = SockRef::from(&stream);
            sock.set_nosigpipe(true).ok();
        }

        // Create writer via try_clone (cross-platform dup)
        let writer_stream = match stream.try_clone() {
            Ok(s) => s,
            Err(e) => {
                log::warn!("[{}] failed to clone client stream: {}", name, e);
                return true;
            }
        };
        let client_id = InterfaceId(self.next_id.fetch_add(1, Ordering::Relaxed));
        let key = crate::interface::loop_writer::read_key(client_id);

        log::trace!(target: crate::logging::PATHING_LOG_TARGET,
            "[{}] backbone client connected: {} → id {}",
            name,
            peer_addr,
            client_id.0
        );

        let write_stall_flag = Arc::new(AtomicBool::new(false));
        let writer: Box<dyn Writer> = Box::new(BackboneWriter {
            stream: writer_stream,
            runtime: Arc::clone(&self.runtime),
            interface_name: name.clone(),
            interface_id: client_id,
            event_tx: self.tx.clone(),
            transmit_buffer: TransmitBuffer::new(),
            egress_controller: EgressController::default(),
            egress_control: EgressControl::new(EGRESS_HIGH_WATERMARK),
            egress_started: Instant::now(),
            stall_started: None,
            disconnect_notified: false,
            write_stall_flag: Arc::clone(&write_stall_flag),
        });

        state.clients.insert(
            key,
            ClientState {
                id: client_id,
                peer_ip,
                peer_port,
                fd: stream.as_raw_fd(),
                connected_at: Instant::now(),
                has_received_data: false,
                write_stall_flag,
                idle_timed_out: false,
                dp_ingress_bytes: 0,
                dp_ingress_packets: 0,
                dp_ingress_hold: None,
                dp_ingress_gated: false,
                dp_ingress_throttle_count: 0,
            },
        );
        state
            .peers
            .entry(peer_ip)
            .or_insert_with(PeerBehaviorState::new)
            .connected_count += 1;
        lock_or_recover(&self.peer_state, "backbone peer state").upsert_snapshot(&state.peers);

        let info = InterfaceInfo {
            id: client_id,
            name: format!("BackboneInterface/{}", client_id.0),
            mode: self.accepted_peer_mode,
            gravity: self.accepted_peer_gravity,
            recursive_prs: self.accepted_peer_recursive_prs,
            announces_from_internal: self.accepted_peer_announces_from_internal,
            announces_to_internal: self.accepted_peer_announces_to_internal,
            out_capable: true,
            in_capable: true,
            bitrate: Some(SERVER_BITRATE_GUESS),
            airtime_profile: None,
            announce_rate_target: None,
            announce_rate_grace: 0,
            announce_rate_penalty: 0.0,
            announce_cap: constants::ANNOUNCE_CAP,
            is_local_client: false,
            wants_tunnel: false,
            tunnel_id: None,
            mtu: SERVER_PEER_MTU,
            ia_freq: 0.0,
            ip_freq: 0.0,
            op_freq: 0.0,
            op_samples: 0,
            started: 0.0,
            ingress_control: self.ingress_control,
        };
        let up = if let Some(template) = &self.dynamic_template {
            Event::DynamicInterfaceUp {
                id: client_id,
                writer,
                registration: Box::new(template.registration(info)),
            }
        } else {
            Event::InterfaceUp(client_id, Some(writer), Some(Box::new(info)))
        };
        let connected = Event::BackbonePeerConnected {
            server_interface_id: self.server_interface_id,
            peer_interface_id: client_id,
            peer_ip,
            peer_port,
        };
        if self.tx.push_from_loop([connected, up]).is_err() {
            state.clients.remove(&key);
            return false;
        }
        cx.add_source(
            key,
            Box::new(PeerSource {
                stream,
                decoder: hdlc::Decoder::reticulum(HW_MTU, self.ifac_size),
                key,
                id: client_id,
                server: Arc::clone(self),
            }),
        );
        true
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct IngressThresholds {
    high: usize,
    immediate: usize,
    mid: usize,
    low: usize,
}

fn ingress_thresholds(data_capacity: usize) -> IngressThresholds {
    // Keep initialization silent; only actual throttle/release transitions are
    // operational events. Upstream removed its accidental critical HWM log.
    let high = ((DP_IC_HIGH_WM_PCT * data_capacity) / 100).max(4);
    IngressThresholds {
        high,
        immediate: high.max(128),
        mid: ((DP_IC_MID_WM_PCT * data_capacity) / 100).max(2),
        low: (DP_IC_LOW_WM_PCT * data_capacity) / 100,
    }
}

fn busiest_ingress_key(producers: impl Iterator<Item = (usize, u64)>) -> Option<usize> {
    producers
        .filter(|(_, packets)| *packets > 0)
        .max_by_key(|(_, packets)| *packets)
        .map(|(key, _)| key)
}

#[derive(Debug, Clone)]
struct PeerBehaviorState {
    blacklisted_until: Option<Instant>,
    blacklist_reason: Option<String>,
    reject_count: u64,
    connected_count: usize,
}

impl PeerBehaviorState {
    fn new() -> Self {
        Self {
            blacklisted_until: None,
            blacklist_reason: None,
            reject_count: 0,
            connected_count: 0,
        }
    }
}

#[derive(Debug, Clone, Default)]
pub struct BackbonePeerMonitor {
    peers: HashMap<IpAddr, PeerBehaviorState>,
}

impl BackbonePeerMonitor {
    pub fn new() -> Self {
        Self {
            peers: HashMap::new(),
        }
    }

    fn upsert_snapshot(&mut self, peers: &HashMap<IpAddr, PeerBehaviorState>) {
        let mut merged = self.peers.clone();

        for (peer_ip, state) in peers {
            let entry = merged
                .entry(*peer_ip)
                .or_insert_with(PeerBehaviorState::new);
            entry.connected_count = state.connected_count;
            entry.reject_count = state.reject_count;
            entry.blacklisted_until = state.blacklisted_until;
            entry.blacklist_reason = state.blacklist_reason.clone();
        }

        merged.retain(|peer_ip, state| {
            peers.contains_key(peer_ip)
                || state.blacklisted_until.is_some()
                || state.reject_count > 0
        });
        self.peers = merged;
    }

    fn sync_into(&self, peers: &mut HashMap<IpAddr, PeerBehaviorState>) {
        for (peer_ip, state) in &self.peers {
            let entry = peers.entry(*peer_ip).or_insert_with(PeerBehaviorState::new);
            entry.blacklisted_until = state.blacklisted_until;
            entry.blacklist_reason = state.blacklist_reason.clone();
            entry.reject_count = state.reject_count;
        }

        peers.retain(|peer_ip, state| {
            if state.connected_count > 0 {
                return true;
            }
            self.peers.contains_key(peer_ip)
        });
    }

    pub fn list(&self, interface_name: &str) -> Vec<BackbonePeerStateEntry> {
        let now = Instant::now();
        let mut entries: Vec<BackbonePeerStateEntry> = self
            .peers
            .iter()
            .map(|(peer_ip, state)| BackbonePeerStateEntry {
                interface_name: interface_name.to_string(),
                peer_ip: *peer_ip,
                connected_count: state.connected_count,
                blacklisted_remaining_secs: state
                    .blacklisted_until
                    .and_then(|until| (until > now).then(|| (until - now).as_secs_f64())),
                blacklist_reason: state.blacklist_reason.clone(),
                reject_count: state.reject_count,
            })
            .collect();
        entries.sort_by_key(|a| a.peer_ip);
        entries
    }

    pub fn clear(&mut self, peer_ip: IpAddr) -> bool {
        self.peers.remove(&peer_ip).is_some()
    }

    pub fn blacklist(&mut self, peer_ip: IpAddr, duration: Duration, reason: String) -> bool {
        let state = self
            .peers
            .entry(peer_ip)
            .or_insert_with(PeerBehaviorState::new);
        state.blacklisted_until = Some(Instant::now() + duration);
        state.blacklist_reason = Some(reason);
        true
    }

    #[cfg(test)]
    pub fn seed_entry(&mut self, entry: BackbonePeerStateEntry) {
        let mut state = PeerBehaviorState::new();
        state.connected_count = entry.connected_count;
        state.reject_count = entry.reject_count;
        state.blacklist_reason = entry.blacklist_reason;
        if let Some(remaining) = entry.blacklisted_remaining_secs {
            state.blacklisted_until = Some(Instant::now() + Duration::from_secs_f64(remaining));
        }
        self.peers.insert(entry.peer_ip, state);
    }
}

#[derive(Clone, Copy)]
enum DisconnectReason {
    RemoteClosed,
    IdleTimeout,
    WriteStall,
}

/// The listening socket, served on the driver thread: accepts peers and
/// runs the server's periodic work from its timer.
struct ListenerSource {
    listener: TcpListener,
    server: Arc<Server>,
    next_interval: Instant,
}

impl LoopSource for ListenerSource {
    fn fd(&self) -> std::os::fd::RawFd {
        self.listener.as_raw_fd()
    }

    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
        let server = &self.server;
        let mut state = server.lock();
        server.sync_peer_state(&mut state);
        for _ in 0..ACCEPT_BURST {
            match self.listener.accept() {
                Ok((stream, peer_addr)) => {
                    if !server.admit(&mut state, stream, peer_addr, cx) {
                        return SourceStatus::Remove;
                    }
                }
                Err(error) => match super::tcp_server::accept_error_step(&error) {
                    super::tcp_server::AcceptStep::Drained => return SourceStatus::Rearm,
                    super::tcp_server::AcceptStep::Next => {}
                    super::tcp_server::AcceptStep::Pause => {
                        log::warn!("[{}] accept error, pausing accepts: {}", server.name, error);
                        return SourceStatus::Pause;
                    }
                },
            }
        }
        SourceStatus::Rearm
    }

    fn deadline(&self) -> Option<Instant> {
        Some(self.next_interval)
    }

    fn on_timer(&mut self, _cx: &mut LoopContext<'_>) -> SourceStatus {
        self.server.interval();
        self.next_interval = Instant::now() + DP_IC_INTERVAL;
        SourceStatus::Rearm
    }
}

/// One accepted peer, read on the driver thread.
struct PeerSource {
    stream: TcpStream,
    decoder: hdlc::Decoder,
    key: usize,
    id: InterfaceId,
    server: Arc<Server>,
}

impl LoopSource for PeerSource {
    fn fd(&self) -> std::os::fd::RawFd {
        self.stream.as_raw_fd()
    }

    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
        let server = Arc::clone(&self.server);
        if server
            .lock()
            .clients
            .get(&self.key)
            .is_some_and(|client| client.dp_ingress_gated)
        {
            // Throttled: the server re-arms it when the hold ends.
            return SourceStatus::Hold;
        }
        let (mut bytes, mut frames) = (0u64, 0u64);
        let read = read_stream_frames(
            self.stream.as_raw_fd(),
            &mut self.decoder,
            cx,
            &server.tx,
            self.id,
            &server.name,
            |read_bytes, read_frames| {
                bytes += read_bytes as u64;
                frames += read_frames as u64;
            },
        );
        let mut state = server.lock();
        if bytes > 0 {
            if let Some(client) = state.clients.get_mut(&self.key) {
                client.has_received_data = true;
                client.dp_ingress_bytes = client.dp_ingress_bytes.saturating_add(bytes);
                client.dp_ingress_packets = client.dp_ingress_packets.saturating_add(frames);
            }
        }
        let status = match read {
            StreamRead::Rearm => SourceStatus::Rearm,
            StreamRead::Suspend => SourceStatus::Suspend,
            StreamRead::Closed | StreamRead::Failed(_) => {
                let reason = match state.clients.get(&self.key) {
                    Some(client) if client.idle_timed_out => DisconnectReason::IdleTimeout,
                    Some(client) if client.write_stall_flag.load(Ordering::Relaxed) => {
                        DisconnectReason::WriteStall
                    }
                    _ => DisconnectReason::RemoteClosed,
                };
                self.disconnect(&mut state, reason);
                return SourceStatus::Remove;
            }
            StreamRead::DriverGone => return SourceStatus::Remove,
        };
        if bytes > 0
            && server.tx.inbound_queue_snapshot().heights[0] >= server.ingress_thresholds.immediate
        {
            if let Some(key) = server.throttle_busiest_ingress(&mut state) {
                if key == self.key {
                    return SourceStatus::Hold;
                }
                server.set_readable(key, state.clients[&key].fd, false);
            }
        }
        status
    }

    fn on_removed(&mut self) {
        // Closed without a disconnect (the receiver went away): forget it so
        // nothing touches the descriptor once it is closed.
        self.server.lock().clients.remove(&self.key);
    }

    fn on_abandoned(&mut self) {
        // The driver was told about this peer; it will never be read.
        let server = Arc::clone(&self.server);
        let mut state = server.lock();
        self.disconnect(&mut state, DisconnectReason::RemoteClosed);
    }
}

impl PeerSource {
    fn disconnect(&self, state: &mut ServerState, reason: DisconnectReason) {
        let server = &self.server;
        let name = &server.name;
        let Some(client) = state.clients.remove(&self.key) else {
            return;
        };
        let client_id = client.id;

        match reason {
            DisconnectReason::RemoteClosed => {
                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
                    "[{}] backbone client {} disconnected",
                    name,
                    client_id.0,
                );
            }
            DisconnectReason::IdleTimeout => {
                log::info!(
                    "[{}] backbone client {} disconnected due to idle timeout",
                    name,
                    client_id.0
                );
            }
            DisconnectReason::WriteStall => {
                // Already logged by the writer that detected the stall.
            }
        }

        let connected_for = client.connected_at.elapsed();
        if let Some(record) = lock_or_recover(&server.fast_flap_state, "backbone fast-flap state")
            .record_disconnect(
                client.peer_ip,
                connected_for,
                &server.fast_flap,
                Instant::now(),
            )
        {
            let frequency = if record.elapsed_since_first.is_zero() {
                None
            } else {
                Some(record.flap_count as f64 / record.elapsed_since_first.as_secs_f64())
            };
            log::debug!(
                "[{}] peer {} fast flapped after {:.3}s ({} flaps{})",
                name,
                client.peer_ip,
                connected_for.as_secs_f64(),
                record.flap_count,
                frequency
                    .map(|value| format!(", {:.3} Hz", value))
                    .unwrap_or_default(),
            );
            if record.blocked {
                let peer = state
                    .peers
                    .entry(client.peer_ip)
                    .or_insert_with(PeerBehaviorState::new);
                peer.blacklisted_until = record.blocked_until;
                peer.blacklist_reason = Some("fast flapping".into());
                if record.newly_blocked {
                    log::warn!(
                        "[{}] blocking further connections from {} due to fast flapping",
                        name,
                        client.peer_ip,
                    );
                }
            }
        }
        let mut events = vec![Event::BackbonePeerDisconnected {
            server_interface_id: server.server_interface_id,
            peer_interface_id: client.id,
            peer_ip: client.peer_ip,
            peer_port: client.peer_port,
            connected_for,
            had_received_data: client.has_received_data,
        }];
        match reason {
            DisconnectReason::IdleTimeout => events.push(Event::BackbonePeerIdleTimeout {
                server_interface_id: server.server_interface_id,
                peer_interface_id: client.id,
                peer_ip: client.peer_ip,
                peer_port: client.peer_port,
                connected_for,
            }),
            DisconnectReason::WriteStall => events.push(Event::BackbonePeerWriteStall {
                server_interface_id: server.server_interface_id,
                peer_interface_id: client.id,
                peer_ip: client.peer_ip,
                peer_port: client.peer_port,
                connected_for,
            }),
            DisconnectReason::RemoteClosed => {}
        }

        if let Some(peer) = state.peers.get_mut(&client.peer_ip) {
            peer.connected_count = peer.connected_count.saturating_sub(1);
        }
        lock_or_recover(&server.peer_state, "backbone peer state").upsert_snapshot(&state.peers);
        // The writer already reported a write stall's interface down.
        if !matches!(reason, DisconnectReason::WriteStall) {
            events.push(Event::InterfaceDown(client_id));
        }
        let _ = server.tx.push_from_loop(events);
    }
}

fn cleanup_peer_state(peers: &mut HashMap<IpAddr, PeerBehaviorState>) {
    let now = Instant::now();
    peers.retain(|_, state| {
        if matches!(state.blacklisted_until, Some(until) if now >= until) {
            state.blacklisted_until = None;
            state.blacklist_reason = None;
        }
        state.blacklisted_until.is_some() || state.connected_count > 0 || state.reject_count > 0
    });
}

fn is_ip_blacklisted(peers: &mut HashMap<IpAddr, PeerBehaviorState>, peer_ip: IpAddr) -> bool {
    let now = Instant::now();
    if let Some(state) = peers.get_mut(&peer_ip) {
        if let Some(until) = state.blacklisted_until {
            if now < until {
                return true;
            }
            state.blacklisted_until = None;
        }
    }
    false
}

fn set_tcp_keepalive(stream: &TcpStream) -> io::Result<()> {
    let sock = SockRef::from(stream);
    let mut keepalive = TcpKeepalive::new()
        .with_time(Duration::from_secs(5))
        .with_interval(Duration::from_secs(2));
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    {
        keepalive = keepalive.with_retries(12);
    }
    sock.set_tcp_keepalive(&keepalive)
}

// ---------------------------------------------------------------------------
// Client mode
// ---------------------------------------------------------------------------

/// Configuration for a backbone client interface.
#[derive(Debug, Clone)]
pub struct BackboneClientConfig {
    pub name: String,
    pub target_host: String,
    pub target_port: u16,
    pub interface_id: InterfaceId,
    pub reconnect_wait: Duration,
    pub max_reconnect_tries: Option<u32>,
    pub connect_timeout: Duration,
    pub transport_identity: Option<String>,
    pub runtime: Arc<Mutex<BackboneClientRuntime>>,
}

#[derive(Debug, Clone)]
pub struct BackboneClientRuntime {
    pub reconnect_wait: Duration,
    pub max_reconnect_tries: Option<u32>,
    pub connect_timeout: Duration,
}

impl BackboneClientRuntime {
    pub fn from_config(config: &BackboneClientConfig) -> Self {
        Self {
            reconnect_wait: config.reconnect_wait,
            max_reconnect_tries: config.max_reconnect_tries,
            connect_timeout: config.connect_timeout,
        }
    }
}

#[derive(Debug, Clone)]
pub struct BackboneClientRuntimeConfigHandle {
    pub interface_name: String,
    pub runtime: Arc<Mutex<BackboneClientRuntime>>,
    pub startup: BackboneClientRuntime,
}

impl Default for BackboneClientConfig {
    fn default() -> Self {
        let mut config = BackboneClientConfig {
            name: String::new(),
            target_host: "127.0.0.1".into(),
            target_port: 4242,
            interface_id: InterfaceId(0),
            reconnect_wait: Duration::from_secs(5),
            max_reconnect_tries: None,
            connect_timeout: Duration::from_secs(5),
            transport_identity: None,
            runtime: Arc::new(Mutex::new(BackboneClientRuntime {
                reconnect_wait: Duration::from_secs(5),
                max_reconnect_tries: None,
                connect_timeout: Duration::from_secs(5),
            })),
        };
        let startup = BackboneClientRuntime::from_config(&config);
        config.runtime = Arc::new(Mutex::new(startup));
        config
    }
}

/// Writer that sends HDLC-framed data over a TCP stream (client mode).
///
/// A fresh writer is constructed for every connection: the initial connect in
/// [`start_client`] and each reconnect in [`ClientLink::reconnect_loop`]. Its
/// [`TransmitBuffer`] and [`EgressController`] therefore always start empty and
/// unstalled, and the stalled writer is dropped together with its socket on
/// teardown, so no egress drain/stall state can survive into the next
/// connection. This structurally covers upstream's `BackboneInterface._dp_ec_reset`
/// (rgit `f0875ac4`), which resets the same counters on a persistent Python
/// initiator instance.
struct BackboneClientWriter {
    stream: TcpStream,
    transmit_buffer: TransmitBuffer,
    egress_control: EgressControl,
}

impl Writer for BackboneClientWriter {
    fn send_frame_confirmed(&mut self, data: &[u8]) -> io::Result<()> {
        if !self
            .transmit_buffer
            .append_with_limit(hdlc::frame(data), Some(EGRESS_HIGH_WATERMARK))
        {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "confirmed frame exceeds backbone buffer",
            ));
        }
        self.transmit_buffer.flush();
        while self.transmit_buffer.sendable_bytes() > 0 {
            if self.transmit_buffer.drain_to(&mut self.stream)? == 0 {
                return Err(io::Error::new(
                    io::ErrorKind::WriteZero,
                    "backbone writer made no progress",
                ));
            }
        }
        Ok(())
    }
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        self.send_frames(&[crate::interface::FrameBytes::Owned(data.to_vec())])
    }

    fn send_frames(&mut self, frames: &[crate::interface::FrameBytes]) -> io::Result<()> {
        for frame in frames {
            let encoded = hdlc::frame(frame);
            let encoded_len = encoded.len();
            if !self
                .transmit_buffer
                .append_with_limit(encoded, Some(EGRESS_HIGH_WATERMARK))
            {
                self.egress_control.record_drop(encoded_len);
            }
        }
        self.transmit_buffer.flush();
        while self.transmit_buffer.sendable_bytes() > 0 {
            if self.transmit_buffer.drain_to(&mut self.stream)? == 0 {
                return Err(io::Error::new(
                    io::ErrorKind::WriteZero,
                    "backbone client writer made no progress",
                ));
            }
        }
        Ok(())
    }

    fn egress_control(&self) -> Option<EgressControl> {
        Some(self.egress_control.clone())
    }

    fn loop_socket(&self) -> Option<std::os::fd::RawFd> {
        Some(self.stream.as_raw_fd())
    }

    fn loop_egress(&mut self) -> Option<LoopEgress> {
        Some(LoopEgress::budget(self.egress_control.clone()))
    }
}

/// Try to connect to the target host:port with timeout.
fn try_connect_client(
    config: &BackboneClientConfig,
    underlay_mark: Option<u32>,
) -> io::Result<TcpStream> {
    let runtime = lock_or_recover(&config.runtime, "backbone runtime").clone();
    let addr_str = format!("{}:{}", config.target_host, config.target_port);
    let addr = addr_str
        .to_socket_addrs()?
        .next()
        .ok_or_else(|| io::Error::new(io::ErrorKind::AddrNotAvailable, "no addresses resolved"))?;

    #[cfg(target_os = "linux")]
    let stream = if underlay_mark.is_some() {
        super::connect_tcp_with_options(&addr, None, underlay_mark, runtime.connect_timeout)?
    } else {
        TcpStream::connect_timeout(&addr, runtime.connect_timeout)?
    };
    #[cfg(not(target_os = "linux"))]
    let stream = TcpStream::connect_timeout(&addr, runtime.connect_timeout)?;
    stream.set_nodelay(true)?;
    set_tcp_keepalive(&stream).ok();

    // Prevent SIGPIPE on macOS when writing to broken pipes
    #[cfg(target_os = "macos")]
    {
        let sock = SockRef::from(&stream);
        sock.set_nosigpipe(true).ok();
    }

    Ok(stream)
}

/// Connect and start the reader thread. Returns the writer for the driver.
pub fn start_client(config: BackboneClientConfig, tx: EventSender) -> io::Result<Box<dyn Writer>> {
    start_client_with_ifac(config, tx, 0, None, ListenerControl::new())
}

pub(crate) fn start_client_with_ifac(
    config: BackboneClientConfig,
    tx: EventSender,
    ifac_size: usize,
    underlay_mark: Option<u32>,
    control: ListenerControl,
) -> io::Result<Box<dyn Writer>> {
    let stream = try_connect_client(&config, underlay_mark)?;
    let writer_stream = stream.try_clone()?;

    let id = config.interface_id;
    log::info!(
        "[{}] backbone client connected to {}:{}",
        config.name,
        config.target_host,
        config.target_port
    );

    // Initial connect: writer is None because it's returned directly to the caller
    let _ = tx.send(Event::InterfaceUp(id, None, None));

    let link = Arc::new(ClientLink {
        config,
        tx,
        ifac_size,
        underlay_mark,
        control,
    });
    if let Err(error) = link.serve(stream) {
        let _ = link.tx.send(Event::InterfaceDown(id));
        return Err(error);
    }

    Ok(Box::new(BackboneClientWriter {
        stream: writer_stream,
        transmit_buffer: TransmitBuffer::new(),
        egress_control: EgressControl::new(EGRESS_HIGH_WATERMARK),
    }))
}

/// What a client connection needs across reconnects.
struct ClientLink {
    config: BackboneClientConfig,
    tx: EventSender,
    ifac_size: usize,
    underlay_mark: Option<u32>,
    control: ListenerControl,
}

impl ClientLink {
    /// Read `stream` on the driver thread from now on.
    fn serve(self: &Arc<Self>, stream: TcpStream) -> io::Result<()> {
        let source = ClientSource {
            stream,
            decoder: hdlc::Decoder::reticulum(HW_MTU, self.ifac_size),
            link: Arc::clone(self),
        };
        self.tx.register_source(
            crate::interface::loop_writer::read_key(self.config.interface_id),
            Box::new(source),
        )
    }

    /// Report the connection down and reconnect on a thread of its own,
    /// which exits once the new connection is served on the driver (or
    /// reconnecting gives up or is stopped).
    fn connection_lost(self: &Arc<Self>) {
        let id = self.config.interface_id;
        let _ = self.tx.push_from_loop([Event::InterfaceDown(id)]);
        let max_tries =
            lock_or_recover(&self.config.runtime, "backbone runtime").max_reconnect_tries;
        if max_tries == Some(0) {
            // Whoever owns this client (the backbone peer pool) reconnects
            // it. A reconnect thread would only wait out a backoff and then
            // report the interface down again, possibly after the owner has
            // already brought it back up under the same id.
            log::debug!(
                "[{}] not reconnecting (max_reconnect_tries = 0)",
                self.config.name
            );
            return;
        }
        let link = Arc::clone(self);
        let spawned = thread::Builder::new()
            .name(format!("bb-reconn-{}", id.0))
            .spawn(move || link.reconnect_loop());
        if let Err(error) = spawned {
            log::error!(
                "[{}] cannot start reconnecting, giving up: {}",
                self.config.name,
                error
            );
        }
    }

    fn reconnect_loop(self: Arc<Self>) {
        loop {
            let Some(stream) =
                client_reconnect(&self.config, &self.tx, self.underlay_mark, &self.control)
            else {
                if !self.control.should_stop() {
                    log::error!("[{}] reconnection failed, giving up", self.config.name);
                }
                return;
            };
            match self.serve(stream) {
                Ok(()) => return,
                Err(error) => {
                    log::warn!(
                        "[{}] cannot read the new connection: {}",
                        self.config.name,
                        error
                    );
                    let _ = self.tx.send(Event::InterfaceDown(self.config.interface_id));
                }
            }
        }
    }
}

/// The connected client socket, read on the driver thread.
struct ClientSource {
    stream: TcpStream,
    decoder: hdlc::Decoder,
    link: Arc<ClientLink>,
}

impl LoopSource for ClientSource {
    fn fd(&self) -> std::os::fd::RawFd {
        self.stream.as_raw_fd()
    }

    fn stopped(&self) -> bool {
        self.link.control.should_stop()
    }

    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
        let link = &self.link;
        if link.control.should_stop() {
            return SourceStatus::Remove;
        }
        let name = &link.config.name;
        match read_stream_frames(
            self.stream.as_raw_fd(),
            &mut self.decoder,
            cx,
            &link.tx,
            link.config.interface_id,
            name,
            |_, _| {},
        ) {
            StreamRead::Rearm => SourceStatus::Rearm,
            StreamRead::Suspend => SourceStatus::Suspend,
            StreamRead::Closed => {
                log::warn!("[{}] connection closed", name);
                link.connection_lost();
                SourceStatus::Remove
            }
            StreamRead::Failed(error) => {
                log::warn!("[{}] read error: {}", name, error);
                link.connection_lost();
                SourceStatus::Remove
            }
            StreamRead::DriverGone => SourceStatus::Remove,
        }
    }
}

/// Maximum backoff multiplier: `base_delay * 2^MAX_BACKOFF_SHIFT`.
/// With a 5 s base this caps at 5 × 2^6 = 320 s ≈ 5 min.
pub(crate) const MAX_BACKOFF_SHIFT: u32 = 6;

/// Delay before reconnect attempt number `attempt` (0-based): `base`
/// doubled per attempt up to `base * 2^MAX_BACKOFF_SHIFT`, with ±25 %
/// jitter to avoid thundering-herd reconnects.
pub(crate) fn reconnect_backoff(base: Duration, attempt: u32) -> Duration {
    let backoff = base * 2u32.pow(attempt.min(MAX_BACKOFF_SHIFT));
    let jitter_range = backoff / 4;
    if jitter_range.as_nanos() == 0 {
        return backoff;
    }
    let offset = Duration::from_nanos(
        (std::hash::RandomState::new().build_hasher().finish() % jitter_range.as_nanos() as u64)
            * 2,
    );
    if offset > jitter_range {
        backoff + (offset - jitter_range)
    } else {
        backoff - (jitter_range - offset)
    }
}

/// Attempt to reconnect with exponential backoff and jitter.
/// Returns the new reader stream on success.
/// Sends the new writer to the driver via InterfaceUp event.
fn client_reconnect(
    config: &BackboneClientConfig,
    tx: &EventSender,
    underlay_mark: Option<u32>,
    control: &ListenerControl,
) -> Option<TcpStream> {
    let mut attempts = 0u32;
    loop {
        let runtime = lock_or_recover(&config.runtime, "backbone runtime").clone();

        let jitter = reconnect_backoff(runtime.reconnect_wait, attempts);
        if !wait_while_active(control, jitter) {
            return None;
        }

        attempts += 1;

        if let Some(max) = runtime.max_reconnect_tries {
            if attempts > max {
                let _ = tx.send(Event::InterfaceDown(config.interface_id));
                return None;
            }
        }

        log::info!(
            "[{}] reconnect attempt {} (backoff {:.1}s) ...",
            config.name,
            attempts,
            jitter.as_secs_f64(),
        );

        match try_connect_client(config, underlay_mark) {
            Ok(new_stream) => {
                if control.should_stop() {
                    return None;
                }
                let writer_stream = match new_stream.try_clone() {
                    Ok(s) => s,
                    Err(e) => {
                        log::warn!("[{}] failed to clone stream: {}", config.name, e);
                        continue;
                    }
                };
                log::info!(
                    "[{}] reconnected after {} attempt(s)",
                    config.name,
                    attempts
                );
                let new_writer: Box<dyn Writer> = Box::new(BackboneClientWriter {
                    stream: writer_stream,
                    transmit_buffer: TransmitBuffer::new(),
                    egress_control: EgressControl::new(EGRESS_HIGH_WATERMARK),
                });
                let _ = tx.send(Event::InterfaceUp(
                    config.interface_id,
                    Some(new_writer),
                    None,
                ));
                return Some(new_stream);
            }
            Err(e) => {
                log::warn!("[{}] reconnect failed: {}", config.name, e);
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Factory
// ---------------------------------------------------------------------------

/// Internal enum used by [`BackboneInterfaceFactory`] to carry either a
/// server or client config through the opaque `InterfaceConfigData` channel.
#[derive(Clone)]
pub(crate) enum BackboneMode {
    Server(BackboneConfig),
    Client(BackboneClientConfig, u8),
}

/// Factory for `BackboneInterface`.
///
/// If the config params contain `"remote"` or `"target_host"` the interface
/// is started in client mode; otherwise it is started as a TCP listener
/// (server mode).
pub struct BackboneInterfaceFactory;

/// Compatibility name used by Reticulum 1.3.8 for a forced client-mode
/// Backbone interface.
pub struct BackboneClientInterfaceFactory;

impl InterfaceFactory for BackboneClientInterfaceFactory {
    fn type_name(&self) -> &str {
        "BackboneClientInterface"
    }

    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        if !params.contains_key("remote") && !params.contains_key("target_host") {
            return Err("BackboneClientInterface requires 'remote' or 'target_host'".into());
        }
        BackboneInterfaceFactory.parse_config(name, id, params)
    }

    fn start(
        &self,
        config: Box<dyn InterfaceConfigData>,
        ctx: StartContext,
    ) -> io::Result<StartResult> {
        BackboneInterfaceFactory.start(config, ctx)
    }
}

fn parse_positive_duration_secs(params: &HashMap<String, String>, key: &str) -> Option<Duration> {
    params
        .get(key)
        .and_then(|v| v.parse::<f64>().ok())
        .filter(|v| *v > 0.0)
        .and_then(|v| Duration::try_from_secs_f64(v).ok())
        .filter(|duration| Instant::now().checked_add(*duration).is_some())
}

fn parse_nonnegative_duration(
    params: &HashMap<String, String>,
    key: &str,
    multiplier: f64,
    default: Duration,
) -> Result<Duration, String> {
    let Some(raw) = params.get(key) else {
        return Ok(default);
    };
    let value = raw
        .parse::<f64>()
        .map_err(|_| format!("invalid {key} '{raw}' (expected a non-negative number)"))?;
    let seconds = value * multiplier;
    if !value.is_finite() || value < 0.0 || !seconds.is_finite() || seconds > u64::MAX as f64 {
        return Err(format!(
            "invalid {key} '{raw}' (expected a finite non-negative number)"
        ));
    }
    let duration = Duration::from_secs_f64(seconds);
    if Instant::now().checked_add(duration).is_none() {
        return Err(format!("invalid {key} '{raw}' (duration is too large)"));
    }
    Ok(duration)
}

fn parse_fast_flap_config(
    params: &HashMap<String, String>,
) -> Result<BackboneFastFlapConfig, String> {
    let defaults = BackboneFastFlapConfig::default();
    let enabled = match params.get("block_fast_flapping") {
        Some(raw) => crate::config::parse_bool_pub(raw)
            .ok_or_else(|| format!("invalid block_fast_flapping '{raw}' (expected yes or no)"))?,
        None => defaults.enabled,
    };
    let connection_threshold = parse_nonnegative_duration(
        params,
        "fast_flapping_threshold",
        1.0,
        defaults.connection_threshold,
    )?;
    let grace = match params.get("fast_flapping_grace") {
        Some(raw) => raw.parse::<u64>().map_err(|_| {
            format!("invalid fast_flapping_grace '{raw}' (expected a non-negative integer)")
        })?,
        None => defaults.grace,
    };
    let block_duration = parse_nonnegative_duration(
        params,
        "fast_flapping_block_time",
        60.0,
        defaults.block_duration,
    )?;

    Ok(BackboneFastFlapConfig {
        enabled,
        connection_threshold,
        grace,
        block_duration,
    })
}

impl InterfaceFactory for BackboneInterfaceFactory {
    fn type_name(&self) -> &str {
        "BackboneInterface"
    }

    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        if let Some(target_host) = params.get("remote").or_else(|| params.get("target_host")) {
            // Client mode
            let target_host = target_host.clone();
            let target_port = params
                .get("target_port")
                .or_else(|| params.get("port"))
                .and_then(|v| v.parse().ok())
                .unwrap_or(4242);
            let transport_identity = params.get("transport_identity").cloned();
            // Like Python, retry forever unless max_reconnect_tries is set.
            let max_reconnect_tries = match params.get("max_reconnect_tries") {
                Some(value) => Some(value.trim().parse::<u32>().map_err(|_| {
                    format!(
                        "invalid Backbone max_reconnect_tries '{}' (expected a non-negative integer)",
                        value
                    )
                })?),
                None => None,
            };
            let priority = match params.get("priority") {
                Some(value) => value.parse::<u8>().map_err(|_| {
                    format!(
                        "invalid Backbone peer priority '{}' (expected 0..100)",
                        value
                    )
                })?,
                None => crate::driver::BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY,
            };
            if priority > 100 {
                return Err(format!(
                    "invalid Backbone peer priority '{}' (expected 0..100)",
                    priority
                ));
            }
            let mut client = BackboneClientConfig {
                name: name.to_string(),
                target_host,
                target_port,
                interface_id: id,
                max_reconnect_tries,
                transport_identity,
                ..BackboneClientConfig::default()
            };
            client.runtime = Arc::new(Mutex::new(BackboneClientRuntime::from_config(&client)));
            Ok(Box::new(BackboneMode::Client(client, priority)))
        } else {
            // Server mode
            let listen_ip = params
                .get("listen_ip")
                .or_else(|| params.get("device"))
                .cloned()
                .unwrap_or_else(|| "0.0.0.0".into());
            let listen_port = params
                .get("listen_port")
                .or_else(|| params.get("port"))
                .and_then(|v| v.parse().ok())
                .unwrap_or(4242);
            let max_connections = params.get("max_connections").and_then(|v| v.parse().ok());
            let idle_timeout = parse_positive_duration_secs(params, "idle_timeout");
            let write_stall_timeout = parse_positive_duration_secs(params, "write_stall_timeout");
            let abuse = BackboneAbuseConfig {
                max_penalty_duration: parse_positive_duration_secs(params, "max_penalty_duration"),
            };
            let fast_flap = parse_fast_flap_config(params)?;
            let mut config = BackboneConfig {
                name: name.to_string(),
                listen_ip,
                listen_port,
                interface_id: id,
                mode: constants::MODE_FULL,
                gravity: 0,
                recursive_prs: false,
                announces_from_internal: true,
                announces_to_internal: None,
                max_connections,
                idle_timeout,
                write_stall_timeout,
                abuse,
                fast_flap,
                fast_flap_state: shared_fast_flap_monitor(),
                ingress_control: IngressControlConfig::enabled(),
                runtime: Arc::new(Mutex::new(BackboneServerRuntime {
                    max_connections: None,
                    idle_timeout: None,
                    write_stall_timeout: None,
                    abuse: BackboneAbuseConfig::default(),
                })),
                peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
            };
            let startup = BackboneServerRuntime::from_config(&config);
            config.runtime = Arc::new(Mutex::new(startup));
            Ok(Box::new(BackboneMode::Server(config)))
        }
    }

    fn start(
        &self,
        config: Box<dyn InterfaceConfigData>,
        ctx: StartContext,
    ) -> io::Result<StartResult> {
        let mode = *config.into_any().downcast::<BackboneMode>().map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                "wrong config type for BackboneInterface",
            )
        })?;

        match mode {
            BackboneMode::Client(cfg, _) => {
                let id = cfg.interface_id;
                let name = cfg.name.clone();
                let info = InterfaceInfo {
                    id,
                    name,
                    mode: ctx.mode,
                    gravity: ctx.gravity,
                    recursive_prs: ctx.recursive_prs,
                    announces_from_internal: ctx.announces_from_internal,
                    announces_to_internal: ctx.announces_to_internal,
                    out_capable: true,
                    in_capable: true,
                    bitrate: Some(CLIENT_BITRATE_GUESS),
                    airtime_profile: None,
                    announce_rate_target: None,
                    announce_rate_grace: 0,
                    announce_rate_penalty: 0.0,
                    announce_cap: constants::ANNOUNCE_CAP,
                    is_local_client: false,
                    wants_tunnel: false,
                    tunnel_id: None,
                    mtu: HW_MTU as u32,
                    ingress_control: ctx.ingress_control,
                    ia_freq: 0.0,
                    ip_freq: 0.0,
                    op_freq: 0.0,
                    op_samples: 0,
                    started: crate::time::now(),
                };
                let ifac_size = ctx.ifac.as_ref().map(|ifac| ifac.size).unwrap_or(0);
                let control = ListenerControl::new();
                let writer = start_client_with_ifac(
                    cfg,
                    ctx.tx,
                    ifac_size,
                    ctx.underlay_mark,
                    control.clone(),
                )?;
                Ok(StartResult::Simple {
                    control: Some(control),
                    id,
                    info,
                    writer,
                    interface_type_name: "BackboneInterface".to_string(),
                })
            }
            BackboneMode::Server(mut cfg) => {
                cfg.ingress_control = ctx.ingress_control;
                cfg.mode = ctx.mode;
                cfg.gravity = ctx.gravity;
                cfg.recursive_prs = ctx.recursive_prs;
                cfg.announces_from_internal = ctx.announces_from_internal;
                cfg.announces_to_internal = ctx.announces_to_internal;
                let parent_id = cfg.interface_id;
                start_with_template(
                    cfg,
                    ctx.tx,
                    ctx.next_dynamic_id,
                    Some(DynamicInterfaceTemplate {
                        parent_id,
                        interface_type: "BackboneInterfacePeer".into(),
                        ifac: ctx.ifac,
                        mode: ctx.mode,
                        gravity: ctx.gravity,
                        recursive_prs: ctx.recursive_prs,
                        announces_from_internal: ctx.announces_from_internal,
                        announces_to_internal: ctx.announces_to_internal,
                    }),
                    ctx.underlay_mark,
                )?;
                Ok(StartResult::Listener { control: None })
            }
        }
    }
}

pub(crate) fn runtime_handle_from_mode(mode: &BackboneMode) -> Option<BackboneRuntimeConfigHandle> {
    match mode {
        BackboneMode::Server(config) => Some(BackboneRuntimeConfigHandle {
            interface_name: config.name.clone(),
            runtime: Arc::clone(&config.runtime),
            startup: BackboneServerRuntime::from_config(config),
        }),
        BackboneMode::Client(_, _) => None,
    }
}

pub(crate) fn peer_state_handle_from_mode(mode: &BackboneMode) -> Option<BackbonePeerStateHandle> {
    match mode {
        BackboneMode::Server(config) => Some(BackbonePeerStateHandle {
            interface_id: config.interface_id,
            interface_name: config.name.clone(),
            mode: config.mode,
            gravity: config.gravity,
            announces_to_internal: config.announces_to_internal,
            peer_state: Arc::clone(&config.peer_state),
            fast_flap: config.fast_flap,
            fast_flap_state: Arc::clone(&config.fast_flap_state),
        }),
        BackboneMode::Client(_, _) => None,
    }
}

pub(crate) fn client_runtime_handle_from_mode(
    mode: &BackboneMode,
) -> Option<BackboneClientRuntimeConfigHandle> {
    match mode {
        BackboneMode::Client(config, _) => Some(BackboneClientRuntimeConfigHandle {
            interface_name: config.name.clone(),
            runtime: Arc::clone(&config.runtime),
            startup: BackboneClientRuntime::from_config(config),
        }),
        BackboneMode::Server(_) => None,
    }
}

pub(crate) fn client_config_from_mode(mode: &BackboneMode) -> Option<BackboneClientConfig> {
    match mode {
        BackboneMode::Client(config, _) => Some(config.clone()),
        BackboneMode::Server(_) => None,
    }
}

pub(crate) fn client_priority_from_mode(mode: &BackboneMode) -> Option<u8> {
    match mode {
        BackboneMode::Client(_, priority) => Some(*priority),
        BackboneMode::Server(_) => None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Read;
    use std::sync::mpsc;
    use std::time::Duration;

    fn find_free_port() -> u16 {
        crate::test_support::port()
    }

    /// Serve the receiver's sockets and timers for `duration` without
    /// taking any event, as a driver busy with other work would.
    fn serve_for(rx: &crate::event::EventReceiver, duration: Duration) {
        let deadline = Instant::now() + duration;
        let mut ready = polling::Events::new();
        while Instant::now() < deadline {
            ready.clear();
            rx.poll_ready(&mut ready).unwrap();
            rx.serve_sources(&ready);
            thread::sleep(Duration::from_millis(2));
        }
    }

    fn test_fast_flap_config() -> BackboneFastFlapConfig {
        BackboneFastFlapConfig {
            enabled: true,
            connection_threshold: Duration::from_secs(20),
            grace: 5,
            block_duration: Duration::from_secs(12 * 60 * 60),
        }
    }

    #[test]
    fn fast_flap_defaults_match_upstream() {
        assert_eq!(BackboneFastFlapConfig::default(), test_fast_flap_config());
    }

    #[test]
    fn sixth_short_connection_blocks_but_fifth_does_not() {
        let config = test_fast_flap_config();
        let mut monitor = BackboneFastFlapMonitor::new();
        let peer: IpAddr = "192.0.2.10".parse().unwrap();
        let start = Instant::now();

        for flap in 1..=5 {
            let record = monitor
                .record_disconnect(
                    peer,
                    Duration::from_secs(19),
                    &config,
                    start + Duration::from_secs(flap),
                )
                .unwrap();
            assert_eq!(record.flap_count, flap);
            assert!(!record.blocked);
            assert!(!monitor.is_blocked(peer, &config, start + Duration::from_secs(flap)));
        }

        let sixth = monitor
            .record_disconnect(
                peer,
                Duration::from_secs(1),
                &config,
                start + Duration::from_secs(6),
            )
            .unwrap();
        assert_eq!(sixth.flap_count, 6);
        assert!(sixth.blocked);
        assert!(sixth.newly_blocked);
        assert!(monitor.is_blocked(peer, &config, start + Duration::from_secs(6)));
    }

    #[test]
    fn every_short_connection_counts_even_after_receiving_data() {
        let config = test_fast_flap_config();
        let mut monitor = BackboneFastFlapMonitor::new();
        let peer: IpAddr = "192.0.2.11".parse().unwrap();
        let now = Instant::now();

        let record = monitor
            .record_disconnect(peer, Duration::from_secs(2), &config, now)
            .unwrap();
        assert_eq!(record.flap_count, 1);
        assert_eq!(monitor.flap_count(peer), 1);
    }

    #[test]
    fn threshold_boundary_and_long_connections_are_not_counted() {
        let config = test_fast_flap_config();
        let mut monitor = BackboneFastFlapMonitor::new();
        let peer: IpAddr = "192.0.2.12".parse().unwrap();
        let now = Instant::now();

        assert!(monitor
            .record_disconnect(peer, Duration::from_secs(20), &config, now)
            .is_none());
        assert!(monitor
            .record_disconnect(peer, Duration::from_secs(21), &config, now)
            .is_none());
        assert_eq!(monitor.flap_count(peer), 0);
    }

    #[test]
    fn disabled_fast_flapping_neither_tracks_nor_blocks() {
        let mut config = test_fast_flap_config();
        config.enabled = false;
        let mut monitor = BackboneFastFlapMonitor::new();
        let peer: IpAddr = "192.0.2.13".parse().unwrap();
        let now = Instant::now();

        for offset in 0..10 {
            assert!(monitor
                .record_disconnect(
                    peer,
                    Duration::from_secs(1),
                    &config,
                    now + Duration::from_secs(offset),
                )
                .is_none());
        }
        assert_eq!(monitor.flap_count(peer), 0);
        assert!(!monitor.is_blocked(peer, &config, now + Duration::from_secs(10)));
        assert_eq!(monitor.blocked_ip_count(&config, now), 0);
    }

    #[test]
    fn flap_blocks_expire_strictly_after_configured_duration() {
        let mut config = test_fast_flap_config();
        config.grace = 0;
        config.block_duration = Duration::from_secs(60);
        let mut monitor = BackboneFastFlapMonitor::new();
        let peer: IpAddr = "192.0.2.14".parse().unwrap();
        let now = Instant::now();

        monitor
            .record_disconnect(peer, Duration::from_secs(1), &config, now)
            .unwrap();
        assert!(monitor.is_blocked(peer, &config, now + Duration::from_secs(60)));
        assert!(!monitor.is_blocked(
            peer,
            &config,
            now + Duration::from_secs(60) + Duration::from_nanos(1)
        ));
        assert_eq!(monitor.flap_count(peer), 0);
    }

    #[test]
    fn fast_flap_history_is_isolated_by_ip() {
        let mut config = test_fast_flap_config();
        config.grace = 1;
        let mut monitor = BackboneFastFlapMonitor::new();
        let first: IpAddr = "192.0.2.15".parse().unwrap();
        let second: IpAddr = "192.0.2.16".parse().unwrap();
        let now = Instant::now();

        for offset in 0..2 {
            monitor
                .record_disconnect(
                    first,
                    Duration::from_secs(1),
                    &config,
                    now + Duration::from_secs(offset),
                )
                .unwrap();
        }
        monitor
            .record_disconnect(second, Duration::from_secs(1), &config, now)
            .unwrap();

        assert!(monitor.is_blocked(first, &config, now + Duration::from_secs(2)));
        assert!(!monitor.is_blocked(second, &config, now + Duration::from_secs(2)));
        assert_eq!(
            monitor.blocked_ip_count(&config, now + Duration::from_secs(2)),
            1
        );
        assert_eq!(
            monitor.blocked_ip_list(&config, now + Duration::from_secs(2)),
            vec![first],
            "tracked peers within the grace allowance must not appear as blocked"
        );
    }

    #[test]
    fn blocked_ip_list_is_sorted_current_and_empty_when_disabled() {
        let mut config = test_fast_flap_config();
        config.grace = 0;
        config.block_duration = Duration::from_secs(60);
        let mut monitor = BackboneFastFlapMonitor::new();
        let first: IpAddr = "203.0.113.20".parse().unwrap();
        let second: IpAddr = "192.0.2.10".parse().unwrap();
        let now = Instant::now();
        monitor
            .record_disconnect(first, Duration::from_secs(1), &config, now)
            .unwrap();
        monitor
            .record_disconnect(second, Duration::from_secs(1), &config, now)
            .unwrap();

        assert_eq!(monitor.blocked_ip_list(&config, now), vec![second, first]);
        assert!(monitor
            .blocked_ip_list(&config, now + Duration::from_secs(61))
            .is_empty());
        config.enabled = false;
        assert!(monitor.blocked_ip_list(&config, now).is_empty());
    }

    #[test]
    fn peer_monitor_snapshot_clears_expired_blacklist_state() {
        let peer: IpAddr = "192.0.2.17".parse().unwrap();
        let mut monitor = BackbonePeerMonitor::new();
        monitor.blacklist(peer, Duration::from_secs(60), "temporary".into());
        let mut peers = HashMap::new();
        monitor.sync_into(&mut peers);
        let state = peers.get_mut(&peer).unwrap();
        state.blacklisted_until = None;
        state.blacklist_reason = None;

        monitor.upsert_snapshot(&peers);

        let entry = monitor.list("test").into_iter().next().unwrap();
        assert!(entry.blacklisted_remaining_secs.is_none());
        assert!(entry.blacklist_reason.is_none());
    }

    fn recv_non_peer_event(
        rx: &crate::event::EventReceiver,
        timeout: Duration,
    ) -> Result<Event, mpsc::RecvTimeoutError> {
        let deadline = Instant::now() + timeout;
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(mpsc::RecvTimeoutError::Timeout);
            }
            let event = rx.recv_timeout(remaining)?;
            match event {
                Event::BackbonePeerConnected { .. }
                | Event::BackbonePeerDisconnected { .. }
                | Event::BackbonePeerIdleTimeout { .. }
                | Event::BackbonePeerWriteStall { .. }
                | Event::BackbonePeerPenalty { .. } => continue,
                other => return Ok(other),
            }
        }
    }

    fn make_server_config(
        port: u16,
        interface_id: u64,
        max_connections: Option<usize>,
        idle_timeout: Option<Duration>,
        write_stall_timeout: Option<Duration>,
        abuse: BackboneAbuseConfig,
    ) -> BackboneConfig {
        let mut config = BackboneConfig {
            name: "test-backbone".into(),
            listen_ip: "127.0.0.1".into(),
            listen_port: port,
            interface_id: InterfaceId(interface_id),
            mode: constants::MODE_FULL,
            gravity: 0,
            recursive_prs: false,
            announces_from_internal: true,
            announces_to_internal: None,
            max_connections,
            idle_timeout,
            write_stall_timeout,
            abuse,
            fast_flap: BackboneFastFlapConfig::default(),
            fast_flap_state: Arc::new(Mutex::new(BackboneFastFlapMonitor::new())),
            ingress_control: IngressControlConfig::enabled(),
            runtime: Arc::new(Mutex::new(BackboneServerRuntime {
                max_connections: None,
                idle_timeout: None,
                write_stall_timeout: None,
                abuse: BackboneAbuseConfig::default(),
            })),
            peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
        };
        let startup = BackboneServerRuntime::from_config(&config);
        config.runtime = Arc::new(Mutex::new(startup));
        config
    }

    #[test]
    fn backbone_accept_connection() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8000));

        let config = make_server_config(port, 80, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::InterfaceUp(id, writer, info) => {
                assert_eq!(id, InterfaceId(8000));
                assert!(writer.is_some());
                assert!(info.is_some());
                let info = info.unwrap();
                assert!(info.out_capable);
                assert!(info.in_capable);
                assert_eq!(info.bitrate, Some(SERVER_BITRATE_GUESS));
                assert_eq!(info.mtu, SERVER_PEER_MTU);
            }
            other => panic!("expected InterfaceUp, got {:?}", other),
        }
    }

    #[test]
    fn backbone_class_defaults_preserve_server_and_client_distinction() {
        assert_eq!(SERVER_BITRATE_GUESS, 100_000_000);
        assert_eq!(SERVER_PEER_MTU, 32_768);
        assert_eq!(CLIENT_BITRATE_GUESS, 100_000_000);
        assert_eq!(HW_MTU, 32_768);
    }

    #[test]
    fn auto_mtu_thresholds_include_exact_bitrate_boundaries() {
        for (bitrate, mtu) in [
            (1_000_000_000, Some(524_288)),
            (750_000_000, Some(262_144)),
            (400_000_000, Some(131_072)),
            (200_000_000, Some(65_536)),
            (100_000_000, Some(32_768)),
            (10_000_000, Some(16_384)),
            (5_000_000, Some(8_192)),
            (2_000_000, Some(4_096)),
            (1_000_000, Some(2_048)),
            (62_500, Some(1_024)),
            (62_499, None),
        ] {
            assert_eq!(auto_mtu_for_bitrate(bitrate), mtu, "bitrate {bitrate}");
        }
    }

    #[test]
    fn backbone_accepted_connection_inherits_server_mode() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8050));

        let mut config =
            make_server_config(port, 85, None, None, None, BackboneAbuseConfig::default());
        config.mode = constants::MODE_GATEWAY;

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::InterfaceUp(_, _, Some(info)) => {
                assert_eq!(info.mode, constants::MODE_GATEWAY);
            }
            other => panic!("expected InterfaceUp with info, got {:?}", other),
        }
    }

    #[test]
    fn dataplane_ingress_thresholds_match_upstream_defaults() {
        assert_eq!(
            ingress_thresholds(1024),
            IngressThresholds {
                high: 921,
                immediate: 921,
                mid: 696,
                low: 102,
            }
        );
        assert_eq!(
            ingress_thresholds(8),
            IngressThresholds {
                high: 7,
                immediate: 128,
                mid: 5,
                low: 0,
            }
        );
    }

    #[test]
    fn dataplane_ingress_selects_the_largest_packet_producer() {
        assert_eq!(
            busiest_ingress_key([(7, 12), (8, 41), (9, 3)].into_iter()),
            Some(8)
        );
        assert_eq!(busiest_ingress_key([(7, 0), (8, 0)].into_iter()), None);
    }

    #[test]
    fn backbone_gates_the_largest_producer_before_data_queue_overflow() {
        let port = find_free_port();
        let capacities = crate::event::InboundQueueCapacities {
            data: 200,
            announce: 8,
            path_request: 8,
            ingress_limited: 8,
        };
        let (tx, rx) = crate::event::channel_with_queue_capacities(16, capacities);
        let observer = tx.clone();
        let next_id = Arc::new(AtomicU64::new(8075));
        let config = make_server_config(port, 80, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));
        let mut client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        let payload = vec![0u8; 32];
        let framed = hdlc::frame(&payload);
        let mut burst = Vec::with_capacity(framed.len() * 180);
        for _ in 0..180 {
            burst.extend_from_slice(&framed);
        }
        client.write_all(&burst).unwrap();

        let deadline = Instant::now() + Duration::from_secs(2);
        while observer.inbound_queue_snapshot().heights[0] < 180 && Instant::now() < deadline {
            serve_for(&rx, Duration::from_millis(5));
        }
        serve_for(&rx, Duration::from_millis(100));
        let gated = observer.inbound_queue_snapshot();
        assert!(
            gated.heights[0] >= 180,
            "DATA queue never reached the high watermark"
        );
        assert_eq!(gated.dropped[0], 0);

        client.write_all(&hdlc::frame(&[1u8; 32])).unwrap();
        serve_for(&rx, Duration::from_millis(300));
        let after_marker = observer.inbound_queue_snapshot();
        assert_eq!(after_marker.heights[0], gated.heights[0]);
        assert_eq!(
            after_marker.dropped[0], gated.dropped[0],
            "the producer remained readable and overflowed the DATA queue"
        );

        for _ in 0..gated.heights[0] {
            let event = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
            assert!(matches!(event, Event::Frame { .. }));
        }
        let released = recv_non_peer_event(&rx, Duration::from_secs(5)).unwrap();
        assert!(
            matches!(released, Event::Frame { data, .. } if data == vec![1u8; 32]),
            "held producer was not released after DATA pressure recovered"
        );
    }

    #[test]
    fn backbone_receive_frame() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8100));

        let config = make_server_config(port, 81, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        // Drain InterfaceUp
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        // The exact Reticulum header minimum must be dropped without
        // preventing recovery of the next valid frame.
        client
            .write_all(&hdlc::frame(&[0x11; rns_core::constants::HEADER_MINSIZE]))
            .unwrap();
        let payload: Vec<u8> = (0..32).collect();
        client.write_all(&hdlc::frame(&payload)).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame {
                interface_id,
                data,
                rssi: _,
                snr: _,
            } => {
                assert_eq!(interface_id, InterfaceId(8100));
                assert_eq!(data, payload);
            }
            other => panic!("expected Frame, got {:?}", other),
        }
    }

    #[test]
    fn backbone_send_to_client() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8200));

        let config = make_server_config(port, 82, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client
            .set_read_timeout(Some(Duration::from_secs(2)))
            .unwrap();

        // Get writer from InterfaceUp
        let event = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
        let mut writer = match event {
            Event::InterfaceUp(_, Some(w), _) => w,
            other => panic!("expected InterfaceUp with writer, got {:?}", other),
        };

        // Send frame via writer
        let payload: Vec<u8> = (0..24).collect();
        writer.send_frame(&payload).unwrap();

        // Read from client
        let mut buf = [0u8; 256];
        let n = client.read(&mut buf).unwrap();
        let expected = hdlc::frame(&payload);
        assert_eq!(&buf[..n], &expected[..]);
    }

    #[test]
    fn backbone_write_backpressure_does_not_starve_reads() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8250));
        let config = make_server_config(port, 82, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
        SockRef::from(&client).set_recv_buffer_size(4096).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
        let mut writer = match event {
            Event::InterfaceUp(_, Some(writer), _) => writer,
            other => panic!("expected InterfaceUp with writer, got {other:?}"),
        };

        let outbound = vec![0x55; 60 * 1024];
        let deadline = Instant::now() + Duration::from_secs(3);
        let mut backpressured = false;
        while Instant::now() < deadline {
            match writer.send_frame(&outbound) {
                Ok(()) => {}
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                    backpressured = true;
                    break;
                }
                Err(error) => panic!("unexpected backbone write error: {error}"),
            }
        }
        assert!(
            backpressured,
            "failed to create outbound write backpressure"
        );

        let inbound: Vec<u8> = (0..32).collect();
        client.write_all(&hdlc::frame(&inbound)).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame { data, .. } => assert_eq!(data, inbound),
            other => panic!("inbound frame was starved by outbound backpressure: {other:?}"),
        }
    }

    #[test]
    fn backbone_multiple_clients() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8300));

        let config = make_server_config(port, 83, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        let mut ids = Vec::new();
        for _ in 0..2 {
            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
            match event {
                Event::InterfaceUp(id, _, _) => ids.push(id),
                other => panic!("expected InterfaceUp, got {:?}", other),
            }
        }

        assert_eq!(ids.len(), 2);
        assert_ne!(ids[0], ids[1]);
    }

    #[test]
    fn backbone_client_disconnect() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8400));

        let config = make_server_config(port, 84, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        // Drain InterfaceUp
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        // Disconnect
        drop(client);

        // Should receive InterfaceDown
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(
            matches!(event, Event::InterfaceDown(InterfaceId(8400))),
            "expected InterfaceDown(8400), got {:?}",
            event
        );
    }

    #[test]
    fn backbone_epoll_multiplexing() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8500));

        let config = make_server_config(port, 85, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let mut client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        // Drain both InterfaceUp events
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        // Both clients send data simultaneously
        let payload1: Vec<u8> = (0..24).collect();
        let payload2: Vec<u8> = (100..130).collect();
        client1.write_all(&hdlc::frame(&payload1)).unwrap();
        client2.write_all(&hdlc::frame(&payload2)).unwrap();

        // Should receive both Frame events
        let mut received = Vec::new();
        for _ in 0..2 {
            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
            match event {
                Event::Frame { data, .. } => received.push(data),
                other => panic!("expected Frame, got {:?}", other),
            }
        }
        assert!(received.contains(&payload1));
        assert!(received.contains(&payload2));
    }

    #[test]
    fn backbone_bind_port() {
        let port = find_free_port();
        let (tx, _rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8600));

        let config = make_server_config(port, 86, None, None, None, BackboneAbuseConfig::default());

        // Should not error
        start(config, tx, next_id).unwrap();
    }

    #[test]
    fn backbone_hdlc_fragmented() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8700));

        let config = make_server_config(port, 87, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client.set_nodelay(true).unwrap();

        // Drain InterfaceUp
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        // Send HDLC frame in two fragments
        let payload: Vec<u8> = (0..32).collect();
        let framed = hdlc::frame(&payload);
        let mid = framed.len() / 2;

        client.write_all(&framed[..mid]).unwrap();
        thread::sleep(Duration::from_millis(50));
        client.write_all(&framed[mid..]).unwrap();

        // Should receive reassembled frame
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame { data, .. } => {
                assert_eq!(data, payload);
            }
            other => panic!("expected Frame, got {:?}", other),
        }
    }

    // -----------------------------------------------------------------------
    // Client mode tests
    // -----------------------------------------------------------------------

    fn make_client_config(port: u16, id: u64) -> BackboneClientConfig {
        BackboneClientConfig {
            name: format!("test-bb-client-{}", port),
            target_host: "127.0.0.1".into(),
            target_port: port,
            interface_id: InterfaceId(id),
            reconnect_wait: Duration::from_millis(100),
            max_reconnect_tries: Some(2),
            connect_timeout: Duration::from_secs(2),
            transport_identity: None,
            runtime: Arc::new(Mutex::new(BackboneClientRuntime {
                reconnect_wait: Duration::from_millis(100),
                max_reconnect_tries: Some(2),
                connect_timeout: Duration::from_secs(2),
            })),
        }
    }

    #[test]
    fn backbone_client_connect() {
        let port = find_free_port();
        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
        let (tx, rx) = crate::event::channel();

        let config = make_client_config(port, 9000);
        let _writer = start_client(config, tx).unwrap();

        let _server_stream = listener.accept().unwrap();

        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceUp(InterfaceId(9000), _, _)));
    }

    #[test]
    fn backbone_client_receive_frame() {
        let port = find_free_port();
        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
        let (tx, rx) = crate::event::channel();

        let config = make_client_config(port, 9100);
        let _writer = start_client(config, tx).unwrap();

        let (mut server_stream, _) = listener.accept().unwrap();

        // Drain InterfaceUp
        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();

        // Client mode applies the same strict lower bound and recovers.
        server_stream
            .write_all(&hdlc::frame(&[0x11; rns_core::constants::HEADER_MINSIZE]))
            .unwrap();
        let payload: Vec<u8> = (0..32).collect();
        server_stream.write_all(&hdlc::frame(&payload)).unwrap();

        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame {
                interface_id,
                data,
                rssi: _,
                snr: _,
            } => {
                assert_eq!(interface_id, InterfaceId(9100));
                assert_eq!(data, payload);
            }
            other => panic!("expected Frame, got {:?}", other),
        }
    }

    #[test]
    fn backbone_client_send_frame() {
        let port = find_free_port();
        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
        let (tx, _rx) = crate::event::channel();

        let config = make_client_config(port, 9200);
        let mut writer = start_client(config, tx).unwrap();

        let (mut server_stream, _) = listener.accept().unwrap();
        server_stream
            .set_read_timeout(Some(Duration::from_secs(2)))
            .unwrap();

        let payload: Vec<u8> = (0..24).collect();
        writer.send_frame(&payload).unwrap();

        let mut buf = [0u8; 256];
        let n = server_stream.read(&mut buf).unwrap();
        let expected = hdlc::frame(&payload);
        assert_eq!(&buf[..n], &expected[..]);
    }

    #[test]
    fn backbone_max_connections_rejects_excess() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8800));

        let config = make_server_config(
            port,
            88,
            Some(2),
            None,
            None,
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        // Connect two clients (at limit)
        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        // Drain both InterfaceUp events
        for _ in 0..2 {
            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
            assert!(matches!(event, Event::InterfaceUp(_, _, _)));
        }

        // Third connection should be accepted at TCP level but immediately dropped
        let client3 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client3
            .set_read_timeout(Some(Duration::from_millis(500)))
            .unwrap();

        // Give server time to reject
        thread::sleep(Duration::from_millis(100));

        // Should NOT receive a third InterfaceUp
        let result = recv_non_peer_event(&rx, Duration::from_millis(500));
        assert!(
            result.is_err(),
            "expected no InterfaceUp for rejected connection, got {:?}",
            result
        );
    }

    #[test]
    fn backbone_max_connections_allows_after_disconnect() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(8900));

        let config = make_server_config(
            port,
            89,
            Some(1),
            None,
            None,
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        // Connect first client
        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceUp(_, _, _)));

        // Disconnect first client
        drop(client1);

        // Wait for InterfaceDown
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(_)));

        // Now a new connection should be accepted
        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(
            matches!(event, Event::InterfaceUp(_, _, _)),
            "expected InterfaceUp after slot freed, got {:?}",
            event
        );
    }

    #[test]
    fn backbone_client_reconnect() {
        let port = find_free_port();
        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
        listener.set_nonblocking(false).unwrap();
        let (tx, rx) = crate::event::channel();

        let config = make_client_config(port, 9300);
        let _writer = start_client(config, tx).unwrap();

        // Accept first connection and immediately close it
        let (server_stream, _) = listener.accept().unwrap();

        // Drain InterfaceUp
        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();

        drop(server_stream);

        // Should get InterfaceDown
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9300))));

        // Accept the reconnection
        let _server_stream2 = listener.accept().unwrap();

        // Should get InterfaceUp again
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceUp(InterfaceId(9300), _, _)));
    }

    #[test]
    fn a_reconnected_backbone_client_is_read_on_the_receiving_thread() {
        let port = find_free_port();
        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
        let (tx, rx) = crate::event::channel();
        let _writer = start_client(make_client_config(port, 9310), tx).unwrap();
        let (first, _) = listener.accept().unwrap();
        assert!(matches!(
            rx.recv_timeout(Duration::from_secs(1)).unwrap(),
            Event::InterfaceUp(InterfaceId(9310), None, _)
        ));
        drop(first);
        assert!(matches!(
            recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap(),
            Event::InterfaceDown(InterfaceId(9310))
        ));
        let (mut second, _) = listener.accept().unwrap();
        assert!(matches!(
            recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap(),
            Event::InterfaceUp(InterfaceId(9310), Some(_), _)
        ));
        let payload: Vec<u8> = (0..32).collect();
        second.write_all(&hdlc::frame(&payload)).unwrap();
        match recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap() {
            Event::Frame { data, .. } => assert_eq!(data, payload),
            other => panic!("expected Frame, got {:?}", other),
        }
        // The reconnect thread exits once the new socket is served here.
        let reconnecting = || {
            std::fs::read_dir("/proc/self/task")
                .unwrap()
                .filter_map(|task| std::fs::read_to_string(task.ok()?.path().join("comm")).ok())
                .any(|name| name.trim_end() == "bb-reconn-9310")
        };
        let deadline = Instant::now() + Duration::from_secs(2);
        while reconnecting() && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(10));
        }
        assert!(!reconnecting());
    }

    #[test]
    fn backbone_client_keeps_reconnecting_until_server_returns() {
        // The remote node restarts: its listener disappears for several
        // reconnect attempts, then comes back on the same port. Without
        // max_reconnect_tries the client must not give up.
        let port = find_free_port();
        let listener = TcpListener::bind(("127.0.0.1", port)).unwrap();
        let (tx, rx) = crate::event::channel();

        let mut config = make_client_config(port, 9360);
        config.reconnect_wait = Duration::from_millis(20);
        config.max_reconnect_tries = None;
        config.connect_timeout = Duration::from_millis(200);
        config.runtime = Arc::new(Mutex::new(BackboneClientRuntime::from_config(&config)));
        let _writer = start_client(config, tx).unwrap();

        let (server_stream, _) = listener.accept().unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
        assert!(matches!(
            event,
            Event::InterfaceUp(InterfaceId(9360), None, _)
        ));

        // Server goes away: connection and listener both close.
        drop(server_stream);
        drop(listener);
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9360))));

        // Several attempts (backoff 20, 40, 80, 160 ms ...) fail meanwhile.
        thread::sleep(Duration::from_millis(400));
        if let Ok(event) = recv_non_peer_event(&rx, Duration::from_millis(10)) {
            panic!("unexpected event while the server is down: {:?}", event);
        }

        let listener = TcpListener::bind(("127.0.0.1", port)).unwrap();
        let (_server_stream, _) = listener.accept().unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(5)).unwrap();
        assert!(
            matches!(event, Event::InterfaceUp(InterfaceId(9360), Some(_), _)),
            "expected reconnect InterfaceUp, got {:?}",
            event
        );
    }

    #[test]
    fn backbone_client_without_reconnect_tries_leaves_reconnecting_to_owner() {
        let port = find_free_port();
        let listener = TcpListener::bind(("127.0.0.1", port)).unwrap();
        let (tx, rx) = crate::event::channel();

        let mut config = make_client_config(port, 9370);
        config.reconnect_wait = Duration::from_millis(20);
        config.max_reconnect_tries = Some(0);
        config.runtime = Arc::new(Mutex::new(BackboneClientRuntime::from_config(&config)));
        let _writer = start_client(config, tx).unwrap();

        let (server_stream, _) = listener.accept().unwrap();
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
        drop(server_stream);

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9370))));
        // No reconnect attempt and no second, late InterfaceDown that could
        // tear down a connection the owner has since re-established.
        listener.set_nonblocking(true).unwrap();
        if let Ok(event) = recv_non_peer_event(&rx, Duration::from_millis(300)) {
            panic!("unexpected event after the connection closed: {:?}", event);
        }
        assert!(listener.accept().is_err());
    }

    #[test]
    fn backbone_client_stop_prevents_reconnect() {
        let port = find_free_port();
        let listener = TcpListener::bind(("127.0.0.1", port)).unwrap();
        let (tx, rx) = crate::event::channel();

        let config = make_client_config(port, 9350);
        let control = ListenerControl::new();
        let _writer = start_client_with_ifac(config, tx, 0, None, control.clone()).unwrap();

        let (server_stream, _) = listener.accept().unwrap();
        // Drain the initial InterfaceUp from the successful connect.
        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();

        control.request_stop();
        drop(server_stream);

        // A detached client must not emit InterfaceDown or reconnect.
        if let Ok(event) = recv_non_peer_event(&rx, Duration::from_millis(700)) {
            panic!("unexpected event after stop: {:?}", event);
        }
    }

    #[test]
    fn backbone_idle_timeout_disconnects_silent_client() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9400));

        let config = make_server_config(
            port,
            94,
            None,
            Some(Duration::from_millis(150)),
            None,
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let client_id = match event {
            Event::InterfaceUp(id, _, _) => id,
            other => panic!("expected InterfaceUp, got {:?}", other),
        };

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
    }

    #[test]
    fn backbone_idle_timeout_ignores_client_after_data() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9500));

        let config = make_server_config(
            port,
            95,
            None,
            Some(Duration::from_millis(200)),
            None,
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let client_id = match event {
            Event::InterfaceUp(id, _, _) => id,
            other => panic!("expected InterfaceUp, got {:?}", other),
        };

        client.write_all(&hdlc::frame(&[1u8; 24])).unwrap();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame {
                interface_id,
                data,
                rssi: _,
                snr: _,
            } => {
                assert_eq!(interface_id, client_id);
                assert_eq!(data, vec![1u8; 24]);
            }
            other => panic!("expected Frame, got {:?}", other),
        }

        let result = recv_non_peer_event(&rx, Duration::from_millis(500));
        assert!(
            result.is_err(),
            "expected no InterfaceDown after client sent data, got {:?}",
            result
        );
    }

    #[test]
    fn backbone_runtime_idle_timeout_updates_live() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9650));

        let config = make_server_config(port, 97, None, None, None, BackboneAbuseConfig::default());
        let runtime = Arc::clone(&config.runtime);

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let client_id = match event {
            Event::InterfaceUp(id, _, _) => id,
            other => panic!("expected InterfaceUp, got {:?}", other),
        };

        {
            let mut runtime = runtime.lock().unwrap();
            runtime.idle_timeout = Some(Duration::from_millis(150));
        }

        let event = recv_non_peer_event(&rx, Duration::from_secs(4)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
    }

    #[test]
    fn backbone_write_stall_timeout_disconnects_unwritable_client() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9660));

        let config = make_server_config(
            port,
            98,
            None,
            None,
            Some(Duration::from_millis(50)),
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client
            .set_read_timeout(Some(Duration::from_millis(100)))
            .unwrap();
        let sock = SockRef::from(&client);
        sock.set_recv_buffer_size(4096).ok();

        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let (client_id, mut writer) = match event {
            Event::InterfaceUp(id, Some(writer), _) => (id, writer),
            other => panic!("expected InterfaceUp with writer, got {:?}", other),
        };

        let payload = vec![0x55; 512 * 1024];
        let deadline = Instant::now() + Duration::from_secs(3);
        let mut stalled = false;
        while Instant::now() < deadline {
            match writer.send_frame(&payload) {
                Ok(()) => {}
                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                    thread::sleep(Duration::from_millis(10));
                }
                Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
                    stalled = true;
                    break;
                }
                Err(e) => panic!("unexpected send error: {}", e),
            }
        }

        assert!(stalled, "expected writer to time out on persistent stall");
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
    }

    #[test]
    fn shutting_down_backbone_writer_closes_the_client_socket() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9670));
        let config = make_server_config(port, 99, None, None, None, BackboneAbuseConfig::default());

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client
            .set_read_timeout(Some(Duration::from_secs(2)))
            .unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let mut writer = match event {
            Event::InterfaceUp(_, Some(writer), _) => writer,
            other => panic!("expected InterfaceUp with writer, got {:?}", other),
        };

        writer.shutdown();

        let mut byte = [0u8; 1];
        assert_eq!(client.read(&mut byte).unwrap(), 0);
    }

    /// Drain events matching a predicate, return the first match.
    fn wait_for<F>(
        rx: &crate::event::EventReceiver,
        timeout: Duration,
        mut pred: F,
    ) -> Option<Event>
    where
        F: FnMut(&Event) -> bool,
    {
        let deadline = Instant::now() + timeout;
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return None;
            }
            match rx.recv_timeout(remaining) {
                Ok(event) if pred(&event) => return Some(event),
                Ok(_) => continue,
                Err(_) => return None,
            }
        }
    }

    #[test]
    fn backbone_write_stall_emits_peer_events() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9700));

        let config = make_server_config(
            port,
            97,
            None,
            None,
            Some(Duration::from_millis(50)), // 50ms stall timeout
            BackboneAbuseConfig::default(),
        );

        start(config, tx, next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        // Connect a client that won't read (will cause write stall)
        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        client
            .set_read_timeout(Some(Duration::from_millis(100)))
            .unwrap();
        let sock = SockRef::from(&client);
        sock.set_recv_buffer_size(4096).ok();

        // Wait for InterfaceUp and grab writer
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        let mut writer = match event {
            Event::InterfaceUp(_, Some(w), _) => w,
            other => panic!("expected InterfaceUp with writer, got {:?}", other),
        };

        // Flood until stall
        let payload = vec![0x55; 512 * 1024];
        let deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() < deadline {
            match writer.send_frame(&payload) {
                Ok(()) | Err(_) => {
                    if Instant::now() + Duration::from_millis(10) > deadline {
                        break;
                    }
                    thread::sleep(Duration::from_millis(10));
                }
            }
        }

        // Should see BackbonePeerWriteStall event
        let stall_event = wait_for(&rx, Duration::from_secs(3), |e| {
            matches!(e, Event::BackbonePeerWriteStall { .. })
        });
        assert!(
            stall_event.is_some(),
            "expected BackbonePeerWriteStall event"
        );
    }

    #[test]
    fn backbone_blacklisted_peer_rejected_on_connect() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9800));

        let config = make_server_config(port, 98, None, None, None, BackboneAbuseConfig::default());
        let peer_state = config.peer_state.clone();

        start(config, tx.clone(), next_id).unwrap();
        thread::sleep(Duration::from_millis(50));

        // First connection should succeed
        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
        assert!(
            matches!(event, Event::InterfaceUp(_, _, _)),
            "first connection should succeed"
        );
        drop(client1);

        // Drain disconnect events
        thread::sleep(Duration::from_millis(100));
        while rx.try_recv().is_ok() {}

        // Blacklist 127.0.0.1 via the peer monitor
        peer_state.lock().unwrap().blacklist(
            "127.0.0.1".parse().unwrap(),
            Duration::from_secs(60),
            "test blacklist".into(),
        );

        // Second connection from same IP should be rejected (no InterfaceUp)
        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
        // Give poll loop time to reject
        thread::sleep(Duration::from_millis(200));

        // Should NOT get an InterfaceUp — connection should have been rejected
        let event = rx.try_recv();
        if let Ok(Event::InterfaceUp(_, _, _)) = event {
            panic!("blacklisted peer should not get InterfaceUp")
        }
    }

    #[test]
    fn backbone_six_fast_flaps_reject_before_allocating_seventh_interface() {
        let port = find_free_port();
        let (tx, rx) = crate::event::channel();
        let next_id = Arc::new(AtomicU64::new(9900));
        let mut config =
            make_server_config(port, 99, None, None, None, BackboneAbuseConfig::default());
        config.fast_flap = test_fast_flap_config();
        let fast_flap_state = Arc::clone(&config.fast_flap_state);
        let peer_state = Arc::clone(&config.peer_state);

        start(config, tx, Arc::clone(&next_id)).unwrap();
        thread::sleep(Duration::from_millis(50));

        for flap in 0..6 {
            let mut client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
            let up = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
            assert!(matches!(up, Event::InterfaceUp(_, _, _)));

            // Receiving traffic must not exempt a short-lived connection.
            if flap == 0 {
                let payload = vec![0x41; 32];
                client.write_all(&hdlc::frame(&payload)).unwrap();
                let frame = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
                assert!(matches!(frame, Event::Frame { data, .. } if data == payload));
            }

            drop(client);
            let down = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
            assert!(matches!(down, Event::InterfaceDown(_)));
        }

        let peer_ip: IpAddr = "127.0.0.1".parse().unwrap();
        assert!(
            lock_or_recover(&fast_flap_state, "test fast-flap state").is_blocked(
                peer_ip,
                &test_fast_flap_config(),
                Instant::now(),
            )
        );
        assert_eq!(next_id.load(Ordering::Relaxed), 9906);
        let entries = lock_or_recover(&peer_state, "test peer state").list("test-backbone");
        let entry = entries
            .iter()
            .find(|entry| entry.peer_ip == peer_ip)
            .expect("fast-flapping peer state");
        assert_eq!(entry.blacklist_reason.as_deref(), Some("fast flapping"));
        assert!(entry.blacklisted_remaining_secs.is_some());

        let _rejected = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
        serve_for(&rx, Duration::from_millis(200));
        assert_eq!(
            next_id.load(Ordering::Relaxed),
            9906,
            "rejection must happen before dynamic interface allocation"
        );
        while let Ok(event) = rx.try_recv() {
            assert!(
                !matches!(event, Event::InterfaceUp(_, _, _)),
                "blocked peer must not register a dynamic interface"
            );
        }
        let entries = lock_or_recover(&peer_state, "test peer state").list("test-backbone");
        assert_eq!(
            entries
                .iter()
                .find(|entry| entry.peer_ip == peer_ip)
                .unwrap()
                .reject_count,
            1
        );
    }

    #[test]
    fn backbone_parse_config_reads_abuse_settings() {
        let factory = BackboneInterfaceFactory;
        let mut params = HashMap::new();
        params.insert("listen_ip".into(), "127.0.0.1".into());
        params.insert("listen_port".into(), "4242".into());
        params.insert("idle_timeout".into(), "15".into());
        params.insert("write_stall_timeout".into(), "45".into());
        params.insert("max_penalty_duration".into(), "3600".into());

        let config = factory
            .parse_config("test-backbone", InterfaceId(97), &params)
            .unwrap();
        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();

        match mode {
            BackboneMode::Server(config) => {
                assert_eq!(config.listen_ip, "127.0.0.1");
                assert_eq!(config.listen_port, 4242);
                assert_eq!(config.idle_timeout, Some(Duration::from_secs(15)));
                assert_eq!(config.write_stall_timeout, Some(Duration::from_secs(45)));
                assert_eq!(
                    config.abuse.max_penalty_duration,
                    Some(Duration::from_secs(3600))
                );
            }
            BackboneMode::Client(_, _) => panic!("expected server config"),
        }
    }

    #[test]
    fn backbone_parse_config_reads_fast_flapping_settings() {
        let factory = BackboneInterfaceFactory;
        let mut params = HashMap::new();
        params.insert("listen_ip".into(), "127.0.0.1".into());
        params.insert("listen_port".into(), "4242".into());
        params.insert("block_fast_flapping".into(), "no".into());
        params.insert("fast_flapping_threshold".into(), "2.5".into());
        params.insert("fast_flapping_grace".into(), "9".into());
        params.insert("fast_flapping_block_time".into(), "1.5".into());

        let config = factory
            .parse_config("test-backbone", InterfaceId(93), &params)
            .unwrap();
        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();
        let BackboneMode::Server(config) = mode else {
            panic!("expected server config");
        };

        assert!(!config.fast_flap.enabled);
        assert_eq!(
            config.fast_flap.connection_threshold,
            Duration::from_secs_f64(2.5)
        );
        assert_eq!(config.fast_flap.grace, 9);
        assert_eq!(config.fast_flap.block_duration, Duration::from_secs(90));
    }

    #[test]
    fn backbone_parse_config_rejects_invalid_fast_flapping_settings() {
        let factory = BackboneInterfaceFactory;
        for (key, value) in [
            ("block_fast_flapping", "perhaps"),
            ("fast_flapping_threshold", "-1"),
            ("fast_flapping_threshold", "NaN"),
            ("fast_flapping_grace", "-1"),
            ("fast_flapping_block_time", "infinite"),
        ] {
            let mut params = HashMap::new();
            params.insert("listen_port".into(), "4242".into());
            params.insert(key.into(), value.into());

            let err = match factory.parse_config("test-backbone", InterfaceId(92), &params) {
                Ok(_) => panic!("{key}={value} should be rejected"),
                Err(err) => err,
            };
            assert!(err.contains(key), "unexpected error: {err}");
        }
    }

    #[test]
    fn parsed_backbone_listeners_share_fast_flap_history() {
        let factory = BackboneInterfaceFactory;
        let mut params = HashMap::new();
        params.insert("listen_port".into(), "4242".into());

        let first = factory
            .parse_config("first", InterfaceId(90), &params)
            .unwrap();
        let second = factory
            .parse_config("second", InterfaceId(91), &params)
            .unwrap();
        let BackboneMode::Server(first) = *first.into_any().downcast::<BackboneMode>().unwrap()
        else {
            panic!("expected server config");
        };
        let BackboneMode::Server(second) = *second.into_any().downcast::<BackboneMode>().unwrap()
        else {
            panic!("expected server config");
        };

        assert!(Arc::ptr_eq(&first.fast_flap_state, &second.fast_flap_state));
    }

    #[test]
    fn backbone_parse_config_reads_client_priority() {
        let factory = BackboneInterfaceFactory;
        let mut params = HashMap::new();
        params.insert("remote".into(), "example.com".into());
        params.insert("target_port".into(), "4242".into());
        params.insert("priority".into(), "87".into());

        let config = factory
            .parse_config("test-backbone-client", InterfaceId(96), &params)
            .unwrap();
        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();

        match mode {
            BackboneMode::Client(_, priority) => assert_eq!(priority, 87),
            BackboneMode::Server(_) => panic!("expected client config"),
        }
    }

    #[test]
    fn backbone_parse_config_reads_client_max_reconnect_tries() {
        let parse = |value: Option<&str>| {
            let mut params = HashMap::new();
            params.insert("remote".into(), "127.0.0.1".into());
            if let Some(value) = value {
                params.insert("max_reconnect_tries".into(), value.into());
            }
            BackboneInterfaceFactory
                .parse_config("peer", InterfaceId(1), &params)
                .map(|config| {
                    let mode = config.into_any().downcast::<BackboneMode>().unwrap();
                    match *mode {
                        BackboneMode::Client(client, _) => client,
                        BackboneMode::Server(_) => panic!("expected client mode"),
                    }
                })
        };

        let client = parse(None).unwrap();
        assert_eq!(client.max_reconnect_tries, None);
        assert_eq!(client.runtime.lock().unwrap().max_reconnect_tries, None);

        let client = parse(Some("7")).unwrap();
        assert_eq!(client.max_reconnect_tries, Some(7));
        assert_eq!(client.runtime.lock().unwrap().max_reconnect_tries, Some(7));

        assert!(parse(Some("-1")).is_err());
        assert!(parse(Some("forever")).is_err());
    }

    #[test]
    fn backbone_parse_config_defaults_client_priority() {
        let factory = BackboneInterfaceFactory;
        let mut params = HashMap::new();
        params.insert("remote".into(), "example.com".into());

        let config = factory
            .parse_config("test-backbone-client", InterfaceId(95), &params)
            .unwrap();
        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();

        match mode {
            BackboneMode::Client(_, priority) => assert_eq!(priority, 60),
            BackboneMode::Server(_) => panic!("expected client config"),
        }
    }

    #[test]
    fn backbone_parse_config_rejects_invalid_client_priority() {
        let factory = BackboneInterfaceFactory;
        for value in ["-1", "101", "fast"] {
            let mut params = HashMap::new();
            params.insert("remote".into(), "example.com".into());
            params.insert("priority".into(), value.into());

            let err = match factory.parse_config("test-backbone-client", InterfaceId(94), &params) {
                Ok(_) => panic!("priority {value} should be rejected"),
                Err(err) => err,
            };
            assert!(
                err.contains("priority"),
                "unexpected error for {value}: {err}"
            );
        }
    }

    #[test]
    fn positive_duration_param_rejects_values_that_are_not_durations() {
        for raw in ["inf", "-inf", "NaN", "1e300", "-1", "0"] {
            let params = HashMap::from([("idle_timeout".to_string(), raw.to_string())]);
            assert_eq!(
                parse_positive_duration_secs(&params, "idle_timeout"),
                None,
                "value {raw} should be ignored"
            );
        }
        let params = HashMap::from([("idle_timeout".to_string(), "2.5".to_string())]);
        assert_eq!(
            parse_positive_duration_secs(&params, "idle_timeout"),
            Some(Duration::from_millis(2500))
        );
    }
}