use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use super::client::Client;
use super::protocol::Info;
use super::machine::MachineRendezvous;
const START_TIMEOUT: Duration = Duration::from_secs(20);
const START_POLL: Duration = Duration::from_millis(50);
const STOP_TIMEOUT: Duration = Duration::from_secs(20);
const BUSY_TIMEOUT: Duration = Duration::from_secs(60);
pub fn log_path(state_dir: &Path) -> PathBuf {
state_dir.join("daemon.log")
}
#[derive(Debug, Clone)]
pub struct Spawn {
pub listen: std::net::IpAddr,
pub http_port: Port,
pub parent_pid: Option<u32>,
pub debug: bool,
pub peer_port: u16,
}
impl Default for Spawn {
fn default() -> Self {
Self {
listen: super::LOOPBACK,
http_port: Port::FirstFreeFrom(super::DEFAULT_HTTP_PORT),
parent_pid: None,
debug: false,
peer_port: crate::peer::DEFAULT_PEER_PORT,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Port {
Exact(u16),
FirstFreeFrom(u16),
}
const PORT_ATTEMPTS: u16 = 16;
impl Port {
fn resolve(self, listen: std::net::IpAddr) -> Result<u16, String> {
match self {
Self::Exact(port) => Ok(port),
Self::FirstFreeFrom(first) => {
for offset in 0..PORT_ATTEMPTS {
let port = first.saturating_add(offset);
if std::net::TcpListener::bind(std::net::SocketAddr::new(listen, port)).is_ok()
{
return Ok(port);
}
}
Err(format!(
"no free port for the workbench between {first} and {}; \
name one with `wsb serve --port`",
first.saturating_add(PORT_ATTEMPTS - 1)
))
}
}
}
}
fn is_stopped(state_dir: &Path, stopping: &Info) -> Result<bool, String> {
let unpublished = match MachineRendezvous::read(state_dir)? {
Some(rendezvous) => rendezvous.pid != stopping.pid,
None => true,
};
if !unpublished {
return Ok(false);
}
let address = std::net::SocketAddr::new(
super::client::dialable(stopping.http_address),
stopping.http_port,
);
Ok(std::net::TcpListener::bind(address).is_ok())
}
pub fn wait_until_stopped(state_dir: &Path, stopping: &Info) -> Result<(), String> {
let address = std::net::SocketAddr::new(
super::client::dialable(stopping.http_address),
stopping.http_port,
);
let deadline = Instant::now() + STOP_TIMEOUT;
loop {
if is_stopped(state_dir, stopping)? {
return Ok(());
}
if Instant::now() >= deadline {
return Err(format!(
"wsb-daemon (pid {}) accepted the request to stop but is still running \
{}s later, and {address} is still its port.\n \
its log is in {}",
stopping.pid,
STOP_TIMEOUT.as_secs(),
log_path(state_dir).display()
));
}
std::thread::sleep(START_POLL);
}
}
pub fn ensure_running(
state_dir: &Path,
worktree_root: &Path,
spawn_with: &Spawn,
) -> Result<(Client, Info, bool), String> {
if let Some(existing) = MachineRendezvous::read(state_dir)? {
let client = Client::from_rendezvous(&existing)?;
if let Some(info) = answer_of(&client, &existing, BUSY_TIMEOUT)? {
require_same_version(&info)?;
client.resolve_worktree(worktree_root)?;
return Ok((client, info, false));
}
MachineRendezvous::remove_if_published_by(state_dir, existing.pid)?;
}
let (child, log_was) = spawn(state_dir, worktree_root, spawn_with)?;
let (client, info) = wait_for_daemon(state_dir, worktree_root, child, log_was, START_TIMEOUT)?;
Ok((client, info, true))
}
fn answer_of(client: &Client, existing: &MachineRendezvous, wait: Duration) -> Result<Option<Info>, String> {
let deadline = Instant::now() + wait;
loop {
if let Ok(info) = client.info() {
return Ok(Some(info));
}
if !still_serving(existing) {
return Ok(None);
}
if Instant::now() >= deadline {
return Err(format!(
"wsb-daemon (pid {}) is serving this machine on port {} and has not answered for \
{}s: it is busy, not gone. Try again when the machine is less loaded, or \
restart it:\n wsb daemon restart",
existing.pid,
existing.http_port,
wait.as_secs()
));
}
std::thread::sleep(START_POLL);
}
}
pub fn still_serving(published: &MachineRendezvous) -> bool {
let address = std::net::SocketAddr::new(super::client::dialable(published.http_address), published.http_port);
crate::ops::process::is_alive(published.pid) && std::net::TcpListener::bind(address).is_err()
}
pub fn daemon_executable() -> Result<PathBuf, String> {
beside_me(Half::Daemon)
}
pub fn cli_executable() -> Result<PathBuf, String> {
beside_me(Half::Cli)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Half {
Cli,
Daemon,
}
impl Half {
pub(crate) fn file_name(self) -> &'static str {
match (self, cfg!(windows)) {
(Self::Cli, false) => "wsb",
(Self::Cli, true) => "wsb.exe",
(Self::Daemon, false) => "wsb-daemon",
(Self::Daemon, true) => "wsb-daemon.exe",
}
}
}
fn beside_me(half: Half) -> Result<PathBuf, String> {
let own = std::env::current_exe()
.map_err(|e| format!("cannot locate this executable: {e}"))?;
let directory = own
.parent()
.ok_or_else(|| format!("{} has no directory", own.display()))?;
beside(directory, half)
}
pub const INSTALL_PAIR: &str = "qrust install wsb wsb-daemon --path . --force";
pub fn beside(directory: &Path, half: Half) -> Result<PathBuf, String> {
let candidate = directory.join(half.file_name());
if candidate.is_file() {
return Ok(candidate);
}
Err(match half {
Half::Daemon => format!(
"the daemon is not installed beside wsb: {} does not exist.\n \
They are two packages in one workspace, installed together from wsb/:\n \
{INSTALL_PAIR}\n \
Or run this command with --no-daemon, which does the work in this \
process and shows it in no workbench.",
candidate.display()
),
Half::Cli => format!(
"wsb is not installed beside the daemon: {} does not exist.\n \
They are two packages in one workspace, installed together from wsb/:\n \
{INSTALL_PAIR}",
candidate.display()
),
})
}
pub fn daemon_arguments(worktree_root: &Path, with: &Spawn) -> Result<Vec<String>, String> {
let mut arguments: Vec<String> = vec![
"--worktree".to_string(),
worktree_root.display().to_string(),
"--port".to_string(),
with.http_port.resolve(with.listen)?.to_string(),
"--listen".to_string(),
with.listen.to_string(),
"--peer-port".to_string(),
with.peer_port.to_string(),
];
if let Some(pid) = with.parent_pid {
arguments.push("--parent-pid".to_string());
arguments.push(pid.to_string());
}
if with.debug {
arguments.push("--debug".to_string());
}
Ok(arguments)
}
fn spawn(
state_dir: &Path,
worktree_root: &Path,
with: &Spawn,
) -> Result<(std::process::Child, u64), String> {
let executable = daemon_executable()?;
std::fs::create_dir_all(state_dir)
.map_err(|e| format!("cannot create {}: {e}", state_dir.display()))?;
let log = log_path(state_dir);
let out = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&log)
.map_err(|e| format!("cannot open {}: {e}", log.display()))?;
let err = out
.try_clone()
.map_err(|e| format!("cannot open {}: {e}", log.display()))?;
let log_was = std::fs::metadata(&log).map(|m| m.len()).unwrap_or(0);
let arguments = daemon_arguments(worktree_root, with)?;
let mut argv = vec![executable.display().to_string()];
argv.extend(arguments.iter().cloned());
let scoped = crate::ops::scope::wrap(
&argv,
&format!("wsb-daemon-{}", crate::ops::scope::worktree_tag(worktree_root)),
crate::ops::scope::DAEMON_PROPERTIES,
);
let mut command = match &scoped {
Some(wrapped) => {
let mut command = crate::git_backend::command_without_repo_scope(&wrapped[0]);
command.args(&wrapped[1..]);
command
}
None => {
let mut command = crate::git_backend::command_without_repo_scope(&executable);
command.args(&arguments);
command
}
};
command
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::from(out))
.stderr(std::process::Stdio::from(err));
command.detached();
crate::platform::withhold_standard_handles()
.map_err(|e| format!("cannot keep this process's output out of the daemon: {e}"))?;
let child = command
.spawn()
.map_err(|e| format!("cannot start {}: {e}", executable.display()))?;
Ok((child, log_was))
}
fn log_since(state_dir: &Path, from: u64) -> Option<String> {
use std::io::{Read, Seek, SeekFrom};
let mut file = std::fs::File::open(log_path(state_dir)).ok()?;
file.seek(SeekFrom::Start(from)).ok()?;
let mut text = String::new();
file.read_to_string(&mut text).ok()?;
let trimmed = text.trim();
if trimmed.is_empty() {
return None;
}
let tail: Vec<&str> = trimmed.lines().rev().take(12).collect();
Some(
tail.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n "),
)
}
fn wait_for_daemon(
state_dir: &Path,
worktree_root: &Path,
mut child: std::process::Child,
log_was: u64,
timeout: Duration,
) -> Result<(Client, Info), String> {
let deadline = Instant::now() + timeout;
let mut last_error = String::from("the daemon published no rendezvous file");
while Instant::now() < deadline {
if let Some(rendezvous) = MachineRendezvous::read(state_dir)? {
let client = Client::from_rendezvous(&rendezvous)?;
match client.info() {
Ok(info) => {
require_same_version(&info)?;
client.resolve_worktree(worktree_root)?;
std::thread::spawn(move || {
let _ = child.wait();
});
return Ok((client, info));
}
Err(error) => last_error = error,
}
}
match child.try_wait() {
Ok(Some(status)) => {
return Err(match log_since(state_dir, log_was) {
Some(said) => format!(
"the daemon stopped immediately ({status}). It said:\n {said}\n \
its full output is in {}",
log_path(state_dir).display()
),
None => format!(
"the daemon stopped immediately ({status}) without saying why.\n \
its output would be in {}",
log_path(state_dir).display()
),
});
}
Ok(None) => {}
Err(error) => {
return Err(format!("cannot tell whether the daemon is running: {error}"))
}
}
std::thread::sleep(START_POLL);
}
let pid = child.id();
let ended = match child.kill().and_then(|()| child.wait()) {
Ok(_) => "it has been stopped".to_string(),
Err(error) => format!("stopping it failed ({error}); it is pid {pid}"),
};
MachineRendezvous::remove_if_published_by(state_dir, pid)?;
Err(format!(
"the daemon did not start within {}s, and {ended}: {last_error}\n \
its output is in {}",
timeout.as_secs(),
log_path(state_dir).display()
))
}
fn require_same_version(info: &Info) -> Result<(), String> {
let mine = crate::get_version();
if info.version == mine {
return Ok(());
}
Err(format!(
"the running daemon is wsb {} and this is wsb {mine}.\n \
They share the worktree database and this protocol, so they are not \
mixed. Stop the old one and this command will start the new one:\n \
wsb daemon stop",
info.version
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test9925_the_pair_is_found_beside_itself_never_on_path() {
let temp = tempfile::tempdir().expect("tempdir");
let directory = temp.path();
let missing_daemon =
beside(directory, Half::Daemon).expect_err("an absent daemon must be refused");
assert!(missing_daemon.contains("wsb-daemon"), "{missing_daemon}");
assert!(missing_daemon.contains(INSTALL_PAIR), "{missing_daemon}");
let missing_cli = beside(directory, Half::Cli).expect_err("an absent wsb must be refused");
assert!(missing_cli.contains("wsb is not installed"), "{missing_cli}");
assert!(missing_cli.contains(INSTALL_PAIR), "{missing_cli}");
for half in [Half::Cli, Half::Daemon] {
std::fs::write(directory.join(half.file_name()), b"#!/bin/sh\n").expect("write");
}
let cli = beside(directory, Half::Cli).expect("wsb is here");
let daemon = beside(directory, Half::Daemon).expect("the daemon is here");
assert_eq!(cli.parent(), Some(directory));
assert_eq!(daemon.parent(), Some(directory));
assert_ne!(cli, daemon, "the two halves must not resolve to one file");
let lonely = tempfile::tempdir().expect("tempdir");
std::fs::write(lonely.path().join(Half::Cli.file_name()), b"x").expect("write");
assert!(beside(lonely.path(), Half::Cli).is_ok());
assert!(beside(lonely.path(), Half::Daemon).is_err());
}
fn info(pid: u32, port: u16) -> Info {
Info {
version: env!("CARGO_PKG_VERSION").to_string(),
pid,
http_address: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
http_port: port,
machine_id: "0000000000000000".to_string(),
machine_name: "test".to_string(),
peer_address: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
peer_port: 0,
peer_refused: None,
worktrees: Vec::new(),
}
}
fn rendezvous(pid: u32, port: u16) -> MachineRendezvous {
MachineRendezvous {
version: env!("CARGO_PKG_VERSION").to_string(),
pid,
http_address: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
http_port: port,
machine_id: "0000000000000000".to_string(),
machine_name: "test".to_string(),
peer_address: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
peer_port: 0,
started_at: 0,
control_token: String::new(),
}
}
fn listening() -> (std::net::TcpListener, u16) {
let listener = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
.expect("a loopback port");
let port = listener.local_addr().expect("the bound address").port();
(listener, port)
}
fn dead_pid() -> u32 {
let me = std::env::current_exe().expect("this test binary");
let mut child = crate::git_backend::command_without_repo_scope(&me)
.arg("--list")
.stdout(std::process::Stdio::null())
.spawn()
.expect("the test binary runs");
let pid = child.id();
child.wait().expect("reaped");
pid
}
#[test]
fn test12916_a_daemon_past_its_start_deadline_is_stopped() {
let dir = tempfile::tempdir().expect("a scratch directory");
#[cfg(unix)]
let starting = crate::git_backend::command_without_repo_scope("sleep").arg("60").spawn();
#[cfg(windows)]
let starting = crate::git_backend::command_without_repo_scope("ping")
.args(["-n", "61", "127.0.0.1"])
.stdout(std::process::Stdio::null())
.spawn();
let child = starting.expect("a process that outlives the deadline");
let pid = child.id();
let refused = wait_for_daemon(dir.path(), dir.path(), child, 0, Duration::from_millis(300))
.expect_err("a daemon that never published has not started");
assert!(refused.contains("has been stopped"), "{refused}");
assert!(
!crate::ops::process::is_alive(pid),
"the daemon that missed its deadline is still running as pid {pid}"
);
}
#[test]
fn test12584_a_busy_daemon_is_not_taken_for_gone() {
let (listener, port) = listening();
let busy = rendezvous(std::process::id(), port);
assert!(still_serving(&busy), "a live process holding its port is serving");
let client = Client::from_rendezvous(&busy).expect("a client");
let refused = answer_of(&client, &busy, Duration::from_secs(1))
.expect_err("a serving daemon that does not answer is refused as busy, not replaced");
assert!(refused.contains("busy, not gone"), "{refused}");
let dead = rendezvous(dead_pid(), port);
assert!(!still_serving(&dead), "a dead process is not serving");
drop(listener);
let freed = rendezvous(std::process::id(), port);
assert!(!still_serving(&freed), "nothing holds the port");
let client = Client::from_rendezvous(&freed).expect("a client");
assert!(
answer_of(&client, &freed, Duration::from_secs(60)).expect("gone is an answer").is_none(),
"a daemon that is gone is reported gone, at once"
);
}
#[test]
fn test9580_a_daemon_still_on_its_port_has_not_stopped() {
let dir = tempfile::tempdir().expect("a scratch directory");
let (_listener, port) = listening();
assert!(
!is_stopped(dir.path(), &info(4242, port)).expect("the state is readable"),
"a daemon that still holds its port has not stopped, whatever the \
rendezvous file says"
);
}
#[test]
fn test9581_a_daemon_that_still_publishes_itself_has_not_stopped() {
let dir = tempfile::tempdir().expect("a scratch directory");
let (listener, port) = listening();
rendezvous(4242, port)
.write(dir.path())
.expect("the rendezvous file is written");
assert!(
!is_stopped(dir.path(), &info(4242, port)).expect("the state is readable"),
"a daemon that still publishes itself has not stopped"
);
drop(listener);
MachineRendezvous::remove_if_published_by(dir.path(), 4242).expect("the rendezvous file is removed");
assert!(
is_stopped(dir.path(), &info(4242, port)).expect("the state is readable"),
"once the port is free and nothing publishes this daemon, it is gone"
);
}
#[test]
fn test9582_another_daemons_rendezvous_is_not_this_one() {
let dir = tempfile::tempdir().expect("a scratch directory");
let (listener, port) = listening();
let replacement = listener.local_addr().expect("the bound address").port() + 1;
rendezvous(9999, replacement)
.write(dir.path())
.expect("the rendezvous file is written");
drop(listener);
assert!(
is_stopped(dir.path(), &info(4242, port)).expect("the state is readable"),
"the daemon that was stopped is gone; the file names its successor"
);
}
}