rns_net/interface/

pipe.rs

//! Pipe interface: subprocess stdin/stdout with HDLC framing.
//!
//! Matches Python `PipeInterface.py`. Spawns a subprocess, communicates
//! via piped stdin/stdout using HDLC framing for packet boundaries.
//! Auto-respawns subprocess on failure.

use std::collections::HashMap;
use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

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

use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};
use crate::event::{Event, EventSender, LoopContext, LoopSource, SourceStatus};
use crate::hdlc;
use crate::interface::{lock_or_recover, Writer};

/// Configuration for a pipe interface.
#[derive(Debug, Clone)]
pub struct PipeConfig {
    pub name: String,
    pub command: String,
    pub respawn_delay: Duration,
    pub interface_id: InterfaceId,
    pub runtime: Arc<Mutex<PipeRuntime>>,
}

#[derive(Debug, Clone)]
pub struct PipeRuntime {
    pub respawn_delay: Duration,
}

impl PipeRuntime {
    pub fn from_config(config: &PipeConfig) -> Self {
        Self {
            respawn_delay: config.respawn_delay,
        }
    }
}

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

impl Default for PipeConfig {
    fn default() -> Self {
        let mut config = PipeConfig {
            name: String::new(),
            command: String::new(),
            respawn_delay: Duration::from_secs(5),
            interface_id: InterfaceId(0),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_secs(5),
            })),
        };
        let startup = PipeRuntime::from_config(&config);
        config.runtime = Arc::new(Mutex::new(startup));
        config
    }
}

/// Writer that sends HDLC-framed data to a subprocess stdin.
struct PipeWriter {
    stdin: std::process::ChildStdin,
}

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

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

/// Start the pipe interface. Spawns subprocess, returns writer.
pub fn start(config: PipeConfig, tx: EventSender) -> io::Result<Box<dyn Writer>> {
    let id = config.interface_id;
    {
        let startup = PipeRuntime::from_config(&config);
        *lock_or_recover(&config.runtime, "pipe runtime") = startup;
    }

    let mut child = spawn_child(&config.command)?;

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| io::Error::other("no stdout from child"))?;
    let stdin = child
        .stdin
        .take()
        .ok_or_else(|| io::Error::other("no stdin from child"))?;

    log::info!(
        "[{}] pipe interface started: {}",
        config.name,
        config.command
    );

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

    if crate::engine::FOLD_IO {
        Arc::new(PipeLink { config, tx }).serve(stdout, child)?;
        return Ok(Box::new(PipeWriter { stdin }));
    }

    // Spawn reader thread
    thread::Builder::new()
        .name(format!("pipe-reader-{}", id.0))
        .spawn(move || {
            reader_loop(stdout, child, id, config, tx);
        })?;

    Ok(Box::new(PipeWriter { stdin }))
}

fn spawn_child(command: &str) -> io::Result<std::process::Child> {
    Command::new("sh")
        .args(["-c", command])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
}

