daemon: centralized architecture with Unix socket IPC

- single daemon process manages all services in memory
- CLI communicates via Unix socket, does not read config
- config hot-reload: daemon polls config.toml, reacts immediately
- rename-safe: removing a service from config kills it, adding starts if enabled
- logs remain file-based for direct reading by rss logs
- removed per-service PID files, supervisor module simplified
This commit is contained in:
Niko Marmeladkov 2026-07-03 11:49:46 +03:00
parent a7ce8405d1
commit b069c527bd
Signed by untrusted user who does not match committer: Niko
GPG key ID: E3B955F9442D44E3
6 changed files with 601 additions and 529 deletions

View file

@ -1,16 +1,25 @@
# rss — RuSt Supervisor # 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. - **Single daemon** (`rss daemon`) manages all services in-process
- **UNIX way** — PID files for state, no IPC, no sockets. - **CLI** (`rss start`, `rss stop`, etc.) talks to the daemon via Unix socket
- **User-space** — everything under `~/.config/rss/` and `~/.local/share/rss/`. - **No PID files** for services — state lives in daemon memory
- **No root** — runs entirely as your user. - **Config hot-reload** — rename/add/remove services live, daemon detects changes
- **Logs** written to `~/.local/share/rss/logs/<name>.log` — readable by CLI directly
## Usage ## Usage
First start the daemon (via systemd or terminal):
```
rss daemon
```
Then manage services:
``` ```
rss start <name> Start a service rss start <name> Start a service
rss stop <name> Stop a service rss stop <name> Stop a service
@ -20,7 +29,6 @@ rss logs <name> [-n N] [-f] View logs (last N, follow)
rss enable <name> [--now] Enable auto-start rss enable <name> [--now] Enable auto-start
rss disable <name> Disable auto-start rss disable <name> Disable auto-start
rss list List all services rss list List all services
rss daemon Start all enabled services
``` ```
## Configuration ## Configuration
@ -34,7 +42,7 @@ command = "your command here"
stop_command = "optional stop command" stop_command = "optional stop command"
timeout_start = 15 timeout_start = 15
timeout_stop = 10 timeout_stop = 10
restart = "on-failure" # no, always, on-failure restart = "on-failure" # no, always, unless-stopped, on-failure
restart_delay = 3 restart_delay = 3
restart_max = 5 # 0 = unlimited restart_max = 5 # 0 = unlimited
oneshot = false oneshot = false
@ -44,7 +52,7 @@ env = { KEY = "value" }
Placeholders in `command` / `stop_command`: `{name}`, `{workdir}`, `{pid}`, `{log_file}`, plus any env var. 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) ## systemd (user mode)

441
src/daemon.rs Normal file
View file

@ -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<String, Proc> = HashMap::new();
let mut last_mtime: Option<SystemTime> = 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<SystemTime> {
std::fs::metadata(path).ok().and_then(|m| m.modified().ok())
}
fn start_enabled(config: &Config, state: &State, procs: &mut HashMap<String, Proc>) {
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<String, Proc>) {
let old_names: Vec<String> = 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<Proc> {
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<String, Proc>) {
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<String, Proc>) {
let names: Vec<String> = procs.keys().cloned().collect();
for name in &names {
if procs.contains_key(name) {
stop(name, procs);
}
}
}
fn check_children(procs: &mut HashMap<String, Proc>) {
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<String, Proc>,
_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();
}
}

40
src/ipc.rs Normal file
View file

@ -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<Vec<String>, 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)
}

View file

