kunobi_daemon/local/

unix_socket.rs

//! Unix listener acquisition and guarded endpoint cleanup.
use crate::ProcessLock;
use std::{
    io,
    os::unix::{
        fs::{FileTypeExt, MetadataExt, PermissionsExt},
        net::UnixListener,
    },
    path::{Path, PathBuf},
    time::Duration,
};

pub use super::Bound;
/// Failed acquisition, preserving filesystem errors separately from contention.
#[derive(Debug)]
#[non_exhaustive]
pub enum BindError {
    /// Failed to prepare the private parent directory.
    Directory(io::Error),
    /// Failed to bind, acquire the serialization lock, or probe the existing socket.
    Bind(io::Error),
    /// Failed to remove a proven stale socket.
    Unlink(io::Error),
}
/// Longest socket path, in bytes, that fits in a Unix socket address on this
/// platform: `sockaddr_un.sun_path` less the NUL that terminates it.
pub const MAX_PATH_BYTES: usize = match crate::socket_path::MAX_SOCKET_PATH_BYTES {
    Some(limit) => limit,
    None => unreachable!(),
};

/// Fails when `socket` is too long to bind or connect to as a Unix socket.
///
/// The OS reports this only as an invalid argument, naming neither the length
/// nor the limit, and a deep application directory hits it without warning.
/// Prefer [`crate::socket_path::SocketDir`], which checks a directory against
/// every socket it will hold before any of them is created.
pub fn check_path(socket: &Path) -> io::Result<()> {
    crate::socket_path::check_socket_path(socket)
}
fn lock_path(path: &Path) -> PathBuf {
    let mut path = path.as_os_str().to_owned();
    path.push(".bind.lock");
    path.into()
}
/// Prepare an application-owned directory before creating its socket.
pub use crate::ensure_private_dir;
fn bind_private(socket: &Path) -> io::Result<UnixListener> {
    // The 0700 parent excludes other users during bind; never alter global umask.
    let listener = crate::spawn_lock::without_spawns(|| UnixListener::bind(socket))?;
    std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o600))?;
    Ok(listener)
}
/// Serialize stale-socket recovery. Only ConnectionRefused permits reclaiming
/// an existing socket; permission failures and unrelated files are preserved.
///
/// # Spawning while binding
///
/// Linux creates the socket close-on-exec in one step. macOS takes two, and a
/// child spawned by another thread between them inherits the listener. It
/// keeps accepting for the child's whole life, after the returned listener is
/// dropped or this process exits, so a successor gets `AlreadyRunning`.
///
/// Every spawn this crate makes waits while the socket is created:
/// [`super::unix::spawn_with_child_waiting`], the `launch` spawns built on it,
/// and [`crate::warmup`]. Other spawns can still capture it. A process that
/// spawns another way while it may be binding should have each child close,
/// or mark close-on-exec, every descriptor above stderr before it execs.
pub fn acquire(socket: &Path) -> Result<Bound<UnixListener>, BindError> {
    // Before creating the directory or taking the lock: a path that cannot be
    // bound should leave nothing behind.
    check_path(socket).map_err(BindError::Bind)?;
    ensure_private_dir(
        socket
            .parent()
            .ok_or_else(|| BindError::Directory(io::ErrorKind::InvalidInput.into()))?,
    )
    .map_err(BindError::Directory)?;
    let Some(_lock) = ProcessLock::try_acquire(lock_path(socket)).map_err(BindError::Bind)? else {
        return Ok(Bound::AlreadyRunning);
    };
    match bind_private(socket) {
        Ok(listener) => return Ok(Bound::Won(listener)),
        Err(error) if error.kind() == io::ErrorKind::AddrInUse => {}
        Err(error) => return Err(BindError::Bind(error)),
    }
    let metadata = std::fs::symlink_metadata(socket).map_err(BindError::Bind)?;
    if !metadata.file_type().is_socket() {
        return Err(BindError::Bind(io::ErrorKind::AlreadyExists.into()));
    }
    let probe = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None)
        .map_err(BindError::Bind)?;
    let address = socket2::SockAddr::unix(socket).map_err(BindError::Bind)?;
    match probe.connect_timeout(&address, Duration::from_millis(250)) {
        Ok(()) => return Ok(Bound::AlreadyRunning),
        Err(error)
            if matches!(
                error.kind(),
                io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
            ) =>
        {
            return Ok(Bound::AlreadyRunning);
        }
        Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => {}
        Err(error) => return Err(BindError::Bind(error)),
    }
    std::fs::remove_file(socket).map_err(BindError::Unlink)?;
    bind_private(socket)
        .map(Bound::Won)
        .map_err(BindError::Bind)
}

