rns_net/interface/

mod.rs

//! Network interface abstractions.

#[cfg(feature = "iface-auto")]
pub mod auto;
#[cfg(feature = "iface-kiss")]
pub mod ax25_kiss;
#[cfg(feature = "iface-backbone")]
pub mod backbone;
#[cfg(feature = "iface-i2p")]
pub mod i2p;
#[cfg(feature = "iface-kiss")]
pub mod kiss_iface;
#[cfg(feature = "iface-local")]
pub mod local;
pub(crate) mod loop_writer;
#[cfg(feature = "iface-pipe")]
pub mod pipe;
pub mod registry;
#[cfg(feature = "iface-rnode")]
pub mod rnode;
#[cfg(feature = "iface-serial")]
pub mod serial_iface;
#[cfg(feature = "iface-tcp")]
pub mod tcp;
#[cfg(feature = "iface-tcp")]
pub mod tcp_server;
pub(crate) mod transmit_buffer;
#[cfg(feature = "iface-udp")]
pub mod udp;
#[cfg(all(feature = "iface-weave", target_os = "linux"))]
pub mod weave;

use std::any::Any;
use std::collections::HashMap;
use std::io;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::TrySendError;
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::thread;
use std::time::{Duration, Instant};

use crate::event::EventSender;
use crate::ifac::IfacState;
use rns_core::transport::types::{InterfaceId, InterfaceInfo};

/// Host callback used where every Reticulum underlay socket must be explicitly
/// excluded from an application VPN, notably Android `VpnService.protect()`.
pub trait SocketProtector: Send + Sync {
    fn protect(&self, fd: std::os::unix::io::RawFd) -> io::Result<()>;
}

fn socket_protector_slot() -> &'static std::sync::RwLock<Option<Arc<dyn SocketProtector>>> {
    static SLOT: std::sync::OnceLock<std::sync::RwLock<Option<Arc<dyn SocketProtector>>>> =
        std::sync::OnceLock::new();
    SLOT.get_or_init(|| std::sync::RwLock::new(None))
}

/// Exclusive process-scoped socket protection registration.
///
/// Android VPN services should run their private node in a dedicated process.
/// A second registration fails instead of replacing a callback used by live
/// sockets.
pub struct SocketProtectorGuard {
    protector: Arc<dyn SocketProtector>,
}

pub fn install_socket_protector(
    protector: Arc<dyn SocketProtector>,
) -> io::Result<SocketProtectorGuard> {
    let mut slot = socket_protector_slot()
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if slot.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::AlreadyExists,
            "an underlay socket protector is already installed",
        ));
    }
    *slot = Some(Arc::clone(&protector));
    Ok(SocketProtectorGuard { protector })
}

#[cfg(test)]
mod socket_protector_tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    static TEST_LOCK: Mutex<()> = Mutex::new(());
    struct CountingProtector(Arc<AtomicUsize>);
    impl SocketProtector for CountingProtector {
        fn protect(&self, _fd: std::os::unix::io::RawFd) -> io::Result<()> {
            self.0.fetch_add(1, Ordering::Relaxed);
            Ok(())
        }
    }

    #[test]
    fn underlay_helper_invokes_installed_protector_without_mark() {
        use std::os::fd::AsRawFd;
        let _lock = TEST_LOCK.lock().unwrap();
        let calls = Arc::new(AtomicUsize::new(0));
        let _guard =
            install_socket_protector(Arc::new(CountingProtector(Arc::clone(&calls)))).unwrap();
        let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
        apply_underlay_mark(socket.as_raw_fd(), None).unwrap();
        assert!(calls.load(Ordering::Relaxed) >= 1);
    }
}

impl Drop for SocketProtectorGuard {
    fn drop(&mut self) {
        let mut slot = socket_protector_slot()
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if slot
            .as_ref()
            .is_some_and(|installed| Arc::ptr_eq(installed, &self.protector))
        {
            *slot = None;
        }
    }
}

fn protect_underlay_socket(fd: std::os::unix::io::RawFd) -> io::Result<()> {
    let protector = socket_protector_slot()
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .clone();
    if let Some(protector) = protector {
        protector.protect(fd)?;
    }
    Ok(())
}

/// Bind a socket to a specific network interface using `SO_BINDTODEVICE`.
///
/// Requires `CAP_NET_RAW` or root on Linux.
#[cfg(target_os = "linux")]
pub fn bind_to_device(fd: std::os::unix::io::RawFd, device: &str) -> io::Result<()> {
    let dev_bytes = device.as_bytes();
    if dev_bytes.len() >= libc::IFNAMSIZ {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("device name too long: {}", device),
        ));
    }
    let ret = unsafe {
        libc::setsockopt(
            fd,
            libc::SOL_SOCKET,
            libc::SO_BINDTODEVICE,
            dev_bytes.as_ptr() as *const libc::c_void,
            dev_bytes.len() as libc::socklen_t,
        )
    };
    if ret != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

/// Apply a Linux firewall mark to an IP underlay socket.
///
/// The mark must be set before `connect()` so policy routing can keep Reticulum
/// underlay traffic outside application-level full tunnels. Configuring a mark
/// is fail-closed: callers must not silently create an unmarked socket when the
/// operation is not permitted.
#[cfg(target_os = "linux")]
pub fn apply_underlay_mark(fd: std::os::unix::io::RawFd, mark: Option<u32>) -> io::Result<()> {
    protect_underlay_socket(fd)?;
    let Some(mark) = mark else {
        return Ok(());
    };
    let ret = unsafe {
        libc::setsockopt(
            fd,
            libc::SOL_SOCKET,
            libc::SO_MARK,
            &mark as *const _ as *const libc::c_void,
            std::mem::size_of::<u32>() as libc::socklen_t,
        )
    };
    if ret != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

#[cfg(not(target_os = "linux"))]
pub fn apply_underlay_mark(fd: std::os::unix::io::RawFd, mark: Option<u32>) -> io::Result<()> {
    protect_underlay_socket(fd)?;
    if mark.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "underlay socket marks are only supported on Linux",
        ));
    }
    Ok(())
}

/// Connect a TCP socket after applying underlay routing options.
#[cfg(target_os = "linux")]
pub fn connect_tcp_with_options(
    addr: &std::net::SocketAddr,
    device: Option<&str>,
    underlay_mark: Option<u32>,
    timeout: Duration,
) -> io::Result<std::net::TcpStream> {
    use std::os::unix::io::AsRawFd;

    let socket = socket2::Socket::new(
        socket2::Domain::for_address(*addr),
        socket2::Type::STREAM,
        Some(socket2::Protocol::TCP),
    )?;
    apply_underlay_mark(socket.as_raw_fd(), underlay_mark)?;
    if let Some(device) = device {
        bind_to_device(socket.as_raw_fd(), device)?;
    }
    socket.connect_timeout(&(*addr).into(), timeout)?;
    Ok(socket.into())
}

/// Validate at node startup that the configured mark can actually be applied.
pub fn validate_underlay_mark(mark: Option<u32>) -> io::Result<()> {
    let Some(_) = mark else {
        return Ok(());
    };
    let socket = socket2::Socket::new(
        socket2::Domain::IPV4,
        socket2::Type::DGRAM,
        Some(socket2::Protocol::UDP),
    )?;
    use std::os::unix::io::AsRawFd;
    apply_underlay_mark(socket.as_raw_fd(), mark)
}

