use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::FileTypeExt;
use std::path::{Path, PathBuf};
use thiserror::Error;
use tracing::warn;
#[cfg(target_os = "macos")]
const UNIX_PATH_MAX: usize = 104;
#[cfg(not(target_os = "macos"))]
const UNIX_PATH_MAX: usize = 108;
pub(crate) const MAX_CONTROL_REQUEST_BYTES: usize = 32;
pub(crate) const CONTROL_CLIENT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
const MAX_CONTROL_RESPONSE_BYTES: usize = 16;
pub const STOP_COMMAND: &[u8] = b"STOP\n";
pub const OK_RESPONSE: &[u8] = b"OK\n";
#[derive(Debug, Error)]
pub enum ControlError {
#[error("control socket path {0:?} exceeds UNIX_PATH_MAX ({UNIX_PATH_MAX} bytes)")]
PathTooLong(PathBuf),
#[error("control socket filesystem error: {0}")]
Io(#[from] std::io::Error),
#[error("control socket returned unexpected acknowledgement")]
BadResponse,
}
#[derive(Debug, Error)]
pub enum ControlSetupError {
#[error(
"could not bind a restrictive greggd control socket; \
primary {primary:?}, fallback {fallback:?}. \
Run greggd from a directory the daemon user can own, or fix \
permissions on the temp directory."
)]
NoSecureControl {
primary: Option<PathBuf>,
fallback: Option<PathBuf>,
},
#[error("failed to register signal handler: {0}")]
SignalRegistration(#[from] std::io::Error),
}
#[must_use]
pub fn config_id_for_path(config_path: &Path) -> String {
const FNV_OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
let bytes = control_identity_path(config_path);
let bytes = bytes.as_os_str().as_bytes();
let mut hash = FNV_OFFSET_BASIS;
for &byte in bytes {
hash ^= u64::from(byte);
hash = hash.wrapping_mul(FNV_PRIME);
}
format!("{hash:016x}")
}
fn control_identity_path(path: &Path) -> PathBuf {
if let Ok(canonical) = std::fs::canonicalize(path) {
return canonical;
}
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()
.unwrap_or_else(|_| PathBuf::from("/"))
.join(path)
};
let mut normalized = PathBuf::new();
for component in absolute.components() {
match component {
std::path::Component::CurDir => {}
std::path::Component::ParentDir => {
let _ = normalized.pop();
}
_ => normalized.push(component.as_os_str()),
}
}
normalized
}
#[must_use]
pub fn primary_control_path(config_path: &Path) -> Option<PathBuf> {
let identity_path = control_identity_path(config_path);
let parent = identity_path.parent()?;
let path = parent.join(format!(
"greggd-{}.control.sock",
config_id_for_path(config_path)
));
if path.as_os_str().len() > UNIX_PATH_MAX {
return None;
}
Some(path)
}
#[must_use]
pub fn fallback_control_path(config_path: &Path) -> Option<PathBuf> {
let path = std::env::temp_dir().join(format!(
"greggd-{}.control.sock",
config_id_for_path(config_path)
));
if path.as_os_str().len() > UNIX_PATH_MAX {
return None;
}
Some(path)
}
#[must_use]
pub fn stop_candidates(config_path: &Path) -> Vec<PathBuf> {
let mut out = Vec::with_capacity(2);
if let Some(primary) = primary_control_path(config_path) {
out.push(primary);
}
if let Some(fallback) = fallback_control_path(config_path) {
if !out.iter().any(|p| p == &fallback) {
out.push(fallback);
}
}
out
}
pub fn remove_control_socket(path: &Path) -> std::io::Result<()> {
match std::fs::metadata(path) {
Ok(meta) if meta.file_type().is_socket() => match std::fs::remove_file(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
},
Ok(_) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
}
}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum StopErrorSeverity {
NotRunning,
Other,
PermissionDenied,
}
fn stop_error_severity(error: &std::io::Error) -> StopErrorSeverity {
match error.kind() {
std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused => {
StopErrorSeverity::NotRunning
}
std::io::ErrorKind::PermissionDenied => StopErrorSeverity::PermissionDenied,
_ => StopErrorSeverity::Other,
}
}
fn record_stop_error(slot: &mut Option<std::io::Error>, error: std::io::Error) {
if slot
.as_ref()
.is_none_or(|previous| stop_error_severity(&error) > stop_error_severity(previous))
{
*slot = Some(error);
}
}
pub fn send_stop(config_path: &Path) -> Result<StopOutcome, ControlError> {
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use std::time::Duration;
const IO_TIMEOUT: Duration = Duration::from_millis(750);
let candidates = stop_candidates(config_path);
let mut last_io_error: Option<std::io::Error> = None;
for candidate in &candidates {
let mut stream = match UnixStream::connect(candidate) {
Ok(stream) => stream,
Err(e) => {
record_stop_error(&mut last_io_error, e);
continue;
}
};
let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
if let Err(e) = stream.write_all(STOP_COMMAND) {
record_stop_error(&mut last_io_error, e);
continue;
}
let mut buf = [0_u8; MAX_CONTROL_RESPONSE_BYTES];
let mut length = 0;
let mut response: Option<Vec<u8>> = None;
while length < buf.len() && response.is_none() {
match stream.read(&mut buf[length..]) {
Ok(0) => break,
Ok(read) => {
length += read;
if let Some(end) = buf[..length].iter().position(|b| *b == b'\n') {
response = Some(buf[..=end].to_vec());
}
}
Err(e) => {
record_stop_error(&mut last_io_error, e);
break;
}
}
}
if response.is_none() && length == buf.len() {
return Err(ControlError::BadResponse);
}
if let Some(bytes) = response {
if bytes.as_slice() == OK_RESPONSE {
return Ok(StopOutcome::Stopped {
path: candidate.clone(),
});
}
return Err(ControlError::BadResponse);
}
record_stop_error(
&mut last_io_error,
std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"control socket accepted STOP but sent no complete response",
),
);
}
match last_io_error {
Some(e)
if e.kind() == std::io::ErrorKind::NotFound
|| e.kind() == std::io::ErrorKind::ConnectionRefused =>
{
Ok(StopOutcome::NotRunning)
}
Some(e) if e.kind() == std::io::ErrorKind::PermissionDenied => Err(ControlError::Io(e)),
Some(e) => {
tracing::warn!(error = ?e, "control socket stop attempt failed unexpectedly");
let detail = match e.kind() {
std::io::ErrorKind::TimedOut => {
format!("timed out waiting for daemon response: {e}")
}
std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::UnexpectedEof
| std::io::ErrorKind::BrokenPipe => {
format!("connection reset/closed mid-response (daemon may have crashed): {e}")
}
_ => format!("unexpected control socket error: {e}"),
};
Ok(StopOutcome::Uncertain { detail })
}
None => Ok(StopOutcome::NotRunning),
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum StopOutcome {
Stopped {
path: PathBuf,
},
NotRunning,
Uncertain {
detail: String,
},
}
#[derive(Debug)]
pub enum ControlBind {
Bound {
path: PathBuf,
listener: tokio::net::UnixListener,
},
NotBound,
}
pub fn bind_listener(config_path: &Path) -> ControlBind {
use tracing::info;
if let Some(primary) = primary_control_path(config_path) {
if let Some(bound) = try_bind_secure(&primary) {
info!(path = %primary.display(), "control socket bound");
return bound;
}
}
if let Some(fallback) = fallback_control_path(config_path) {
if Some(&fallback) != primary_control_path(config_path).as_ref() {
if let Some(bound) = try_bind_secure(&fallback) {
info!(path = %fallback.display(), "control socket bound (fallback)");
return bound;
}
}
}
warn!("control socket not bound; daemon will only respond to signals");
ControlBind::NotBound
}
#[must_use]
pub fn stale_connect_error(kind: std::io::ErrorKind) -> bool {
matches!(
kind,
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
)
}
fn try_bind_secure(path: &Path) -> Option<ControlBind> {
use std::os::unix::fs::PermissionsExt;
prepare_final_path(path)?;
if let Some(parent) = path.parent() {
if let Err(e) = std::fs::create_dir_all(parent) {
warn!(
parent = %parent.display(),
error = %e,
"control socket parent directory creation failed"
);
return None;
}
}
let listener = match tokio::net::UnixListener::bind(path) {
Ok(listener) => listener,
Err(e) => {
warn!(
path = %path.display(),
error = %e,
"control socket bind failed"
);
return None;
}
};
if let Err(e) = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) {
warn!(
path = %path.display(),
error = %e,
"control socket permission update failed; closing listener"
);
drop(listener);
let _ = remove_control_socket(path);
return None;
}
match std::fs::metadata(path) {
Ok(meta) => {
let mode = meta.permissions().mode() & 0o777;
if mode != 0o600 {
warn!(
path = %path.display(),
mode = format!("{mode:o}"),
"control socket permissions are not 0600 after chmod; closing listener"
);
drop(listener);
let _ = remove_control_socket(path);
return None;
}
}
Err(e) => {
warn!(
path = %path.display(),
error = %e,
"control socket metadata check failed; closing listener"
);
drop(listener);
let _ = remove_control_socket(path);
return None;
}
}
Some(ControlBind::Bound {
path: path.to_path_buf(),
listener,
})
}
fn prepare_final_path(path: &Path) -> Option<()> {
match std::fs::metadata(path) {
Ok(meta) => {
let ft = meta.file_type();
if !ft.is_socket() {
warn!(
path = %path.display(),
"control socket path exists but is not a socket; skipping"
);
return None;
}
match std::os::unix::net::UnixStream::connect(path) {
Ok(_) => {
None
}
Err(e) if stale_connect_error(e.kind()) => {
let _ = std::fs::remove_file(path);
Some(())
}
Err(e) => {
warn!(
path = %path.display(),
error = %e,
"control socket connect failed with non-stale classification; \
leaving existing entry in place"
);
None
}
}
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(()),
Err(e) => {
warn!(
path = %path.display(),
error = %e,
"control socket metadata failed"
);
None
}
}
}
pub fn spawn_stop_task(
listener: tokio::net::UnixListener,
path: PathBuf,
notify: tokio::sync::oneshot::Sender<std::io::Result<&'static str>>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let guard = ControlSocketGuard { path: path.clone() };
let result = stop_loop(listener).await;
let _ = notify.send(result);
drop(guard);
})
}
struct ControlSocketGuard {
path: PathBuf,
}
impl Drop for ControlSocketGuard {
fn drop(&mut self) {
if let Err(e) = crate::control::remove_control_socket(&self.path) {
tracing::warn!(
path = %self.path.display(),
error = %e,
"control socket guard cleanup failed"
);
}
}
}
async fn stop_loop(listener: tokio::net::UnixListener) -> std::io::Result<&'static str> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut accept_backoff = std::time::Duration::from_millis(20);
loop {
let (mut stream, _) = match listener.accept().await {
Ok(pair) => {
accept_backoff = std::time::Duration::from_millis(20);
pair
}
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::Interrupted
| std::io::ErrorKind::WouldBlock
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::TimedOut
) =>
{
warn!(%error, "transient control listener error; retrying");
tokio::time::sleep(accept_backoff).await;
accept_backoff = accept_backoff
.saturating_mul(2)
.min(std::time::Duration::from_secs(1));
continue;
}
Err(error) => {
warn!(%error, "control listener became unavailable; disabling local stop control");
return Err(error);
}
};
let received = tokio::time::timeout(CONTROL_CLIENT_TIMEOUT, async {
let mut buf = [0_u8; MAX_CONTROL_REQUEST_BYTES];
let mut length = 0;
let mut received: Option<Vec<u8>> = None;
while length < buf.len() && received.is_none() {
match stream.read(&mut buf[length..]).await {
Ok(0) => break,
Err(error) => {
warn!(error = %error, "control socket client read failed; dropping connection");
break;
}
Ok(read) => {
length += read;
if let Some(end) = buf[..length].iter().position(|b| *b == 10) {
received = Some(buf[..=end].to_vec());
}
}
}
}
received
})
.await
.ok()
.flatten();
if received.as_deref() == Some(STOP_COMMAND) {
let _ = tokio::time::timeout(CONTROL_CLIENT_TIMEOUT, async {
let _ = stream.write_all(OK_RESPONSE).await;
let _ = stream.flush().await;
let _ = stream.shutdown().await;
})
.await;
return Ok("control-stop");
}
let _ = stream.shutdown().await;
}
}
pub async fn wait_for_stop_task(
receiver: tokio::sync::oneshot::Receiver<std::io::Result<&'static str>>,
) -> Option<&'static str> {
match receiver.await {
Ok(Ok(reason)) => Some(reason),
Ok(Err(e)) => {
tracing::warn!(error = %e, "control stop task ended with I/O error");
None
}
Err(_) => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::os::unix::fs::PermissionsExt;
fn temp_config_path(tag: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"gd{tag}-{}-{}.toml",
std::process::id(),
std::thread::current().name().unwrap_or("main"),
))
}
fn repo_temp_dir(tag: &str) -> PathBuf {
let dir = std::env::current_dir()
.unwrap()
.join("target")
.join(format!("gd{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
#[test]
fn config_id_for_path_is_deterministic_and_hex_encoded() {
let cfg_path = temp_config_path("id-deterministic");
let id = config_id_for_path(&cfg_path);
assert_eq!(id.len(), 16);
assert!(id
.chars()
.all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()));
assert_eq!(id, config_id_for_path(&cfg_path));
}
#[test]
fn config_id_changes_only_when_path_changes() {
let a = temp_config_path("id-a");
let b = temp_config_path("id-b");
assert_ne!(config_id_for_path(&a), config_id_for_path(&b));
}
#[test]
fn existing_file_relative_and_absolute_spellings_share_identity() {
let dir = repo_temp_dir("path-spellings");
let config = make_config_file(&dir, "greggd.toml");
let current_dir = std::env::current_dir().unwrap();
let relative = PathBuf::from(".").join(config.strip_prefix(current_dir).unwrap());
assert!(relative.is_relative());
assert_eq!(
config_id_for_path(&relative),
config_id_for_path(&config),
"relative and absolute spellings of one existing file must converge"
);
assert_eq!(
primary_control_path(&relative),
primary_control_path(&config),
"equivalent spellings must select the same primary socket"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn existing_file_symlink_and_target_share_identity() {
let dir = fresh_temp_dir("symlink-identity");
let target = make_config_file(&dir, "target.toml");
let link_dir = dir.join("links");
std::fs::create_dir_all(&link_dir).unwrap();
let link = link_dir.join("config-link.toml");
std::os::unix::fs::symlink(&target, &link).unwrap();
assert_eq!(
config_id_for_path(&link),
config_id_for_path(&target),
"symlink and target spellings must converge"
);
assert_eq!(primary_control_path(&link), primary_control_path(&target));
assert_eq!(fallback_control_path(&link), fallback_control_path(&target));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn existing_different_files_keep_distinct_identities() {
let dir = fresh_temp_dir("different-files");
let a = make_config_file(&dir, "a.toml");
let b = make_config_file(&dir, "b.toml");
assert_ne!(config_id_for_path(&a), config_id_for_path(&b));
assert_ne!(primary_control_path(&a), primary_control_path(&b));
assert_ne!(fallback_control_path(&a), fallback_control_path(&b));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn config_contents_do_not_change_existing_file_identity() {
let dir = fresh_temp_dir("content-identity");
let config = make_config_file(&dir, "greggd.toml");
let before = config_id_for_path(&config);
std::fs::write(&config, b"host = \"127.0.0.1\"\nport = 11311\n").unwrap();
assert_eq!(before, config_id_for_path(&config));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn absent_absolute_path_has_deterministic_identity_without_creation() {
let dir = std::env::temp_dir().join(format!("gdmissing-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let config = dir.join("greggd.toml");
assert!(!config.exists());
assert_eq!(config_id_for_path(&config), config_id_for_path(&config));
let primary = primary_control_path(&config).unwrap();
let fallback = fallback_control_path(&config).unwrap();
assert!(primary.as_os_str().len() <= UNIX_PATH_MAX);
assert!(fallback.as_os_str().len() <= UNIX_PATH_MAX);
}
#[test]
fn primary_path_derives_from_config_path_in_same_directory() {
let dir = std::env::temp_dir().join(format!("gd-id-cf-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let a = dir.join("a.toml");
let b = dir.join("b.toml");
let pa = primary_control_path(&a);
let pb = primary_control_path(&b);
assert!(
pa.is_some() && pb.is_some(),
"both primary paths must fit in UNIX_PATH_MAX; got {pa:?} / {pb:?}"
);
assert_ne!(
pa, pb,
"different config files in the same directory must produce different control identities"
);
assert_ne!(
fallback_control_path(&a),
fallback_control_path(&b),
"fallback identities must also differ"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn primary_path_is_config_adjacent_and_within_sun_path_limit() {
let cfg_path = temp_config_path("primary-adjacent");
let primary = primary_control_path(&cfg_path).unwrap();
let name = primary.file_name().unwrap().to_str().unwrap();
assert!(name.starts_with("greggd-"));
assert!(name.ends_with(".control.sock"));
assert!(primary.as_os_str().len() <= UNIX_PATH_MAX);
}
#[test]
fn fallback_path_is_deterministic_and_in_temp_dir() {
let cfg_path = temp_config_path("fallback-deterministic");
let first = fallback_control_path(&cfg_path).unwrap();
let second = fallback_control_path(&cfg_path).unwrap();
assert_eq!(first, second);
assert!(first.starts_with(std::env::temp_dir()));
assert!(first.as_os_str().len() <= UNIX_PATH_MAX);
let name = first.file_name().unwrap().to_str().unwrap();
assert!(name.starts_with("greggd-"));
assert!(name.ends_with(".control.sock"));
}
#[test]
fn stop_candidates_returns_primary_then_fallback() {
let parent = std::env::temp_dir().join(format!("gd-cand-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&parent);
std::fs::create_dir_all(&parent).unwrap();
let cfg_path = parent.join("greggd.toml");
let candidates = stop_candidates(&cfg_path);
let primary = primary_control_path(&cfg_path);
let fallback = fallback_control_path(&cfg_path);
assert!(
primary.is_some() && fallback.is_some(),
"both candidates must fit in UNIX_PATH_MAX; primary={primary:?} fallback={fallback:?}"
);
assert_ne!(
primary, fallback,
"primary and fallback must not alias each other"
);
assert_eq!(candidates.len(), 2);
assert_eq!(candidates[0], primary.unwrap());
assert_eq!(candidates[1], fallback.unwrap());
let _ = std::fs::remove_dir_all(&parent);
}
#[test]
fn remove_control_socket_is_a_noop_for_missing_paths() {
let missing =
std::env::temp_dir().join(format!("greggd-no-such-{}.sock", std::process::id()));
remove_control_socket(&missing).unwrap();
}
#[test]
fn remove_control_socket_leaves_regular_files_alone() {
let dir = std::env::temp_dir().join(format!(
"greggd-control-remove-test-{}-{}",
std::process::id(),
std::thread::current().name().unwrap_or("main"),
));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let regular = dir.join("greggd-control-regular-file");
std::fs::write(®ular, b"not a socket").unwrap();
remove_control_socket(®ular).unwrap();
assert!(regular.exists());
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn wire_constants_have_expected_format() {
assert_eq!(STOP_COMMAND, b"STOP\n");
assert_eq!(OK_RESPONSE, b"OK\n");
}
#[test]
fn stale_connect_error_classifies_only_documented_kinds() {
assert!(stale_connect_error(std::io::ErrorKind::ConnectionRefused));
assert!(stale_connect_error(std::io::ErrorKind::NotFound));
assert!(!stale_connect_error(std::io::ErrorKind::PermissionDenied));
assert!(!stale_connect_error(std::io::ErrorKind::TimedOut));
assert!(!stale_connect_error(std::io::ErrorKind::AddrInUse));
assert!(!stale_connect_error(std::io::ErrorKind::Other));
}
fn fresh_temp_dir(name: &str) -> PathBuf {
use std::sync::atomic::{AtomicUsize, Ordering};
static NEXT_TEMP_DIR: AtomicUsize = AtomicUsize::new(0);
let id = NEXT_TEMP_DIR.fetch_add(1, Ordering::Relaxed);
let dir = std::env::temp_dir().join(format!("gd{name}-{}-{id}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
fn make_config_file(dir: &Path, name: &str) -> PathBuf {
let cfg = dir.join(name);
std::fs::write(&cfg, b"").unwrap();
cfg
}
#[test]
fn bind_listener_prefers_config_adjacent_path_when_available() {
let dir = fresh_temp_dir("bind-primary");
let cfg = make_config_file(&dir, "greggd.toml");
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let bound = rt.block_on(async { bind_listener(&cfg) });
match bound {
ControlBind::Bound { path, .. } => {
assert_eq!(path, primary_control_path(&cfg).unwrap());
assert!(path.exists());
assert_eq!(
std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
0o600,
"control socket must have restrictive permissions"
);
}
ControlBind::NotBound => panic!("primary bind should have succeeded"),
}
remove_control_socket(&primary_control_path(&cfg).unwrap()).unwrap();
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn bind_listener_falls_back_when_config_parent_is_not_writable() {
let dir = fresh_temp_dir("bind-fallback");
let cfg = make_config_file(&dir, "greggd.toml");
let primary = primary_control_path(&cfg).unwrap();
std::fs::write(&primary, b"blocker").unwrap();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let bound = rt.block_on(async { bind_listener(&cfg) });
match bound {
ControlBind::Bound { path, .. } => {
assert_eq!(path, fallback_control_path(&cfg).unwrap());
assert!(path.exists());
}
ControlBind::NotBound => panic!("fallback bind should have succeeded"),
}
let _ = std::fs::remove_file(&primary);
remove_control_socket(&fallback_control_path(&cfg).unwrap()).unwrap();
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn bind_listener_skips_live_primary_listener() {
let dir = fresh_temp_dir("bind-live");
let cfg = make_config_file(&dir, "greggd.toml");
let primary = primary_control_path(&cfg).unwrap();
let live = std::os::unix::net::UnixListener::bind(&primary).unwrap();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let bound = rt.block_on(async { bind_listener(&cfg) });
match bound {
ControlBind::Bound { path, .. } => {
assert_eq!(
path,
fallback_control_path(&cfg).unwrap(),
"live primary listener must be left alone"
);
}
ControlBind::NotBound => panic!("fallback should have bound when primary is live"),
}
drop(live);
remove_control_socket(&primary).unwrap();
remove_control_socket(&fallback_control_path(&cfg).unwrap()).unwrap();
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn send_stop_reports_not_running_when_no_socket_is_present() {
let dir = fresh_temp_dir("send-stop-missing");
let cfg = make_config_file(&dir, "greggd.toml");
let outcome = send_stop(&cfg).unwrap();
assert_eq!(outcome, StopOutcome::NotRunning);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn stop_error_bookkeeping_never_downgrades_severity() {
let mut slot = None;
record_stop_error(
&mut slot,
std::io::Error::from(std::io::ErrorKind::PermissionDenied),
);
record_stop_error(
&mut slot,
std::io::Error::from(std::io::ErrorKind::NotFound),
);
assert_eq!(
slot.as_ref().map(std::io::Error::kind),
Some(std::io::ErrorKind::PermissionDenied)
);
let mut slot = None;
record_stop_error(
&mut slot,
std::io::Error::from(std::io::ErrorKind::TimedOut),
);
record_stop_error(
&mut slot,
std::io::Error::from(std::io::ErrorKind::NotFound),
);
assert_eq!(
slot.as_ref().map(std::io::Error::kind),
Some(std::io::ErrorKind::TimedOut)
);
record_stop_error(
&mut slot,
std::io::Error::from(std::io::ErrorKind::PermissionDenied),
);
assert_eq!(
slot.as_ref().map(std::io::Error::kind),
Some(std::io::ErrorKind::PermissionDenied)
);
}
#[test]
fn send_stop_treats_silent_close_as_uncertain_with_diagnostic() {
let dir = fresh_temp_dir("send-stop-silent");
let cfg = make_config_file(&dir, "greggd.toml");
let primary = primary_control_path(&cfg).unwrap();
let listener = std::os::unix::net::UnixListener::bind(&primary).unwrap();
std::thread::spawn(move || {
if let Ok((stream, _)) = listener.accept() {
use std::io::Read;
let mut stream = stream;
let mut buf = [0_u8; 32];
let _ = stream.read(&mut buf);
drop(stream);
}
});
let outcome = send_stop(&cfg);
assert!(
matches!(outcome, Ok(StopOutcome::Uncertain { .. })),
"silent close must be uncertain, got {outcome:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn send_stop_delivers_stop_and_receives_ok_response() {
let dir = fresh_temp_dir("send-stop-ok");
let cfg = make_config_file(&dir, "greggd.toml");
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let outcome = rt.block_on(async {
let bound = bind_listener(&cfg);
let (path, listener) = match bound {
ControlBind::Bound { path, listener } => (path, listener),
ControlBind::NotBound => panic!("expected bound listener"),
};
let (tx, rx) = tokio::sync::oneshot::channel();
let _task = spawn_stop_task(listener, path.clone(), tx);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let cfg_clone = cfg.clone();
let client = tokio::task::spawn_blocking(move || send_stop(&cfg_clone));
let reason = rx.await.expect("control task must signal completion");
let outcome = client.await.expect("client task must complete");
(outcome, reason)
});
let (outcome, reason) = outcome;
assert!(matches!(outcome, Ok(StopOutcome::Stopped { .. })));
assert!(reason.is_ok());
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn timed_out_control_client_does_not_block_later_stop() {
let dir = fresh_temp_dir("control-timeout");
let cfg = make_config_file(&dir, "greggd.toml");
let bound = bind_listener(&cfg);
let (path, listener) = match bound {
ControlBind::Bound { path, listener } => (path, listener),
ControlBind::NotBound => panic!("expected bound listener"),
};
let (tx, rx) = tokio::sync::oneshot::channel();
let _task = spawn_stop_task(listener, path.clone(), tx);
let silent = tokio::net::UnixStream::connect(&path)
.await
.expect("silent client connects");
tokio::time::sleep(CONTROL_CLIENT_TIMEOUT + std::time::Duration::from_millis(50)).await;
drop(silent);
let cfg_for_stop = cfg.clone();
let client = tokio::task::spawn_blocking(move || send_stop(&cfg_for_stop));
let outcome = client.await.expect("stop client task completes");
let reason = rx.await.expect("control task signals later valid stop");
assert!(matches!(outcome, Ok(StopOutcome::Stopped { .. })));
assert_eq!(reason.expect("valid stop succeeds"), "control-stop");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn bind_listener_rebinds_stale_primary_socket_file() {
let dir = fresh_temp_dir("rebind-stale");
let cfg = make_config_file(&dir, "greggd.toml");
let primary = primary_control_path(&cfg).unwrap();
let stale = std::os::unix::net::UnixListener::bind(&primary).unwrap();
drop(stale);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let bound = rt.block_on(async { bind_listener(&cfg) });
match bound {
ControlBind::Bound { path, .. } => {
assert_eq!(path, primary);
}
ControlBind::NotBound => panic!("bind_listener must rebind stale primary"),
}
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn send_stop_rejects_malformed_responses() {
let dir = fresh_temp_dir("send-stop-malformed");
let cfg = make_config_file(&dir, "greggd.toml");
let primary = primary_control_path(&cfg).unwrap();
let listener = std::os::unix::net::UnixListener::bind(&primary).unwrap();
std::thread::spawn(move || {
if let Ok((mut stream, _)) = listener.accept() {
use std::io::{Read, Write};
let mut buf = [0_u8; 32];
let _ = stream.read(&mut buf);
let _ = stream.write_all(b"NOPE\n");
let _ = stream.flush();
}
});
let outcome = send_stop(&cfg);
assert!(matches!(outcome, Err(ControlError::BadResponse)));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn cross_config_stop_isolates_two_daemons_in_same_directory() {
let dir = fresh_temp_dir("cross-stop");
let cfg_a = make_config_file(&dir, "a.toml");
let cfg_b = make_config_file(&dir, "b.toml");
let primary_a = primary_control_path(&cfg_a).unwrap();
let primary_b = primary_control_path(&cfg_b).unwrap();
assert_ne!(primary_a, primary_b);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let (b_stopped, b_reason_ok, a_socket_present, a_final_stop) = rt.block_on(async {
let bound_a = bind_listener(&cfg_a);
let bound_b = bind_listener(&cfg_b);
let (
ControlBind::Bound {
path: path_a,
listener: listener_a,
},
ControlBind::Bound {
path: path_b,
listener: listener_b,
},
) = (bound_a, bound_b)
else {
panic!("both control listeners must bind concurrently")
};
assert_eq!(path_a, primary_a);
assert_eq!(path_b, primary_b);
let (tx_a, mut rx_a) = tokio::sync::oneshot::channel();
let (tx_b, rx_b) = tokio::sync::oneshot::channel();
let _task_a = spawn_stop_task(listener_a, path_a.clone(), tx_a);
let _task_b = spawn_stop_task(listener_b, path_b.clone(), tx_b);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let cfg_b_for_stop_b = cfg_b.clone();
let client_b = tokio::task::spawn_blocking(move || send_stop(&cfg_b_for_stop_b));
let stopped_b = client_b.await.expect("client task must complete");
let reason_b = rx_b.await.expect("B control task must signal completion");
let a_still_alive = matches!(
rx_a.try_recv(),
Err(tokio::sync::oneshot::error::TryRecvError::Empty)
);
let a_socket_present = a_still_alive && primary_a.exists();
let cfg_a_for_stop_a = cfg_a.clone();
let client_a = tokio::task::spawn_blocking(move || send_stop(&cfg_a_for_stop_a));
let stopped_a = client_a.await.expect("client task must complete");
let reason_a = rx_a.await.expect("A control task must signal completion");
(
stopped_b,
reason_b.is_ok(),
a_socket_present,
(stopped_a, reason_a.is_ok()),
)
});
match &b_stopped {
Ok(StopOutcome::Stopped { path }) => assert_eq!(
path, &primary_b,
"send_stop(cfg_b) must resolve on the B primary path"
),
other => panic!("send_stop(cfg_b) must succeed; got {other:?}"),
}
assert!(b_reason_ok, "daemon B control task must complete cleanly");
assert!(
a_socket_present,
"daemon A's primary socket must remain on disk after sending STOP to daemon B"
);
assert!(
matches!(a_final_stop.0, Ok(StopOutcome::Stopped { .. })),
"daemon A must respond to a follow-up send_stop(cfg_a)"
);
assert!(
a_final_stop.1,
"daemon A control task must complete cleanly"
);
remove_control_socket(&primary_a).unwrap();
remove_control_socket(&primary_b).unwrap();
remove_control_socket(&fallback_control_path(&cfg_a).unwrap()).unwrap();
remove_control_socket(&fallback_control_path(&cfg_b).unwrap()).unwrap();
let _ = std::fs::remove_dir_all(&dir);
}
}