/// Reader loop: reads from subprocess stdout, HDLC-decodes, dispatches events.
/// On subprocess exit, attempts respawn.
fn reader_loop(
    mut stdout: std::process::ChildStdout,
    mut child: std::process::Child,
    id: InterfaceId,
    config: PipeConfig,
    tx: EventSender,
) {
    let mut decoder = hdlc::Decoder::new();
    let mut buf = [0u8; 4096];

    loop {
        match stdout.read(&mut buf) {
            Ok(0) => {
                // EOF — subprocess exited
                let _ = child.wait();
                log::warn!("[{}] subprocess terminated", config.name);
                let _ = tx.send(Event::InterfaceDown(id));
                match respawn(&config, &tx) {
                    Some((new_stdout, new_child)) => {
                        stdout = new_stdout;
                        child = new_child;
                        decoder = hdlc::Decoder::new();
                        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()
                    {
                        // Driver shut down
                        let _ = child.kill();
                        return;
                    }
                }
            }
            Err(e) => {
                log::warn!("[{}] pipe read error: {}", config.name, e);
                let _ = child.kill();
                let _ = child.wait();
                let _ = tx.send(Event::InterfaceDown(id));
                match respawn(&config, &tx) {
                    Some((new_stdout, new_child)) => {
                        stdout = new_stdout;
                        child = new_child;
                        decoder = hdlc::Decoder::new();
                        continue;
                    }
                    None => return,
                }
            }
        }
    }
}

/// Attempt to respawn the subprocess after a delay.
fn respawn(
    config: &PipeConfig,
    tx: &EventSender,
) -> Option<(std::process::ChildStdout, std::process::Child)> {
    loop {
        let respawn_delay = lock_or_recover(&config.runtime, "pipe runtime").respawn_delay;
        thread::sleep(respawn_delay);
        log::info!(
            "[{}] attempting to respawn subprocess: {}",
            config.name,
            config.command
        );

        match spawn_child(&config.command) {
            Ok(mut child) => {
                let stdout = match child.stdout.take() {
                    Some(s) => s,
                    None => {
                        let _ = child.kill();
                        let _ = child.wait();
                        continue;
                    }
                };
                let stdin = match child.stdin.take() {
                    Some(s) => s,
                    None => {
                        let _ = child.kill();
                        let _ = child.wait();
                        continue;
                    }
                };

                let new_writer: Box<dyn Writer> = Box::new(PipeWriter { stdin });
                if tx
                    .send(Event::InterfaceUp(
                        config.interface_id,
                        Some(new_writer),
                        None,
                    ))
                    .is_err()
                {
                    return None; // Driver shut down
                }
                log::info!("[{}] subprocess respawned", config.name);
                return Some((stdout, child));
            }
            Err(e) => {
                log::warn!("[{}] respawn failed: {}", config.name, e);
            }
        }
    }
}

/// What reading a subprocess on the driver loop needs across respawns (see
/// [`crate::engine::FOLD_IO`]).
struct PipeLink {
    config: PipeConfig,
    tx: EventSender,
}

impl PipeLink {
    /// Read `stdout` on the driver thread from now on.
    fn serve(
        self: &Arc<Self>,
        stdout: std::process::ChildStdout,
        child: std::process::Child,
    ) -> io::Result<()> {
        crate::interface::stream_read::set_nonblocking(stdout.as_raw_fd())?;
        self.tx.register_source(
            crate::interface::loop_writer::read_key(self.config.interface_id),
            Box::new(PipeSource {
                stdout,
                child: Some(child),
                decoder: hdlc::Decoder::new(),
                link: Arc::clone(self),
            }),
        )
    }

    /// Report the interface down, and on a thread of its own wait for the
    /// subprocess, respawn it, and serve the new one on the driver.
    fn subprocess_lost(self: &Arc<Self>, mut child: std::process::Child, kill: bool) {
        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!("pipe-respawn-{}", id.0))
            .spawn(move || {
                if kill {
                    let _ = child.kill();
                }
                let _ = child.wait();
                while let Some((stdout, child)) = respawn(&link.config, &link.tx) {
                    match link.serve(stdout, child) {
                        Ok(()) => return,
                        Err(e) => {
                            log::warn!("[{}] cannot read the subprocess: {}", link.config.name, e);
                            let _ = link.tx.send(Event::InterfaceDown(id));
                        }
                    }
                }
            });
        if let Err(e) = spawned {
            log::error!(
                "[{}] cannot start respawning, giving up: {}",
                self.config.name,
                e
            );
        }
    }
}

