use camino::{Utf8Path, Utf8PathBuf};
use std::{
fmt, io,
process::ExitStatus,
str::FromStr,
thread,
time::{Duration, Instant},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[must_use]
pub enum Readiness<T> {
Ready(T),
Pending,
}
#[derive(Debug)]
#[non_exhaustive]
pub enum PollError {
Malformed {
path: Utf8PathBuf,
contents: Vec<u8>,
source: Box<dyn std::error::Error + Send + Sync>,
},
Io {
path: Utf8PathBuf,
source: io::Error,
},
ProcessExited {
path: Utf8PathBuf,
status: ExitStatus,
},
ChildTryWait {
path: Utf8PathBuf,
source: io::Error,
},
}
impl PollError {
pub fn path(&self) -> &Utf8Path {
match self {
PollError::Malformed { path, .. }
| PollError::Io { path, .. }
| PollError::ProcessExited { path, .. }
| PollError::ChildTryWait { path, .. } => path,
}
}
}
impl fmt::Display for PollError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
PollError::Malformed { path, contents, .. } => {
write!(
f,
"malformed port file {path}: {:?}",
String::from_utf8_lossy(contents)
)
}
PollError::Io { path, .. } => {
write!(f, "failed to read port file {path}")
}
PollError::ProcessExited { path, status } => {
write!(
f,
"process exited before writing port file {path} \
(status: {status})"
)
}
PollError::ChildTryWait { path, .. } => {
write!(
f,
"polling whether process was alive for port file {path}"
)
}
}
}
}
impl std::error::Error for PollError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
PollError::Malformed { source, .. } => Some(source.as_ref()),
PollError::Io { source, .. } => Some(source),
PollError::ProcessExited { .. } => None,
PollError::ChildTryWait { source, .. } => Some(source),
}
}
}
#[derive(Debug)]
struct MissingTrailingNewline;
impl fmt::Display for MissingTrailingNewline {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("file does not end with a newline")
}
}
impl std::error::Error for MissingTrailingNewline {}
pub fn poll_once<T>(
path: &Utf8Path,
exited: io::Result<Option<ExitStatus>>,
read: io::Result<Vec<u8>>,
) -> Result<Readiness<T>, PollError>
where
T: FromStr,
T::Err: std::error::Error + Send + Sync + 'static,
{
match read {
Ok(bytes) => {
let contents = match String::from_utf8(bytes) {
Ok(contents) => contents,
Err(error) => {
let source = error.utf8_error();
return Err(PollError::Malformed {
path: path.to_owned(),
contents: error.into_bytes(),
source: Box::new(source),
});
}
};
let Some(payload) = contents.strip_suffix('\n') else {
return Err(PollError::Malformed {
path: path.to_owned(),
contents: contents.into_bytes(),
source: Box::new(MissingTrailingNewline),
});
};
match payload.parse::<T>() {
Ok(value) => Ok(Readiness::Ready(value)),
Err(source) => Err(PollError::Malformed {
path: path.to_owned(),
contents: contents.into_bytes(),
source: Box::new(source),
}),
}
}
Err(e) if e.kind() == io::ErrorKind::NotFound => match exited {
Ok(Some(status)) => {
Err(PollError::ProcessExited { path: path.to_owned(), status })
}
Ok(None) => Ok(Readiness::Pending),
Err(e) if e.kind() == io::ErrorKind::Interrupted => {
Ok(Readiness::Pending)
}
Err(source) => {
Err(PollError::ChildTryWait { path: path.to_owned(), source })
}
},
Err(e) if e.kind() == io::ErrorKind::Interrupted => {
Ok(Readiness::Pending)
}
Err(e) if is_sharing_violation(&e) => Ok(Readiness::Pending),
Err(source) => Err(PollError::Io { path: path.to_owned(), source }),
}
}
#[cfg(windows)]
fn is_sharing_violation(error: &io::Error) -> bool {
error.raw_os_error() == Some(32)
}
#[cfg(not(windows))]
fn is_sharing_violation(_error: &io::Error) -> bool {
false
}
fn parent_dir_to_check(path: &Utf8Path) -> Option<&Utf8Path> {
let parent = path.parent()?;
if parent.as_str().is_empty() { None } else { Some(parent) }
}
#[derive(Debug)]
enum ParentDirProblem {
Missing(Utf8PathBuf),
NotADirectory(Utf8PathBuf),
}
impl ParentDirProblem {
fn into_error(self, path: &Utf8Path, elapsed: Duration) -> WaitForError {
match self {
ParentDirProblem::Missing(parent) => {
WaitForError::MissingParentDirectory {
path: path.to_owned(),
parent,
elapsed,
}
}
ParentDirProblem::NotADirectory(parent) => {
WaitForError::ParentNotADirectory {
path: path.to_owned(),
parent,
elapsed,
}
}
}
}
}
fn classify_parent_dir(
parent: &Utf8Path,
metadata: io::Result<std::fs::Metadata>,
) -> Option<ParentDirProblem> {
match metadata {
Ok(metadata) if metadata.is_dir() => None,
Ok(_) => Some(ParentDirProblem::NotADirectory(parent.to_owned())),
Err(error) if error.kind() == io::ErrorKind::NotFound => {
Some(ParentDirProblem::Missing(parent.to_owned()))
}
Err(_) => None,
}
}
fn check_parent_dir(path: &Utf8Path) -> Option<ParentDirProblem> {
let parent = parent_dir_to_check(path)?;
classify_parent_dir(parent, std::fs::metadata(parent))
}
#[cfg(feature = "tokio")]
async fn check_parent_dir_async(path: &Utf8Path) -> Option<ParentDirProblem> {
let parent = parent_dir_to_check(path)?;
classify_parent_dir(parent, tokio::fs::metadata(parent).await)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PollInterval(pub Duration);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Timeout(pub Duration);
#[derive(Debug)]
#[non_exhaustive]
pub enum WaitForError {
Poll {
error: PollError,
elapsed: Duration,
},
MissingParentDirectory {
path: Utf8PathBuf,
parent: Utf8PathBuf,
elapsed: Duration,
},
ParentNotADirectory {
path: Utf8PathBuf,
parent: Utf8PathBuf,
elapsed: Duration,
},
TimedOut {
path: Utf8PathBuf,
elapsed: Duration,
},
}
impl WaitForError {
pub fn path(&self) -> &Utf8Path {
match self {
WaitForError::Poll { error, .. } => error.path(),
WaitForError::MissingParentDirectory { path, .. }
| WaitForError::ParentNotADirectory { path, .. }
| WaitForError::TimedOut { path, .. } => path,
}
}
pub fn elapsed(&self) -> Duration {
match self {
WaitForError::Poll { elapsed, .. }
| WaitForError::MissingParentDirectory { elapsed, .. }
| WaitForError::ParentNotADirectory { elapsed, .. }
| WaitForError::TimedOut { elapsed, .. } => *elapsed,
}
}
}
impl fmt::Display for WaitForError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
WaitForError::Poll { error, elapsed } => {
write!(f, "{error} (after {elapsed:?})")
}
WaitForError::MissingParentDirectory { path, parent, elapsed } => {
write!(
f,
"parent directory {parent} does not exist, so port file \
{path} can never appear (after {elapsed:?})"
)
}
WaitForError::ParentNotADirectory { path, parent, elapsed } => {
write!(
f,
"parent path {parent} is not a directory, so port file \
{path} can never appear (after {elapsed:?})"
)
}
WaitForError::TimedOut { path, elapsed } => {
write!(
f,
"timed out after {elapsed:?} waiting for port file {path}"
)
}
}
}
}
impl std::error::Error for WaitForError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
WaitForError::Poll { error, .. } => {
std::error::Error::source(error)
}
WaitForError::MissingParentDirectory { .. }
| WaitForError::ParentNotADirectory { .. }
| WaitForError::TimedOut { .. } => None,
}
}
}
pub fn wait_for_blocking<T>(
path: &Utf8Path,
mut child_try_wait: impl FnMut() -> io::Result<Option<ExitStatus>>,
poll_interval: PollInterval,
timeout: Timeout,
) -> Result<T, WaitForError>
where
T: FromStr,
T::Err: std::error::Error + Send + Sync + 'static,
{
let start = Instant::now();
loop {
let exited = child_try_wait();
let read = std::fs::read(path);
match poll_once::<T>(path, exited, read) {
Ok(Readiness::Ready(value)) => return Ok(value),
Ok(Readiness::Pending) => {
if let Some(problem) = check_parent_dir(path) {
return Err(problem.into_error(path, start.elapsed()));
}
}
Err(error) => {
return Err(WaitForError::Poll {
error,
elapsed: start.elapsed(),
});
}
}
let elapsed = start.elapsed();
if elapsed >= timeout.0 {
return Err(WaitForError::TimedOut {
path: path.to_owned(),
elapsed,
});
}
thread::sleep(poll_interval.0.min(timeout.0 - elapsed));
}
}
#[cfg(feature = "tokio")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "tokio")))]
pub async fn wait_for<T>(
path: &Utf8Path,
mut child_try_wait: impl FnMut() -> io::Result<Option<ExitStatus>>,
poll_interval: PollInterval,
timeout: Timeout,
) -> Result<T, WaitForError>
where
T: FromStr,
T::Err: std::error::Error + Send + Sync + 'static,
{
let start = tokio::time::Instant::now();
loop {
let exited = child_try_wait();
let read = tokio::fs::read(path).await;
match poll_once::<T>(path, exited, read) {
Ok(Readiness::Ready(value)) => return Ok(value),
Ok(Readiness::Pending) => {
if let Some(problem) = check_parent_dir_async(path).await {
return Err(problem.into_error(path, start.elapsed()));
}
}
Err(error) => {
return Err(WaitForError::Poll {
error,
elapsed: start.elapsed(),
});
}
}
let elapsed = start.elapsed();
if elapsed >= timeout.0 {
return Err(WaitForError::TimedOut {
path: path.to_owned(),
elapsed,
});
}
tokio::time::sleep(poll_interval.0.min(timeout.0 - elapsed)).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::SocketAddr;
fn missing() -> io::Result<Vec<u8>> {
Err(io::Error::from(io::ErrorKind::NotFound))
}
fn bytes(contents: &str) -> io::Result<Vec<u8>> {
Ok(contents.as_bytes().to_vec())
}
#[cfg(unix)]
fn exited_status() -> ExitStatus {
use std::os::unix::process::ExitStatusExt;
ExitStatus::from_raw(1 << 8)
}
#[cfg(windows)]
fn exited_status() -> ExitStatus {
use std::os::windows::process::ExitStatusExt;
ExitStatus::from_raw(1)
}
#[test]
fn parses_bound_socket_addr() {
let path = Utf8Path::new("port");
let addr: SocketAddr = "[::1]:4676".parse().unwrap();
assert_eq!(
poll_once::<SocketAddr>(path, Ok(None), bytes("[::1]:4676\n"))
.unwrap(),
Readiness::Ready(addr)
);
}
#[test]
fn empty_file_is_malformed() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(path, Ok(None), Ok(Vec::new()))
.unwrap_err();
let PollError::Malformed { path: p, contents, source } = err else {
panic!("expected Malformed, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(contents, b"");
assert!(
source.downcast_ref::<MissingTrailingNewline>().is_some(),
"empty file is missing its trailing newline: {source}"
);
}
#[test]
fn whitespace_only_file_is_malformed() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(path, Ok(None), bytes(" \n"))
.unwrap_err();
let PollError::Malformed { path: p, contents, .. } = err else {
panic!("expected Malformed, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(contents, b" \n");
}
#[test]
fn non_utf8_file_is_malformed() {
let path = Utf8Path::new("port");
let err =
poll_once::<SocketAddr>(path, Ok(None), Ok(vec![0xff, b'\n']))
.unwrap_err();
let PollError::Malformed { path: p, contents, source } = err else {
panic!("expected Malformed, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(contents, b"\xff\n");
assert!(
source.downcast_ref::<std::str::Utf8Error>().is_some(),
"source should be a Utf8Error: {source}"
);
}
#[test]
fn missing_trailing_newline_is_malformed() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(path, Ok(None), bytes("[::1]:4676"))
.unwrap_err();
let PollError::Malformed { path: p, contents, .. } = err else {
panic!("expected Malformed, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(contents, b"[::1]:4676");
}
#[test]
fn only_trailing_newline_is_stripped() {
let path = Utf8Path::new("port");
assert_eq!(
poll_once::<String>(path, Ok(None), bytes(" spaced \n")).unwrap(),
Readiness::Ready(" spaced ".to_owned())
);
}
#[test]
fn missing_file_is_pending_while_alive() {
let path = Utf8Path::new("port");
assert_eq!(
poll_once::<SocketAddr>(path, Ok(None), missing()).unwrap(),
Readiness::Pending
);
}
#[test]
fn missing_file_after_exit_is_permanent() {
let path = Utf8Path::new("port");
let err =
poll_once::<SocketAddr>(path, Ok(Some(exited_status())), missing())
.unwrap_err();
let PollError::ProcessExited { path: p, .. } = err else {
panic!("expected ProcessExited, got {err:?}");
};
assert_eq!(p, path);
}
#[test]
fn malformed_file_is_permanent() {
let path = Utf8Path::new("port");
let err =
poll_once::<SocketAddr>(path, Ok(None), bytes("not-an-addr\n"))
.unwrap_err();
let PollError::Malformed { path: p, contents, .. } = err else {
panic!("expected Malformed, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(contents, b"not-an-addr\n");
}
#[test]
fn malformed_file_wins_over_exit() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(
path,
Ok(Some(exited_status())),
bytes("garbage\n"),
)
.unwrap_err();
let PollError::Malformed { .. } = err else {
panic!("unexpected error: {err:?}");
};
}
#[test]
fn valid_file_wins_over_exit() {
let path = Utf8Path::new("port");
let addr: SocketAddr = "[::1]:4676".parse().unwrap();
assert_eq!(
poll_once::<SocketAddr>(
path,
Ok(Some(exited_status())),
bytes("[::1]:4676\n")
)
.unwrap(),
Readiness::Ready(addr)
);
}
#[test]
fn io_error_is_permanent() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(
path,
Ok(None),
Err(io::Error::from(io::ErrorKind::PermissionDenied)),
)
.unwrap_err();
let PollError::Io { path: p, source } = err else {
panic!("expected Io, got {err:?}");
};
assert_eq!(p, path);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
#[test]
fn child_try_wait_error_is_permanent() {
let path = Utf8Path::new("port");
let err = poll_once::<SocketAddr>(
path,
Err(io::Error::from(io::ErrorKind::Other)),
missing(),
)
.unwrap_err();
let PollError::ChildTryWait { path: p, .. } = err else {
panic!("expected ChildTryWait, got {err:?}");
};
assert_eq!(p, path);
}
#[test]
fn interrupted_read_is_pending() {
let path = Utf8Path::new("port");
assert_eq!(
poll_once::<SocketAddr>(
path,
Ok(None),
Err(io::Error::from(io::ErrorKind::Interrupted)),
)
.unwrap(),
Readiness::Pending
);
}
#[test]
fn interrupted_child_try_wait_is_pending() {
let path = Utf8Path::new("port");
assert_eq!(
poll_once::<SocketAddr>(
path,
Err(io::Error::from(io::ErrorKind::Interrupted)),
missing(),
)
.unwrap(),
Readiness::Pending
);
}
#[test]
#[cfg(windows)]
fn sharing_violation_is_pending() {
let path = Utf8Path::new("port");
assert_eq!(
poll_once::<SocketAddr>(
path,
Ok(None),
Err(io::Error::from_raw_os_error(32)),
)
.unwrap(),
Readiness::Pending
);
}
#[test]
fn wait_for_error_poll_is_transparent() {
let inner = PollError::Io {
path: Utf8PathBuf::from("port"),
source: io::Error::from(io::ErrorKind::PermissionDenied),
};
let inner_message = inner.to_string();
let wrapped =
WaitForError::Poll { error: inner, elapsed: Duration::ZERO };
assert_eq!(wrapped.to_string(), format!("{inner_message} (after 0ns)"));
let source = std::error::Error::source(&wrapped)
.expect("Poll delegates to the PollError's source");
assert!(
source.downcast_ref::<io::Error>().is_some(),
"source is the underlying io::Error, not the PollError: {source}"
);
}
#[test]
fn parent_dir_to_check_skips_bare_and_root_paths() {
assert_eq!(
parent_dir_to_check(Utf8Path::new("port")),
None,
"a bare file name has no directory to check"
);
assert_eq!(
parent_dir_to_check(Utf8Path::new("/")),
None,
"the root has no parent"
);
assert_eq!(
parent_dir_to_check(Utf8Path::new("dir/port")),
Some(Utf8Path::new("dir"))
);
assert_eq!(
parent_dir_to_check(Utf8Path::new("/port")),
Some(Utf8Path::new("/"))
);
}
#[test]
fn check_parent_dir_reports_missing_directory() {
let dir = camino_tempfile::tempdir().unwrap();
let present = dir.path().join("port");
assert!(
check_parent_dir(&present).is_none(),
"an existing parent directory is not an error"
);
let missing = dir.path().join("missing_dir").join("port");
let problem = check_parent_dir(&missing)
.expect("a missing parent directory is a permanent problem");
let ParentDirProblem::Missing(parent) = &problem else {
panic!("expected Missing, got {problem:?}");
};
assert_eq!(parent, &dir.path().join("missing_dir"));
let error = problem.into_error(&missing, Duration::from_secs(1));
let WaitForError::MissingParentDirectory { path, parent, elapsed } =
&error
else {
panic!("expected MissingParentDirectory, got {error:?}");
};
assert_eq!(path, &missing);
assert_eq!(parent, &dir.path().join("missing_dir"));
assert_eq!(*elapsed, Duration::from_secs(1));
}
#[test]
fn check_parent_dir_reports_file_in_parent_position() {
let dir = camino_tempfile::tempdir().unwrap();
let occupied = dir.path().join("occupied");
std::fs::write(&occupied, "not a directory").unwrap();
let path = occupied.join("port");
let problem = check_parent_dir(&path)
.expect("a file in the parent position is a permanent problem");
let ParentDirProblem::NotADirectory(parent) = &problem else {
panic!("expected NotADirectory, got {problem:?}");
};
assert_eq!(parent, &occupied);
let error = problem.into_error(&path, Duration::from_secs(1));
let WaitForError::ParentNotADirectory { path: p, parent, elapsed } =
&error
else {
panic!("expected ParentNotADirectory, got {error:?}");
};
assert_eq!(p, &path);
assert_eq!(parent, &occupied);
assert_eq!(*elapsed, Duration::from_secs(1));
}
}