rns_net/interface/

serial_iface.rs

//! Serial interface with HDLC framing.
//!
//! Matches Python `SerialInterface.py` — opens a serial port,
//! reads with HDLC framing, reconnects on failure.

use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::sync::Arc;
use std::thread;
use std::time::Duration;

use rns_core::transport::types::InterfaceId;

use crate::event::{Event, EventSender, LoopContext, LoopSource, SourceStatus};
use crate::hdlc;
use crate::interface::stream_read::{read_device_frames, StreamRead};
use crate::interface::{
    poll_readable, wait_while_active, ListenerControl, UnavailableWriter, Writer,
};
use crate::serial::{Parity, SerialConfig, SerialPort};

/// Upstream `SerialInterface.HW_MTU`.
const HW_MTU: usize = 564;

/// HDLC decoder bounded by the serial hardware MTU plus on-wire IFAC bytes.
fn frame_decoder(ifac_size: usize) -> hdlc::Decoder {
    hdlc::Decoder::reticulum(HW_MTU, ifac_size)
}

/// Configuration for a Serial interface.
#[derive(Debug, Clone)]
pub struct SerialIfaceConfig {
    pub name: String,
    pub port: String,
    pub speed: u32,
    pub data_bits: u8,
    pub parity: Parity,
    pub stop_bits: u8,
    pub interface_id: InterfaceId,
}

impl Default for SerialIfaceConfig {
    fn default() -> Self {
        SerialIfaceConfig {
            name: String::new(),
            port: String::new(),
            speed: 9600,
            data_bits: 8,
            parity: Parity::None,
            stop_bits: 1,
            interface_id: InterfaceId(0),
        }
    }
}

/// Writer that sends HDLC-framed data over a serial port.
struct SerialWriter {
    file: std::fs::File,
}

impl Writer for SerialWriter {
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        self.file.write_all(&hdlc::frame(data))
    }

    fn loop_socket(&self) -> Option<std::os::fd::RawFd> {
        // With interface I/O folded the driver writes the port itself.
        crate::engine::FOLD_IO.then(|| self.file.as_raw_fd())
    }
}

/// Start the serial interface. Opens the port, spawns reader thread.
/// Returns the writer for the driver.
pub fn start(config: SerialIfaceConfig, tx: EventSender) -> io::Result<Box<dyn Writer>> {
    start_with_control(config, tx, 0).map(|(writer, _)| writer)
}

fn start_with_control(
    config: SerialIfaceConfig,
    tx: EventSender,
    ifac_size: usize,
) -> io::Result<(Box<dyn Writer>, ListenerControl)> {
    let serial_config = SerialConfig {
        path: config.port.clone(),
        baud: config.speed,
        data_bits: config.data_bits,
        parity: config.parity,
        stop_bits: config.stop_bits,
    };

    let control = ListenerControl::new();
    let port = match SerialPort::open(&serial_config) {
        Ok(port) => port,
        Err(error) => {
            log::warn!(
                "[{}] could not open serial port {}: {}; retrying",
                config.name,
                config.port,
                error
            );
            let retry_control = control.clone();
            thread::Builder::new()
                .name(format!("serial-reconnect-{}", config.interface_id.0))
                .spawn(move || {
                    if let Some(reader) = reconnect(&config, &tx, &retry_control) {
                        if crate::engine::FOLD_IO {
                            let link = Arc::new(SerialLink {
                                config,
                                tx,
                                control: retry_control,
                                ifac_size,
                            });
                            link.serve_or_retry(reader);
                            return;
                        }
                        let id = config.interface_id;
                        reader_loop(reader, id, config, tx, retry_control, ifac_size);
                    }
                })?;
            return Ok((Box::new(UnavailableWriter), control));
        }
    };
    let reader_file = port.reader()?;
    let writer_file = port.writer()?;

    let id = config.interface_id;

    // Signal interface up
    let _ = tx.send(Event::InterfaceUp(id, None, None));

    // Short delay matching Python's configure_device sleep
    thread::sleep(Duration::from_millis(500));

    if crate::engine::FOLD_IO {
        let link = Arc::new(SerialLink {
            config,
            tx,
            control: control.clone(),
            ifac_size,
        });
        link.serve(reader_file)?;
        return Ok((Box::new(SerialWriter { file: writer_file }), control));
    }

    // Spawn reader thread
    let reader_control = control.clone();
    thread::Builder::new()
        .name(format!("serial-reader-{}", id.0))
        .spawn(move || {
            reader_loop(reader_file, id, config, tx, reader_control, ifac_size);
        })?;

    Ok((Box::new(SerialWriter { file: writer_file }), control))
}

