use std::convert::Infallible;
use std::ffi::OsString;
use std::fs::File;
use std::io;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::time::{Duration, Instant};
use crate::local::ProcessHandle;
use crate::readiness;
use crate::readiness::channel::{Channel, ENV};
pub const POLL: Duration = readiness::POLL_INTERVAL;
#[derive(Debug)]
#[non_exhaustive]
pub struct SpawnWait {
pub live: bool,
pub spawn_error: Option<io::Error>,
}
pub fn spawn(command: &mut Command) -> io::Result<Child> {
#[cfg(unix)]
{
crate::local::unix::spawn_with_child_waiting(command)
}
#[cfg(windows)]
{
let _spawn = crate::spawn_lock::spawning();
let _guard = crate::local::windows::StdioInheritGuard::suppress()?;
command.spawn()
}
}
pub fn spawn_detached(command: &mut Command) -> io::Result<Child> {
command
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
spawn(command)
}
#[derive(Debug, Default)]
pub enum DaemonOutput {
#[default]
Null,
File(File),
}
#[derive(Clone, Debug, PartialEq)]
enum EnvChange {
Set(OsString, OsString),
Remove(OsString),
}
#[derive(Debug)]
pub struct DaemonCommand {
program: PathBuf,
args: Vec<OsString>,
env: Vec<EnvChange>,
current_dir: Option<PathBuf>,
stdout: DaemonOutput,
stderr: DaemonOutput,
readiness: bool,
}
impl DaemonCommand {
pub fn new(program: impl AsRef<Path>) -> Self {
Self {
program: program.as_ref().to_path_buf(),
args: Vec::new(),
env: Vec::new(),
current_dir: None,
stdout: DaemonOutput::Null,
stderr: DaemonOutput::Null,
readiness: false,
}
}
pub fn arg(&mut self, arg: impl Into<OsString>) -> &mut Self {
self.args.push(arg.into());
self
}
pub fn args<I, S>(&mut self, args: I) -> &mut Self
where
I: IntoIterator<Item = S>,
S: Into<OsString>,
{
self.args.extend(args.into_iter().map(Into::into));
self
}
pub fn env(&mut self, name: impl Into<OsString>, value: impl Into<OsString>) -> &mut Self {
self.env.push(EnvChange::Set(name.into(), value.into()));
self
}
pub fn env_remove(&mut self, name: impl Into<OsString>) -> &mut Self {
self.env.push(EnvChange::Remove(name.into()));
self
}
pub fn current_dir(&mut self, dir: impl AsRef<Path>) -> &mut Self {
self.current_dir = Some(dir.as_ref().to_path_buf());
self
}
pub fn stdout(&mut self, target: DaemonOutput) -> &mut Self {
self.stdout = target;
self
}
pub fn stderr(&mut self, target: DaemonOutput) -> &mut Self {
self.stderr = target;
self
}
pub fn readiness_channel(&mut self) -> &mut Self {
self.readiness = true;
self
}
pub fn spawn(&self) -> io::Result<DaemonChild> {
#[cfg(unix)]
{
use std::os::fd::AsRawFd;
let mut command = Command::new(&self.program);
command.args(&self.args).stdin(Stdio::null());
for change in &self.env {
match change {
EnvChange::Set(name, value) => command.env(name, value),
EnvChange::Remove(name) => command.env_remove(name),
};
}
if let Some(dir) = &self.current_dir {
command.current_dir(dir);
}
command.stdout(unix_stdio(&self.stdout)?);
command.stderr(unix_stdio(&self.stderr)?);
let readiness = if self.readiness {
let (reader, child_end) = crate::local::unix::ready_pipe()?;
command.env(ENV, child_end.as_raw_fd().to_string());
Some((Channel::start(reader)?, child_end))
} else {
command.env_remove(ENV);
None
};
let keep = readiness
.as_ref()
.map(|(_, child_end)| child_end.as_raw_fd());
let (inner, exit) =
crate::local::unix::spawn_in_new_session_watched(&mut command, keep)?;
Ok(DaemonChild {
inner: crate::candidate::Reaped::new(inner),
exit,
readiness: readiness.map(|(channel, _child_end)| channel),
})
}
#[cfg(windows)]
{
use crate::local::command_line::EnvChange as WideChange;
use crate::local::windows::ready;
use crate::local::windows_spawn::{Spawn, Target, spawn};
use std::os::windows::ffi::OsStrExt;
let wide = |text: &OsString| text.encode_wide().collect::<Vec<u16>>();
let mut env: Vec<WideChange> = self
.env
.iter()
.map(|change| match change {
EnvChange::Set(name, value) => WideChange::Set(wide(name), wide(value)),
EnvChange::Remove(name) => WideChange::Remove(wide(name)),
})
.collect();
let name: Vec<u16> = ENV.encode_utf16().collect();
let readiness = if self.readiness {
let (pipe_name, pipe) = ready::create()?;
env.push(WideChange::Set(name, pipe_name.encode_utf16().collect()));
let (started, process) = std::sync::mpsc::channel();
let channel = Channel::start(ready::Reader::new(pipe, process))?;
Some((channel, started))
} else {
env.push(WideChange::Remove(name));
None
};
fn target(output: &DaemonOutput) -> Target<'_> {
match output {
DaemonOutput::Null => Target::Null,
DaemonOutput::File(file) => Target::File(file),
}
}
let mut inner = spawn(&Spawn {
program: &self.program,
args: &self.args,
env: &env,
current_dir: self.current_dir.as_deref(),
stdout: target(&self.stdout),
stderr: target(&self.stderr),
})?;
let handles = (|| {
let exit = ProcessHandle::open(
crate::ProcessId::new(inner.id()).expect("spawned daemon has a nonzero PID"),
)?;
let readiness = match readiness {
Some((channel, started)) => {
let _ = started.send(inner.watch()?);
Some(channel)
}
None => None,
};
Ok((exit, readiness))
})();
match handles {
Ok((exit, readiness)) => Ok(DaemonChild {
inner,
exit,
readiness,
}),
Err(error) => {
let _ = inner.kill();
Err(error)
}
}
}
}
}
impl From<&Command> for DaemonCommand {
fn from(command: &Command) -> Self {
let mut daemon = Self::new(command.get_program());
daemon.args(command.get_args());
for (name, value) in command.get_envs() {
match value {
Some(value) => daemon.env(name, value),
None => daemon.env_remove(name),
};
}
if let Some(dir) = command.get_current_dir() {
daemon.current_dir(dir);
}
daemon
}
}
#[cfg(unix)]
fn unix_stdio(target: &DaemonOutput) -> io::Result<Stdio> {
Ok(match target {
DaemonOutput::Null => Stdio::null(),
DaemonOutput::File(file) => Stdio::from(file.try_clone()?),
})
}
#[derive(Debug)]
pub struct DaemonChild {
#[cfg(unix)]
inner: crate::candidate::Reaped,
#[cfg(windows)]
inner: crate::local::windows_spawn::WindowsChild,
exit: ProcessHandle,
readiness: Option<Channel>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Startup {
Live,
Exited,
TimedOut,
}
impl DaemonChild {
pub fn id(&self) -> crate::ProcessId {
crate::ProcessId::new(self.inner.id()).expect("spawned daemon has a nonzero PID")
}
pub fn take_readiness(&mut self) -> Option<Channel> {
self.readiness.take()
}
pub fn in_callers_job(&self) -> bool {
#[cfg(unix)]
{
false
}
#[cfg(windows)]
{
self.inner.in_callers_job()
}
}
pub fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
self.inner.try_wait()
}
pub fn wait(&mut self) -> io::Result<ExitStatus> {
self.inner.wait()
}
pub fn wait_until(&mut self, deadline: Instant) -> io::Result<Option<ExitStatus>> {
if self.exits_by(deadline)? {
self.inner.wait().map(Some)
} else {
Ok(None)
}
}
fn exits_by(&mut self, deadline: Instant) -> io::Result<bool> {
crate::local::wait_child(|| self.inner.try_wait(), Some(&mut self.exit), deadline)
}
pub fn exit_handle(&mut self) -> &mut ProcessHandle {
&mut self.exit
}
pub fn wait_until_live(
&mut self,
deadline: Instant,
mut is_live: impl FnMut() -> bool,
) -> io::Result<Startup> {
loop {
if is_live() {
return Ok(Startup::Live);
}
let now = Instant::now();
if now >= deadline {
return Ok(Startup::TimedOut);
}
if self.exits_by(deadline.min(now + POLL))? {
return Ok(if is_live() {
Startup::Live
} else {
Startup::Exited
});
}
}
}
pub fn kill(&mut self) -> io::Result<()> {
self.inner.kill()
}
}
pub fn wait_until_live(deadline: Instant, mut is_live: impl FnMut() -> bool) -> bool {
readiness::wait_until(deadline, |_| Ok::<_, Infallible>(is_live().then_some(())))
.ok()
.flatten()
.is_some()
}
pub fn spawn_and_wait(
command: &mut Command,
budget: Duration,
is_live: impl FnMut() -> bool,
) -> SpawnWait {
let spawn_error = spawn_detached(command).err();
let live = wait_until_live(Instant::now() + budget, is_live);
SpawnWait { live, spawn_error }
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn wait_until_live_fails_when_the_probe_never_succeeds() {
let started = Instant::now();
assert!(!wait_until_live(
Instant::now() + Duration::from_millis(300),
|| false,
));
assert!(started.elapsed() >= Duration::from_millis(250));
assert!(started.elapsed() < Duration::from_secs(3));
}
#[test]
fn wait_until_live_returns_as_soon_as_the_probe_succeeds() {
let mut n = 0;
assert!(wait_until_live(
Instant::now() + Duration::from_secs(2),
|| {
n += 1;
n >= 2
}
));
assert!(n >= 2);
}
#[test]
fn a_missing_binary_is_reported_on_spawn_error() {
let mut command = Command::new("/this/does/not/exist-kunobi-daemon-launch");
let outcome = spawn_and_wait(&mut command, Duration::from_millis(80), || false);
assert!(!outcome.live);
assert!(outcome.spawn_error.is_some());
}
#[cfg(unix)]
#[test]
fn a_stale_socket_file_is_not_live() {
let dir = tempfile::tempdir().unwrap();
let sock = dir.path().join("b.sock");
{
let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();
drop(listener);
}
assert!(sock.exists());
let stale_deadline = Instant::now() + Duration::from_secs(1);
while std::os::unix::net::UnixStream::connect(&sock).is_ok() {
assert!(
Instant::now() < stale_deadline,
"closed listener stayed live"
);
#[expect(
clippy::disallowed_methods,
reason = "Bounded test probe waits for the closed Unix listener to stop accepting."
)]
std::thread::sleep(Duration::from_millis(10));
}
assert!(!wait_until_live(
Instant::now() + Duration::from_millis(300),
|| std::os::unix::net::UnixStream::connect(&sock).is_ok(),
));
}
#[cfg(unix)]
#[test]
fn spawn_and_wait_still_sees_a_peer_another_client_started() {
let dir = tempfile::tempdir().unwrap();
let sock = dir.path().join("b.sock");
let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();
let mut command = Command::new("/this/does/not/exist-kunobi-daemon-launch");
let outcome = spawn_and_wait(&mut command, Duration::from_secs(2), || {
std::os::unix::net::UnixStream::connect(&sock).is_ok()
});
assert!(outcome.live);
assert!(outcome.spawn_error.is_some());
drop(listener);
}
#[cfg(unix)]
#[test]
fn spawn_and_wait_waits_for_a_peer_that_binds_later() {
let dir = tempfile::tempdir().unwrap();
let sock = dir.path().join("b.sock");
let delayed_socket = sock.clone();
let (stop_tx, stop_rx) = std::sync::mpsc::channel();
let delayed = std::thread::spawn(move || {
#[expect(
clippy::disallowed_methods,
reason = "The fixture binds after startup to exercise retrying a failed liveness probe."
)]
std::thread::sleep(Duration::from_millis(250));
let listener = std::os::unix::net::UnixListener::bind(delayed_socket).unwrap();
let _ = stop_rx.recv_timeout(Duration::from_secs(2));
listener
});
let mut command = Command::new("/this/does/not/exist-kunobi-daemon-launch");
let outcome = spawn_and_wait(&mut command, Duration::from_secs(2), || {
std::os::unix::net::UnixStream::connect(&sock).is_ok()
});
stop_tx.send(()).unwrap();
drop(delayed.join().unwrap());
assert!(outcome.live);
}
#[cfg(unix)]
fn sleeper() -> DaemonChild {
let mut command = DaemonCommand::new("/bin/sleep");
command.arg("30");
command.spawn().unwrap()
}
#[cfg(unix)]
#[test]
#[ignore = "helper process for the SIGINT test"]
fn detach_sigint_helper() {
let Some(out) = std::env::var_os("KDAEMON_DETACH_HELPER_OUT") else {
return;
};
let child = sleeper();
std::fs::write(&out, child.id().to_string()).unwrap();
#[expect(
clippy::disallowed_methods,
reason = "The signal-test helper must stay alive until the parent sends SIGINT."
)]
std::thread::sleep(Duration::from_secs(30));
}
#[cfg(unix)]
#[test]
fn a_daemon_survives_sigint_to_its_callers_process_group() {
use std::os::unix::process::CommandExt;
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("pid");
let mut helper = Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"launch::tests::detach_sigint_helper",
"--ignored",
"--nocapture",
"--test-threads=1",
])
.env("KDAEMON_DETACH_HELPER_OUT", &out)
.stdout(Stdio::null())
.process_group(0)
.spawn()
.unwrap();
let deadline = Instant::now() + Duration::from_secs(20);
let daemon: u32 = loop {
if let Ok(pid) = std::fs::read_to_string(&out)
&& let Ok(pid) = pid.trim().parse()
{
break pid;
}
assert!(Instant::now() < deadline, "helper never started the daemon");
#[expect(
clippy::disallowed_methods,
reason = "The cross-process fixture reports its daemon PID through a file, with a deadline."
)]
std::thread::sleep(Duration::from_millis(20));
};
let mut daemon_exit = ProcessHandle::open(crate::ProcessId::new(daemon).unwrap()).unwrap();
crate::local::unix::interrupt_process_group(helper.id()).unwrap();
let status = helper.wait().unwrap();
let died = daemon_exit
.wait_until(Instant::now() + Duration::from_secs(2))
.unwrap();
let _ = Command::new("kill")
.args(["-KILL", &daemon.to_string()])
.status();
assert!(!status.success(), "SIGINT did not reach the helper's group");
assert!(!died, "the daemon died with its caller's process group");
}
#[cfg(unix)]
fn reaches_eof_within(read: File, timeout: Duration) -> bool {
use std::io::Read;
let (done, finished) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let mut read = read;
let mut buffer = [0u8; 64];
while matches!(read.read(&mut buffer), Ok(n) if n > 0) {}
let _ = done.send(());
});
finished.recv_timeout(timeout).is_ok()
}
#[cfg(unix)]
#[test]
fn a_daemon_does_not_keep_the_callers_pipe_open() {
let (read, write) = crate::local::unix::inheritable_pipe();
let mut child = sleeper();
drop(write);
let eof = reaches_eof_within(read, Duration::from_secs(5));
child.kill().unwrap();
child.wait().unwrap();
assert!(eof, "the daemon kept the caller's pipe open");
}
#[cfg(unix)]
#[test]
fn a_plain_spawn_would_keep_the_callers_pipe_open() {
let (read, write) = crate::local::unix::inheritable_pipe();
let mut child = Command::new("/bin/sleep")
.arg("30")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.unwrap();
drop(write);
let eof = reaches_eof_within(read, Duration::from_secs(1));
child.kill().unwrap();
child.wait().unwrap();
assert!(!eof, "a plain spawn is expected to inherit the write end");
}
#[cfg(unix)]
#[test]
fn a_daemon_gets_its_output_file_env_and_directory() {
let dir = tempfile::tempdir().unwrap();
let log = dir.path().join("daemon.log");
let file = File::create(&log).unwrap();
assert!(std::env::var_os("CARGO_MANIFEST_DIR").is_some());
let mut command = DaemonCommand::new("/bin/sh");
command
.args([
"-c",
"echo \"$KDAEMON_SET|${CARGO_MANIFEST_DIR:-unset}|$(pwd)\" >&2",
])
.env("KDAEMON_SET", "set-by-builder")
.env_remove("CARGO_MANIFEST_DIR")
.current_dir(dir.path())
.stderr(DaemonOutput::File(file));
let mut child = command.spawn().unwrap();
assert!(child.wait().unwrap().success());
let written = std::fs::read_to_string(&log).unwrap();
let expected_dir = dir.path().canonicalize().unwrap();
assert_eq!(
written.trim(),
format!("set-by-builder|unset|{}", expected_dir.display())
);
}
#[test]
fn a_daemon_command_takes_a_command_s_program_arguments_environment_and_directory() {
let mut command = Command::new("exampled");
command
.args(["serve", "--quiet"])
.env("KDAEMON_SET", "set-by-command")
.env_remove("KDAEMON_REMOVED")
.current_dir("daemon-dir");
let daemon = DaemonCommand::from(&command);
assert_eq!(daemon.program, Path::new("exampled"));
assert_eq!(daemon.args, ["serve", "--quiet"]);
assert_eq!(daemon.env.len(), 2);
assert!(daemon.env.contains(&EnvChange::Set(
"KDAEMON_SET".into(),
"set-by-command".into()
)));
assert!(
daemon
.env
.contains(&EnvChange::Remove("KDAEMON_REMOVED".into()))
);
assert_eq!(daemon.current_dir.as_deref(), Some(Path::new("daemon-dir")));
}
#[test]
fn a_missing_daemon_binary_is_a_spawn_error() {
let error = DaemonCommand::new("/this/does/not/exist-kunobi-daemon")
.spawn()
.unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::NotFound);
}
#[cfg(unix)]
struct Held {
fifo: PathBuf,
_dir: tempfile::TempDir,
}
#[cfg(unix)]
impl Held {
fn release(&self) {
std::fs::write(&self.fifo, b"go\n").unwrap();
}
}
#[cfg(unix)]
impl Drop for Held {
fn drop(&mut self) {
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
if let Ok(mut fifo) = std::fs::OpenOptions::new()
.write(true)
.custom_flags(libc::O_NONBLOCK)
.open(&self.fifo)
{
let _ = fifo.write_all(b"go\n");
}
}
}
#[cfg(unix)]
fn held_daemon() -> (DaemonChild, Held) {
let dir = tempfile::tempdir().unwrap();
let fifo = dir.path().join("release");
assert!(
Command::new("mkfifo")
.arg(&fifo)
.status()
.unwrap()
.success()
);
let mut command = DaemonCommand::new("/bin/sh");
command
.args(["-c", "read line < \"$1\"; exit 3", "sh"])
.arg(&fifo);
(command.spawn().unwrap(), Held { fifo, _dir: dir })
}
#[cfg(unix)]
#[test]
fn a_daemon_that_exits_before_it_is_live_ends_the_wait() {
let (mut child, held) = held_daemon();
held.release();
let started = Instant::now();
let outcome = child
.wait_until_live(started + Duration::from_secs(60), || false)
.unwrap();
assert_eq!(outcome, Startup::Exited);
assert!(started.elapsed() < Duration::from_secs(30));
assert_eq!(child.wait().unwrap().code(), Some(3));
}
#[cfg(unix)]
#[test]
fn a_live_daemon_is_reported_as_soon_as_a_probe_succeeds() {
let (mut child, held) = held_daemon();
let mut probes = 0;
let started = Instant::now();
let outcome = child
.wait_until_live(started + Duration::from_secs(60), || {
probes += 1;
probes == 3
})
.unwrap();
assert_eq!(outcome, Startup::Live);
assert_eq!(probes, 3);
assert!(started.elapsed() < Duration::from_secs(30));
assert!(child.wait_until(Instant::now()).unwrap().is_none());
held.release();
let status = child
.wait_until(Instant::now() + Duration::from_secs(30))
.unwrap()
.expect("the daemon exited once released");
assert_eq!(status.code(), Some(3));
}
#[cfg(unix)]
#[test]
fn a_daemon_that_never_becomes_live_times_out_without_busy_probing() {
let (mut child, held) = held_daemon();
let mut probes = 0;
let deadline = Instant::now() + Duration::from_millis(200);
let outcome = child
.wait_until_live(deadline, || {
probes += 1;
false
})
.unwrap();
assert_eq!(outcome, Startup::TimedOut);
assert!(Instant::now() >= deadline);
assert!(probes <= 20, "{probes} probes in 200 ms");
held.release();
child.wait().unwrap();
}
#[cfg(unix)]
#[test]
fn a_daemon_that_exits_while_another_serves_is_live() {
let (mut child, held) = held_daemon();
held.release();
child.exit_handle().wait().unwrap();
assert_eq!(child.exit_handle().pid(), child.id());
let mut probes = 0;
let outcome = child
.wait_until_live(Instant::now() + Duration::from_secs(60), || {
probes += 1;
probes == 2
})
.unwrap();
assert_eq!(outcome, Startup::Live);
assert_eq!(child.wait().unwrap().code(), Some(3));
}
#[cfg(unix)]
#[allow(unsafe_code)]
fn reaped(pid: crate::ProcessId) -> bool {
let pid = libc::pid_t::try_from(pid.get()).unwrap();
let result = unsafe { libc::kill(pid, 0) };
result == -1 && io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH)
}
#[cfg(unix)]
fn reaped_soon(pid: crate::ProcessId) -> bool {
let deadline = Instant::now() + Duration::from_secs(10);
crate::backoff::poll(|| Ok(reaped(pid)), Some(deadline)).unwrap()
}
#[cfg(unix)]
#[test]
fn a_dropped_daemon_keeps_running_and_is_reaped_once_it_exits() {
let (child, held) = held_daemon();
let pid = child.id();
let mut exit = ProcessHandle::open(pid).unwrap();
drop(child);
assert!(
!exit
.wait_until(Instant::now() + Duration::from_millis(200))
.unwrap(),
"dropping the daemon's handle stopped it"
);
held.release();
assert!(exit.wait_until(Instant::now() + SILENCE).unwrap());
assert!(reaped_soon(pid), "the exited daemon was left a zombie");
}
#[test]
fn a_daemon_taken_as_a_candidate_is_stopped_when_dropped_unselected() {
let child = silent_daemon().spawn().unwrap();
let pid = child.id();
let mut exit = ProcessHandle::open(pid).unwrap();
let mut candidate = crate::Candidate::from(child);
assert_eq!(candidate.id(), pid);
assert_eq!(candidate.exit_handle().unwrap().pid(), pid);
assert!(candidate.try_wait().unwrap().is_none());
assert!(candidate.wait_until(Instant::now()).unwrap().is_none());
drop(candidate);
assert!(
exit.has_exited().unwrap(),
"the unselected daemon was not stopped"
);
}
#[cfg(unix)]
#[test]
fn a_selected_daemon_candidate_keeps_running_and_is_reaped_once_it_exits() {
let (child, held) = held_daemon();
let pid = child.id();
let mut exit = ProcessHandle::open(pid).unwrap();
let mut candidate = crate::Candidate::from(child);
candidate.selected();
drop(candidate);
assert!(
!exit
.wait_until(Instant::now() + Duration::from_millis(200))
.unwrap(),
"dropping a selected candidate stopped it"
);
held.release();
assert!(exit.wait_until(Instant::now() + SILENCE).unwrap());
assert!(reaped_soon(pid), "the exited daemon was left a zombie");
}
#[cfg(unix)]
#[test]
fn spawn_detached_execs() {
let bin = ["/usr/bin/true", "/bin/true"]
.into_iter()
.find(|path| std::path::Path::new(path).exists());
let Some(bin) = bin else {
return;
};
let mut command = Command::new(bin);
let mut child = spawn_detached(&mut command).unwrap();
assert!(child.wait().unwrap().success());
}
use crate::readiness::channel::{NotReady, Notifier};
const SILENCE: Duration = Duration::from_secs(60);
fn progress_then_ready(mut notifier: Notifier) -> io::Result<()> {
notifier.progress("")?;
notifier.progress("loading")?;
notifier.ready()
}
#[test]
#[ignore = "helper process for the readiness tests"]
#[allow(unsafe_code)]
fn readiness_helper() {
let (Some(mode), Some(out)) = (
std::env::var_os("KDAEMON_READY_HELPER"),
std::env::var_os("KDAEMON_READY_HELPER_OUT"),
) else {
return;
};
let taken = unsafe { Notifier::from_env() };
let left = std::env::var_os(ENV).is_some();
let again = unsafe { Notifier::from_env() };
let report = match (mode.to_str(), taken) {
_ if left => "the variable was left in the environment".to_owned(),
_ if !matches!(again, Ok(None)) => format!("taken twice: {again:?}"),
(Some("progress-then-ready"), Ok(Some(notifier))) => {
match progress_then_ready(notifier) {
Ok(()) => "sent".to_owned(),
Err(error) => error.to_string(),
}
}
(Some("exit-before-ready"), Ok(Some(_notifier))) => "exiting".to_owned(),
(Some("no-channel"), Ok(None)) => "no channel".to_owned(),
(mode, taken) => format!("unexpected: {mode:?} {taken:?}"),
};
std::fs::write(out, report).unwrap();
}
fn readiness_helper_daemon(mode: &str, out: &Path) -> DaemonCommand {
let mut command = DaemonCommand::new(std::env::current_exe().unwrap());
command
.args([
"--exact",
"launch::tests::readiness_helper",
"--ignored",
"--nocapture",
"--test-threads=1",
])
.env("KDAEMON_READY_HELPER", mode)
.env("KDAEMON_READY_HELPER_OUT", out.as_os_str());
command
}
fn silent_daemon() -> DaemonCommand {
#[cfg(unix)]
{
let mut command = DaemonCommand::new("/bin/sleep");
command.arg("30");
command
}
#[cfg(windows)]
{
let root = std::env::var_os("SystemRoot").unwrap_or_else(|| "C:\\Windows".into());
let mut command =
DaemonCommand::new(PathBuf::from(root).join("System32").join("PING.EXE"));
command.args(["-n", "30", "127.0.0.1"]);
command
}
}
#[test]
fn a_daemon_reports_progress_then_ready() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("report");
let mut child = readiness_helper_daemon("progress-then-ready", &out)
.readiness_channel()
.spawn()
.unwrap();
let mut channel = child.take_readiness().expect("the channel was asked for");
assert!(
child.take_readiness().is_none(),
"the channel is taken once"
);
assert_eq!(channel.wait(SILENCE), Ok(()));
assert_eq!(channel.progress_reports(), 2);
assert_eq!(channel.last_progress().as_deref(), Some("loading"));
assert!(child.wait().unwrap().success());
assert_eq!(std::fs::read_to_string(&out).unwrap(), "sent");
}
#[test]
fn a_daemon_that_exits_before_ready_is_noticed_at_once() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("report");
let mut child = readiness_helper_daemon("exit-before-ready", &out)
.readiness_channel()
.spawn()
.unwrap();
let mut channel = child.take_readiness().unwrap();
assert!(
child
.exit_handle()
.wait_until(Instant::now() + SILENCE)
.unwrap()
);
assert_eq!(channel.wait(SILENCE), Err(NotReady::Died));
assert_eq!(
child
.wait_until_live(Instant::now() + SILENCE, || false)
.unwrap(),
Startup::Exited,
);
child.wait().unwrap();
assert_eq!(std::fs::read_to_string(&out).unwrap(), "exiting");
}
#[test]
fn a_killed_daemon_ends_its_channel() {
let mut child = silent_daemon().readiness_channel().spawn().unwrap();
let mut channel = child.take_readiness().unwrap();
assert_eq!(channel.wait(Duration::ZERO), Err(NotReady::Silent));
child.kill().unwrap();
child.wait().unwrap();
assert_eq!(channel.wait(SILENCE), Err(NotReady::Died));
}
#[test]
fn without_opting_in_the_daemon_gets_no_channel() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("report");
let mut command = readiness_helper_daemon("no-channel", &out);
command.env(ENV, "5");
let mut child = command.spawn().unwrap();
assert!(child.take_readiness().is_none());
assert!(child.wait().unwrap().success());
assert_eq!(std::fs::read_to_string(&out).unwrap(), "no channel");
}
#[cfg(unix)]
#[test]
fn a_daemon_with_a_readiness_channel_still_does_not_keep_the_callers_pipe_open() {
let (read, write) = crate::local::unix::inheritable_pipe();
let mut command = DaemonCommand::new("/bin/sleep");
command.arg("30").readiness_channel();
let mut child = command.spawn().unwrap();
drop(write);
let eof = reaches_eof_within(read, Duration::from_secs(5));
child.kill().unwrap();
child.wait().unwrap();
assert!(eof, "the channel's exception let the caller's pipe through");
}
#[cfg(feature = "async")]
#[tokio::test]
async fn a_daemon_s_readiness_can_be_awaited() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("report");
let mut child = readiness_helper_daemon("progress-then-ready", &out)
.readiness_channel()
.spawn()
.unwrap();
let mut channel = child.take_readiness().unwrap();
assert_eq!(channel.wait_async(SILENCE).await, Ok(()));
assert_eq!(channel.progress_reports(), 2);
assert!(child.wait().unwrap().success());
}
}