kunobi-daemon 0.4.2

//! 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)]
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 = if cfg!(any(target_os = "linux", target_os = "android")) {
    107
} else {
    103
};

/// 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.
/// Callers can check before creating anything, and add their own advice.
pub fn check_path(socket: &Path) -> io::Result<()> {
    use std::os::unix::ffi::OsStrExt;
    let bytes = socket.as_os_str().as_bytes().len();
    if bytes > MAX_PATH_BYTES {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!(
                "socket path is {bytes} bytes, but a Unix socket path can be at most \
                 {MAX_PATH_BYTES} on this platform: {}",
                socket.display()
            ),
        ));
    }
    Ok(())
}
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 fn ensure_private_dir(dir: &Path) -> io::Result<()> {
    std::fs::create_dir_all(dir)?;
    let metadata = std::fs::symlink_metadata(dir)?;
    if !metadata.is_dir() || metadata.uid() != super::unix::own_uid() {
        return Err(io::ErrorKind::PermissionDenied.into());
    }
    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))
}
fn bind_private(socket: &Path) -> io::Result<UnixListener> {
    // The 0700 parent excludes other users during bind; never alter global umask.
    let listener = 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.
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");
        // Under the lock: a directory created while a sibling test holds a
        // narrowed umask comes out 0o600 and is untraversable.
        std::fs::create_dir_all(&dir).unwrap();
        let sock = dir.join("b.sock");

        // Bind then drop the listener WITHOUT unlinking - exactly what a killed
        // process leaves behind.
        {
            let l = UnixListener::bind(&sock).unwrap();
            drop(l);
        }
        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");
        // Under the lock: a directory created while a sibling test holds a
        // narrowed umask comes out 0o600 and is untraversable.
        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");
        // Under the lock: a directory created while a sibling test holds a
        // narrowed umask comes out 0o600 and is untraversable.
        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");
        // Under the lock: a directory created while a sibling test holds a
        // narrowed umask comes out 0o600 and is untraversable.
        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 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);
                std::thread::spawn(move || {
                    match acquire(&sock) {
                        Ok(Bound::Won(l)) => {
                            won.fetch_add(1, Ordering::SeqCst);
                            // Hold it, so the losers meet a LIVE socket.
                            std::thread::sleep(std::time::Duration::from_millis(200));
                            drop(l);
                        }
                        Ok(Bound::AlreadyRunning) => {
                            stood_down.fetch_add(1, Ordering::SeqCst);
                        }
                        Err(_) => {
                            failed.fetch_add(1, Ordering::SeqCst);
                        }
                    }
                })
            })
            .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);
    }
}