use std::{
fs,
io::{self, ErrorKind, Read, Write},
os::unix::{
fs::{DirBuilderExt, MetadataExt, PermissionsExt},
net::{UnixListener, UnixStream},
process::CommandExt,
},
panic::{AssertUnwindSafe, catch_unwind},
path::{Path, PathBuf},
process::Stdio,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
mpsc::channel,
},
thread,
time::Duration,
};
use nix::{
fcntl::{Flock, FlockArg},
sys::signal::{Signal, kill},
unistd::Pid,
};
use crate::{
core::{LoopExit, Wake, run_loop},
frame::{read_frame, write_frame},
protocol::{
Command, Event, LaunchContext, PROTOCOL_VERSION, decode_command, decode_event,
decode_hello, encode_command, encode_event, encode_hello,
},
supervisor::Supervisor,
};
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(5);
const HELLO_PROBE: Duration = Duration::from_secs(1);
fn runtime_dir() -> PathBuf {
resolve_runtime_dir(
std::env::var("FLEETCOM_RUNTIME_DIR").ok(),
std::env::var("XDG_RUNTIME_DIR").ok(),
std::env::temp_dir(),
nix::unistd::getuid().as_raw(),
)
}
fn resolve_runtime_dir(
override_dir: Option<String>,
xdg: Option<String>,
tmp: PathBuf,
uid: u32,
) -> PathBuf {
if let Some(d) = override_dir {
return PathBuf::from(d);
}
if let Some(d) = xdg
&& !d.is_empty()
{
return PathBuf::from(d).join("fleetcom");
}
tmp.join(format!("fleetcom-{uid}"))
}
fn socket_path() -> PathBuf {
runtime_dir().join("default.sock")
}
fn ensure_runtime_dir(dir: &Path) -> io::Result<()> {
match fs::symlink_metadata(dir) {
Ok(md) => {
if !md.file_type().is_dir() {
return Err(io::Error::new(
ErrorKind::AlreadyExists,
"runtime path exists but is not a directory",
));
}
if md.uid() != nix::unistd::getuid().as_raw() {
return Err(io::Error::new(
ErrorKind::PermissionDenied,
"runtime dir is not owned by this user",
));
}
if md.permissions().mode() & 0o077 != 0 {
fs::set_permissions(dir, fs::Permissions::from_mode(0o700))?;
}
Ok(())
}
Err(e) if e.kind() == ErrorKind::NotFound => fs::DirBuilder::new()
.recursive(true)
.mode(0o700)
.create(dir),
Err(e) => Err(e),
}
}
fn read_frame_bounded(stream: &mut UnixStream, timeout: Duration) -> io::Result<(u8, Vec<u8>)> {
stream.set_read_timeout(Some(timeout))?;
let res = read_frame(stream);
stream.set_read_timeout(None)?;
res
}
fn is_timeout(e: &io::Error) -> bool {
matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut)
}
fn hello_read_error(e: io::Error) -> io::Error {
if e.kind() == ErrorKind::UnexpectedEof {
io::Error::new(
ErrorKind::ConnectionAborted,
"the daemon closed the connection during the handshake (it may be \
shutting down); rerun fleetcom to start a fresh one",
)
} else {
e
}
}
fn check_hello_ack(kind: u8, payload: &[u8]) -> io::Result<()> {
match decode_event(kind, payload) {
Some(Event::HelloOk) => Ok(()),
Some(Event::Status(msg)) => Err(io::Error::other(msg)),
_ => Err(io::Error::other(
"daemon predates the protocol handshake (stale daemon from an older \
fleetcom); run 'fleetcom --kill' and retry",
)),
}
}
pub fn connect_ready() -> io::Result<UnixStream> {
let mut stream = connect_or_autostart()?;
let done = Arc::new(AtomicBool::new(false));
{
let done = Arc::clone(&done);
thread::spawn(move || {
thread::sleep(HELLO_PROBE);
if !done.load(Ordering::Relaxed) {
eprintln!(
"fleetcom: the daemon is serving another client; waiting \
to attach (Ctrl-C to abort)"
);
}
});
}
let (kind, payload) = encode_hello(&LaunchContext::here());
write_frame(&mut stream, kind, &payload)?;
let reply = read_frame(&mut stream);
done.store(true, Ordering::Relaxed);
let (kind, payload) = reply.map_err(hello_read_error)?;
check_hello_ack(kind, &payload)?;
Ok(stream)
}
fn busy_daemon_error(e: io::Error) -> io::Error {
if is_timeout(&e) {
io::Error::new(
ErrorKind::TimedOut,
"the daemon is serving another client; retry after it detaches",
)
} else {
e
}
}
pub fn connect_ready_bounded() -> io::Result<UnixStream> {
let mut stream = connect_or_autostart()?;
stream.set_write_timeout(Some(HANDSHAKE_TIMEOUT))?;
let (kind, payload) = encode_hello(&LaunchContext::here());
write_frame(&mut stream, kind, &payload).map_err(busy_daemon_error)?;
stream.set_write_timeout(None)?;
let (kind, payload) = read_frame_bounded(&mut stream, HANDSHAKE_TIMEOUT)
.map_err(|e| hello_read_error(busy_daemon_error(e)))?;
check_hello_ack(kind, &payload)?;
Ok(stream)
}
fn connect_or_autostart() -> io::Result<UnixStream> {
let path = socket_path();
if let Ok(s) = UnixStream::connect(&path) {
return Ok(s);
}
spawn_daemon()?;
for _ in 0..100 {
if let Ok(s) = UnixStream::connect(&path) {
return Ok(s);
}
thread::sleep(Duration::from_millis(10));
}
Err(io::Error::new(
ErrorKind::TimedOut,
"daemon did not come up",
))
}
fn spawn_daemon() -> io::Result<()> {
let exe = std::env::current_exe()?;
let dir = runtime_dir();
ensure_runtime_dir(&dir)?;
let log = fs::File::create(dir.join("daemon.log")).ok();
let mut cmd = std::process::Command::new(exe);
cmd.arg("--daemon")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(log.map(Stdio::from).unwrap_or_else(Stdio::null))
.process_group(0);
cmd.spawn()?;
Ok(())
}
pub fn run_kill() -> io::Result<()> {
let lock_path = runtime_dir().join("daemon.lock");
let Ok(file) = fs::OpenOptions::new()
.read(true)
.write(true)
.open(&lock_path)
else {
eprintln!("fleetcom: no daemon running");
return Ok(());
};
let mut file = match Flock::lock(file, FlockArg::LockExclusiveNonblock) {
Ok(_held) => {
eprintln!("fleetcom: no daemon running");
return Ok(());
}
Err((file, _)) => file,
};
let mut pid_str = String::new();
file.read_to_string(&mut pid_str)?;
let Some(pid) = pid_str.trim().parse::<i32>().ok().filter(|p| *p > 0) else {
return kill_via_socket();
};
match kill(Pid::from_raw(pid), Signal::SIGTERM) {
Ok(()) | Err(nix::errno::Errno::ESRCH) => {}
Err(e) => return Err(io::Error::other(e)),
}
for _ in 0..200 {
match Flock::lock(file, FlockArg::LockExclusiveNonblock) {
Ok(_held) => return Ok(()),
Err((f, _)) => file = f,
}
thread::sleep(Duration::from_millis(50));
}
Err(io::Error::new(
ErrorKind::TimedOut,
"daemon did not exit after SIGTERM",
))
}
fn kill_via_socket() -> io::Result<()> {
let path = socket_path();
match UnixStream::connect(&path) {
Ok(mut s) => {
let (kind, payload) = encode_hello(&LaunchContext::here());
write_frame(&mut s, kind, &payload)?;
let (kind, payload) = read_frame(&mut s)?;
check_hello_ack(kind, &payload)?;
let (kind, payload) = encode_command(&Command::Shutdown);
write_frame(&mut s, kind, &payload)?;
let mut buf = [0u8; 256];
while s.read(&mut buf).map(|n| n > 0).unwrap_or(false) {}
Ok(())
}
Err(_) => {
eprintln!("fleetcom: no daemon running");
Ok(())
}
}
}
pub fn run_daemon() -> io::Result<()> {
let dir = runtime_dir();
ensure_runtime_dir(&dir)?; let path = dir.join("default.sock");
let lock_file = fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(false) .open(dir.join("daemon.lock"))?;
let mut lock = match Flock::lock(lock_file, FlockArg::LockExclusiveNonblock) {
Ok(l) => l,
Err(_) => return Ok(()), };
lock.set_len(0)?;
lock.write_all(std::process::id().to_string().as_bytes())?;
let _ = fs::remove_file(&path);
let listener = UnixListener::bind(&path)?;
fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
let mut sup = Supervisor::new(24, 80);
let term = Arc::new(AtomicBool::new(false));
{
use signal_hook::consts::{SIGHUP, SIGINT, SIGTERM};
signal_hook::flag::register(SIGTERM, Arc::clone(&term))?;
signal_hook::flag::register(SIGINT, Arc::clone(&term))?;
signal_hook::flag::register(SIGHUP, Arc::clone(&term))?;
}
listener.set_nonblocking(true)?;
const IDLE_REAP: Duration = Duration::from_millis(100);
loop {
if term.load(Ordering::Relaxed) {
sup.apply(Command::Shutdown);
break;
}
match listener.accept() {
Ok((stream, _)) => {
stream.set_nonblocking(false)?;
if serve_client(&mut sup, stream, &term) == ServeOutcome::Shutdown {
break;
}
}
Err(e) if e.kind() == ErrorKind::WouldBlock || transient_accept_error(&e) => {
sup.reap();
thread::sleep(IDLE_REAP);
}
Err(e) => {
eprintln!("fleetcom: accept failed, shutting down: {e}");
break;
}
}
}
let _ = fs::remove_file(&path);
Ok(())
}
fn transient_accept_error(e: &io::Error) -> bool {
use nix::errno::Errno;
matches!(
e.raw_os_error(),
Some(code) if code == Errno::EMFILE as i32
|| code == Errno::ENFILE as i32
|| code == Errno::EINTR as i32
|| code == Errno::ECONNABORTED as i32
)
}
#[derive(PartialEq)]
enum ServeOutcome {
Disconnected,
Shutdown,
}
fn hello_version(kind: u8, payload: &[u8]) -> Option<u32> {
let v = jzon::parse(std::str::from_utf8(payload).ok()?).ok()?;
match kind {
crate::frame::KIND_HELLO => v["v"].as_u32(),
crate::frame::KIND_CONTROL if v["t"].as_str() == Some("hello") => v["v"].as_u32(),
_ => None,
}
}
fn handshake(stream: &mut UnixStream) -> Result<LaunchContext, String> {
let (kind, payload) = read_frame_bounded(stream, HANDSHAKE_TIMEOUT)
.map_err(|e| format!("no valid hello received: {e}"))?;
let mismatch = |version: u32| {
format!(
"protocol mismatch: daemon {} speaks v{PROTOCOL_VERSION}, client speaks \
v{version}; run 'fleetcom --kill' and retry",
env!("CARGO_PKG_VERSION"),
)
};
match decode_hello(kind, &payload) {
Some((PROTOCOL_VERSION, ctx)) => Ok(ctx),
Some((version, _)) => Err(mismatch(version)),
None => match hello_version(kind, &payload) {
Some(version) if version != PROTOCOL_VERSION => Err(mismatch(version)),
_ => Err(format!(
"daemon {} requires a hello handshake (older client?); upgrade the \
client or run 'fleetcom --kill' and retry",
env!("CARGO_PKG_VERSION"),
)),
},
}
}
fn serve_client(sup: &mut Supervisor, stream: UnixStream, stop: &AtomicBool) -> ServeOutcome {
let mut stream = stream;
match handshake(&mut stream) {
Ok(ctx) => {
sup.set_launch_context(ctx);
let (kind, payload) = encode_event(&Event::HelloOk);
if write_frame(&mut stream, kind, &payload).is_err() {
return ServeOutcome::Disconnected;
}
}
Err(reason) => {
eprintln!("fleetcom: refusing client: {reason}");
let (kind, payload) = encode_event(&Event::Status(reason));
let _ = write_frame(&mut stream, kind, &payload);
return ServeOutcome::Disconnected;
}
}
let Ok(read) = stream.try_clone() else {
return ServeOutcome::Disconnected;
};
let (wake_tx, wake_rx) = channel::<Wake>();
sup.set_waker(wake_tx.clone());
thread::spawn(move || {
let mut read = read;
while let Ok((kind, payload)) = read_frame(&mut read) {
if let Some(cmd) = decode_command(kind, &payload)
&& wake_tx.send(Wake::Cmd(cmd)).is_err()
{
return;
}
}
let _ = wake_tx.send(Wake::Hangup);
});
let mut write = stream;
let _ = write.set_write_timeout(Some(Duration::from_secs(5)));
let outcome = catch_unwind(AssertUnwindSafe(|| {
run_loop(sup, &wake_rx, stop, |ev| {
let (kind, payload) = encode_event(ev);
write_frame(&mut write, kind, &payload).is_ok()
})
}));
sup.clear_waker();
sup.clear_watch();
match outcome {
Ok(LoopExit::Shutdown) => ServeOutcome::Shutdown,
Ok(LoopExit::ClientGone) => ServeOutcome::Disconnected,
Err(_) => {
eprintln!("fleetcom: serve loop panicked; client dropped, fleet kept");
ServeOutcome::Disconnected
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn temp(tag: &str) -> PathBuf {
let d =
std::env::temp_dir().join(format!("fleetcom_daemon_test_{tag}_{}", std::process::id()));
let _ = fs::remove_dir_all(&d);
fs::create_dir_all(&d).unwrap();
d
}
#[test]
fn ensure_runtime_dir_rejects_symlink() {
let base = temp("symlink");
let target = base.join("target");
fs::create_dir(&target).unwrap();
let link = base.join("runtime");
std::os::unix::fs::symlink(&target, &link).unwrap();
assert!(ensure_runtime_dir(&link).is_err());
let _ = fs::remove_dir_all(&base);
}
#[test]
fn ensure_runtime_dir_rejects_plain_file() {
let base = temp("file");
let path = base.join("runtime");
fs::write(&path, b"x").unwrap();
assert!(ensure_runtime_dir(&path).is_err());
let _ = fs::remove_dir_all(&base);
}
#[test]
fn transient_accept_errors_are_classified() {
use nix::errno::Errno;
for errno in [
Errno::EMFILE,
Errno::ENFILE,
Errno::EINTR,
Errno::ECONNABORTED,
] {
assert!(
transient_accept_error(&io::Error::from_raw_os_error(errno as i32)),
"{errno} should be transient"
);
}
assert!(!transient_accept_error(&io::Error::from_raw_os_error(
Errno::EBADF as i32
)));
assert!(!transient_accept_error(&io::Error::other("no raw errno")));
}
#[test]
fn runtime_dir_resolution_order() {
let tmp = PathBuf::from("/tmpdir");
assert_eq!(
resolve_runtime_dir(
Some("/override".into()),
Some("/xdg".into()),
tmp.clone(),
501
),
PathBuf::from("/override")
);
assert_eq!(
resolve_runtime_dir(None, Some("/run/user/501".into()), tmp.clone(), 501),
PathBuf::from("/run/user/501/fleetcom")
);
assert_eq!(
resolve_runtime_dir(None, Some(String::new()), tmp.clone(), 501),
PathBuf::from("/tmpdir/fleetcom-501")
);
assert_eq!(
resolve_runtime_dir(None, None, tmp, 42),
PathBuf::from("/tmpdir/fleetcom-42")
);
}
#[test]
fn ensure_runtime_dir_creates_private_dir() {
let base = temp("create");
let path = base.join("runtime");
ensure_runtime_dir(&path).unwrap();
let mode = fs::symlink_metadata(&path).unwrap().permissions().mode();
assert_eq!(mode & 0o777, 0o700, "dir must be private");
ensure_runtime_dir(&path).unwrap();
let _ = fs::remove_dir_all(&base);
}
}