/// Writable end of an interface. Held by the driver.
///
/// Each implementation wraps a socket + framing.
pub trait Writer: Send {
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()>;

    /// Write one frame without silently dropping it. WouldBlock means the
    /// frame was accepted and `flush` must finish it before success is reported.
    /// Writers with intentional drop policies must override this method.
    fn send_frame_confirmed(&mut self, data: &[u8]) -> io::Result<()> {
        self.send_frame(data)
    }

    /// Internal async-writer admission. Leaves completion untouched on failure.
    #[doc(hidden)]
    fn enqueue_confirmed(
        &mut self,
        _data: &[u8],
        _completion: &mut Option<crate::link_send::Completion>,
    ) -> io::Result<()> {
        Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "confirmed sends require an async interface writer",
        ))
    }

    /// Like [`Writer::send_frame`], but a writer with its own worker may
    /// postpone waking it until [`Writer::wake`]. The driver uses this for
    /// forwarded frames and wakes each touched writer once per event batch.
    fn send_frame_deferred(&mut self, data: &[u8]) -> io::Result<()> {
        self.send_frame(data)
    }

    /// Deferred send of shared frame bytes; writers with a worker queue may
    /// retain the shared buffer instead of copying it.
    fn send_shared_deferred(
        &mut self,
        data: &rns_core::transport::types::PacketBytes,
    ) -> io::Result<()> {
        self.send_frame_deferred(data)
    }

    /// Wake a worker postponed by [`Writer::send_frame_deferred`].
    fn wake(&mut self) {}

    /// Write out what a driver-run writer has buffered (see `loop_writer`).
    /// The driver calls this on every writer before it waits for more
    /// input, so frames from one pass leave in as few writes as possible.
    fn flush_pending(&mut self) {}

    fn send_frames(&mut self, frames: &[crate::interface::FrameBytes]) -> io::Result<()> {
        for frame in frames {
            self.send_frame(frame)?;
        }
        Ok(())
    }

    /// Resume bytes accepted by a previous call that returned `WouldBlock`.
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }

    /// Share an exact byte admission budget with the async producer when this
    /// writer uses dataplane egress control.
    fn egress_control(&self) -> Option<EgressControl> {
        None
    }

    /// A stream socket the driver may write directly, without a worker
    /// thread (see `loop_writer`). Writers returning one must only frame
    /// with HDLC and write the result to this descriptor.
    fn loop_socket(&self) -> Option<std::os::fd::RawFd> {
        None
    }

    /// Byte-budget egress rules for a [`Writer::loop_socket`] writer, taken
    /// once when the driver starts writing it. Without them the driver
    /// applies its default backpressure.
    #[doc(hidden)]
    fn loop_egress(&mut self) -> Option<loop_writer::LoopEgress> {
        None
    }

    /// Called by the driver once per tick, so driver-run writers can detect
    /// a peer that stopped reading even when nothing new is sent.
    #[doc(hidden)]
    fn tick(&mut self) {}

    fn shutdown(&mut self) {}
}

/// Hasher for maps keyed by [`InterfaceId`], looked up several times per
/// forwarded packet. IDs are small trusted integers, so a multiplicative mix
/// replaces SipHash, whose per-lookup cost showed up in forwarding profiles.
#[derive(Default, Clone, Copy)]
pub struct InterfaceIdHasher(u64);

impl std::hash::Hasher for InterfaceIdHasher {
    fn write(&mut self, bytes: &[u8]) {
        for byte in bytes {
            self.write_u64(u64::from(*byte));
        }
    }

    fn write_u64(&mut self, value: u64) {
        self.0 = (self.0 ^ value).wrapping_mul(0x9E37_79B9_7F4A_7C15);
    }

    fn finish(&self) -> u64 {
        self.0 ^ (self.0 >> 29)
    }
}

/// Map keyed by interface ID using [`InterfaceIdHasher`].
pub type InterfaceMap<V> =
    HashMap<InterfaceId, V, std::hash::BuildHasherDefault<InterfaceIdHasher>>;

pub const DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY: usize = 256;

/// Longest a sleeping writer worker goes without rechecking its queue, a
/// backstop for wakeups postponed by `send_frame_deferred`.
const WRITER_RECHECK: Duration = Duration::from_millis(5);

/// Longest time a producer waits for a busy async writer to free queue space
/// or byte budget before treating the frame as undeliverable.
///
/// A healthy writer drains its queue in well under a millisecond, so this
/// bound converts momentary bursts into backpressure (the producer pauses and
/// the inbound queue fills) instead of silent loss. A writer that stays full
/// for the whole bound is treated as stalled: the frame is refused and the
/// driver's per-interface send deferral protects other interfaces.
pub const ASYNC_WRITER_BACKPRESSURE_WAIT: Duration = Duration::from_millis(20);

#[derive(Clone)]
#[doc(hidden)]
pub struct EgressControl {
    byte_limit: usize,
    reserved_bytes: Arc<AtomicUsize>,
    stalled: Arc<AtomicBool>,
    dropped_frames: Arc<AtomicU64>,
    dropped_bytes: Arc<AtomicU64>,
}

impl EgressControl {
    pub(crate) fn new(byte_limit: usize) -> Self {
        Self {
            byte_limit,
            reserved_bytes: Arc::new(AtomicUsize::new(0)),
            stalled: Arc::new(AtomicBool::new(false)),
            dropped_frames: Arc::new(AtomicU64::new(0)),
            dropped_bytes: Arc::new(AtomicU64::new(0)),
        }
    }

    fn try_reserve(&self, bytes: usize) -> bool {
        if self.stalled.load(Ordering::Relaxed) {
            return false;
        }
        let mut current = self.reserved_bytes.load(Ordering::Relaxed);
        loop {
            let Some(next) = current.checked_add(bytes) else {
                return false;
            };
            if next > self.byte_limit {
                return false;
            }
            match self.reserved_bytes.compare_exchange_weak(
                current,
                next,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => return true,
                Err(observed) => current = observed,
            }
        }
    }

    fn release(&self, bytes: usize) {
        self.reserved_bytes.fetch_sub(bytes, Ordering::Relaxed);
    }

    pub(crate) fn stalled(&self) -> bool {
        self.stalled.load(Ordering::Relaxed)
    }

    pub(crate) fn set_stalled(&self, stalled: bool) {
        self.stalled.store(stalled, Ordering::Relaxed);
    }

    pub(crate) fn record_drop(&self, bytes: usize) {
        self.dropped_frames.fetch_add(1, Ordering::Relaxed);
        self.dropped_bytes
            .fetch_add(bytes as u64, Ordering::Relaxed);
    }
}

/// Return whether an interface type normally operates on a shared medium.
///
/// This is a type-level hint: runtime configuration can still describe links
/// with different physical properties. It mirrors Reticulum's interface-class
/// defaults and is suitable for discovery/capability metadata.
pub fn shared_medium_hint(interface_type: &str) -> bool {
    matches!(
        interface_type,
        "AX25KISSInterface"
            | "KISSInterface"
            | "PipeInterface"
            | "RNodeInterface"
            | "RNodeMultiInterface"
            | "SerialInterface"
            | "UDPInterface"
    )
}

