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};
#[cfg(test)]
use std::io::Write;
use std::io::{self, Read};
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 polling::{Event as PollEvent, Events, Poller};
use socket2::{SockRef, TcpKeepalive};

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

use crate::event::{Event, EventSender};
use crate::hdlc;
use crate::interface::transmit_buffer::{
    EgressController, TransmitBuffer, EGRESS_DEAD_TIME, EGRESS_HIGH_WATERMARK,
};
use crate::interface::{
    lock_or_recover, poll_readable, 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;

const fn auto_mtu_for_bitrate(bitrate: u64) -> Option<u32> {
    if bitrate >= 1_000_000_000 {
        Some(524_288)
    } else if bitrate >= 750_000_000 {
        Some(262_144)
    } else if bitrate >= 400_000_000 {
        Some(131_072)
    } else if bitrate >= 200_000_000 {
        Some(65_536)
    } else if bitrate >= 100_000_000 {
        Some(32_768)
    } else if bitrate >= 10_000_000 {
        Some(16_384)
    } else if bitrate >= 5_000_000 {
        Some(8_192)
    } else if bitrate >= 2_000_000 {
        Some(4_096)
    } else if bitrate >= 1_000_000 {
        Some(2_048)
    } else if bitrate >= 62_500 {
        Some(1_024)
    } else {
        None
    }
}

/// 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(&[data.to_vec()])
    }

    fn send_frames(&mut self, frames: &[Vec<u8>]) -> 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 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 TCP listener, spawns poll 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 name = config.name.clone();
    let server_interface_id = config.interface_id;
    let runtime = Arc::clone(&config.runtime);
    let peer_state = Arc::clone(&config.peer_state);
    let fast_flap = config.fast_flap;
    let fast_flap_state = Arc::clone(&config.fast_flap_state);
    let ingress_control = config.ingress_control;
    let accepted_peer_mode = config.mode;
    let accepted_peer_gravity = config.gravity;
    let accepted_peer_recursive_prs = config.recursive_prs;
    let accepted_peer_announces_from_internal = config.announces_from_internal;
    let accepted_peer_announces_to_internal = config.announces_to_internal;
    let ifac_size = dynamic_template
        .as_ref()
        .and_then(|template| template.ifac.as_ref())
        .map(|ifac| ifac.size)
        .unwrap_or(0);
    thread::Builder::new()
        .name(format!("backbone-poll-{}", config.interface_id.0))
        .spawn(move || {
            if let Err(e) = poll_loop(PollLoopContext {
                listener,
                name,
                server_interface_id,
                tx,
                next_id,
                runtime,
                peer_state,
                fast_flap,
                fast_flap_state,
                ingress_control,
                accepted_peer_mode,
                accepted_peer_gravity,
                accepted_peer_recursive_prs,
                accepted_peer_announces_from_internal,
                accepted_peer_announces_to_internal,
                dynamic_template,
                ifac_size,
                underlay_mark,
            }) {
                log::error!("backbone poll loop error: {}", e);
            }
        })?;

    Ok(())
}

/// Per-client tracking state.
struct ClientState {
    id: InterfaceId,
    peer_ip: IpAddr,
    peer_port: u16,
    stream: TcpStream,
    decoder: hdlc::Decoder,
    connected_at: Instant,
    has_received_data: bool,
    write_stall_flag: Arc<AtomicBool>,
    dp_ingress_bytes: u64,
    dp_ingress_packets: u64,
    dp_ingress_hold: Option<Instant>,
    dp_ingress_gated: bool,
    dp_ingress_throttle_count: u64,
}

#[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)
}

fn throttle_busiest_ingress(
    poller: &Poller,
    clients: &mut HashMap<usize, ClientState>,
    sample_span: Duration,
) -> io::Result<Option<InterfaceId>> {
    let Some(key) = busiest_ingress_key(
        clients
            .iter()
            .map(|(key, client)| (*key, client.dp_ingress_packets)),
    ) else {
        return Ok(None);
    };
    if clients
        .get(&key)
        .is_some_and(|client| client.dp_ingress_gated)
    {
        return Ok(None);
    }

    let allocation_divisor = clients.len().max(DP_IC_IF_HEADROOM) as f64 * DP_IC_PENALTY;
    let hold = Duration::from_secs_f64(
        (sample_span.as_secs_f64().max(0.001) * allocation_divisor).max(0.001),
    );
    let client = clients.get_mut(&key).expect("selected client disappeared");
    poller.modify(&client.stream, PollEvent::none(key))?;
    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(),
    );
    Ok(Some(client.id))
}