/// Reader thread: reads from serial, HDLC-decodes, sends frames to driver.
fn reader_loop(
    mut reader: std::fs::File,
    id: InterfaceId,
    config: SerialIfaceConfig,
    tx: EventSender,
    control: ListenerControl,
    ifac_size: usize,
) {
    let mut decoder = frame_decoder(ifac_size);
    let mut buf = [0u8; 4096];

    loop {
        if control.should_stop() {
            return;
        }
        match poll_readable(reader.as_raw_fd(), Duration::from_millis(200)) {
            Ok(false) => continue,
            Ok(true) => {}
            Err(e) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] serial poll error: {}", config.name, e);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &control) {
                    Some(new_reader) => {
                        reader = new_reader;
                        decoder = frame_decoder(ifac_size);
                        continue;
                    }
                    None => return,
                }
            }
        }
        match reader.read(&mut buf) {
            Ok(0) => {
                if control.should_stop() {
                    return;
                }
                // EOF — port closed
                log::warn!("[{}] serial port closed", config.name);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &control) {
                    Some(new_reader) => {
                        reader = new_reader;
                        decoder = frame_decoder(ifac_size);
                        continue;
                    }
                    None => return,
                }
            }
            Ok(n) => {
                for frame in decoder.feed(&buf[..n]) {
                    if tx
                        .send(Event::Frame {
                            interface_id: id,
                            data: frame.into(),
                            rssi: None,
                            snr: None,
                        })
                        .is_err()
                    {
                        return; // driver shut down
                    }
                }
            }
            Err(e) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] serial read error: {}", config.name, e);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &control) {
                    Some(new_reader) => {
                        reader = new_reader;
                        decoder = frame_decoder(ifac_size);
                        continue;
                    }
                    None => return,
                }
            }
        }
    }
}

/// What a serial port read on the driver loop needs across reconnects (see
/// [`crate::engine::FOLD_IO`]).
struct SerialLink {
    config: SerialIfaceConfig,
    tx: EventSender,
    control: ListenerControl,
    ifac_size: usize,
}

impl SerialLink {
    /// Read `reader` on the driver thread from now on.
    fn serve(self: &Arc<Self>, reader: std::fs::File) -> io::Result<()> {
        crate::interface::stream_read::set_nonblocking(reader.as_raw_fd())?;
        self.tx.register_source(
            crate::interface::loop_writer::read_key(self.config.interface_id),
            Box::new(SerialSource {
                reader,
                decoder: frame_decoder(self.ifac_size),
                link: Arc::clone(self),
            }),
        )
    }

    /// [`SerialLink::serve`] a reopened port, reopening it again if that
    /// fails. Runs on a reconnect thread.
    fn serve_or_retry(self: &Arc<Self>, mut reader: std::fs::File) {
        loop {
            match self.serve(reader) {
                Ok(()) => return,
                Err(e) => {
                    log::warn!(
                        "[{}] cannot read the reopened port: {}",
                        self.config.name,
                        e
                    );
                    let _ = self.tx.send(Event::InterfaceDown(self.config.interface_id));
                }
            }
            match reconnect(&self.config, &self.tx, &self.control) {
                Some(new_reader) => reader = new_reader,
                None => return,
            }
        }
    }

    /// Report the port down and reopen it on a thread of its own, which
    /// exits once the reopened port is served on the driver.
    fn connection_lost(self: &Arc<Self>) {
        let id = self.config.interface_id;
        let _ = self.tx.push_from_loop([Event::InterfaceDown(id)]);
        let link = Arc::clone(self);
        let spawned = thread::Builder::new()
            .name(format!("serial-reconnect-{}", id.0))
            .spawn(move || {
                if let Some(reader) = reconnect(&link.config, &link.tx, &link.control) {
                    link.serve_or_retry(reader);
                }
            });
        if let Err(e) = spawned {
            log::error!(
                "[{}] cannot start reconnecting, giving up: {}",
                self.config.name,
                e
            );
        }
    }
}

/// The serial port, read on the driver thread.
struct SerialSource {
    reader: std::fs::File,
    decoder: hdlc::Decoder,
    link: Arc<SerialLink>,
}

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

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

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