pub(crate) fn lock_or_recover<'a, T>(mutex: &'a Mutex<T>, label: &str) -> MutexGuard<'a, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => {
            log::warn!("recovering poisoned mutex: {}", label);
            poisoned.into_inner()
        }
    }
}

#[derive(Clone, Default)]
pub struct ListenerControl {
    stop: Arc<AtomicBool>,
    running: Option<Arc<AtomicBool>>,
}

impl ListenerControl {
    pub fn new() -> Self {
        Self {
            stop: Arc::new(AtomicBool::new(false)),
            running: None,
        }
    }

    pub(crate) fn with_running_flag(running: Arc<AtomicBool>) -> Self {
        Self {
            stop: Arc::new(AtomicBool::new(false)),
            running: Some(running),
        }
    }

    pub fn request_stop(&self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(running) = &self.running {
            running.store(false, Ordering::Relaxed);
        }
    }

    pub fn should_stop(&self) -> bool {
        self.stop.load(Ordering::Relaxed)
    }

    pub(crate) fn same_instance(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.stop, &other.stop)
    }
}

/// Wait for serial input without keeping a detached reader blocked forever.
pub(crate) fn poll_readable(fd: std::os::fd::RawFd, timeout: Duration) -> io::Result<bool> {
    let mut pfd = libc::pollfd {
        fd,
        events: libc::POLLIN,
        revents: 0,
    };
    let timeout_ms = i32::try_from(timeout.as_millis()).unwrap_or(i32::MAX);
    let result = unsafe { libc::poll(&mut pfd, 1, timeout_ms) };
    if result < 0 {
        let error = io::Error::last_os_error();
        return if error.kind() == io::ErrorKind::Interrupted {
            Ok(false)
        } else {
            Err(error)
        };
    }
    if pfd.revents & libc::POLLNVAL != 0 {
        return Err(io::Error::new(
            io::ErrorKind::BrokenPipe,
            "serial port closed",
        ));
    }
    Ok(result > 0)
}

pub(crate) fn wait_while_active(control: &ListenerControl, duration: Duration) -> bool {
    let deadline = Instant::now() + duration;
    while !control.should_stop() {
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            return true;
        }
        thread::sleep(remaining.min(Duration::from_millis(100)));
    }
    false
}

pub(crate) struct UnavailableWriter;

impl Writer for UnavailableWriter {
    fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
        Err(io::Error::new(
            io::ErrorKind::NotConnected,
            "interface is waiting for its port",
        ))
    }
}

#[derive(Clone, Default)]
pub struct AsyncWriterMetrics {
    queued_frames: Arc<AtomicUsize>,
    worker_alive: Arc<AtomicBool>,
    egress_control: Option<EgressControl>,
    /// Signalled by the worker whenever it frees queue slots or byte budget.
    space: Arc<(Mutex<()>, Condvar)>,
    /// Producers inside a backpressure wait. Incremented under the `space`
    /// lock before the producer's queue attempt, so a worker that frees space
    /// after a failed attempt always observes it and notifies.
    space_waiters: Arc<AtomicUsize>,
}

impl AsyncWriterMetrics {
    fn notify_space(&self) {
        if self.space_waiters.load(Ordering::SeqCst) == 0 {
            return;
        }
        let (lock, condvar) = &*self.space;
        let _guard = lock_or_recover(lock, "async writer space");
        condvar.notify_all();
    }

    pub fn queued_frames(&self) -> usize {
        self.queued_frames.load(Ordering::Relaxed)
    }

    pub fn worker_alive(&self) -> bool {
        self.worker_alive.load(Ordering::Relaxed)
    }

