Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611//! Daemon socket lifecycle: endpoint naming and identity live in//! [`crate::platform`]; the bind/connect calls go through the shared transport//! traits in `pa_types::platform`, so Unix socket files today and named pipes later//! differ only in the implementation module.
use std::path::Path;use std::time::Duration;
#[cfg(unix)]use anyhow::anyhow;use anyhow::Result;
#[cfg(unix)]pub use crate::platform::socket_dir;pub use crate::platform::{ default_daemon_socket_path, socket_identity, worker_socket_path, SocketIdentity,};
/// Try to connect to an endpoint within `timeout`; true when a peer accepts.pub async fn can_connect(path: &Path, timeout: Duration) -> bool { let connect = pa_types::platform::transport::connect_transport(path); match tokio::time::timeout(timeout, connect).await { Ok(Ok(stream)) => { drop(stream); true } _ => false, }}
/// Staleness after which the cleanup lock of a crashed holder is reclaimed./// Unix only: every taker of the cleanup lock sits behind the unix stale-file wall.#[cfg(unix)]const LOCK_STALE_AFTER: Duration = Duration::from_secs(5);/// Live-lock retry cadence and cap (600 retries): ~15s total.#[cfg(unix)]const LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(25);#[cfg(unix)]const LOCK_RETRIES: u32 = 600;
/// Acquire the cross-process cleanup lock (proper-lockfile's empty/// `{path}.lock` directory): held for the whole probe/unlink sequence, so a/// competing startup worker cannot bind a live listener in between.#[cfg(unix)]async fn acquire_cleanup_lock(path: &Path) -> Result<pa_core::platform::LockDir> { for attempt in 0..=LOCK_RETRIES { match pa_core::platform::LockDir::acquire(path, LOCK_STALE_AFTER) { Ok(lock) => return Ok(lock), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} Err(error) => return Err(anyhow!("Daemon socket cleanup lock: {error}")), } if attempt == LOCK_RETRIES { break; } tokio::time::sleep(LOCK_RETRY_INTERVAL).await; } Err(anyhow!( "Timed out waiting for the daemon socket cleanup lock: {}", path.display() ))}
/// Remove a stale socket file after verifying nothing is listening./// Unix only: a stale socket file blocks `bind`; named pipes have no filesystem/// residue, so preparing the path is a no-op there.////// # Errors////// Returns an error when the parent directory cannot be created, a live listener/// answers, the cleanup lock cannot be acquired, or the cleanup fails.#[cfg(unix)]pub async fn prepare_socket_path(path: &Path) -> Result<()> { if let Some(parent) = path.parent() { crate::paths::ensure_dir(parent)?; } // lstat, not `Path::exists()`: a dangling symlink still blocks `bind` // while `exists()` (which follows links) denies it. match std::fs::symlink_metadata(path) { Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), Err(error) => return Err(anyhow!("Daemon socket path stat failed: {error}")), Ok(_) => {} } // Quick refusal before taking the cross-process lock: a second daemon // fails fast instead of queueing. if can_connect(path, Duration::from_millis(250)).await { return Err(anyhow!("Daemon socket already in use: {}", path.display())); } let _lock = acquire_cleanup_lock(path).await?; prepare_locked_socket_path(path).await}
/// Probe + grace wait + unlink for a probed-stale socket file; the caller owns the cleanup lock.#[cfg(unix)]async fn prepare_locked_socket_path(path: &Path) -> Result<()> { use std::os::unix::fs::FileTypeExt; let Ok(metadata) = std::fs::symlink_metadata(path) else { return Ok(()); }; if !metadata.file_type().is_socket() { return Err(anyhow!( "Daemon socket path exists and is not a socket: {}", path.display() )); } let stale_identity = SocketIdentity { dev: std::os::unix::fs::MetadataExt::dev(&metadata), ino: std::os::unix::fs::MetadataExt::ino(&metadata), }; if can_connect(path, Duration::from_millis(250)).await { return Err(anyhow!("Daemon socket already in use: {}", path.display())); } let deadline = tokio::time::Instant::now() + Duration::from_secs(1); while tokio::time::Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(25)).await; if !path.exists() { return Ok(()); } match socket_identity(path) { None => return Ok(()), Some(current) if current == stale_identity => {} Some(_) => { return Err(anyhow!( "Daemon socket changed ownership while waiting for cleanup: {}", path.display() )) } } if can_connect(path, Duration::from_millis(250)).await { return Err(anyhow!("Daemon socket already in use: {}", path.display())); } } unlink_stale_socket(path, stale_identity).await}
/// Final gate before unlinking a probed-stale socket: refuse while a live/// listener answers, and remove only the exact inode that was probed stale —/// a file replaced between the probe and the unlink stays untouched. The/// identity gate covers processes that do not take the cleanup lock.#[cfg(unix)]async fn unlink_stale_socket(path: &Path, expected: SocketIdentity) -> Result<()> { if can_connect(path, Duration::from_millis(250)).await { return Err(anyhow!("Daemon socket already in use: {}", path.display())); } match socket_identity(path) { None => Ok(()), Some(current) if current == expected => { std::fs::remove_file(path)?; Ok(()) } Some(_) => Err(anyhow!( "Daemon socket changed ownership while waiting for cleanup: {}", path.display() )), }}
/// Windows arm of [`prepare_socket_path`]: named-pipe endpoints have no/// filesystem residue (the first listener creates the pipe), so preparing/// the path is a no-op.////// # Errors////// Does not error: there is no path to prepare for a named pipe.// The signature stays async for the shared unix callers (the await is// the unix arm's own; the pipe arm has no path to prepare).#[cfg(not(unix))]#[cfg_attr(not(unix), allow(clippy::unused_async))]pub async fn prepare_socket_path(_path: &Path) -> Result<()> { Ok(())}
/// Remove the socket file when it still belongs to this supervisor/// incarnation (no-op for named pipes). The remove runs best-effort under/// the cleanup lock: contention means another daemon owns the socket path.pub fn cleanup_socket_path(path: &Path, expected_identity: Option<SocketIdentity>) { if !path.exists() { return; } #[cfg(unix)] let Ok(_cleanup_lock) = pa_core::platform::LockDir::acquire(path, LOCK_STALE_AFTER) else { return; }; if let Some(expected) = expected_identity { match socket_identity(path) { Some(current) if current == expected => {} _ => return, } } let _ = std::fs::remove_file(path);}
/// The exit cleanup after the owner's own listener is closed (the TS/// graceful-shutdown sequence closes before cleanup). A successor's live/// listener must survive even when a poisoned bind-time identity capture/// names the successor's inode. A nonblocking connect can distinguish a/// definitely closed listener (`ECONNREFUSED`) from a saturated backlog/// (`EAGAIN` on Linux); unknown outcomes preserve the socket path. Only/// after definite refusal may the existing cleanup lock and identity gate/// unlink the stale, still-ours socket. TS cleanup checks identity alone,/// so the poisoned-capture case remains a disclosed TS difference.////// A caller without a captured identity never unlinks: `None` skips/// `cleanup_socket_path`'s inode gate, so a replacement that binds the/// path between this probe and that remove would lose its live file to/// an identity-less unlink. The worker's exit paths wait for the serve/// handshake's confirmation - which always follows the identity capture -/// so a registration-refusal exit inside the bind->capture window still/// reads its own captured identity; `None` reaches here only from exits/// before the bind (no listener, no file) or from platforms without a/// file identity (named pipes), and the cleanup stays a no-op.#[cfg(unix)]pub fn cleanup_socket_path_after_close(path: &Path, expected_identity: Option<SocketIdentity>) { if expected_identity.is_none() || !path.exists() || !pa_types::platform::transport::unix_listener_definitely_closed(path) { return; } cleanup_socket_path(path, expected_identity);}
#[cfg(not(unix))]pub fn cleanup_socket_path_after_close(_path: &Path, _expected_identity: Option<SocketIdentity>) {}
/// Restrict the bound socket file to its owner (Unix mode 0o600; Windows/// named pipes use ACLs on the pipe object instead).pub fn restrict_socket_path(path: &Path) { let _ = pa_core::platform::perms::restrict_file(path);}
/// The bind-capture gap seam (the `PA_DAEMON_EVENT_LOG` seam family): a/// replacement landing between the bind and the bind-time identity capture/// poisons the captured identity - the exact residual the/// close-listener exit cleanup exists to survive. Production leaves the/// gap unset, so the bind and the capture stay back-to-back; the/// poisoned-capture oracle sets the gap so the replacement provably lands/// in the window instead of racing microseconds.pub const BIND_CAPTURE_GAP_ENV: &str = "PA_DAEMON_BIND_CAPTURE_GAP_MS";
/// Sleep the bounded bind-capture fault-injection gap in debug builds only./// Production binaries never pause startup between bind and identity capture.pub async fn bind_capture_gap() { #[cfg(debug_assertions)] if let Ok(raw) = std::env::var(BIND_CAPTURE_GAP_ENV) { if let Ok(ms @ 1..=2_000) = raw.parse::<u64>() { tokio::time::sleep(Duration::from_millis(ms)).await; } }}
#[cfg(all(test, unix))]mod tests { use super::*; use pa_types::platform::transport::bind_transport;
/// Bind and drop the listener: the socket file outlives the fd with /// nobody listening - exactly a crashed worker's residue. async fn bind_stale_socket(path: &Path) { drop(bind_transport(path).await.expect("bind stale socket")); }
#[tokio::test] async fn missing_path_prepares_as_a_no_op() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); prepare_socket_path(&socket).await.unwrap(); assert!(!socket.exists()); }
#[tokio::test] async fn a_dangling_symlink_is_rejected_as_not_a_socket() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); std::os::unix::fs::symlink(dir.path().join("missing.sock"), &socket).unwrap(); let error = prepare_socket_path(&socket).await.unwrap_err(); assert!(error.to_string().contains("not a socket"), "{error}"); assert!(std::fs::symlink_metadata(&socket).is_ok()); }
#[tokio::test] async fn non_socket_file_at_the_path_is_refused_and_preserved() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); std::fs::write(&socket, b"not a socket").unwrap(); let error = prepare_socket_path(&socket).await.unwrap_err(); assert!(error.to_string().contains("not a socket"), "{error}"); assert!(socket.exists()); }
#[tokio::test] async fn live_listener_is_never_unlinked() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let listener = bind_transport(&socket).await.unwrap(); let error = prepare_socket_path(&socket).await.unwrap_err(); assert!(error.to_string().contains("already in use"), "{error}"); assert!(socket.exists()); assert!(can_connect(&socket, Duration::from_millis(250)).await); drop(listener); }
/// Dropping the bound listener is the graceful close the exit /// cleanups run before their unlink (the TS `server.close` step, /// daemon-mode.ts:8011-8018): the bind releases (a fresh connect is /// refused) while the socket FILE survives (the close never /// unlinks), an in-flight accepted stream keeps serving across the /// close, and the listener's own fd is closed while the in-flight /// stream's fd stays open - no fd is leaked on the bound socket /// across the exit sequence. /// The fd-leak oracle reads `/proc/self/fd`, so it runs where that /// exists: the neighbors' `target_os = "linux"` gate, not bare `unix` /// (macOS has no `/proc` and the fd probes would always read false). #[cfg(target_os = "linux")] #[tokio::test] async fn dropping_the_listener_is_the_graceful_exit_close() { use std::io::Write; use std::os::fd::AsRawFd; use tokio::io::{AsyncReadExt, AsyncWriteExt};
fn fd_exists(fd: std::os::fd::RawFd) -> bool { std::fs::read_link(format!("/proc/self/fd/{fd}")).is_ok() }
let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let listener = tokio::net::UnixListener::bind(&socket).unwrap(); // One accepted connection in flight: the client connected, the // listener accepted; the stream must survive the listener's close. let mut client = std::os::unix::net::UnixStream::connect(&socket).unwrap(); let mut accepted = listener.accept().await.unwrap().0; let listener_fd = listener.as_raw_fd(); let accepted_fd = accepted.as_raw_fd(); assert!(fd_exists(listener_fd)); // The leak check compares the fd's /proc TARGET (the socket's // anon inode), never the bare fd number: parallel tests recycle // fd numbers the moment a close lands, so a bare-number probe // misreads another test's fresh fd as a leak. The exact socket // object is what a leak would still reference. let listener_target = std::fs::read_link(format!("/proc/self/fd/{listener_fd}")) .expect("the bound listener's fd target before the drop"); drop(listener); let leaked_at_fd = std::fs::read_link(format!("/proc/self/fd/{listener_fd}")); assert!( !matches!(&leaked_at_fd, Ok(target) if *target == listener_target), "the listener's socket object is no longer referenced at the dropped fd: no fd leaked on the bound socket" ); assert!( fd_exists(accepted_fd), "the in-flight accepted stream's fd survives the close" ); assert!(socket.exists(), "the close never unlinks the file"); assert!( !can_connect(&socket, Duration::from_millis(250)).await, "the bind released at the drop" ); // The in-flight stream still moves bytes across the close. client.write_all(b"ping").unwrap(); let mut buffer = [0u8; 4]; accepted.read_exact(&mut buffer).await.unwrap(); assert_eq!(&buffer, b"ping"); assert!(accepted.flush().await.is_ok()); }
/// The exit cleanup after the owner's listener closed spares a LIVE /// successor even when the expected identity matches that /// successor's file exactly - the poisoned capture a replacement /// landing in the bind->capture window produces - and still unlinks /// the dead file the matching identity describes (the still-ours /// direction: a respawn does not wait out the stale-socket path). #[tokio::test] async fn exit_cleanup_after_close_spares_a_live_successor_with_a_matching_identity() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); // The owner's own bind, closed exactly like the exit sequences // close it before their cleanup. let owner = bind_transport(&socket).await.unwrap(); drop(owner); // The poisoned capture: the successor binds the path after the // owner's file is renamed aside, and the "captured" identity is // the successor's own file (a replacement landing in the // bind->capture window stores exactly this). let aside = dir.path().join("owner.sock"); std::fs::rename(&socket, &aside).unwrap(); let successor = bind_transport(&socket).await.unwrap(); let poisoned = socket_identity(&socket).unwrap(); cleanup_socket_path_after_close(&socket, Some(poisoned.clone())); assert!( socket.exists(), "a live successor is never unlinked, even with a matching identity" ); assert!(can_connect(&socket, Duration::from_millis(250)).await); drop(successor); std::fs::remove_file(&aside).unwrap(); }
/// The still-ours direction of the close cleanup: the matching /// identity now describes a dead file the probe passed as definitely /// closed. Linux-only because `unix_listener_definitely_closed` only /// rules `ECONNREFUSED` definitive there (a saturated BSD/macOS /// backlog also refuses, so those platforms never reach this arm). #[cfg(target_os = "linux")] #[tokio::test] async fn exit_cleanup_after_close_unlinks_the_dead_still_ours_file() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let owner = bind_transport(&socket).await.unwrap(); let identity = socket_identity(&socket).unwrap(); drop(owner); cleanup_socket_path_after_close(&socket, Some(identity)); assert!(!socket.exists(), "the dead still-ours file is unlinked"); }
/// Off Linux the refused probe is ambiguous (a saturated backlog /// also refuses), so the close cleanup preserves the dead still-ours /// file; the next bind's stale-socket prepare clears it instead. #[cfg(not(target_os = "linux"))] #[tokio::test] async fn exit_cleanup_off_linux_preserves_the_dead_file_until_the_next_bind() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let owner = bind_transport(&socket).await.unwrap(); let identity = socket_identity(&socket).unwrap(); drop(owner); cleanup_socket_path_after_close(&socket, Some(identity)); assert!( socket.exists(), "off Linux the close cleanup never claims a dead file from the probe alone" ); prepare_socket_path(&socket).await.unwrap(); assert!( !socket.exists(), "the next bind's stale-socket prepare clears the preserved dead file" ); }
/// A full accept queue is not proof of a dead listener: the successor's /// inode can exactly match a poisoned bind-time identity capture. #[cfg(target_os = "linux")] #[tokio::test] async fn exit_cleanup_preserves_a_backlogged_successor_with_a_matching_identity() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let owner = bind_transport(&socket).await.unwrap(); drop(owner); let aside = dir.path().join("owner.sock"); std::fs::rename(&socket, &aside).unwrap(); let successor = std::os::unix::net::UnixListener::bind(&socket).unwrap(); socket2::SockRef::from(&successor).listen(1).unwrap(); let poisoned = socket_identity(&socket).unwrap(); let mut queued = Vec::new(); // Never accept: hold each successful connection until the queue fills. for _ in 0..4 { match tokio::time::timeout( Duration::from_millis(100), tokio::net::UnixStream::connect(&socket), ) .await { Ok(Ok(stream)) => queued.push(stream), _ => break, } } assert!(!queued.is_empty()); assert!( !can_connect(&socket, Duration::from_millis(100)).await, "the successor's queue must be saturated for this oracle" ); cleanup_socket_path_after_close(&socket, Some(poisoned.clone())); assert!( socket.exists(), "a live backlogged successor must not be unlinked" ); drop(successor); cleanup_socket_path_after_close(&socket, Some(poisoned)); assert!( !socket.exists(), "the same inode unlinks after its listener closes" ); drop(queued); std::fs::remove_file(&aside).unwrap(); }
#[cfg(target_os = "linux")] #[tokio::test] async fn exit_cleanup_unlinks_a_closed_listener_on_a_long_socket_path() { let dir = tempfile::TempDir::new().unwrap(); let deep = dir.path().join("a".repeat(80)).join("b".repeat(80)); std::fs::create_dir_all(&deep).unwrap(); let socket = deep.join("daemon.sock"); let owner = bind_transport(&socket).await.unwrap(); let identity = socket_identity(&socket).unwrap(); drop(owner); cleanup_socket_path_after_close(&socket, Some(identity)); assert!(!socket.exists(), "deep stale path still unlinks"); }
#[tokio::test] async fn stale_socket_file_is_removed_and_the_path_rebinds() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); bind_stale_socket(&socket).await; assert!(socket.exists()); prepare_socket_path(&socket).await.unwrap(); assert!(!socket.exists(), "stale socket file must be unlinked"); bind_transport(&socket) .await .expect("bind after stale cleanup"); }
#[tokio::test] async fn unlink_refuses_a_live_listener_even_when_marked_stale() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let listener = bind_transport(&socket).await.unwrap(); let stale = socket_identity(&socket).unwrap(); let error = unlink_stale_socket(&socket, stale).await.unwrap_err(); assert!(error.to_string().contains("already in use"), "{error}"); assert!(socket.exists()); drop(listener); }
#[tokio::test] async fn unlink_refuses_a_replaced_file_with_a_new_identity() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); bind_stale_socket(&socket).await; let stale = socket_identity(&socket).unwrap(); // Move the probed file aside instead of unlinking it: its inode stays // allocated, so the replacement is guaranteed a different inode (a freed // inode could be handed straight back to the replacement). let aside = dir.path().join("probed.sock"); std::fs::rename(&socket, &aside).unwrap(); bind_stale_socket(&socket).await; assert_ne!(socket_identity(&socket).unwrap(), stale); let error = unlink_stale_socket(&socket, stale).await.unwrap_err(); assert!(error.to_string().contains("changed ownership"), "{error}"); assert!(socket.exists(), "the replacement socket file must survive"); std::fs::remove_file(&aside).unwrap(); }
#[tokio::test] async fn unlink_is_a_no_op_when_the_file_is_already_gone() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); unlink_stale_socket(&socket, SocketIdentity { dev: 0, ino: 0 }) .await .unwrap(); }
#[tokio::test] async fn cleanup_waits_while_a_rival_startup_holds_the_lock() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); bind_stale_socket(&socket).await; // A rival startup worker owns the cleanup lock: a fresh empty // `{path}.lock` directory, exactly what LockDir::acquire sees as live. let rival_lock = dir.path().join("daemon.sock.lock"); std::fs::create_dir(&rival_lock).unwrap(); let socket_arg = socket.clone(); let mut pending = tokio::spawn(async move { prepare_socket_path(&socket_arg).await }); assert!( tokio::time::timeout(Duration::from_millis(150), &mut pending) .await .is_err() ); // The rival releases: the queued cleanup proceeds and frees the lock. std::fs::remove_dir(&rival_lock).unwrap(); pending.await.unwrap().unwrap(); assert!(!socket.exists(), "stale socket file must be unlinked"); assert!( !rival_lock.exists(), "the cleanup lock must be released after use" ); }
#[test] fn cleanup_is_deferred_while_a_rival_holds_the_lock() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap(); let identity = socket_identity(&socket).unwrap(); let rival_lock = dir.path().join("daemon.sock.lock"); std::fs::create_dir(&rival_lock).unwrap(); cleanup_socket_path(&socket, Some(identity.clone())); assert!(socket.exists()); std::fs::remove_dir(&rival_lock).unwrap(); cleanup_socket_path(&socket, Some(identity)); assert!(!socket.exists()); drop(listener); }
#[tokio::test] async fn a_stale_cleanup_lock_of_a_crashed_holder_is_reclaimed() { let dir = tempfile::TempDir::new().unwrap(); let socket = dir.path().join("daemon.sock"); bind_stale_socket(&socket).await; let crashed_lock = dir.path().join("daemon.sock.lock"); std::fs::create_dir(&crashed_lock).unwrap(); let six_seconds_ago = filetime::FileTime::from_system_time(std::time::SystemTime::now() - LOCK_STALE_AFTER); filetime::set_file_mtime(&crashed_lock, six_seconds_ago).unwrap(); // A lock whose holder crashed (mtime past LOCK_STALE_AFTER) is reclaimed instead of // waiting out the full retry budget. tokio::time::timeout(Duration::from_secs(2), prepare_socket_path(&socket)) .await .unwrap() .unwrap(); assert!(!socket.exists(), "stale socket file must be unlinked"); }}