/// Attempt to reconnect the serial port. Returns new reader file on success.
fn reconnect(
    config: &SerialIfaceConfig,
    tx: &EventSender,
    control: &ListenerControl,
) -> Option<std::fs::File> {
    loop {
        if !wait_while_active(control, Duration::from_secs(5)) {
            return None;
        }
        log::info!(
            "[{}] attempting to reconnect serial port {}...",
            config.name,
            config.port
        );

        let serial_config = SerialConfig {
            path: config.port.clone(),
            baud: config.speed,
            data_bits: config.data_bits,
            parity: config.parity,
            stop_bits: config.stop_bits,
        };

        match SerialPort::open(&serial_config) {
            Ok(port) => match (port.reader(), port.writer()) {
                (Ok(reader), Ok(writer_file)) => {
                    if control.should_stop() {
                        return None;
                    }
                    log::info!("[{}] serial port reconnected", config.name);
                    let new_writer: Box<dyn Writer> = Box::new(SerialWriter { file: writer_file });
                    let _ = tx.send(Event::InterfaceUp(
                        config.interface_id,
                        Some(new_writer),
                        None,
                    ));
                    if !wait_while_active(control, Duration::from_millis(500)) {
                        return None;
                    }
                    return Some(reader);
                }
                _ => {
                    log::warn!(
                        "[{}] failed to get reader/writer from serial port",
                        config.name
                    );
                }
            },
            Err(e) => {
                log::warn!("[{}] serial reconnect failed: {}", config.name, e);
            }
        }
    }
}

// --- Factory implementation ---

use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};
use rns_core::transport::types::InterfaceInfo;
use std::collections::HashMap;

/// Factory for `SerialInterface`.
pub struct SerialFactory;

impl InterfaceFactory for SerialFactory {
    fn type_name(&self) -> &str {
        "SerialInterface"
    }

    fn default_ifac_size(&self) -> usize {
        8
    }

    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        let port = params
            .get("port")
            .cloned()
            .ok_or_else(|| "SerialInterface requires 'port'".to_string())?;

        let speed = params
            .get("speed")
            .and_then(|v| v.parse().ok())
            .unwrap_or(9600u32);

        let data_bits = params
            .get("databits")
            .and_then(|v| v.parse().ok())
            .unwrap_or(8u8);

        let parity = params
            .get("parity")
            .map(|v| match v.to_lowercase().as_str() {
                "e" | "even" => crate::serial::Parity::Even,
                "o" | "odd" => crate::serial::Parity::Odd,
                _ => crate::serial::Parity::None,
            })
            .unwrap_or(crate::serial::Parity::None);

        let stop_bits = params
            .get("stopbits")
            .and_then(|v| v.parse().ok())
            .unwrap_or(1u8);