/// Remove only the socket inode this handle observed, under the bind lock.
/// An exiting generation cannot unlink an endpoint created by its successor.
pub struct SocketGuard {
    path: PathBuf,
    device: u64,
    inode: u64,
}
impl SocketGuard {
    /// Capture immediately after binding, before publishing discovery.
    pub fn capture(path: &Path) -> io::Result<Self> {
        let metadata = std::fs::symlink_metadata(path)?;
        if !metadata.file_type().is_socket() {
            return Err(io::ErrorKind::InvalidInput.into());
        }
        Ok(Self {
            path: path.to_owned(),
            device: metadata.dev(),
            inode: metadata.ino(),
        })
    }
}
impl Drop for SocketGuard {
    fn drop(&mut self) {
        let Ok(Some(_lock)) = ProcessLock::try_acquire(lock_path(&self.path)) else {
            return;
        };
        if std::fs::symlink_metadata(&self.path).is_ok_and(|metadata| {
            metadata.file_type().is_socket()
                && metadata.dev() == self.device
                && metadata.ino() == self.inode
        }) {
            let _ = std::fs::remove_file(&self.path);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    fn temp_dir(name: &str) -> std::path::PathBuf {
        let mut p = std::env::temp_dir();
        p.push(format!("daemon-single-{}-{name}", std::process::id()));
        let _ = std::fs::remove_dir_all(&p);
        p
    }

    #[test]
    fn a_path_fits_up_to_the_limit_and_not_one_byte_more() {
        let fits = format!("/{}", "a".repeat(MAX_PATH_BYTES - 1));
        assert_eq!(fits.len(), MAX_PATH_BYTES);
        check_path(Path::new(&fits)).unwrap();

        let error = check_path(Path::new(&format!("{fits}b"))).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
        let message = error.to_string();
        assert!(
            message.contains(&format!(
                "{} bytes, but a Unix socket path can be at most {MAX_PATH_BYTES}",
                MAX_PATH_BYTES + 1
            )),
            "{message}"
        );
    }

    #[test]
    fn the_limit_matches_this_platform_socket_address() {
        let expected = if cfg!(any(target_os = "linux", target_os = "android")) {
            107
        } else {
            103
        };
        assert_eq!(MAX_PATH_BYTES, expected);
    }

    #[test]
    fn the_limit_is_the_one_the_os_enforces() {
        let dir = temp_dir("limit");
        std::fs::create_dir_all(&dir).unwrap();
        let prefix = dir.join("s").as_os_str().len();
        // Pad the name so the whole path lands exactly on the limit, then one past it.
        let at_limit = dir.join(format!("s{}", "a".repeat(MAX_PATH_BYTES - prefix)));
        assert_eq!(at_limit.as_os_str().len(), MAX_PATH_BYTES);
        UnixListener::bind(&at_limit).expect("the limit itself binds");
        let past = dir.join(format!("s{}", "a".repeat(MAX_PATH_BYTES - prefix + 1)));
        assert!(
            UnixListener::bind(&past).is_err(),
            "one byte past the limit must not bind"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn an_overlong_path_is_refused_before_anything_is_created() {
        let dir = temp_dir("overlong");
        let sock = dir.join("a".repeat(MAX_PATH_BYTES));
        match acquire(&sock) {
            Err(BindError::Bind(error)) => {
                assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
                assert!(error.to_string().contains("at most"), "{error}");
            }
            other => panic!("expected a bind error, got {other:?}"),
        }
        assert!(
            !dir.exists(),
            "no directory may be created for a path that cannot bind"
        );
    }

    #[test]
    fn the_first_binder_wins() {
        let dir = temp_dir("first");
        let sock = dir.join("b.sock");
        match acquire(&sock).unwrap() {
            Bound::Won(_l) => {}
            other => panic!("expected to win, got {other:?}"),
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_second_binder_stands_down_while_the_first_is_listening() {
        let dir = temp_dir("second");
        let sock = dir.join("b.sock");
        let _first = match acquire(&sock).unwrap() {
            Bound::Won(l) => l,
            other => panic!("first should win, got {other:?}"),
        };

        match acquire(&sock).unwrap() {
            Bound::AlreadyRunning => {}
            other => panic!("second should stand down, got {other:?}"),
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_stale_socket_left_by_a_dead_broker_is_reclaimed() {
        // Without this the system is permanently brokerless after any crash:
        // the path exists so `bind` fails, but nothing is listening so no client
        // can ever connect.
        let dir = temp_dir("stale");
        std::fs::create_dir_all(&dir).unwrap();
        let sock = dir.join("b.sock");

        // Leave a socket inode that nothing listens on, as a killed process
        // does. Bind without listen: a child spawned concurrently by another
        // test inherits every descriptor this process holds until it execs or
        // exits, and a listening socket it held would accept the probe below,
        // so the test failed intermittently with AlreadyRunning. A socket that
        // never listened refuses a connection whoever holds it.
        {
            let never_listened =
                socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None).unwrap();
            never_listened
                .bind(&socket2::SockAddr::unix(&sock).unwrap())
                .unwrap();
        }
        assert!(sock.exists(), "fixture should leave the inode behind");

        match acquire(&sock).unwrap() {
            Bound::Won(_l) => {}
            other => panic!("a stale socket should be reclaimed, got {other:?}"),
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_live_broker_is_never_unlinked() {
        // The dangerous mistake: unlinking on EADDRINUSE alone. That races a
        // healthy broker and silently detaches every client attached to it.
        let dir = temp_dir("nounlink");
        let sock = dir.join("b.sock");
        let _first = match acquire(&sock).unwrap() {
            Bound::Won(l) => l,
            other => panic!("{other:?}"),
        };

        let before = std::fs::metadata(&sock).unwrap();
        let _ = acquire(&sock).unwrap();
        let after = std::fs::metadata(&sock).unwrap();

        use std::os::unix::fs::MetadataExt;
        assert_eq!(
            before.ino(),
            after.ino(),
            "the live broker's socket was replaced"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_regular_file_at_the_socket_path_is_preserved() {
        // An unrelated regular file must survive a failed acquisition.
        let dir = temp_dir("regular");
        std::fs::create_dir_all(&dir).unwrap();
        let sock = dir.join("b.sock");
        let mut f = std::fs::File::create(&sock).unwrap();
        f.write_all(b"not a socket").unwrap();
        drop(f);

        assert!(acquire(&sock).is_err());
        assert_eq!(std::fs::read(&sock).unwrap(), b"not a socket");

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn the_socket_is_owner_only() {
        let dir = temp_dir("perms");
        let sock = dir.join("b.sock");
        let _l = match acquire(&sock).unwrap() {
            Bound::Won(l) => l,
            other => panic!("{other:?}"),
        };

        let mode = std::fs::metadata(&sock).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "socket is 0o{mode:o}, expected 0o600");

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn the_parent_directory_is_owner_only_even_if_it_already_existed() {
        // `create_dir_all` is a no-op on an existing directory, so a directory
        // created by an older build - or by a user - would keep its permissions
        // and never be checked.
        let dir = temp_dir("dirperms");
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap();

        let sock = dir.join("b.sock");
        let _l = acquire(&sock).unwrap();

        let mode = std::fs::metadata(&dir).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o700, "directory left at 0o{mode:o}");

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn many_concurrent_acquirers_produce_exactly_one_winner() {
        // The thundering herd: N relays start at once against no broker. Exactly
        // one must win; every other must stand down cleanly rather than error.
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let dir = temp_dir("herd");
        std::fs::create_dir_all(&dir).unwrap();
        let sock = Arc::new(dir.join("b.sock"));

        let won = Arc::new(AtomicUsize::new(0));
        let stood_down = Arc::new(AtomicUsize::new(0));
        let failed = Arc::new(AtomicUsize::new(0));

        let acquired = Arc::new(std::sync::Barrier::new(15));
        let handles: Vec<_> = (0..15)
            .map(|_| {
                let sock = Arc::clone(&sock);
                let won = Arc::clone(&won);
                let stood_down = Arc::clone(&stood_down);
                let failed = Arc::clone(&failed);
                let acquired = Arc::clone(&acquired);
                std::thread::spawn(move || {
                    let outcome = acquire(&sock);
                    match &outcome {
                        Ok(Bound::Won(_)) => {
                            won.fetch_add(1, Ordering::SeqCst);
                        }
                        Ok(Bound::AlreadyRunning) => {
                            stood_down.fetch_add(1, Ordering::SeqCst);
                        }
                        Err(_) => {
                            failed.fetch_add(1, Ordering::SeqCst);
                        }
                    }
                    // Keep the winning listener alive until every contender
                    // has inspected it, regardless of how it is scheduled.
                    acquired.wait();
                    drop(outcome);
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(failed.load(Ordering::SeqCst), 0, "some acquirers errored");
        assert_eq!(
            won.load(Ordering::SeqCst),
            1,
            "expected exactly one winner, got {} (stood down: {})",
            won.load(Ordering::SeqCst),
            stood_down.load(Ordering::SeqCst)
        );
        assert_eq!(stood_down.load(Ordering::SeqCst), 14);

        let _ = std::fs::remove_dir_all(&dir);
    }
}