    /// Exact encoded bytes waiting in the async queue or concrete writer.
    pub fn tx_buffered(&self) -> usize {
        self.egress_control
            .as_ref()
            .map(|control| control.reserved_bytes.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    pub fn tx_drops(&self) -> u64 {
        self.egress_control
            .as_ref()
            .map(|control| control.dropped_frames.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    pub fn tx_dropped_bytes(&self) -> u64 {
        self.egress_control
            .as_ref()
            .map(|control| control.dropped_bytes.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    pub fn tx_stalled(&self) -> bool {
        self.egress_control
            .as_ref()
            .is_some_and(EgressControl::stalled)
    }
}

/// Bounded frame queue between the driver and one interface writer thread.
///
/// The worker drains every queued frame in one lock acquisition. This
/// replaces a std sync channel whose receiver spins while it observes a send
/// in progress, which on a single CPU burned the rest of the time slice.
struct WriterQueue {
    state: Mutex<WriterQueueState>,
    ready: Condvar,
    capacity: usize,
}

struct WriterQueueState {
    frames: std::collections::VecDeque<QueuedFrame>,
    sender_closed: bool,
    receiver_closed: bool,
    receiver_waiting: bool,
}

impl WriterQueue {
    fn new(capacity: usize) -> Arc<Self> {
        Arc::new(Self {
            state: Mutex::new(WriterQueueState {
                frames: std::collections::VecDeque::new(),
                sender_closed: false,
                receiver_closed: false,
                receiver_waiting: false,
            }),
            ready: Condvar::new(),
            capacity: capacity.max(1),
        })
    }

    /// Wait until frames are queued, then move all of them into `out`.
    /// Returns false once the sender is gone and nothing remains.
    fn recv_all(&self, out: &mut Vec<QueuedFrame>) -> bool {
        let mut state = lock_or_recover(&self.state, "async writer queue");
        while state.frames.is_empty() {
            if state.sender_closed {
                return false;
            }
            state.receiver_waiting = true;
            // Producers may postpone the wakeup (see send_frame_deferred);
            // the periodic recheck bounds any delay if one is never sent.
            state = match self.ready.wait_timeout(state, WRITER_RECHECK) {
                Ok((state, _)) => state,
                Err(poisoned) => poisoned.into_inner().0,
            };
            state.receiver_waiting = false;
        }
        out.extend(state.frames.drain(..));
        true
    }

    /// Wake the worker if it sleeps while frames are queued.
    fn kick(&self) {
        let state = lock_or_recover(&self.state, "async writer queue");
        let wake = state.receiver_waiting && !state.frames.is_empty();
        drop(state);
        if wake {
            self.ready.notify_one();
        }
    }

    fn close_receiver(&self) {
        lock_or_recover(&self.state, "async writer queue").receiver_closed = true;
    }
}

/// Producer handle; dropping it lets the worker exit after draining.
struct WriterTx(Arc<WriterQueue>);

impl WriterTx {
    /// Queue a frame. Returns whether the worker still needs waking: false
    /// when it was woken here (`notify`) or was not waiting.
    fn try_send(
        &self,
        frame: QueuedFrame,
        notify: bool,
    ) -> Result<bool, TrySendError<QueuedFrame>> {
        let mut state = lock_or_recover(&self.0.state, "async writer queue");
        if state.receiver_closed {
            return Err(TrySendError::Disconnected(frame));
        }
        if state.frames.len() >= self.0.capacity {
            return Err(TrySendError::Full(frame));
        }
        state.frames.push_back(frame);
        let waiting = state.receiver_waiting;
        drop(state);
        if waiting && notify {
            self.0.ready.notify_one();
        }
        Ok(waiting && !notify)
    }
}

impl Drop for WriterTx {
    fn drop(&mut self) {
        lock_or_recover(&self.0.state, "async writer queue").sender_closed = true;
        self.0.ready.notify_one();
    }
}

struct AsyncWriter {
    tx: WriterTx,
    metrics: AsyncWriterMetrics,
    /// Set after a backpressure wait times out. Until a frame is admitted
    /// again, unconfirmed sends make a single attempt instead of waiting,
    /// so a stalled peer costs the producer at most one wait.
    skip_wait: bool,
    /// Set while a deferred send is in progress.
    defer_wake: bool,
    /// A deferred send left the worker asleep with frames queued.
    wake_pending: bool,
}

enum Admission {
    Done,
    Refused(QueuedFrame),
}

/// Frame bytes queued for a writer: owned, or shared with the driver's
/// forwarding buffer so queueing a forwarded frame does not copy it.
pub enum FrameBytes {
    Owned(Vec<u8>),
    Shared(rns_core::transport::types::PacketBytes),
}

impl std::ops::Deref for FrameBytes {
    type Target = [u8];

    fn deref(&self) -> &[u8] {
        match self {
            FrameBytes::Owned(bytes) => bytes,
            FrameBytes::Shared(bytes) => bytes,
        }
    }
}

struct QueuedFrame {
    data: FrameBytes,
    reserved_bytes: usize,
    completion: Option<crate::link_send::Completion>,
}

impl Writer for AsyncWriter {
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        self.enqueue(data, &mut None)
    }

    fn send_frame_deferred(&mut self, data: &[u8]) -> io::Result<()> {
        self.defer_wake = true;
        let result = self.enqueue(data, &mut None);
        self.defer_wake = false;
        result
    }

    fn send_shared_deferred(
        &mut self,
        data: &rns_core::transport::types::PacketBytes,
    ) -> io::Result<()> {
        self.defer_wake = true;
        let result = self.enqueue_unconfirmed(FrameBytes::Shared(data.clone()));
        self.defer_wake = false;
        result
    }

    fn wake(&mut self) {
        if std::mem::take(&mut self.wake_pending) {
            self.tx.0.kick();
        }
    }

    fn enqueue_confirmed(
        &mut self,
        data: &[u8],
        completion: &mut Option<crate::link_send::Completion>,
    ) -> io::Result<()> {
        self.enqueue(data, completion)
    }
}

impl AsyncWriter {
    /// Admit an unconfirmed frame, waiting up to
    /// [`ASYNC_WRITER_BACKPRESSURE_WAIT`] while the worker is busy.
    ///
    /// A stalled peer (as reported by the byte-budget controller) is refused
    /// immediately so that one dead connection cannot repeatedly pause the
    /// driver. Over-budget frames that can never fit are dropped as before.
    fn enqueue_unconfirmed(&mut self, data: FrameBytes) -> io::Result<()> {
        let mut frame = QueuedFrame {
            data,
            reserved_bytes: 0,
            completion: None,
        };
        // Fast path without the space lock: almost every frame fits.
        match self.try_admit(frame)? {
            Admission::Done => return Ok(()),
            Admission::Refused(returned) => frame = returned,
        }
        if self.skip_wait {
            return self.refuse(frame);
        }
        // The worker must run to free space; wake it if a deferred send left
        // it asleep.
        self.wake();
        let space = Arc::clone(&self.metrics.space);
        let (lock, condvar) = &*space;
        let mut guard = lock_or_recover(lock, "async writer space");
        // Register before retrying: a worker that frees space after this
        // retry fails observes the waiter and must take the lock to notify,
        // which it can only do once this thread is waiting.
        self.metrics.space_waiters.fetch_add(1, Ordering::SeqCst);
        let deadline = Instant::now() + ASYNC_WRITER_BACKPRESSURE_WAIT;
        let result = loop {
            match self.try_admit(frame) {
                Err(error) => break Err(error),
                Ok(Admission::Done) => break Ok(()),
                Ok(Admission::Refused(returned)) => frame = returned,
            }
            let now = Instant::now();
            if now >= deadline {
                self.skip_wait = true;
                break self.refuse(frame);
            }
            guard = match condvar.wait_timeout(guard, deadline - now) {
                Ok((guard, _)) => guard,
                Err(poisoned) => poisoned.into_inner().0,
            };
        };
        self.metrics.space_waiters.fetch_sub(1, Ordering::SeqCst);
        drop(guard);
        result
    }

    /// One admission attempt. Frames that can never be admitted (over the
    /// byte limit, or a stalled peer) are dropped and counted, as before.
    fn try_admit(&mut self, mut frame: QueuedFrame) -> io::Result<Admission> {
        if !self.metrics.worker_alive() {
            return Err(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "interface writer worker is offline",
            ));
        }
        if let Some(control) = &self.metrics.egress_control {
            let bytes = crate::hdlc::framed_len(&frame.data);
            if bytes > control.byte_limit || control.stalled() {
                control.record_drop(bytes);
                return Ok(Admission::Done);
            }
            if !control.try_reserve(bytes) {
                return Ok(Admission::Refused(frame));
            }
            frame.reserved_bytes = bytes;
        }
        self.metrics.queued_frames.fetch_add(1, Ordering::Relaxed);
        match self.tx.try_send(frame, !self.defer_wake) {
            Ok(needs_wake) => {
                self.skip_wait = false;
                self.wake_pending |= needs_wake;
                Ok(Admission::Done)
            }
            Err(TrySendError::Full(mut returned)) => {
                self.metrics.queued_frames.fetch_sub(1, Ordering::Relaxed);
                if let Some(control) = &self.metrics.egress_control {
                    control.release(returned.reserved_bytes);
                }
                returned.reserved_bytes = 0;
                Ok(Admission::Refused(returned))
            }
            Err(TrySendError::Disconnected(returned)) => {
                self.metrics.queued_frames.fetch_sub(1, Ordering::Relaxed);
                if let Some(control) = &self.metrics.egress_control {
                    control.release(returned.reserved_bytes);
                }
                self.metrics.worker_alive.store(false, Ordering::Relaxed);
                Err(io::Error::new(
                    io::ErrorKind::BrokenPipe,
                    "interface writer worker disconnected",
                ))
            }
        }
    }

    /// Give up on a frame that did not fit. A frame refused for byte budget
    /// is dropped and counted (the pre-existing egress-control policy); a
    /// frame refused for queue slots is reported as would-block.
    fn refuse(&self, frame: QueuedFrame) -> io::Result<()> {
        if let Some(control) = &self.metrics.egress_control {
            let bytes = crate::hdlc::framed_len(&frame.data);
            if self.metrics.queued_frames() < self.tx.0.capacity {
                control.record_drop(bytes);
                return Ok(());
            }
        }
        Err(io::Error::new(
            io::ErrorKind::WouldBlock,
            "interface writer queue is full",
        ))
    }

    fn enqueue(
        &mut self,
        data: &[u8],
        completion: &mut Option<crate::link_send::Completion>,
    ) -> io::Result<()> {
        if !self.metrics.worker_alive() {
            return Err(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "interface writer worker is offline",
            ));
        }

        // Unconfirmed frames (transit forwarding, announces) wait briefly for a
        // busy worker instead of being refused at the first full queue.
        // Confirmed sends keep their immediate WouldBlock contract: their
        // callers already retry through the completion machinery.
        if completion.is_none() {
            return self.enqueue_unconfirmed(FrameBytes::Owned(data.to_vec()));
        }

        let reserved_bytes = if let Some(control) = &self.metrics.egress_control {
            let framed_bytes = crate::hdlc::framed_len(data);
            if !control.try_reserve(framed_bytes) {
                if completion.is_some() {
                    if framed_bytes > control.byte_limit {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidInput,
                            "frame exceeds interface byte budget",
                        ));
                    }
                    return Err(io::Error::new(
                        io::ErrorKind::WouldBlock,
                        "interface egress budget is full",
                    ));
                }
                control.record_drop(framed_bytes);
                return Ok(());
            }
            framed_bytes
        } else {
            0
        };

        // Publish accounting before the frame becomes visible to the worker.
        self.metrics.queued_frames.fetch_add(1, Ordering::Relaxed);
        match self.tx.try_send(
            QueuedFrame {
                data: FrameBytes::Owned(data.to_vec()),
                reserved_bytes,
                completion: completion.take(),
            },
            true,
        ) {
            Ok(_) => Ok(()),
            Err(TrySendError::Full(frame)) => {
                *completion = frame.completion;
                self.metrics.queued_frames.fetch_sub(1, Ordering::Relaxed);
                if let Some(control) = &self.metrics.egress_control {
                    control.release(frame.reserved_bytes);
                }
                Err(io::Error::new(
                    io::ErrorKind::WouldBlock,
                    "interface writer queue is full",
                ))
            }
            Err(TrySendError::Disconnected(frame)) => {
                *completion = frame.completion;
                self.metrics.queued_frames.fetch_sub(1, Ordering::Relaxed);
                if let Some(control) = &self.metrics.egress_control {
                    control.release(frame.reserved_bytes);
                }
                self.metrics.worker_alive.store(false, Ordering::Relaxed);
                Err(io::Error::new(
                    io::ErrorKind::BrokenPipe,
                    "interface writer worker disconnected",
                ))
            }
        }
    }
}

pub fn wrap_async_writer(
    writer: Box<dyn Writer>,
    interface_id: InterfaceId,
    interface_name: &str,
    event_tx: EventSender,
    queue_capacity: usize,
) -> (Box<dyn Writer>, AsyncWriterMetrics) {
    if let Some(fd) = writer.loop_socket() {
        return loop_writer::wrap(writer, fd, interface_id, interface_name, event_tx);
    }
    let egress_control = writer.egress_control();
    let queue = WriterQueue::new(queue_capacity);
    let (tx, rx) = (WriterTx(Arc::clone(&queue)), queue);
    let metrics = AsyncWriterMetrics {
        queued_frames: Arc::new(AtomicUsize::new(0)),
        worker_alive: Arc::new(AtomicBool::new(true)),
        egress_control,
        space: Arc::default(),
        space_waiters: Arc::default(),
    };
    let metrics_thread = metrics.clone();
    let name = interface_name.to_string();

    let spawn_result = thread::Builder::new()
        .name(format!("iface-writer-{}", interface_id.0))
        .spawn(move || async_writer_loop(writer, rx, interface_id, name, event_tx, metrics_thread));

    if let Err(err) = spawn_result {
        metrics.worker_alive.store(false, Ordering::Relaxed);
        log::error!(
            "[{}:{}] failed to spawn async writer thread: {}",
            interface_name,
            interface_id.0,
            err
        );
        return (Box::new(DirectWriterFallback), metrics);
    }

    (
        Box::new(AsyncWriter {
            tx,
            metrics: metrics.clone(),
            skip_wait: false,
            defer_wake: false,
            wake_pending: false,
        }),
        metrics,
    )
}

struct DirectWriterFallback;

impl Writer for DirectWriterFallback {
    fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
        Err(io::Error::other("interface writer worker unavailable"))
    }
}