fn release_one_ingress(
    poller: &Poller,
    clients: &mut HashMap<usize, ClientState>,
    now: Instant,
) -> io::Result<Option<InterfaceId>> {
    let Some(key) = clients
        .iter()
        .find(|(_, client)| {
            client.dp_ingress_gated
                && client
                    .dp_ingress_hold
                    .is_none_or(|hold_until| now >= hold_until)
        })
        .map(|(key, _)| *key)
    else {
        return Ok(None);
    };
    let client = clients.get_mut(&key).expect("selected client disappeared");
    poller.modify(&client.stream, PollEvent::readable(key))?;
    client.dp_ingress_gated = false;
    client.dp_ingress_hold = None;
    log::info!(
        "released ingress throttle on BackboneInterface/{}",
        client.id.0
    );
    Ok(Some(client.id))
}

#[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,
}

/// Main poll event loop.
struct PollLoopContext {
    listener: TcpListener,
    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>,
}

fn poll_loop(context: PollLoopContext) -> io::Result<()> {
    let PollLoopContext {
        listener,
        name,
        server_interface_id,
        tx,
        next_id,
        runtime,
        peer_state,
        fast_flap,
        fast_flap_state,
        ingress_control,
        accepted_peer_mode,
        accepted_peer_gravity,
        accepted_peer_recursive_prs,
        accepted_peer_announces_from_internal,
        accepted_peer_announces_to_internal,
        dynamic_template,
        ifac_size,
        underlay_mark,
    } = context;
    // 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 poller = Poller::new()?;

    const LISTENER_KEY: usize = 0;

    // SAFETY: listener outlives its registration in the poller.
    unsafe { poller.add(&listener, PollEvent::readable(LISTENER_KEY))? };

    let mut clients: HashMap<usize, ClientState> = HashMap::new();
    let mut peers: HashMap<IpAddr, PeerBehaviorState> = HashMap::new();
    let mut events = Events::new();
    let mut next_key: usize = 1;
    let mut ingress_snapshot_at = Instant::now();

    loop {
        let runtime_snapshot = runtime.lock().unwrap().clone();
        let max_connections = runtime_snapshot.max_connections;
        let idle_timeout = runtime_snapshot.idle_timeout;
        {
            let mut monitor = peer_state.lock().unwrap();
            monitor.sync_into(&mut peers);
            cleanup_peer_state(&mut peers);
            monitor.upsert_snapshot(&peers);
        }

        events.clear();
        poller.wait(&mut events, Some(DP_IC_INTERVAL))?;

        for ev in events.iter() {
            if ev.key == LISTENER_KEY {
                // Accept new connections
                loop {
                    match listener.accept() {
                        Ok((stream, peer_addr)) => {
                            if let Err(error) =
                                super::apply_underlay_mark(stream.as_raw_fd(), underlay_mark)
                            {
                                log::error!(
                                    "[{}] failed to mark accepted underlay socket: {}",
                                    name,
                                    error
                                );
                                drop(stream);
                                continue;
                            }
                            let peer_ip = peer_addr.ip();
                            let peer_port = peer_addr.port();

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

                            if let Some(max) = max_connections {
                                if clients.len() >= max {
                                    log::warn!(
                                        "[{}] max connections ({}) reached, rejecting {}",
                                        name,
                                        max,
                                        peer_addr
                                    );
                                    drop(stream);
                                    continue;
                                }
                            }

                            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();
                            }

                            let key = next_key;
                            next_key += 1;
                            let client_id = InterfaceId(next_id.fetch_add(1, Ordering::Relaxed));

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

                            // Register client with poller
                            // SAFETY: stream is stored in ClientState and outlives registration.
                            if let Err(e) = unsafe { poller.add(&stream, PollEvent::readable(key)) }
                            {
                                log::warn!("[{}] failed to add client to poller: {}", name, e);
                                continue; // stream drops, closing socket
                            }

                            // 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);
                                    let _ = poller.delete(&stream);
                                    continue; // stream drops
                                }
                            };
                            let write_stall_flag = Arc::new(AtomicBool::new(false));
                            let writer: Box<dyn Writer> = Box::new(BackboneWriter {
                                stream: writer_stream,
                                runtime: Arc::clone(&runtime),
                                interface_name: name.clone(),
                                interface_id: client_id,
                                event_tx: 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),
                            });

                            clients.insert(
                                key,
                                ClientState {
                                    id: client_id,
                                    peer_ip,
                                    peer_port,
                                    stream,
                                    decoder: hdlc::Decoder::reticulum(HW_MTU, ifac_size),
                                    connected_at: Instant::now(),
                                    has_received_data: false,
                                    write_stall_flag,
                                    dp_ingress_bytes: 0,
                                    dp_ingress_packets: 0,
                                    dp_ingress_hold: None,
                                    dp_ingress_gated: false,
                                    dp_ingress_throttle_count: 0,
                                },
                            );
                            peers
                                .entry(peer_ip)
                                .or_insert_with(PeerBehaviorState::new)
                                .connected_count += 1;
                            peer_state.lock().unwrap().upsert_snapshot(&peers);
                            let _ = tx.send(Event::BackbonePeerConnected {
                                server_interface_id,
                                peer_interface_id: client_id,
                                peer_ip,
                                peer_port,
                            });

                            let info = InterfaceInfo {
                                id: client_id,
                                name: format!("BackboneInterface/{}", client_id.0),
                                mode: accepted_peer_mode,
                                gravity: accepted_peer_gravity,
                                recursive_prs: accepted_peer_recursive_prs,
                                announces_from_internal: accepted_peer_announces_from_internal,
                                announces_to_internal: 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,
                            };

                            let event = if let Some(template) = &dynamic_template {
                                Event::DynamicInterfaceUp {
                                    id: client_id,
                                    writer,
                                    registration: template.registration(info),
                                }
                            } else {
                                Event::InterfaceUp(client_id, Some(writer), Some(info))
                            };
                            if tx.send(event).is_err() {
                                // Driver shut down
                                cleanup(&poller, &clients, &listener);
                                return Ok(());
                            }
                        }
                        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
                        Err(e) => {
                            log::warn!("[{}] accept error: {}", name, e);
                            break;
                        }
                    }
                }
                // Re-arm listener (oneshot semantics)
                poller.modify(&listener, PollEvent::readable(LISTENER_KEY))?;
            } else if clients.contains_key(&ev.key) {
                let key = ev.key;
                let mut should_remove = false;
                let mut client_id = InterfaceId(0);

                let mut buf = [0u8; 4096];
                let read_result = {
                    let client = clients.get_mut(&key).unwrap();
                    client.stream.read(&mut buf)
                };

                match read_result {
                    Ok(0) | Err(_) => {
                        if let Some(c) = clients.get(&key) {
                            client_id = c.id;
                        }
                        should_remove = true;
                    }
                    Ok(n) => {
                        let client = clients.get_mut(&key).unwrap();
                        client_id = client.id;
                        client.has_received_data = true;
                        client.dp_ingress_bytes = client.dp_ingress_bytes.saturating_add(n as u64);
                        let decoded = client.decoder.feed_with_diagnostics(&buf[..n]);
                        client.dp_ingress_packets = client
                            .dp_ingress_packets
                            .saturating_add(decoded.frames.len() as u64);
                        for frame_len in decoded.invalid_frame_lengths {
                            log::debug!(
                            "[{}] invalid HDLC frame of {} bytes received from client {}, dropping frame",
                            name,
                            frame_len,
                            client_id.0,
                        );
                        }
                        for frame in decoded.frames {
                            if tx
                                .send(Event::Frame {
                                    interface_id: client_id,
                                    data: frame,
                                    rssi: None,
                                    snr: None,
                                })
                                .is_err()
                            {
                                cleanup(&poller, &clients, &listener);
                                return Ok(());
                            }
                        }
                        let queue = tx.inbound_queue_snapshot();
                        if queue.heights[0] >= ingress_thresholds.immediate {
                            let _ = throttle_busiest_ingress(
                                &poller,
                                &mut clients,
                                ingress_snapshot_at.elapsed(),
                            )?;
                        }
                    }
                }

                if should_remove {
                    let reason = if clients
                        .get(&key)
                        .is_some_and(|c| c.write_stall_flag.load(Ordering::Relaxed))
                    {
                        DisconnectReason::WriteStall
                    } else {
                        DisconnectReason::RemoteClosed
                    };
                    disconnect_client(DisconnectContext {
                        poller: &poller,
                        clients: &mut clients,
                        peers: &mut peers,
                        name: &name,
                        server_interface_id,
                        tx: &tx,
                        peer_state: &peer_state,
                        fast_flap: &fast_flap,
                        fast_flap_state: &fast_flap_state,
                        key,
                        client_id,
                        reason,
                    });
                } else if let Some(client) = clients.get(&key) {
                    // Re-arm client (oneshot semantics)
                    if !client.dp_ingress_gated {
                        poller.modify(&client.stream, PollEvent::readable(key))?;
                    }
                }
            }
        }

        if ingress_snapshot_at.elapsed() >= DP_IC_INTERVAL {
            let now = Instant::now();
            let queue = tx.inbound_queue_snapshot();
            if queue.heights[0] > ingress_thresholds.mid {
                let _ =
                    throttle_busiest_ingress(&poller, &mut clients, ingress_snapshot_at.elapsed())?;
            } else if queue.heights[0] < ingress_thresholds.low {
                let _ = release_one_ingress(&poller, &mut clients, now)?;
            }
            for client in clients.values_mut() {
                client.dp_ingress_bytes = 0;
                client.dp_ingress_packets = 0;
            }
            ingress_snapshot_at = now;
        }

        if let Some(timeout) = idle_timeout {
            let now = Instant::now();
            let timed_out: Vec<(usize, InterfaceId)> = clients
                .iter()
                .filter_map(|(&key, client)| {
                    if client.has_received_data || now.duration_since(client.connected_at) < timeout
                    {
                        None
                    } else {
                        Some((key, client.id))
                    }
                })
                .collect();

            for (key, client_id) in timed_out {
                disconnect_client(DisconnectContext {
                    poller: &poller,
                    clients: &mut clients,
                    peers: &mut peers,
                    name: &name,
                    server_interface_id,
                    tx: &tx,
                    peer_state: &peer_state,
                    fast_flap: &fast_flap,
                    fast_flap_state: &fast_flap_state,
                    key,
                    client_id,
                    reason: DisconnectReason::IdleTimeout,
                });
            }
        }
    }
}

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
}

