Introduce a client-server model

Cargo.toml

@@ -20,5 +20,7 @@ krun-sys = { path = "crates/krun-sys", default-features = false }
 log = { version = "0.4.21", default-features = false }
 nix = { version = "0.28.0", default-features = false }
 rustix = { version = "0.38.34", default-features = false }
+serde = { version = "1.0.203", default-features = false }
+serde_json = { version = "1.0.117", default-features = false }
 tempfile = { version = "3.10.1", default-features = false }
 utils = { path = "crates/utils", default-features = false }

crates/krun-server/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "krun-server"
+version = { workspace = true }
+authors = { workspace = true }
+edition = { workspace = true }
+rust-version = { workspace = true }
+description = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true, features = ["std"] }
+bpaf = { workspace = true, features = [] }
+env_logger = { workspace = true, features = ["auto-color", "humantime", "unstable-kv"] }
+log = { workspace = true, features = ["kv"] }
+nix = { workspace = true, features = ["socket"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+utils = { workspace = true, features = [] }
+
+[features]
+default = []

crates/krun-server/src/bin/krun-server.rs

@@ -0,0 +1,33 @@
+use std::net::TcpListener;
+use std::os::fd::AsRawFd;
+use std::process::Command;
+
+use anyhow::Result;
+use krun_server::cli_options::options;
+use krun_server::server::start_server;
+use nix::sys::socket::{shutdown, Shutdown};
+
+fn main() -> Result<()> {
+    env_logger::init();
+
+    let options = options().run();
+
+    let listener = TcpListener::bind(format!("127.0.0.1:{}", &options.server_port))?;
+    let listener_fd = listener.as_raw_fd();
+
+    let server_thread = start_server(listener);
+
+    let status = Command::new(&options.command)
+        .args(options.command_args)
+        .status();
+    if let Err(err) = status {
+        println!("Error executing command {}: {}", options.command, err);
+    }
+
+    shutdown(listener_fd, Shutdown::Both)?;
+    if let Err(err) = server_thread.join() {
+        println!("Error waiting for server thread termination: {err:?}");
+    }
+
+    Ok(())
+}

crates/krun-server/src/cli_options.rs

@@ -0,0 +1,40 @@
+use bpaf::{any, construct, env, positional, OptionParser, Parser};
+
+#[derive(Clone, Debug)]
+pub struct Options {
+    pub server_port: u32,
+    pub command: String,
+    pub command_args: Vec<String>,
+}
+
+pub fn options() -> OptionParser<Options> {
+    let server_port = env("KRUN_SERVER_PORT")
+        .short('p')
+        .help("TCP port to listen for command launch requests")
+        .argument("SERVER_PORT")
+        .fallback(3333)
+        .display_fallback();
+    let command = positional("COMMAND");
+    let command_args = any::<String, _, _>("COMMAND_ARGS", |arg| {
+        (!["--help", "-h"].contains(&&*arg)).then_some(arg)
+    })
+    .many();
+
+    construct!(Options {
+        server_port,
+        // positionals
+        command,
+        command_args,
+    })
+    .to_options()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn check_options() {
+        options().check_invariants(false)
+    }
+}

crates/krun-server/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod cli_options;
+pub mod server;

crates/krun-server/src/server.rs

@@ -0,0 +1,87 @@
+use std::{
+    collections::HashMap,
+    env,
+    fs::File,
+    io::{BufRead, BufReader, Write},
+    net::{TcpListener, TcpStream},
+    path::Path,
+    process::Command,
+    thread::{self, JoinHandle},
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use anyhow::{anyhow, Result};
+use log::debug;
+use utils::launch::Launch;
+
+pub fn start_server(listener: TcpListener) -> JoinHandle<()> {
+    thread::spawn(move || {
+        if let Err(err) = work(listener) {
+            debug!("krun server thread is terminating: {err:?}")
+        }
+    })
+}
+
+fn work(listener: TcpListener) -> Result<()> {
+    for stream in listener.incoming() {
+        let stream = stream?;
+
+        if let Err(e) = handle_connection(stream) {
+            println!("Error processing client request: {e}");
+        }
+    }
+
+    Ok(())
+}
+
+fn read_request(mut stream: &TcpStream) -> Result<Launch> {
+    let mut buf_reader = BufReader::new(&mut stream);
+    let mut buf = String::new();
+    loop {
+        if buf_reader.read_line(&mut buf)? == 0 {
+            return Err(anyhow!("Unexpected EOF"));
+        }
+        if buf.contains("EOM") {
+            let launch: Launch = serde_json::from_str(&buf[..buf.len() - 5])?;
+            return Ok(launch);
+        }
+    }
+}
+
+fn handle_connection(mut stream: TcpStream) -> Result<()> {
+    let mut envs: HashMap<String, String> = env::vars().collect();
+
+    let launch = read_request(&stream)?;
+    envs.extend(launch.envs);
+
+    let (stdout, stderr) = {
+        let base = if envs.contains_key("XDG_RUNTIME_DIR") {
+            Path::new(&envs["XDG_RUNTIME_DIR"])
+        } else {
+            Path::new("/tmp")
+        };
+        let ts = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis();
+        let path_stdout = base.join(format!("krun-{}-{}.stdout", launch.command, ts));
+        let path_stderr = base.join(format!("krun-{}-{}.stderr", launch.command, ts));
+        (
+            File::create_new(path_stdout)?,
+            File::create_new(path_stderr)?,
+        )
+    };
+
+    let err = Command::new(&launch.command)
+        .args(launch.args)
+        .envs(envs)
+        .stdout(stdout)
+        .stderr(stderr)
+        .spawn();
+    if let Err(err) = err {
+        let error = format!("Error executing command {}: {}", launch.command, err);
+        let _ = stream.write_all(error.as_bytes());
+    } else {
+        let _ = stream.write_all(b"OK");
+    }
+    let _ = stream.flush();
+
+    Ok(())
+}

crates/krun/Cargo.toml

@@ -16,6 +16,8 @@ krun-sys = { workspace = true, features = [] }
 log = { workspace = true, features = ["kv"] }
 nix = { workspace = true, features = ["user"] }
 rustix = { workspace = true, features = ["process", "std", "use-libc-auxv"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
 utils = { workspace = true, features = [] }
 
 [features]