fn async_writer_loop(
    mut writer: Box<dyn Writer>,
    rx: Arc<WriterQueue>,
    interface_id: InterfaceId,
    interface_name: String,
    event_tx: EventSender,
    metrics: AsyncWriterMetrics,
) {
    let mut batch = Vec::new();
    while rx.recv_all(&mut batch) {
        let queued = std::mem::take(&mut batch);
        let reserved_bytes = queued.iter().map(|frame| frame.reserved_bytes).sum();
        metrics
            .queued_frames
            .fetch_sub(queued.len(), Ordering::Relaxed);
        metrics.notify_space();
        if event_tx.link_send_pool().in_flight() > 0 {
            let _ = event_tx.try_send(crate::event::Event::LinkWriterReady);
        }

        let mut result = Ok(());
        if queued.iter().any(|frame| frame.completion.is_some()) {
            for mut frame in queued {
                let confirmed = frame.completion.is_some();
                if frame.completion.as_ref().is_some_and(|c| c.is_finished()) {
                    result = Err(io::Error::new(
                        io::ErrorKind::Interrupted,
                        "transmission cancelled by shutdown",
                    ));
                }
                if result.is_ok() {
                    result = if frame.completion.is_some() {
                        writer.send_frame_confirmed(&frame.data)
                    } else {
                        writer.send_frame(&frame.data)
                    };
                    while result
                        .as_ref()
                        .is_err_and(|e| e.kind() == io::ErrorKind::WouldBlock)
                    {
                        if frame.completion.as_ref().is_some_and(|c| c.is_finished()) {
                            result = Err(io::Error::new(
                                io::ErrorKind::Interrupted,
                                "transmission cancelled by shutdown",
                            ));
                            break;
                        }
                        thread::sleep(Duration::from_millis(1));
                        result = writer.flush();
                    }
                }
                if let Some(completion) = frame.completion.take() {
                    completion.finish(
                        result.as_ref().map(|_| ()).map_err(|e| {
                            crate::link_send::LinkSendError::WriteFailed(e.to_string())
                        }),
                    );
                }
                // Packet-level policy/validation errors do not imply a broken
                // connection (e.g. a Local client that temporarily holds TX).
                if confirmed
                    && result.as_ref().is_err_and(|error| {
                        matches!(
                            error.kind(),
                            io::ErrorKind::PermissionDenied
                                | io::ErrorKind::InvalidInput
                                | io::ErrorKind::Unsupported
                        )
                    })
                {
                    result = Ok(());
                }
            }
        } else {
            let frames: Vec<_> = queued.into_iter().map(|frame| frame.data).collect();
            result = writer.send_frames(&frames);
        }
        while result
            .as_ref()
            .is_err_and(|error| error.kind() == io::ErrorKind::WouldBlock)
        {
            thread::sleep(Duration::from_millis(1));
            result = writer.flush();
        }

        if let Some(control) = &metrics.egress_control {
            control.release(reserved_bytes);
        }
        metrics.notify_space();
        if event_tx.link_send_pool().in_flight() > 0 {
            let _ = event_tx.try_send(crate::event::Event::LinkWriterReady);
        }

        if let Err(err) = result {
            metrics.worker_alive.store(false, Ordering::Relaxed);
            log::warn!(
                "[{}:{}] async writer exiting after send failure: {}",
                interface_name,
                interface_id.0,
                err
            );
            writer.shutdown();
            rx.close_receiver();
            metrics.notify_space();
            let _ = event_tx.send(crate::event::Event::InterfaceDown(interface_id));
            return;
        }
    }

    rx.close_receiver();
    metrics.worker_alive.store(false, Ordering::Relaxed);
    metrics.notify_space();
}

pub use crate::common::interface_stats::{InterfaceStats, TrafficRates, ANNOUNCE_SAMPLE_MAX};