struct DisconnectContext<'a> {
    poller: &'a Poller,
    clients: &'a mut HashMap<usize, ClientState>,
    peers: &'a mut HashMap<IpAddr, PeerBehaviorState>,
    name: &'a str,
    server_interface_id: InterfaceId,
    tx: &'a EventSender,
    peer_state: &'a Arc<Mutex<BackbonePeerMonitor>>,
    fast_flap: &'a BackboneFastFlapConfig,
    fast_flap_state: &'a Arc<Mutex<BackboneFastFlapMonitor>>,
    key: usize,
    client_id: InterfaceId,
    reason: DisconnectReason,
}

fn disconnect_client(context: DisconnectContext<'_>) {
    let DisconnectContext {
        poller,
        clients,
        peers,
        name,
        server_interface_id,
        tx,
        peer_state,
        fast_flap,
        fast_flap_state,
        key,
        client_id,
        reason,
    } = context;
    let Some(client) = clients.remove(&key) else {
        return;
    };

    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 BackboneWriter::disconnect_for_write_stall
        }
    }

    let _ = poller.delete(&client.stream);
    // client.stream closes on drop
    let connected_for = client.connected_at.elapsed();
    if let Some(record) = lock_or_recover(fast_flap_state, "backbone fast-flap state")
        .record_disconnect(client.peer_ip, connected_for, 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 state = peers
                .entry(client.peer_ip)
                .or_insert_with(PeerBehaviorState::new);
            state.blacklisted_until = record.blocked_until;
            state.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 _ = tx.send(Event::BackbonePeerDisconnected {
        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 => {
            let _ = tx.send(Event::BackbonePeerIdleTimeout {
                server_interface_id,
                peer_interface_id: client.id,
                peer_ip: client.peer_ip,
                peer_port: client.peer_port,
                connected_for,
            });
        }
        DisconnectReason::WriteStall => {
            let _ = tx.send(Event::BackbonePeerWriteStall {
                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(state) = peers.get_mut(&client.peer_ip) {
        state.connected_count = state.connected_count.saturating_sub(1);
    }
    peer_state.lock().unwrap().upsert_snapshot(peers);
    // Writer already sent InterfaceDown for write stalls; avoid duplicate.
    if !matches!(reason, DisconnectReason::WriteStall) {
        let _ = tx.send(Event::InterfaceDown(client_id));
    }
}

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)
}

fn cleanup(poller: &Poller, clients: &HashMap<usize, ClientState>, listener: &TcpListener) {
    for client in clients.values() {
        let _ = poller.delete(&client.stream);
    }
    let _ = poller.delete(listener);
}

// ---------------------------------------------------------------------------
// 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).
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(&[data.to_vec()])
    }

    fn send_frames(&mut self, frames: &[Vec<u8>]) -> 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())
    }
}

