diff --git a/README.md b/README.md index 34bfc94..9ad6993 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,25 @@ # rss — RuSt Supervisor -Lightweight, per-user process supervisor. No daemon, no root, no virtualization. +Lightweight, per-user process supervisor. No root, no virtualization. -## Philosophy +## Architecture -- **KISS** — one binary, one job. Each service gets its own supervisor process. -- **UNIX way** — PID files for state, no IPC, no sockets. -- **User-space** — everything under `~/.config/rss/` and `~/.local/share/rss/`. -- **No root** — runs entirely as your user. +- **Single daemon** (`rss daemon`) manages all services in-process +- **CLI** (`rss start`, `rss stop`, etc.) talks to the daemon via Unix socket +- **No PID files** for services — state lives in daemon memory +- **Config hot-reload** — rename/add/remove services live, daemon detects changes +- **Logs** written to `~/.local/share/rss/logs/.log` — readable by CLI directly ## Usage +First start the daemon (via systemd or terminal): + +``` +rss daemon +``` + +Then manage services: + ``` rss start Start a service rss stop Stop a service @@ -20,7 +29,6 @@ rss logs [-n N] [-f] View logs (last N, follow) rss enable [--now] Enable auto-start rss disable Disable auto-start rss list List all services -rss daemon Start all enabled services ``` ## Configuration @@ -34,7 +42,7 @@ command = "your command here" stop_command = "optional stop command" timeout_start = 15 timeout_stop = 10 -restart = "on-failure" # no, always, on-failure +restart = "on-failure" # no, always, unless-stopped, on-failure restart_delay = 3 restart_max = 5 # 0 = unlimited oneshot = false @@ -44,7 +52,7 @@ env = { KEY = "value" } Placeholders in `command` / `stop_command`: `{name}`, `{workdir}`, `{pid}`, `{log_file}`, plus any env var. -See `config.example.toml` for a full example. +Edit `config.toml` any time — the daemon picks up changes live. ## systemd (user mode) diff --git a/src/daemon.rs b/src/daemon.rs new file mode 100644 index 0000000..c701475 --- /dev/null +++ b/src/daemon.rs @@ -0,0 +1,441 @@ +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, SystemTime}; +use std::sync::atomic::{AtomicBool, Ordering}; + +use crate::config::{Config, RestartPolicy, ServiceConfig}; +use crate::state::State; + +static STOP: AtomicBool = AtomicBool::new(false); + +extern "C" fn handle_sigterm(_: i32) { + STOP.store(true, Ordering::Relaxed); +} + +struct Proc { + child: Child, + pid: u32, + cfg: ServiceConfig, + restart_count: u32, +} + +pub fn run(config_path: String) { + // Signal handler for graceful shutdown + let handler = nix::sys::signal::SigHandler::Handler(handle_sigterm); + let sig_action = nix::sys::signal::SigAction::new( + handler, + nix::sys::signal::SaFlags::SA_RESTART, + nix::sys::signal::SigSet::empty(), + ); + unsafe { + let _ = nix::sys::signal::sigaction(nix::sys::signal::Signal::SIGTERM, &sig_action); + let _ = nix::sys::signal::sigaction(nix::sys::signal::Signal::SIGINT, &sig_action); + } + + // Write daemon PID + let pid_path = crate::ipc::daemon_pid_path(); + State::write_pid(&pid_path).ok(); + + // Ensure socket path is clean + let sock_path = crate::ipc::socket_path(); + if sock_path.exists() { + let _ = std::fs::remove_file(&sock_path); + } + if let Some(parent) = sock_path.parent() { + let _ = std::fs::create_dir_all(parent); + } + + // Load config + let mut config = match Config::load(&config_path) { + Ok(c) => c, + Err(e) => { + eprintln!("error: config: {}", e); + return; + } + }; + + let listener = UnixListener::bind(&sock_path).expect("bind socket"); + listener.set_nonblocking(true).ok(); + + let state = State::new(); + let mut procs: HashMap = HashMap::new(); + let mut last_mtime: Option = config_mtime(&config_path); + + // Start enabled services + start_enabled(&config, &state, &mut procs); + + // Main loop + loop { + if STOP.load(Ordering::Relaxed) { + stop_all(&mut procs); + let _ = std::fs::remove_file(&sock_path); + let _ = std::fs::remove_file(&pid_path); + return; + } + + // Poll config for changes + if let Some(mtime) = config_mtime(&config_path) { + if Some(mtime) != last_mtime { + last_mtime = Some(mtime); + let new_config = Config::load(&config_path); + if let Ok(new_config) = new_config { + reconcile(&new_config, &state, &mut procs); + config = new_config; + } + } + } + + // Check children + check_children(&mut procs); + + // Accept socket connections + while let Ok((stream, _)) = listener.accept() { + handle_client(stream, &config, &state, &mut procs, &config_path); + } + + std::thread::sleep(Duration::from_millis(200)); + } +} + +fn config_mtime(path: &str) -> Option { + std::fs::metadata(path).ok().and_then(|m| m.modified().ok()) +} + +fn start_enabled(config: &Config, state: &State, procs: &mut HashMap) { + if let Ok(enabled) = state.list_enabled() { + for name in &enabled { + if let Some(cfg) = config.service.get(name) { + if !procs.contains_key(name) { + if let Some(p) = spawn(name, cfg) { + procs.insert(name.clone(), p); + } + } + } + } + } +} + +fn reconcile(new_config: &Config, state: &State, procs: &mut HashMap) { + let old_names: Vec = procs.keys().cloned().collect(); + + for name in &old_names { + if !new_config.service.contains_key(name) { + stop(name, procs); + } + } + + for (name, cfg) in &new_config.service { + if !procs.contains_key(name) && state.is_enabled(name) { + if let Some(p) = spawn(name, cfg) { + procs.insert(name.clone(), p); + } + } + } +} + +fn spawn(name: &str, cfg: &ServiceConfig) -> Option { + let log_path = format!( + "{}/.local/share/rss/logs/{}.log", + std::env::var("HOME").unwrap_or_default(), + name + ); + if let Some(parent) = std::path::Path::new(&log_path).parent() { + let _ = std::fs::create_dir_all(parent); + } + + let cmd = cfg.resolve_command(name, &log_path); + let mut child = match Command::new("sh") + .arg("-c") + .arg(&cmd) + .current_dir(&cfg.workdir) + .envs(&cfg.env) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .stdin(Stdio::null()) + .spawn() + { + Ok(c) => c, + Err(e) => { + append_log(&log_path, &format!("E: spawn failed: {}", e)); + return None; + } + }; + + let pid = child.id(); + let log_out = log_path.clone(); + let log_err = log_path.clone(); + + if let Some(stdout) = child.stdout.take() { + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + if let Ok(l) = line { + append_log(&log_out, &format!("O: {}", l)); + } + } + }); + } + if let Some(stderr) = child.stderr.take() { + std::thread::spawn(move || { + for line in BufReader::new(stderr).lines() { + if let Ok(l) = line { + append_log(&log_err, &format!("E: {}", l)); + } + } + }); + } + + append_log(&log_path, &format!("O: started (pid {})", pid)); + + Some(Proc { + child, + pid, + cfg: cfg.clone(), + restart_count: 0, + }) +} + +fn stop(name: &str, procs: &mut HashMap) { + let p = match procs.get(name) { + Some(p) => p, + None => return, + }; + + let pid = p.pid; + let log_path = format!( + "{}/.local/share/rss/logs/{}.log", + std::env::var("HOME").unwrap_or_default(), + name + ); + + // stop_command + if let Some(cmd) = p.cfg.resolve_stop_command(name, pid, &log_path) { + let _ = Command::new("sh").arg("-c").arg(&cmd).output(); + } + + // signal + let sig = crate::supervisor::parse_signal(&p.cfg.stop_signal); + let _ = nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid as i32), sig); + + // wait + let deadline = std::time::Instant::now() + Duration::from_secs(p.cfg.timeout_stop); + while std::time::Instant::now() < deadline { + if let Ok(Some(_)) = procs.get_mut(name).unwrap().child.try_wait() { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + + // force kill + if let Ok(None) = procs.get_mut(name).unwrap().child.try_wait() { + let _ = nix::sys::signal::kill( + nix::unistd::Pid::from_raw(pid as i32), + nix::sys::signal::Signal::SIGKILL, + ); + let _ = procs.get_mut(name).unwrap().child.wait(); + } else { + let _ = procs.get_mut(name).unwrap().child.wait(); + } + + append_log(&log_path, "O: stopped"); + procs.remove(name); +} + +fn stop_all(procs: &mut HashMap) { + let names: Vec = procs.keys().cloned().collect(); + for name in &names { + if procs.contains_key(name) { + stop(name, procs); + } + } +} + +fn check_children(procs: &mut HashMap) { + let mut dead = Vec::new(); + let mut restarts = Vec::new(); + + for (name, p) in procs.iter_mut() { + if let Ok(Some(status)) = p.child.try_wait() { + let log_path = format!( + "{}/.local/share/rss/logs/{}.log", + std::env::var("HOME").unwrap_or_default(), + name + ); + + let code = status + .code() + .map(|c| format!("code {}", c)) + .unwrap_or_else(|| "signal".to_string()); + + append_log(&log_path, &format!("O: exited with {}", code)); + + if p.cfg.oneshot { + dead.push(name.clone()); + continue; + } + + let should = match p.cfg.restart { + RestartPolicy::No => false, + RestartPolicy::Always | RestartPolicy::UnlessStopped => true, + RestartPolicy::OnFailure => !status.success(), + }; + + let max = if p.cfg.restart_max == 0 { + u32::MAX + } else { + p.cfg.restart_max + }; + + if should && p.restart_count < max { + p.restart_count += 1; + let max_str = if p.cfg.restart_max == 0 { + "unlimited".to_string() + } else { + p.cfg.restart_max.to_string() + }; + append_log( + &log_path, + &format!( + "O: restarting in {}s (attempt {}/{})", + p.cfg.restart_delay, + p.restart_count, + max_str + ), + ); + restarts.push((name.clone(), p.cfg.clone())); + dead.push(name.clone()); + } else { + if p.cfg.restart != RestartPolicy::No { + append_log(&log_path, "O: max restarts reached, exiting"); + } + dead.push(name.clone()); + } + } + } + + for name in &dead { + procs.remove(name); + } + for (name, cfg) in &restarts { + std::thread::sleep(Duration::from_secs(cfg.restart_delay)); + if let Some(p) = spawn(name, cfg) { + procs.insert(name.clone(), p); + } + } +} + +fn handle_client( + mut stream: UnixStream, + config: &Config, + state: &State, + procs: &mut HashMap, + _config_path: &str, +) { + let mut line = String::new(); + let mut reader = BufReader::new(&stream); + if reader.read_line(&mut line).is_err() { + return; + } + let line = line.trim(); + + let response = match line { + l if l.starts_with("start ") => { + let name = &l[6..]; + if !config.service.contains_key(name) { + format!("error: service '{}' not found", name) + } else if procs.contains_key(name) { + format!("error: '{}' already running", name) + } else { + match spawn(name, &config.service[name]) { + Some(p) => { + let pid = p.pid; + procs.insert(name.to_string(), p); + format!("ok {}", pid) + } + None => format!("error: failed to spawn '{}'", name), + } + } + } + l if l.starts_with("stop ") => { + let name = &l[5..]; + if !procs.contains_key(name) { + format!("error: '{}' not running", name) + } else { + stop(name, procs); + "ok".to_string() + } + } + l if l.starts_with("restart ") => { + let name = &l[8..]; + if procs.contains_key(name) { + stop(name, procs); + } + if let Some(cfg) = config.service.get(name) { + match spawn(name, cfg) { + Some(p) => { + let pid = p.pid; + procs.insert(name.to_string(), p); + format!("ok {}", pid) + } + None => format!("error: failed to spawn '{}'", name), + } + } else { + format!("error: service '{}' not found", name) + } + } + "list" | "list " => { + let mut out = Vec::new(); + for (name, _cfg) in &config.service { + let enabled = if state.is_enabled(name) { "yes" } else { "no" }; + let info = if let Some(p) = procs.get(name) { + format!("running {} pid={} enabled={}", name, p.pid, enabled) + } else { + format!("stopped {} pid=0 enabled={}", name, enabled) + }; + out.push(info); + } + out.join("\n") + } + l if l.starts_with("status ") => { + let name = &l[7..]; + let enabled = if state.is_enabled(name) { "yes" } else { "no" }; + if let Some(p) = procs.get(name) { + format!("running {} pid={} enabled={}", name, p.pid, enabled) + } else if config.service.contains_key(name) { + format!("stopped {} pid=0 enabled={}", name, enabled) + } else { + format!("error: service '{}' not found", name) + } + } + "status" => { + let mut out = Vec::new(); + for (name, _) in &config.service { + let enabled = if state.is_enabled(name) { "yes" } else { "no" }; + let info = if let Some(p) = procs.get(name) { + format!("running {} pid={} enabled={}", name, p.pid, enabled) + } else { + format!("stopped {} pid=0 enabled={}", name, enabled) + }; + out.push(info); + } + out.join("\n") + } + _ => format!("error: unknown command '{}'", line), + }; + + let _ = writeln!(stream, "{}", response); + let _ = stream.flush(); +} + +fn append_log(path: &str, msg: &str) { + let ts = chrono::Local::now().format("%Y-%m-%d %H:%M:%S"); + if let Ok(mut f) = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + { + let _ = writeln!(f, "[{}] {}", ts, msg); + let _ = f.flush(); + } +} diff --git a/src/ipc.rs b/src/ipc.rs new file mode 100644 index 0000000..9d7279c --- /dev/null +++ b/src/ipc.rs @@ -0,0 +1,40 @@ +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; + +pub fn socket_path() -> PathBuf { + let home = std::env::var("HOME").expect("HOME not set"); + PathBuf::from(format!("{}/.local/share/rss/rss.sock", home)) +} + +pub fn daemon_pid_path() -> PathBuf { + let home = std::env::var("HOME").expect("HOME not set"); + PathBuf::from(format!("{}/.local/share/rss/daemon.pid", home)) +} + +pub fn send_cmd(cmd: &str) -> Result, String> { + let path = socket_path(); + + if path.exists() { + if UnixStream::connect(&path).is_err() { + let _ = std::fs::remove_file(&path); + } + } + + let mut stream = UnixStream::connect(&path).map_err(|e| format!("daemon not running: {}", e))?; + writeln!(stream, "{}", cmd).map_err(|e| format!("write: {}", e))?; + stream.flush().map_err(|e| format!("flush: {}", e))?; + + let mut lines = Vec::new(); + for line in BufReader::new(&stream).lines() { + let line = line.map_err(|e| format!("read: {}", e))?; + if line == "ok" { + return Ok(lines); + } + if let Some(msg) = line.strip_prefix("error:") { + return Err(msg.trim().to_string()); + } + lines.push(line); + } + Ok(lines) +} diff --git a/src/main.rs b/src/main.rs index 47bb9ef..c62921b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,60 +1,35 @@ mod cli; mod config; +mod daemon; +mod ipc; mod log; mod state; mod supervisor; -use std::process::{Command, Stdio}; - use clap::Parser; use cli::{Cli, Commands}; -use config::Config; use state::State; -use supervisor::{parse_signal, process_running, wait_for_exit}; fn main() { let cli = Cli::parse(); - // Internal supervise subcommand — must run before config loading? - if let Commands::Supervise { ref name } = cli.command { - let config_path = resolve_config_path(cli.config.as_deref()); - let config = Config::load(&config_path).unwrap_or_else(|e| { - eprintln!("error: config: {}", e); - std::process::exit(1); - }); - let cfg = match config.service.get(name) { - Some(c) => c, - None => { - eprintln!("error: service '{}' not found in config", name); - std::process::exit(1); - } - }; - supervisor::run_supervisor(name, cfg); - return; - } - - let config_path = resolve_config_path(cli.config.as_deref()); - let config = Config::load(&config_path).unwrap_or_else(|e| { - eprintln!("error: config: {}", e); - std::process::exit(1); - }); - - let state = State::new(); - match cli.command { - Commands::Start { name } => cmd_start(&name, &config, &config_path), - Commands::Stop { name } => cmd_stop(&name, &config, &state), - Commands::Restart { name } => { - cmd_stop(&name, &config, &state); - cmd_start(&name, &config, &config_path); + Commands::Daemon => { + let config_path = resolve_config_path(cli.config.as_deref()); + daemon::run(config_path); + } + Commands::Start { ref name } => cmd_start(name), + Commands::Stop { ref name } => cmd_stop(name), + Commands::Restart { ref name } => cmd_restart(name), + Commands::Status { ref name } => cmd_status(name.as_deref()), + Commands::List => cmd_list(), + Commands::Logs { ref name, lines, follow } => cmd_logs(name, lines, follow), + Commands::Enable { ref name, now } => cmd_enable(name, now), + Commands::Disable { ref name } => cmd_disable(name), + Commands::Supervise { .. } => { + eprintln!("error: supervise is internal, use daemon"); + std::process::exit(1); } - Commands::Status { name } => cmd_status(name.as_deref(), &config, &state), - Commands::Logs { name, lines, follow } => cmd_logs(&name, lines, follow), - Commands::Enable { name, now } => cmd_enable(&name, now, &config, &state, &config_path), - Commands::Disable { name } => cmd_disable(&name, &state), - Commands::List => cmd_list(&config, &state), - Commands::Daemon => cmd_daemon(&config, &state, &config_path), - Commands::Supervise { .. } => unreachable!(), } } @@ -69,171 +44,116 @@ fn resolve_config_path(custom: Option<&str>) -> String { format!("{}/.config/rss/config.toml", home) } -fn cmd_start(name: &str, config: &Config, config_path: &str) { - if !config.service.contains_key(name) { - eprintln!("error: service '{}' not found in config", name); - std::process::exit(1); - } - let state = State::new(); - let svc_pid = state.service_pid_path(name); - if State::read_pid(&svc_pid).map_or(false, |p| process_running(p)) { - eprintln!("error: '{}' is already running", name); - std::process::exit(1); - } - let self_exe = std::env::current_exe().unwrap_or_else(|_| { - eprintln!("error: cannot determine binary path"); - std::process::exit(1); - }); - - let child = Command::new(&self_exe) - .arg("supervise") - .arg(name) - .env("RSS_CONFIG", config_path) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .stdin(Stdio::null()) - .spawn() - .unwrap_or_else(|e| { - eprintln!("error: failed to start supervisor: {}", e); +fn cmd_start(name: &str) { + match ipc::send_cmd(&format!("start {}", name)) { + Ok(lines) => { + if let Some(first) = lines.first() { + println!("{}", first); + } + } + Err(e) => { + eprintln!("error: {}", e); std::process::exit(1); - }); - - println!("{}", child.id()); - // Detach — forget the child so it becomes orphaned and adopted by init - std::mem::forget(child); + } + } } -fn cmd_stop(name: &str, config: &Config, state: &State) { - let cfg = match config.service.get(name) { - Some(c) => c, - None => { - eprintln!("error: service '{}' not found", name); - return; +fn cmd_stop(name: &str) { + match ipc::send_cmd(&format!("stop {}", name)) { + Ok(_) => {} + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); } + } +} + +fn cmd_restart(name: &str) { + match ipc::send_cmd(&format!("restart {}", name)) { + Ok(lines) => { + if let Some(first) = lines.first() { + println!("{}", first); + } + } + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); + } + } +} + +fn cmd_status(name: Option<&str>) { + let cmd = match name { + Some(n) => format!("status {}", n), + None => "status".to_string(), }; - - let svc_pid_path = state.service_pid_path(name); - let sup_pid_path = state.supervise_pid_path(name); - let log_path = state.log_path(name); - let log_str = log_path.to_str().unwrap_or(""); - - // Tell supervisor to stop (sets TERMINATE flag) first - if let Some(pid) = State::read_pid(&sup_pid_path) { - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGTERM, - ); - } - } - - // Stop child process - if let Some(pid) = State::read_pid(&svc_pid_path) { - if process_running(pid) { - // Run stop_command if configured - if let Some(stop_cmd) = cfg.resolve_stop_command(name, pid, log_str) { - let _ = Command::new("sh").arg("-c").arg(&stop_cmd).current_dir(&cfg.workdir).output(); - } - - // Send configured signal - let sig = parse_signal(&cfg.stop_signal); - let _ = nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid as i32), sig); - - // Wait for graceful exit - wait_for_exit(pid, cfg.timeout_stop); - - // Force kill if still alive - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGKILL, - ); - wait_for_exit(pid, 3); + match ipc::send_cmd(&cmd) { + Ok(lines) => { + for line in lines { + println!("{}", line); } } - } - - // Wait for supervisor to exit (it sees TERMINATE + child dead) - if let Some(pid) = State::read_pid(&sup_pid_path) { - if process_running(pid) { - wait_for_exit(pid, 3); - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGKILL, - ); - wait_for_exit(pid, 3); - } - } - } - - // Cleanup PID files - State::remove_pid(&svc_pid_path); - State::remove_pid(&sup_pid_path); -} - -fn cmd_status(name: Option<&str>, config: &Config, state: &State) { - if let Some(name) = name { - print_service_status(name, config, state); - } else { - let mut names: Vec<&String> = config.service.keys().collect(); - names.sort(); - for name in names { - print_service_status(name, config, state); + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); } } } -fn print_service_status(name: &str, _config: &Config, state: &State) { - let svc_pid_path = state.service_pid_path(name); - let sup_pid_path = state.supervise_pid_path(name); - let enabled = state.is_enabled(name); - - let svc_pid = State::read_pid(&svc_pid_path); - let sup_pid = State::read_pid(&sup_pid_path); - - let svc_running = svc_pid.map_or(false, process_running); - let sup_running = sup_pid.map_or(false, process_running); - - let status = if svc_running { - "running" - } else if sup_running { - "restarting" - } else { - "stopped" - }; - - let enabled_str = if enabled { "+" } else { "-" }; - println!( - "{} {} pid={} supervisor={} enabled={}", - status, - name, - svc_pid.map_or(0, |p| p), - sup_pid.map_or(0, |p| p), - enabled_str, - ); +fn cmd_list() { + match ipc::send_cmd("list") { + Ok(lines) => { + if lines.is_empty() { + println!("no services configured"); + return; + } + for line in &lines { + let parts: Vec<&str> = line.splitn(4, ' ').collect(); + if parts.len() >= 4 { + println!("{:<12} {:<7} {:>6} {}", parts[0], parts[1], parts[2], parts[3]); + } else { + println!("{}", line); + } + } + } + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); + } + } } fn cmd_logs(name: &str, mut lines: Option, follow: bool) { if lines.is_none() && follow { lines = Some(10); } + let state = State::new(); let log_path = state.log_path(name); - let log_str = log_path.to_str().unwrap_or(""); if !log_path.exists() { eprintln!("error: no logs for '{}'", name); std::process::exit(1); } - if let Err(e) = log::show_logs(log_str, lines, follow) { + if let Err(e) = log::show_logs(log_path.to_str().unwrap_or(""), lines, follow) { eprintln!("error: reading logs: {}", e); std::process::exit(1); } } -fn cmd_enable(name: &str, now: bool, config: &Config, state: &State, config_path: &str) { +fn cmd_enable(name: &str, now: bool) { + let state = State::new(); + + let config_path = resolve_config_path(None); + let config = match config::Config::load(&config_path) { + Ok(c) => c, + Err(e) => { + eprintln!("error: config: {}", e); + std::process::exit(1); + } + }; + if !config.service.contains_key(name) { eprintln!("error: service '{}' not found in config", name); return; @@ -247,120 +167,12 @@ fn cmd_enable(name: &str, now: bool, config: &Config, state: &State, config_path println!("enabled {}", name); if now { - cmd_start(name, config, config_path); + cmd_start(name); } } -fn cmd_disable(name: &str, state: &State) { +fn cmd_disable(name: &str) { + let state = State::new(); state.disable(name); println!("disabled {}", name); } - -fn cmd_list(config: &Config, state: &State) { - cleanup_orphans(config, state); - let mut names: Vec<&String> = config.service.keys().collect(); - names.sort(); - - if names.is_empty() { - println!("no services configured"); - return; - } - - println!("{:<12} {:<7} {:>6} {:>6} {}", "status", "name", "pid", "super", "enabled"); - println!("{}", "-".repeat(50)); - for name in names { - let svc_pid_path = state.service_pid_path(name); - let sup_pid_path = state.supervise_pid_path(name); - let enabled = state.is_enabled(name); - - let svc_pid = State::read_pid(&svc_pid_path); - let sup_pid = State::read_pid(&sup_pid_path); - - let svc_running = svc_pid.map_or(false, process_running); - let sup_running = sup_pid.map_or(false, process_running); - - let status = if svc_running { - "running" - } else if sup_running { - "restart" - } else { - "stopped" - }; - - let enabled_str = if enabled { "yes" } else { "no" }; - println!( - "{:<12} {:<7} {:>6} {:>6} {}", - status, - name, - svc_pid.map_or(0, |p| p), - sup_pid.map_or(0, |p| p), - enabled_str, - ); - } -} - -fn cleanup_orphans(config: &Config, state: &State) { - for name in State::list_pid_names(&state.supervise_dir()) { - if !config.service.contains_key(&name) { - stop_orphan(&name, state); - } - } - for name in State::list_pid_names(&state.service_dir()) { - if !config.service.contains_key(&name) { - State::remove_pid(&state.service_pid_path(&name)); - } - } -} - -fn stop_orphan(name: &str, state: &State) { - let svc = state.service_pid_path(name); - let sup = state.supervise_pid_path(name); - - if let Some(pid) = State::read_pid(&svc) { - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGTERM, - ); - wait_for_exit(pid, 5); - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGKILL, - ); - wait_for_exit(pid, 3); - } - } - } - if let Some(pid) = State::read_pid(&sup) { - if process_running(pid) { - let _ = nix::sys::signal::kill( - nix::unistd::Pid::from_raw(pid as i32), - nix::sys::signal::Signal::SIGKILL, - ); - wait_for_exit(pid, 3); - } - } - State::remove_pid(&svc); - State::remove_pid(&sup); -} - -fn cmd_daemon(config: &Config, state: &State, config_path: &str) { - cleanup_orphans(config, state); - - let enabled = state.list_enabled().unwrap_or_default(); - if enabled.is_empty() { - println!("no enabled services"); - return; - } - - for name in &enabled { - let svc_pid_path = state.service_pid_path(name); - let already = State::read_pid(&svc_pid_path).map_or(false, |p| process_running(p)); - if already { - continue; - } - println!("{}", name); - cmd_start(name, config, config_path); - } -} diff --git a/src/state.rs b/src/state.rs index db64768..d3716c8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,7 +1,7 @@ use std::path::PathBuf; pub struct State { - state_dir: PathBuf, + pub state_dir: PathBuf, config_dir: PathBuf, } @@ -14,66 +14,19 @@ impl State { } } - pub fn supervise_pid_path(&self, name: &str) -> PathBuf { - self.state_dir.join("supervise").join(format!("{}.pid", name)) - } - - pub fn service_pid_path(&self, name: &str) -> PathBuf { - self.state_dir.join("service").join(format!("{}.pid", name)) - } - pub fn log_path(&self, name: &str) -> PathBuf { self.state_dir.join("logs").join(format!("{}.log", name)) } - pub fn supervise_dir(&self) -> PathBuf { - self.state_dir.join("supervise") - } - - pub fn service_dir(&self) -> PathBuf { - self.state_dir.join("service") - } - pub fn enabled_path(&self, name: &str) -> PathBuf { self.config_dir.join("enabled").join(name) } - pub fn list_pid_names(dir: &PathBuf) -> Vec { - let mut names = Vec::new(); - if let Ok(entries) = std::fs::read_dir(dir) { - for entry in entries.flatten() { - let fname = entry.file_name(); - let fname = fname.to_string_lossy().to_string(); - if let Some(stripped) = fname.strip_suffix(".pid") { - names.push(stripped.to_string()); - } - } - } - names - } - pub fn write_pid(path: &PathBuf) -> std::io::Result<()> { - Self::write_pid_with(path, std::process::id()) - } - - pub fn write_pid_with(path: &PathBuf, pid: u32) -> std::io::Result<()> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } - std::fs::write(path, format!("{}", pid)) - } - - pub fn read_pid(path: &PathBuf) -> Option { - let content = std::fs::read_to_string(path).ok()?; - content.trim().parse().ok() - } - - pub fn remove_pid(path: &PathBuf) { - let _ = std::fs::remove_file(path); - } - - pub fn process_exists(pid: u32) -> bool { - std::path::Path::new(&format!("/proc/{}", pid)).exists() + std::fs::write(path, format!("{}", std::process::id())) } pub fn is_enabled(&self, name: &str) -> bool { diff --git a/src/supervisor.rs b/src/supervisor.rs index fca1674..9d58ca9 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -1,172 +1,4 @@ -use std::io::{BufRead, Write}; -use std::process::{Child, Command, Stdio}; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; - -use chrono::Local; -use nix::sys::signal::{self, SigAction, SigHandler, Signal}; - -use crate::config::{RestartPolicy, ServiceConfig}; -use crate::state::State; - -pub static TERMINATE: AtomicBool = AtomicBool::new(false); - -extern "C" fn sigterm_handler(_: i32) { - TERMINATE.store(true, Ordering::Relaxed); -} - -pub fn setup_signal_handler() { - let handler = SigHandler::Handler(sigterm_handler); - let action = SigAction::new(handler, nix::sys::signal::SaFlags::SA_RESTART, signal::SigSet::empty()); - unsafe { - let _ = signal::sigaction(Signal::SIGTERM, &action); - let _ = signal::sigaction(Signal::SIGINT, &action); - } -} - -pub fn run_supervisor(name: &str, cfg: &ServiceConfig) { - setup_signal_handler(); - - let state = State::new(); - let supervise_pid = state.supervise_pid_path(name); - let service_pid = state.service_pid_path(name); - let log_path = state.log_path(name); - - State::write_pid(&supervise_pid).ok(); - - if let Some(parent) = log_path.parent() { - std::fs::create_dir_all(parent).ok(); - } - - let mut restarts: u32 = 0; - let max_restarts = if cfg.restart_max == 0 { u32::MAX } else { cfg.restart_max }; - - loop { - let cmd = cfg.resolve_command(name, log_path.to_str().unwrap_or("")); - let log_file = log_path.to_str().unwrap_or("").to_string(); - - let mut child = match spawn_child(name, cfg, &cmd, &log_file) { - Ok(c) => c, - Err(e) => { - append_log(&log_file, &format!("E: failed to spawn: {}", e)); - break; - } - }; - - let child_pid = child.id(); - State::write_pid_with(&service_pid, child_pid).ok(); - - append_log(&log_file, &format!("O: started (pid {})", child_pid)); - - let status = child.wait(); - - match status { - Ok(status) => { - let code = status.code().map(|c| format!("code {}", c)).unwrap_or_else(|| "signal".to_string()); - append_log(&log_file, &format!("O: exited with {}", code)); - - if TERMINATE.load(Ordering::Relaxed) { - append_log(&log_file, "O: supervisor received stop signal, exiting"); - break; - } - - if cfg.oneshot { - append_log(&log_file, "O: oneshot service completed, exiting"); - break; - } - - let should_restart = match cfg.restart { - RestartPolicy::No => false, - RestartPolicy::Always | RestartPolicy::UnlessStopped => true, - RestartPolicy::OnFailure => !status.success(), - }; - - if should_restart && restarts < max_restarts { - restarts += 1; - let max_str = if cfg.restart_max == 0 { - "unlimited".to_string() - } else { - cfg.restart_max.to_string() - }; - append_log( - &log_file, - &format!("O: restarting in {}s (attempt {}/{})", cfg.restart_delay, restarts, max_str), - ); - std::thread::sleep(Duration::from_secs(cfg.restart_delay)); - } else { - if cfg.restart != RestartPolicy::No { - append_log(&log_file, "O: max restarts reached, exiting"); - } - break; - } - } - Err(e) => { - append_log(&log_file, &format!("E: wait error: {}", e)); - break; - } - } - } - - State::remove_pid(&service_pid); - State::remove_pid(&supervise_pid); - - // Give pipe-reader threads time to flush - std::thread::sleep(Duration::from_millis(200)); -} - -fn spawn_child( - _name: &str, - cfg: &ServiceConfig, - cmd: &str, - log_file: &str, -) -> std::io::Result { - let mut child = Command::new("sh") - .arg("-c") - .arg(cmd) - .current_dir(&cfg.workdir) - .envs(&cfg.env) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .stdin(Stdio::null()) - .spawn()?; - - let stdout = child.stdout.take().unwrap(); - let stderr = child.stderr.take().unwrap(); - - let log_out = log_file.to_string(); - std::thread::spawn(move || { - let reader = std::io::BufReader::new(stdout); - for line in reader.lines() { - if let Ok(line) = line { - append_log(&log_out, &format!("O: {}", line)); - } - } - }); - - let log_err = log_file.to_string(); - std::thread::spawn(move || { - let reader = std::io::BufReader::new(stderr); - for line in reader.lines() { - if let Ok(line) = line { - append_log(&log_err, &format!("E: {}", line)); - } - } - }); - - Ok(child) -} - -fn append_log(path: &str, msg: &str) { - if let Ok(mut file) = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(path) - { - let ts = Local::now().format("%Y-%m-%d %H:%M:%S"); - let _ = writeln!(file, "[{}] {}", ts, msg); - let _ = file.flush(); - } -} +use nix::sys::signal::Signal; pub fn parse_signal(name: &str) -> Signal { match name.to_uppercase().as_str() { @@ -180,17 +12,3 @@ pub fn parse_signal(name: &str) -> Signal { _ => Signal::SIGTERM, } } - -pub fn process_running(pid: u32) -> bool { - State::process_exists(pid) -} - -pub fn wait_for_exit(pid: u32, timeout_secs: u64) { - let deadline = std::time::Instant::now() + Duration::from_secs(timeout_secs); - while std::time::Instant::now() < deadline { - if !process_running(pid) { - return; - } - std::thread::sleep(Duration::from_millis(100)); - } -}