@ -1,60 +1,35 @@
mod cli; mod cli;
mod config; mod config;
mod daemon;
mod ipc;
mod log; mod log;
mod state; mod state;
mod supervisor; mod supervisor;
use std::process::{Command, Stdio};
use clap::Parser; use clap::Parser;
use cli::{Cli, Commands}; use cli::{Cli, Commands};
use config::Config;
use state::State; use state::State;
use supervisor::{parse_signal, process_running, wait_for_exit};
fn main() { fn main() {
let cli = Cli::parse(); 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 { match cli.command {
Commands::Start { name } => cmd_start(&name, &config, &config_path), Commands::Daemon => {
Commands::Stop { name } => cmd_stop(&name, &config, &state), let config_path = resolve_config_path(cli.config.as_deref());
Commands::Restart { name } => { daemon::run(config_path);
cmd_stop(&name, &config, &state); }
cmd_start(&name, &config, &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) format!("{}/.config/rss/config.toml", home)
} }
fn cmd_start(name: &str, config: &Config, config_path: &str) { fn cmd_start(name: &str) {
if !config.service.contains_key(name) { match ipc::send_cmd(&format!("start {}", name)) {
eprintln!("error: service '{}' not found in config", name); Ok(lines) => {
if let Some(first) = lines.first() {
println!("{}", first);
}
}
Err(e) => {
eprintln!("error: {}", e);
std::process::exit(1); 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);
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) { fn cmd_stop(name: &str) {
let cfg = match config.service.get(name) { match ipc::send_cmd(&format!("stop {}", name)) {
Some(c) => c, Ok(_) => {}
None => { Err(e) => {
eprintln!("error: service '{}' not found", name); 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(),
};
match ipc::send_cmd(&cmd) {
Ok(lines) => {
for line in lines {
println!("{}", line);
}
}
Err(e) => {
eprintln!("error: {}", e);
std::process::exit(1);
}
}
}
fn cmd_list() {
match ipc::send_cmd("list") {
Ok(lines) => {
if lines.is_empty() {
println!("no services configured");
return; return;
} }
}; for line in &lines {
let parts: Vec<&str> = line.splitn(4, ' ').collect();
let svc_pid_path = state.service_pid_path(name); if parts.len() >= 4 {
let sup_pid_path = state.supervise_pid_path(name); println!("{:<12} {:<7} {:>6} {}", parts[0], parts[1], parts[2], parts[3]);
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);
}
}
}
// 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 { } else {
let mut names: Vec<&String> = config.service.keys().collect(); println!("{}", line);
names.sort();
for name in names {
print_service_status(name, config, state);
} }
} }
} }
Err(e) => {
fn print_service_status(name: &str, _config: &Config, state: &State) { eprintln!("error: {}", e);
let svc_pid_path = state.service_pid_path(name); std::process::exit(1);
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_logs(name: &str, mut lines: Option<usize>, follow: bool) { fn cmd_logs(name: &str, mut lines: Option<usize>, follow: bool) {
if lines.is_none() && follow { if lines.is_none() && follow {
lines = Some(10); lines = Some(10);
} }
let state = State::new(); let state = State::new();
let log_path = state.log_path(name); let log_path = state.log_path(name);
let log_str = log_path.to_str().unwrap_or("");
if !log_path.exists() { if !log_path.exists() {
eprintln!("error: no logs for '{}'", name); eprintln!("error: no logs for '{}'", name);
std::process::exit(1); 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); eprintln!("error: reading logs: {}", e);
std::process::exit(1); 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) { if !config.service.contains_key(name) {
eprintln!("error: service '{}' not found in config", name); eprintln!("error: service '{}' not found in config", name);
return; return;
@ -247,120 +167,12 @@ fn cmd_enable(name: &str, now: bool, config: &Config, state: &State, config_path
println!("enabled {}", name); println!("enabled {}", name);
if now { 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); state.disable(name);
println!("disabled {}", 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);
}
}

View file

@ -1,7 +1,7 @@
use std::path::PathBuf; use std::path::PathBuf;
pub struct State { pub struct State {
state_dir: PathBuf, pub state_dir: PathBuf,
config_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 { pub fn log_path(&self, name: &str) -> PathBuf {
self.state_dir.join("logs").join(format!("{}.log", name)) 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 { pub fn enabled_path(&self, name: &str) -> PathBuf {
self.config_dir.join("enabled").join(name) self.config_dir.join("enabled").join(name)
} }
pub fn list_pid_names(dir: &PathBuf) -> Vec<String> {
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<()> { 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() { if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?; std::fs::create_dir_all(parent)?;
} }
std::fs::write(path, format!("{}", pid)) std::fs::write(path, format!("{}", std::process::id()))
}
pub fn read_pid(path: &PathBuf) -> Option<u32> {
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()
} }
pub fn is_enabled(&self, name: &str) -> bool { pub fn is_enabled(&self, name: &str) -> bool {

View file

@ -1,172 +1,4 @@
use std::io::{BufRead, Write}; use nix::sys::signal::Signal;
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<Child> {
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();
}
}
pub fn parse_signal(name: &str) -> Signal { pub fn parse_signal(name: &str) -> Signal {
match name.to_uppercase().as_str() { match name.to_uppercase().as_str() {
@ -180,17 +12,3 @@ pub fn parse_signal(name: &str) -> Signal {
_ => Signal::SIGTERM, _ => 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));
}
}