/// 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 = config.runtime.lock().unwrap().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 reader_stream = stream.try_clone()?;
    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));

    thread::Builder::new()
        .name(format!("backbone-client-{}", id.0))
        .spawn(move || {
            client_reader_loop(reader_stream, config, tx, ifac_size, underlay_mark, control);
        })?;

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

/// Reader thread: reads from socket, HDLC-decodes, sends frames to driver.
/// On disconnect, attempts reconnection. The socket intentionally remains in
/// blocking mode: this dedicated thread owns reads, while a cloned stream and
/// the interface writer path own writes.
fn client_reader_loop(
    mut stream: TcpStream,
    config: BackboneClientConfig,
    tx: EventSender,
    ifac_size: usize,
    underlay_mark: Option<u32>,
    control: ListenerControl,
) {
    let id = config.interface_id;
    let mut decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
    let mut buf = [0u8; 4096];

    loop {
        if control.should_stop() {
            return;
        }
        match poll_readable(stream.as_raw_fd(), Duration::from_millis(200)) {
            Ok(false) => continue,
            Ok(true) => {}
            Err(error) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] backbone poll error: {}", config.name, error);
                let _ = tx.send(Event::InterfaceDown(id));
                match client_reconnect(&config, &tx, underlay_mark, &control) {
                    Some(new_stream) => {
                        stream = new_stream;
                        decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
                        continue;
                    }
                    None => return,
                }
            }
        }
        match stream.read(&mut buf) {
            Ok(0) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] connection closed", config.name);
                let _ = tx.send(Event::InterfaceDown(id));
                match client_reconnect(&config, &tx, underlay_mark, &control) {
                    Some(new_stream) => {
                        stream = new_stream;
                        decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
                        continue;
                    }
                    None => {
                        log::error!("[{}] reconnection failed, giving up", config.name);
                        return;
                    }
                }
            }
            Ok(n) => {
                let decoded = decoder.feed_with_diagnostics(&buf[..n]);
                for frame_len in decoded.invalid_frame_lengths {
                    log::debug!(
                        "[{}] invalid HDLC frame of {} bytes received, dropping frame",
                        config.name,
                        frame_len,
                    );
                }
                for frame in decoded.frames {
                    if tx
                        .send(Event::Frame {
                            interface_id: id,
                            data: frame,
                            rssi: None,
                            snr: None,
                        })
                        .is_err()
                    {
                        return;
                    }
                }
            }
            Err(e) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] read error: {}", config.name, e);
                let _ = tx.send(Event::InterfaceDown(id));
                match client_reconnect(&config, &tx, underlay_mark, &control) {
                    Some(new_stream) => {
                        stream = new_stream;
                        decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
                        continue;
                    }
                    None => {
                        log::error!("[{}] reconnection failed, giving up", config.name);
                        return;
                    }
                }
            }
        }
    }
}

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

/// 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 = config.runtime.lock().unwrap().clone();

        let shift = attempts.min(MAX_BACKOFF_SHIFT);
        let backoff = runtime.reconnect_wait * 2u32.pow(shift);
        // Add ±25 % jitter to avoid thundering-herd reconnects.
        let jitter_range = backoff / 4;
        let jitter = if jitter_range.as_nanos() > 0 {
            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)
            }
        } else {
            backoff
        };
        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)
        .map(Duration::from_secs_f64)
}

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();
            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
                ));
            }
            Ok(Box::new(BackboneMode::Client(
                BackboneClientConfig {
                    name: name.to_string(),
                    target_host,
                    target_port,
                    interface_id: id,
                    transport_identity,
                    ..BackboneClientConfig::default()
                },
                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::sync::mpsc;
    use std::time::Duration;

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

    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 {
            thread::sleep(Duration::from_millis(5));
        }
        thread::sleep(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();
        thread::sleep(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 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();
        thread::sleep(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_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}"
            );
        }
    }
}