use crate::common::management::InterfaceStatusView;

/// Everything the driver tracks per interface.
pub struct InterfaceEntry {
    pub id: InterfaceId,
    pub info: InterfaceInfo,
    pub writer: Box<dyn Writer>,
    pub async_writer_metrics: Option<AsyncWriterMetrics>,
    /// Administrative enable/disable state.
    pub enabled: bool,
    pub online: bool,
    /// True for dynamically spawned interfaces (e.g. TCP server clients).
    /// These are fully removed on InterfaceDown rather than just marked offline.
    pub dynamic: bool,
    /// IFAC state for this interface, if access codes are enabled.
    pub ifac: Option<IfacState>,
    /// Traffic statistics.
    pub stats: InterfaceStats,
    /// Human-readable interface type string (e.g. "TCPClientInterface").
    pub interface_type: String,
    /// Next time a send should be retried after a transient WouldBlock.
    pub send_retry_at: Option<Instant>,
    /// Current retry backoff for transient send failures.
    pub send_retry_backoff: Duration,
}

/// Result of starting an interface via a factory.
// Boxing `Simple` would change the published factory result representation.
#[allow(clippy::large_enum_variant)]
pub enum StartResult {
    /// One writer, registered immediately (TcpClient, Udp, Serial, etc.)
    Simple {
        id: InterfaceId,
        info: InterfaceInfo,
        writer: Box<dyn Writer>,
        interface_type_name: String,
        control: Option<ListenerControl>,
    },
    /// Spawns a listener; dynamic interfaces arrive via Event::InterfaceUp (TcpServer, Auto, I2P, etc.)
    Listener { control: Option<ListenerControl> },
    /// Multiple subinterfaces from one config (RNode).
    Multi {
        subinterfaces: Vec<SubInterface>,
        control: Option<ListenerControl>,
    },
}

/// A single subinterface returned from a multi-interface factory.
pub struct SubInterface {
    pub id: InterfaceId,
    pub info: InterfaceInfo,
    pub writer: Box<dyn Writer>,
    pub interface_type_name: String,
}

/// Context passed to [`InterfaceFactory::start()`].
pub struct StartContext {
    pub tx: EventSender,
    pub next_dynamic_id: Arc<AtomicU64>,
    pub mode: u8,
    pub gravity: i64,
    pub recursive_prs: bool,
    pub announces_from_internal: bool,
    pub announces_to_internal: Option<bool>,
    pub ingress_control: rns_core::transport::types::IngressControlConfig,
    pub ifac: Option<IfacState>,
    /// Linux `SO_MARK` value applied to IP underlay sockets before connect.
    pub underlay_mark: Option<u32>,
}

/// Opaque interface config data. Each factory downcasts to its concrete type.
pub trait InterfaceConfigData: Send + Any {
    fn as_any(&self) -> &dyn Any;
    fn into_any(self: Box<Self>) -> Box<dyn Any>;
}

/// ConfigObj section passed to factories that need nested `[[[sections]]]`.
pub struct ConfigSection<'a> {
    pub params: &'a HashMap<String, String>,
    pub children: &'a [crate::config::ParsedSubinterface],
}

#[derive(Debug, Clone)]
pub struct DynamicInterfaceTemplate {
    pub parent_id: InterfaceId,
    pub interface_type: String,
    pub ifac: Option<IfacState>,
    pub mode: u8,
    pub gravity: i64,
    pub recursive_prs: bool,
    pub announces_from_internal: bool,
    pub announces_to_internal: Option<bool>,
}

impl DynamicInterfaceTemplate {
    pub fn registration(
        &self,
        mut info: InterfaceInfo,
    ) -> crate::event::DynamicInterfaceRegistration {
        info.mode = self.mode;
        info.gravity = self.gravity;
        info.recursive_prs = self.recursive_prs;
        info.announces_from_internal = self.announces_from_internal;
        info.announces_to_internal = self.announces_to_internal;
        crate::event::DynamicInterfaceRegistration {
            info,
            interface_type: self.interface_type.clone(),
            parent_id: self.parent_id,
            telemetry: crate::event::InterfaceTelemetry::default(),
            ifac: self.ifac.clone(),
        }
    }
}

impl<T: Send + 'static> InterfaceConfigData for T {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn into_any(self: Box<Self>) -> Box<dyn Any> {
        self
    }
}

/// Factory that can parse config and start an interface type.
pub trait InterfaceFactory: Send + Sync {
    /// Config-file type name, e.g. "TCPClientInterface".
    fn type_name(&self) -> &str;

    /// Default IFAC size (bytes). 8 for serial/kiss/rnode, 16 for others.
    fn default_ifac_size(&self) -> usize {
        16
    }

    /// Parse from key-value params (config file or external).
    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String>;

    /// Parse a complete ConfigObj interface section. Flat factories inherit
    /// the compatibility default; multi-interface factories can inspect children.
    fn parse_config_section(
        &self,
        name: &str,
        id: InterfaceId,
        section: ConfigSection<'_>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        self.parse_config(name, id, section.params)
    }

    /// Start the interface from parsed config.
    fn start(
        &self,
        config: Box<dyn InterfaceConfigData>,
        ctx: StartContext,
    ) -> io::Result<StartResult>;
}