        Ok(Box::new(SerialIfaceConfig {
            name: name.to_string(),
            port,
            speed,
            data_bits,
            parity,
            stop_bits,
            interface_id: id,
        }))
    }

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

        let id = serial_config.interface_id;
        let name = serial_config.name.clone();
        let bitrate = Some(serial_config.speed as u64);

        let info = InterfaceInfo {
            stamp_costs: Default::default(),
            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,
            airtime_profile: None,
            announce_rate_target: None,
            announce_rate_grace: 0,
            announce_rate_penalty: 0.0,
            announce_cap: rns_core::constants::ANNOUNCE_CAP,
            is_local_client: false,
            wants_tunnel: false,
            tunnel_id: None,
            mtu: rns_core::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: crate::time::now(),
        };

        let ifac_size = ctx.ifac.as_ref().map_or(0, |ifac| ifac.size);
        let (writer, control) = start_with_control(serial_config, ctx.tx, ifac_size)?;

        Ok(StartResult::Simple {
            control: Some(control),
            id,
            info,
            writer,
            interface_type_name: "SerialInterface".to_string(),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::serial::open_pty_pair;
    use std::os::unix::io::{AsRawFd, FromRawFd};
    use std::sync::mpsc;
    use std::time::Duration;

    /// Helper: poll an fd for reading with timeout (ms). Returns true if data available.
    fn poll_read(fd: i32, timeout_ms: i32) -> bool {
        let mut pfd = libc::pollfd {
            fd,
            events: libc::POLLIN,
            revents: 0,
        };
        let ret = unsafe { libc::poll(&mut pfd, 1, timeout_ms) };
        ret > 0
    }

    #[test]
    fn serial_hdlc_roundtrip() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let mut slave_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        // Write an HDLC frame to the master side
        let payload: Vec<u8> = (0..32).collect();
        let framed = hdlc::frame(&payload);
        master_file.write_all(&framed).unwrap();
        master_file.flush().unwrap();

        // Read from slave using HDLC decoder
        assert!(
            poll_read(slave_file.as_raw_fd(), 2000),
            "should have data available"
        );

        let mut decoder = hdlc::Decoder::new();
        let mut buf = [0u8; 4096];
        let n = slave_file.read(&mut buf).unwrap();
        let frames = decoder.feed(&buf[..n]);
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0], payload);
    }

    #[test]
    fn serial_writer_frames() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();

        let writer_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let mut writer = SerialWriter { file: writer_file };

        let payload = vec![0x01, 0x02, 0x7E, 0x7D, 0x03]; // includes bytes that need HDLC escaping
        writer.send_frame(&payload).unwrap();

        // Read from master
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        assert!(poll_read(master_file.as_raw_fd(), 2000), "should have data");

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

    #[test]
    fn serial_fragmented_read() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let (tx, rx) = mpsc::channel::<Event>();

        // Write a frame in two parts to the master
        let payload: Vec<u8> = (0..32).collect();
        let framed = hdlc::frame(&payload);
        let mid = framed.len() / 2;

        // Spawn reader thread first (it will block waiting for data)
        let reader_thread = thread::spawn(move || {
            let mut reader = slave_file;
            let mut decoder = hdlc::Decoder::new();
            let mut buf = [0u8; 4096];

            loop {
                match reader.read(&mut buf) {
                    Ok(n) if n > 0 => {
                        if let Some(frame) = decoder.feed(&buf[..n]).into_iter().next() {
                            let _ = tx.send(Event::Frame {
                                interface_id: InterfaceId(0),
                                data: frame.into(),
                                rssi: None,
                                snr: None,
                            });
                            return;
                        }
                    }
                    _ => return,
                }
            }
        });

        // Write first half
        master_file.write_all(&framed[..mid]).unwrap();
        master_file.flush().unwrap();

        // Small delay then write second half
        thread::sleep(Duration::from_millis(50));
        master_file.write_all(&framed[mid..]).unwrap();
        master_file.flush().unwrap();

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

        let _ = reader_thread.join();
    }

    #[test]
    fn serial_reconnect_on_close() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();

        let (tx, rx) = mpsc::channel::<Event>();
        let id = InterfaceId(42);

        // Spawn a reader thread on the slave side
        let slave_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let reader_thread = thread::spawn(move || {
            let mut reader = slave_file;
            let mut buf = [0u8; 4096];
            let mut decoder = hdlc::Decoder::new();
            loop {
                match reader.read(&mut buf) {
                    Ok(0) => {
                        let _ = tx.send(Event::InterfaceDown(id));
                        return;
                    }
                    Ok(n) => {
                        for frame in decoder.feed(&buf[..n]) {
                            let _ = tx.send(Event::Frame {
                                interface_id: id,
                                data: frame.into(),
                                rssi: None,
                                snr: None,
                            });
                        }
                    }
                    Err(_) => {
                        let _ = tx.send(Event::InterfaceDown(id));
                        return;
                    }
                }
            }
        });

        // Close the master fd — this should cause the reader to get EOF
        unsafe { libc::close(master_fd) };

        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert!(matches!(event, Event::InterfaceDown(InterfaceId(42))));

        let _ = reader_thread.join();
    }

    #[test]
    fn detach_stops_idle_serial_reader_without_reconnect() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let (tx, rx) = crate::event::channel();
        let control = ListenerControl::new();
        let reader_control = control.clone();
        let config = SerialIfaceConfig {
            name: "detach-test".into(),
            interface_id: InterfaceId(71),
            ..Default::default()
        };

        let reader = thread::spawn(move || {
            reader_loop(slave, InterfaceId(71), config, tx, reader_control, 0)
        });
        thread::sleep(Duration::from_millis(50));
        control.request_stop();
        reader.join().unwrap();
        drop(master);
        assert!(rx.try_recv().is_err());
    }

    #[test]
    fn missing_serial_port_recovers_after_it_appears() {
        let tempdir = tempfile::tempdir().unwrap();
        let port_path = tempdir.path().join("serial-port");
        let (tx, rx) = crate::event::channel();
        let config = SerialIfaceConfig {
            name: "late-port".into(),
            port: port_path.display().to_string(),
            interface_id: InterfaceId(73),
            ..Default::default()
        };

        let (_writer, control) = start_with_control(config, tx, 0).unwrap();
        assert!(rx.try_recv().is_err());

        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave_path = std::fs::read_link(format!("/proc/self/fd/{slave_fd}")).unwrap();
        std::os::unix::fs::symlink(slave_path, &port_path).unwrap();
        let slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let event = rx.recv_timeout(Duration::from_secs(8)).unwrap();
        assert!(matches!(
            event,
            Event::InterfaceUp(InterfaceId(73), Some(_), None)
        ));
        control.request_stop();
        drop(slave);
        drop(master);
    }

    fn thread_named(prefix: &str) -> bool {
        std::fs::read_dir("/proc/self/task").unwrap().any(|task| {
            let comm = task.unwrap().path().join("comm");
            std::fs::read_to_string(comm).is_ok_and(|comm| comm.trim().starts_with(prefix))
        })
    }

    #[test]
    fn folded_serial_port_runs_on_the_driver_loop() {
        if !crate::engine::FOLD_IO {
            return;
        }
        let tempdir = tempfile::tempdir().unwrap();
        let port_path = tempdir.path().join("serial-port");
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave_path = std::fs::read_link(format!("/proc/self/fd/{slave_fd}")).unwrap();
        std::os::unix::fs::symlink(slave_path, &port_path).unwrap();
        let slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let (tx, rx) = crate::event::channel();
        let config = SerialIfaceConfig {
            name: "folded-serial".into(),
            port: port_path.display().to_string(),
            interface_id: InterfaceId(74),
            ..Default::default()
        };
        let (writer, control) = start_with_control(config, tx.clone(), 0).unwrap();
        assert!(matches!(
            rx.recv_timeout(Duration::from_secs(2)).unwrap(),
            Event::InterfaceUp(InterfaceId(74), None, None)
        ));
        assert!(writer.loop_socket().is_some());
        let (mut writer, _) =
            crate::interface::wrap_async_writer(writer, InterfaceId(74), "serial", tx, 16);
        assert!(!thread_named("serial-reader-7"));
        assert!(!thread_named("iface-writer-74"));

        // Inbound, in two pieces, read by the receiver's wait.
        let payload: Vec<u8> = (0..32).collect();
        let framed = hdlc::frame(&payload);
        master.write_all(&framed[..10]).unwrap();
        master.write_all(&framed[10..]).unwrap();
        match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
            Event::Frame {
                interface_id, data, ..
            } => {
                assert_eq!(interface_id, InterfaceId(74));
                assert_eq!(&*data, &payload[..]);
            }
            other => panic!("expected Frame, got {other:?}"),
        }

        // Outbound, written by the driver.
        let outbound: Vec<u8> = (100..140).collect();
        writer.send_frame(&outbound).unwrap();
        writer.flush_pending();
        assert!(poll_read(master.as_raw_fd(), 2000));
        let mut buf = [0u8; 256];
        let n = master.read(&mut buf).unwrap();
        assert_eq!(&buf[..n], &hdlc::frame(&outbound)[..]);

        // The port going away is reported, and reopening starts.
        drop(master);
        match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
            Event::InterfaceDown(id) => assert_eq!(id, InterfaceId(74)),
            other => panic!("expected InterfaceDown, got {other:?}"),
        }
        control.request_stop();
        drop(slave);
    }

    #[test]
    fn frame_decoder_accepts_frames_up_to_hw_mtu_plus_ifac() {
        let ifac_size = 8;
        let mut decoder = frame_decoder(ifac_size);
        let payload = vec![0x3c; HW_MTU + ifac_size];
        let decoded = decoder.feed_with_diagnostics(&hdlc::frame(&payload));
        assert_eq!(decoded.frames, vec![payload]);
        assert!(decoded.invalid_frame_lengths.is_empty());
    }

    #[test]
    fn frame_decoder_drops_frame_above_hw_mtu_as_invalid() {
        let mut decoder = frame_decoder(0);
        let decoded = decoder.feed_with_diagnostics(&hdlc::frame(&vec![0x3c; HW_MTU + 1]));
        assert!(decoded.frames.is_empty());
        assert_eq!(decoded.invalid_frame_lengths, vec![HW_MTU + 1]);
    }
}