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::thread;
use std::time::Duration;

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

use crate::event::{Event, EventSender};
use crate::hdlc;
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))
    }
}

/// 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) {
                        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));

    // 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,
                            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,
                }
            }
        }
    }
}

/// 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 {
            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,
                                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),
            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,
                                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);
    }

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