impl InterfaceStatusView for InterfaceEntry {
    fn id(&self) -> InterfaceId {
        self.id
    }
    fn info(&self) -> &InterfaceInfo {
        &self.info
    }
    fn online(&self) -> bool {
        self.online
    }
    fn stats(&self) -> &InterfaceStats {
        &self.stats
    }
    fn tx_drops(&self) -> u64 {
        self.async_writer_metrics
            .as_ref()
            .map(AsyncWriterMetrics::tx_drops)
            .unwrap_or(0)
    }
    fn tx_dropped_bytes(&self) -> u64 {
        self.async_writer_metrics
            .as_ref()
            .map(AsyncWriterMetrics::tx_dropped_bytes)
            .unwrap_or(0)
    }
    fn tx_stalled(&self) -> bool {
        self.async_writer_metrics
            .as_ref()
            .is_some_and(AsyncWriterMetrics::tx_stalled)
    }
    fn tx_buffered(&self) -> usize {
        self.async_writer_metrics
            .as_ref()
            .map(AsyncWriterMetrics::tx_buffered)
            .unwrap_or(0)
    }
}

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

    #[test]
    fn shared_medium_hint_matches_interface_class_defaults() {
        for interface_type in [
            "AX25KISSInterface",
            "KISSInterface",
            "PipeInterface",
            "RNodeInterface",
            "RNodeMultiInterface",
            "SerialInterface",
            "UDPInterface",
        ] {
            assert!(shared_medium_hint(interface_type), "{interface_type}");
        }
        for interface_type in [
            "AutoInterface",
            "BackboneInterface",
            "I2PInterface",
            "LocalInterface",
            "TCPClientInterface",
            "TCPServerInterface",
        ] {
            assert!(!shared_medium_hint(interface_type), "{interface_type}");
        }
    }
    use crate::event::Event;
    use rns_core::constants;
    use std::sync::mpsc;

    struct MockWriter {
        sent: Vec<Vec<u8>>,
    }

    impl MockWriter {
        fn new() -> Self {
            MockWriter { sent: Vec::new() }
        }
    }

    impl Writer for MockWriter {
        fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
            self.sent.push(data.to_vec());
            Ok(())
        }
    }

    #[test]
    fn interface_entry_construction() {
        let entry = InterfaceEntry {
            id: InterfaceId(1),
            info: InterfaceInfo {
                id: InterfaceId(1),
                name: String::new(),
                mode: constants::MODE_FULL,
                gravity: 0,
                recursive_prs: false,
                announces_from_internal: true,
                announces_to_internal: None,
                out_capable: true,
                in_capable: true,
                bitrate: None,
                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: constants::MTU as u32,
                ia_freq: 0.0,
                ip_freq: 0.0,
                op_freq: 0.0,
                op_samples: 0,
                started: 0.0,
                ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
            },
            writer: Box::new(MockWriter::new()),
            async_writer_metrics: None,
            enabled: true,
            online: false,
            dynamic: false,
            ifac: None,
            stats: InterfaceStats::default(),
            interface_type: String::new(),
            send_retry_at: None,
            send_retry_backoff: Duration::ZERO,
        };
        assert_eq!(entry.id, InterfaceId(1));
        assert!(!entry.online);
        assert!(!entry.dynamic);
    }

    #[test]
    fn dynamic_registration_inherits_announces_to_internal() {
        let template = DynamicInterfaceTemplate {
            parent_id: InterfaceId(7),
            interface_type: "dynamic-test".into(),
            ifac: None,
            mode: constants::MODE_GATEWAY,
            gravity: -3,
            recursive_prs: true,
            announces_from_internal: false,
            announces_to_internal: Some(true),
        };
        let info = InterfaceInfo {
            id: InterfaceId(8),
            name: "child".into(),
            mode: constants::MODE_FULL,
            gravity: 0,
            recursive_prs: false,
            announces_from_internal: true,
            announces_to_internal: None,
            out_capable: true,
            in_capable: true,
            bitrate: None,
            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: constants::MTU as u32,
            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
            ia_freq: 0.0,
            ip_freq: 0.0,
            op_freq: 0.0,
            op_samples: 0,
            started: 0.0,
        };

        let registration = template.registration(info);

        assert_eq!(registration.info.mode, constants::MODE_GATEWAY);
        assert_eq!(registration.info.gravity, -3);
        assert!(registration.info.recursive_prs);
        assert!(!registration.info.announces_from_internal);
        assert_eq!(registration.info.announces_to_internal, Some(true));
    }

    #[test]
    fn mock_writer_captures_bytes() {
        let mut writer = MockWriter::new();
        writer.send_frame(b"hello").unwrap();
        writer.send_frame(b"world").unwrap();
        assert_eq!(writer.sent.len(), 2);
        assert_eq!(writer.sent[0], b"hello");
        assert_eq!(writer.sent[1], b"world");
    }

    #[test]
    fn writer_send_frame_produces_output() {
        let mut writer = MockWriter::new();
        let data = vec![0x01, 0x02, 0x03];
        writer.send_frame(&data).unwrap();
        assert_eq!(writer.sent[0], data);
    }

    struct BlockingWriter {
        entered_tx: mpsc::Sender<()>,
        release_rx: mpsc::Receiver<()>,
    }

    #[test]
    fn confirmed_writer_waits_for_flush_and_reports_write_failures() {
        use crate::link_send::{Completion, LinkSendError, SendPool};
        use futures::FutureExt;
        struct FlushWriter {
            entered: mpsc::Sender<()>,
            release: mpsc::Receiver<()>,
            writes: Arc<AtomicUsize>,
        }
        impl Writer for FlushWriter {
            fn send_frame(&mut self, _: &[u8]) -> io::Result<()> {
                self.writes.fetch_add(1, Ordering::SeqCst);
                Err(io::ErrorKind::WouldBlock.into())
            }
            fn flush(&mut self) -> io::Result<()> {
                self.entered.send(()).unwrap();
                self.release.recv().unwrap();
                Ok(())
            }
        }
        let (event_tx, _rx) = crate::event::channel();
        let (entered, entered_rx) = mpsc::channel();
        let (release_tx, release) = mpsc::channel();
        let writes = Arc::new(AtomicUsize::new(0));
        let pool = SendPool::new(1);
        let (completion, mut receipt) = Completion::new(pool.try_acquire().unwrap());
        let (mut writer, _) = wrap_async_writer(
            Box::new(FlushWriter {
                entered,
                release,
                writes: writes.clone(),
            }),
            InterfaceId(999),
            "flush test",
            event_tx.clone(),
            1,
        );
        writer
            .enqueue_confirmed(b"once", &mut Some(completion))
            .unwrap();
        entered_rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert!((&mut receipt).now_or_never().is_none());
        release_tx.send(()).unwrap();
        receipt.wait().unwrap();
        assert_eq!(
            writes.load(Ordering::SeqCst),
            1,
            "flush must not resubmit the frame"
        );

        let (completion, receipt) = Completion::new(pool.try_acquire().unwrap());
        let (mut writer, _) = wrap_async_writer(
            Box::new(FailingWriter {
                shutdown_called: Arc::new(AtomicBool::new(false)),
            }),
            InterfaceId(998),
            "failure test",
            event_tx,
            1,
        );
        writer
            .enqueue_confirmed(b"fail", &mut Some(completion))
            .unwrap();
        assert_eq!(
            receipt.wait(),
            Err(LinkSendError::WriteFailed("boom".into()))
        );
    }

    impl Writer for BlockingWriter {
        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
            let _ = self.entered_tx.send(());
            let _ = self.release_rx.recv();
            Ok(())
        }
    }

    struct EgressControlledBlockingWriter {
        inner: BlockingWriter,
        control: EgressControl,
    }

    impl Writer for EgressControlledBlockingWriter {
        fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
            self.inner.send_frame(data)
        }

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

    struct FailingWriter {
        shutdown_called: Arc<AtomicBool>,
    }

    struct RecordingWriter {
        sent_tx: mpsc::Sender<Vec<u8>>,
    }

    impl Writer for RecordingWriter {
        fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
            self.sent_tx
                .send(data.to_vec())
                .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "test receiver closed"))
        }
    }

    struct BatchRecordingWriter {
        batch_tx: mpsc::Sender<usize>,
        first_release: Option<mpsc::Receiver<()>>,
    }

    impl Writer for BatchRecordingWriter {
        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
            unreachable!("the async worker must use batch delivery")
        }

        fn send_frames(&mut self, frames: &[crate::interface::FrameBytes]) -> io::Result<()> {
            self.batch_tx.send(frames.len()).unwrap();
            if let Some(release) = self.first_release.take() {
                release.recv().unwrap();
            }
            Ok(())
        }
    }

    impl Writer for FailingWriter {
        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
            Err(io::Error::new(io::ErrorKind::BrokenPipe, "boom"))
        }

        fn shutdown(&mut self) {
            self.shutdown_called.store(true, Ordering::Relaxed);
        }
    }

    #[test]
    fn async_writer_returns_wouldblock_when_queue_is_full() {
        let (event_tx, _event_rx) = crate::event::channel();
        let (entered_tx, entered_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let (mut writer, metrics) = wrap_async_writer(
            Box::new(BlockingWriter {
                entered_tx,
                release_rx,
            }),
            InterfaceId(7),
            "test",
            event_tx,
            1,
        );

        writer.send_frame(&[1]).unwrap();
        entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        writer.send_frame(&[2]).unwrap();
        assert_eq!(metrics.queued_frames(), 1);
        let err = writer.send_frame(&[3]).unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);

        release_tx.send(()).unwrap();
        entered_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("the queued frame must wake the writer without another send trigger");
        assert_eq!(metrics.queued_frames(), 0);
        release_tx.send(()).unwrap();
    }

    #[test]
    fn deferred_sends_are_delivered_on_wake_or_by_the_recheck_backstop() {
        struct Recorder(mpsc::Sender<Vec<u8>>);
        impl Writer for Recorder {
            fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
                self.0.send(data.to_vec()).unwrap();
                Ok(())
            }
        }
        let (event_tx, _event_rx) = crate::event::channel();
        let (sent_tx, sent_rx) = mpsc::channel();
        let (mut writer, _metrics) = wrap_async_writer(
            Box::new(Recorder(sent_tx)),
            InterfaceId(9),
            "test",
            event_tx,
            8,
        );
        // Let the worker go to sleep on the empty queue.
        std::thread::sleep(Duration::from_millis(20));
        writer.send_frame_deferred(&[1]).unwrap();
        writer.send_frame_deferred(&[2]).unwrap();
        writer.wake();
        assert_eq!(
            sent_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
            vec![1]
        );
        assert_eq!(
            sent_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
            vec![2]
        );

        std::thread::sleep(Duration::from_millis(20));
        writer.send_frame_deferred(&[3]).unwrap();
        // No wake: the periodic recheck still delivers the frame.
        assert_eq!(
            sent_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
            vec![3]
        );
    }

    #[test]
    fn async_writer_waits_briefly_for_space_then_fails_fast_while_stalled() {
        let (event_tx, _event_rx) = crate::event::channel();
        let (entered_tx, entered_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let (mut writer, metrics) = wrap_async_writer(
            Box::new(BlockingWriter {
                entered_tx,
                release_rx,
            }),
            InterfaceId(8),
            "test",
            event_tx,
            1,
        );
        writer.send_frame(&[1]).unwrap();
        entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        writer.send_frame(&[2]).unwrap();

        // The worker frees the slot while the producer is waiting: the frame
        // is admitted instead of being refused.
        let release = release_tx.clone();
        let releaser = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(5));
            release.send(()).unwrap();
        });
        writer.send_frame(&[3]).unwrap();
        releaser.join().unwrap();
        entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        assert_eq!(metrics.queued_frames(), 1);

        // A writer that stays full costs one bounded wait; later sends do not
        // wait again until a frame is admitted.
        let started = Instant::now();
        let err = writer.send_frame(&[4]).unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
        assert!(started.elapsed() >= ASYNC_WRITER_BACKPRESSURE_WAIT);
        let started = Instant::now();
        let err = writer.send_frame(&[5]).unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
        assert!(started.elapsed() < ASYNC_WRITER_BACKPRESSURE_WAIT);

        for _ in 0..3 {
            let _ = release_tx.send(());
        }
    }

    #[test]
    fn async_writer_shares_exact_egress_byte_valve_and_stall_gate() {
        let (event_tx, _event_rx) = crate::event::channel();
        let (entered_tx, entered_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let control = EgressControl::new(4);
        let (mut writer, metrics) = wrap_async_writer(
            Box::new(EgressControlledBlockingWriter {
                inner: BlockingWriter {
                    entered_tx,
                    release_rx,
                },
                control: control.clone(),
            }),
            InterfaceId(8),
            "egress-test",
            event_tx,
            8,
        );

        // Two ordinary bytes occupy exactly four HDLC bytes on wire.
        writer.send_frame(&[1, 2]).unwrap();
        entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        assert_eq!(control.reserved_bytes.load(Ordering::Relaxed), 4);
        assert_eq!(metrics.tx_buffered(), 4);

        // The hard valve drops instead of retaining another four-byte frame.
        writer.send_frame(&[3, 4]).unwrap();
        assert_eq!(control.reserved_bytes.load(Ordering::Relaxed), 4);
        assert_eq!(control.dropped_frames.load(Ordering::Relaxed), 1);
        assert_eq!(control.dropped_bytes.load(Ordering::Relaxed), 4);
        assert_eq!(metrics.tx_drops(), 1);
        assert_eq!(metrics.tx_dropped_bytes(), 4);

        release_tx.send(()).unwrap();
        let deadline = Instant::now() + Duration::from_secs(1);
        while control.reserved_bytes.load(Ordering::Relaxed) != 0 {
            assert!(
                Instant::now() < deadline,
                "egress reservation did not drain"
            );
            thread::yield_now();
        }

        control.set_stalled(true);
        assert!(metrics.tx_stalled());
        writer.send_frame(&[5]).unwrap();
        assert_eq!(control.dropped_frames.load(Ordering::Relaxed), 2);
        assert_eq!(control.dropped_bytes.load(Ordering::Relaxed), 7);
        assert_eq!(metrics.queued_frames(), 0);
    }

    #[test]
    fn async_writer_reports_interface_down_after_worker_failure() {
        let (event_tx, event_rx) = crate::event::channel();
        let shutdown_called = Arc::new(AtomicBool::new(false));
        let (mut writer, metrics) = wrap_async_writer(
            Box::new(FailingWriter {
                shutdown_called: Arc::clone(&shutdown_called),
            }),
            InterfaceId(9),
            "fail",
            event_tx,
            2,
        );

        writer.send_frame(&[1]).unwrap();
        let event = event_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9))));
        assert!(!metrics.worker_alive());
        assert!(shutdown_called.load(Ordering::Relaxed));
    }

    #[test]
    fn async_writer_preserves_order_through_fifty_thousand_frame_burst() {
        const FRAME_COUNT: usize = 50_000;
        let (event_tx, _event_rx) = crate::event::channel();
        let (sent_tx, sent_rx) = mpsc::channel();
        let (mut writer, metrics) = wrap_async_writer(
            Box::new(RecordingWriter { sent_tx }),
            InterfaceId(11),
            "burst",
            event_tx,
            FRAME_COUNT,
        );

        for index in 0..FRAME_COUNT {
            let payload = (index as u64).to_be_bytes();
            writer.send_frame(&payload).unwrap();
        }

        for index in 0..FRAME_COUNT {
            let received = sent_rx.recv_timeout(Duration::from_secs(2)).unwrap();
            assert_eq!(received, (index as u64).to_be_bytes());
        }
        assert!(metrics.worker_alive());
    }

    #[test]
    fn async_writer_coalesces_frames_waiting_behind_an_active_batch() {
        let (event_tx, _event_rx) = crate::event::channel();
        let (batch_tx, batch_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let (mut writer, _) = wrap_async_writer(
            Box::new(BatchRecordingWriter {
                batch_tx,
                first_release: Some(release_rx),
            }),
            InterfaceId(12),
            "batch",
            event_tx,
            16,
        );

        writer.send_frame(&[0]).unwrap();
        assert_eq!(batch_rx.recv_timeout(Duration::from_secs(1)).unwrap(), 1);
        for index in 1..=10 {
            writer.send_frame(&[index]).unwrap();
        }
        release_tx.send(()).unwrap();

        assert_eq!(batch_rx.recv_timeout(Duration::from_secs(1)).unwrap(), 10);
    }
}