/// A subprocess's stdout, read on the driver thread.
struct PipeSource {
    stdout: std::process::ChildStdout,
    /// Taken when the subprocess is handed on: to the respawn thread, or
    /// killed when the source goes away.
    child: Option<std::process::Child>,
    decoder: hdlc::Decoder,
    link: Arc<PipeLink>,
}

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

    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
        use crate::interface::stream_read::{read_device_frames, StreamRead};
        let link = &self.link;
        let name = &link.config.name;
        let kill = match read_device_frames(
            self.stdout.as_raw_fd(),
            &mut self.decoder,
            cx,
            &link.tx,
            link.config.interface_id,
            name,
            |_, _| {},
        ) {
            StreamRead::Rearm => return SourceStatus::Rearm,
            StreamRead::Suspend => return SourceStatus::Suspend,
            StreamRead::DriverGone => return SourceStatus::Remove,
            StreamRead::Closed => {
                log::warn!("[{}] subprocess terminated", name);
                false
            }
            StreamRead::Failed(e) => {
                log::warn!("[{}] pipe read error: {}", name, e);
                true
            }
        };
        if let Some(child) = self.child.take() {
            link.subprocess_lost(child, kill);
        }
        SourceStatus::Remove
    }

    fn on_removed(&mut self) {
        // The driver is going away: so does the subprocess, as with the
        // reader thread.
        if let Some(mut child) = self.child.take() {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

/// Factory for [`PipeInterface`] instances.
pub struct PipeFactory;

impl InterfaceFactory for PipeFactory {
    fn type_name(&self) -> &str {
        "PipeInterface"
    }

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

        let respawn_delay = match params.get("respawn_delay") {
            Some(v) => {
                let ms: u64 = v
                    .parse()
                    .map_err(|_| format!("invalid respawn_delay: {}", v))?;
                Duration::from_millis(ms)
            }
            None => Duration::from_secs(5),
        };

        Ok(Box::new(PipeConfig {
            name: name.to_string(),
            command,
            respawn_delay,
            interface_id: id,
            runtime: Arc::new(Mutex::new(PipeRuntime { respawn_delay })),
        }))
    }

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

        let id = pipe_config.interface_id;
        let info = InterfaceInfo {
            stamp_costs: Default::default(),
            id,
            name: pipe_config.name.clone(),
            mode: ctx.mode,
            gravity: ctx.gravity,
            recursive_prs: ctx.recursive_prs,
            announces_from_internal: ctx.announces_from_internal,
            announces_to_internal: ctx.announces_to_internal,
            out_capable: true,
            in_capable: true,
            bitrate: Some(1_000_000),
            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 writer = start(pipe_config, ctx.tx)?;

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

pub(crate) fn pipe_runtime_handle_from_config(config: &PipeConfig) -> PipeRuntimeConfigHandle {
    PipeRuntimeConfigHandle {
        interface_name: config.name.clone(),
        runtime: Arc::clone(&config.runtime),
        startup: PipeRuntime::from_config(config),
    }
}

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

    #[test]
    fn pipe_start_and_receive() {
        // Use `cat` as a loopback subprocess
        let (tx, rx) = crate::event::channel();
        let config = PipeConfig {
            name: "test-pipe".into(),
            command: "cat".into(),
            respawn_delay: Duration::from_secs(1),
            interface_id: InterfaceId(100),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_secs(1),
            })),
        };

        let mut writer = start(config, tx).unwrap();

        // Drain InterfaceUp
        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert!(matches!(
            event,
            Event::InterfaceUp(InterfaceId(100), None, None)
        ));

        // Send a packet (>= 19 bytes for HDLC minimum)
        let payload: Vec<u8> = (0..32).collect();
        writer.send_frame(&payload).unwrap();

        // Should receive Frame event (cat echos back the HDLC frame)
        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        match event {
            Event::Frame {
                interface_id,
                data,
                rssi: _,
                snr: _,
            } => {
                assert_eq!(interface_id, InterfaceId(100));
                assert_eq!(data, payload.into());
            }
            other => panic!("expected Frame, got {:?}", other),
        }
    }

    #[test]
    fn pipe_writer_sends() {
        // Verify the writer wraps data in HDLC
        let (tx, rx) = crate::event::channel();
        let config = PipeConfig {
            name: "test-pipe-writer".into(),
            command: "cat".into(),
            respawn_delay: Duration::from_secs(1),
            interface_id: InterfaceId(101),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_secs(1),
            })),
        };

        let mut writer = start(config, tx).unwrap();
        let _ = rx.recv_timeout(Duration::from_secs(2)).unwrap(); // drain InterfaceUp

        // Write data and verify we get it back as HDLC frame
        let payload: Vec<u8> = (10..42).collect();
        writer.send_frame(&payload).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),
        }
    }

    #[test]
    fn pipe_subprocess_exit() {
        // Use a command that exits immediately
        let (tx, rx) = crate::event::channel();
        let config = PipeConfig {
            name: "test-pipe-exit".into(),
            command: "true".into(),                 // exits immediately with 0
            respawn_delay: Duration::from_secs(60), // long delay so we catch InterfaceDown
            interface_id: InterfaceId(102),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_secs(60),
            })),
        };

        let _writer = start(config, tx).unwrap();

        // Should get InterfaceUp then InterfaceDown
        let mut got_down = false;
        for _ in 0..5 {
            match rx.recv_timeout(Duration::from_secs(2)) {
                Ok(Event::InterfaceDown(InterfaceId(102))) => {
                    got_down = true;
                    break;
                }
                Ok(_) => continue,
                Err(_) => break,
            }
        }
        assert!(
            got_down,
            "should receive InterfaceDown after subprocess exits"
        );
    }

    #[test]
    fn folded_pipe_runs_on_the_driver_loop_and_respawns() {
        if !crate::engine::FOLD_IO {
            return;
        }
        let (tx, rx) = crate::event::channel();
        let config = PipeConfig {
            name: "test-pipe-folded".into(),
            // Echoes one framed 32-byte packet, then exits.
            command: "head -c 34".into(),
            respawn_delay: Duration::from_millis(100),
            interface_id: InterfaceId(104),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_millis(100),
            })),
        };
        let writer = start(config, tx.clone()).unwrap();
        assert!(writer.loop_socket().is_some());
        let (mut writer, _) =
            crate::interface::wrap_async_writer(writer, InterfaceId(104), "pipe", tx.clone(), 16);
        assert!(matches!(
            rx.recv_timeout(Duration::from_secs(2)).unwrap(),
            Event::InterfaceUp(InterfaceId(104), None, None)
        ));
        let threads = || {
            std::fs::read_dir("/proc/self/task")
                .unwrap()
                .filter_map(|task| std::fs::read_to_string(task.unwrap().path().join("comm")).ok())
                .collect::<Vec<_>>()
        };
        assert!(!threads().iter().any(|t| t.starts_with("pipe-reader-104")));

        // Written and read back by the driver.
        let payload: Vec<u8> = (0..32).collect();
        writer.send_frame(&payload).unwrap();
        writer.flush_pending();
        match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
            Event::Frame { data, .. } => assert_eq!(data, payload.clone().into()),
            other => panic!("expected Frame, got {other:?}"),
        }

        // The subprocess exits after one packet: the interface goes down,
        // and a new one is spawned and served.
        let mut down = false;
        let mut new_writer = None;
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while new_writer.is_none() && std::time::Instant::now() < deadline {
            match rx.recv_timeout(Duration::from_millis(500)) {
                Ok(Event::InterfaceDown(InterfaceId(104))) => down = true,
                Ok(Event::InterfaceUp(InterfaceId(104), Some(writer), None)) => {
                    new_writer = Some(writer)
                }
                _ => {}
            }
        }
        assert!(down);
        let (mut writer, _) = crate::interface::wrap_async_writer(
            new_writer.expect("respawned"),
            InterfaceId(104),
            "pipe",
            tx,
            16,
        );
        writer.send_frame(&payload).unwrap();
        writer.flush_pending();
        match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
            Event::Frame { data, .. } => assert_eq!(data, payload.into()),
            other => panic!("expected Frame, got {other:?}"),
        }
    }

    #[test]
    fn pipe_config_defaults() {
        let config = PipeConfig::default();
        assert_eq!(config.respawn_delay, Duration::from_secs(5));
        assert_eq!(config.interface_id, InterfaceId(0));
        assert!(config.command.is_empty());
    }

    #[test]
    fn pipe_invalid_command() {
        let (tx, _rx) = crate::event::channel();
        let config = PipeConfig {
            name: "test-pipe-bad".into(),
            command: "/nonexistent_rns_test_binary_that_does_not_exist_xyz".into(),
            respawn_delay: Duration::from_secs(60),
            interface_id: InterfaceId(103),
            runtime: Arc::new(Mutex::new(PipeRuntime {
                respawn_delay: Duration::from_secs(60),
            })),
        };

        // sh -c <nonexistent> will start sh successfully but the child exits immediately
        // For a truly invalid binary we need to check that the process fails
        // Actually, sh -c "<nonexistent>" will still spawn sh successfully
        // Let's test with a different approach: verify it doesn't panic
        let result = start(config, tx);
        // sh will spawn successfully even if the inner command fails
        // The real test is that it doesn't panic
        assert!(result.is_ok());
    }
}