orbit_core/

shm.rs

//! POSIX shared-memory helpers — V1 substrate for cross-process rings.
//!
//! Wraps `shm_open` / `ftruncate` / `mmap` / `munmap` / `shm_unlink`
//! into a small, RAII-friendly API. Unix-only; Windows support is
//! a separate concern (Win32 named file mapping) that can land later.
//!
//! ## Naming
//!
//! Segments are named `/orbit-{fleet}-{kind}-{uid}` — fleet name from
//! the embedder, KIND from `OrbitTyped::KIND`, UID from `geteuid()`.
//! UID-scoping avoids the `/dev/shm` sticky-bit cross-user collision
//! problem (a stale segment owned by one user blocks another from
//! `shm_unlink`-ing it on next boot).
//!
//! Rings that require a process-recoverable writer lock also open a
//! companion `orbit-{fleet}-{kind}-{uid}.lock` file. It carries no
//! ring data or state; it only supplies a regular-file inode for `flock`,
//! because advisory locking on a POSIX SHM descriptor is not uniformly
//! supported across the Unix targets Orbit serves. An unlocked stale
//! companion file is safe to reuse.
//!
//! Those files live in a per-uid directory — `$XDG_RUNTIME_DIR/orbit-{uid}`
//! where the session provides one, `/tmp/orbit-{uid}` otherwise — created
//! `0700` and checked on every lock. They were once in `/tmp` directly, which
//! made them squattable: see [`lock_dir`].
//!
//! ## Lifetime
//!
//! [`ShmRegion`] owns the mapped pointer and unmaps on drop. It does
//! NOT `shm_unlink` on drop — the segment lives until an explicit
//! [`ShmRegion::unlink`] call. This matches POSIX convention: a
//! segment with mapped users is not removed; `shm_unlink` only
//! prevents *new* opens, the current mapping stays valid until the
//! last process unmaps.

#![cfg(unix)]

use std::ffi::CString;
use std::fs::OpenOptions;
use std::io;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::ptr::NonNull;
use std::sync::atomic::{AtomicBool, Ordering};

/// Namespace used by Orbit POSIX shared-memory objects.
pub const SHM_NAMESPACE: &str = "orbit";

/// Maximum POSIX access allowed for one SHM object, independent of its layout.
///
/// New objects are owned by the effective uid. On macOS, the selected group
/// must be the creator's effective gid and the requested mode is passed directly
/// to `shm_open` (with the platform's native creation-mask semantics).
/// Other Unix targets create privately, then set the requested group and exact
/// group-sharing mode. `OwnerOnly` uses the existing `shm_open(..., 0600)` path.
/// Orbit never changes `umask` or process credentials, or chmods/chowns an
/// existing object. This does not
/// isolate processes running under the same uid.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum ShmAccessPolicy {
    /// Owner read/write, no group or other access (`0600`).
    #[default]
    OwnerOnly,
    /// Owner read/write and selected group read (`0640`).
    GroupRead { gid: u32 },
    /// Owner and selected group read/write (`0660`). Trust every group writer.
    GroupReadWrite { gid: u32 },
}

impl ShmAccessPolicy {
    pub const fn mode(self) -> u32 {
        match self {
            Self::OwnerOnly => 0o600,
            Self::GroupRead { .. } => 0o640,
            Self::GroupReadWrite { .. } => 0o660,
        }
    }

    pub const fn gid(self) -> Option<u32> {
        match self {
            Self::OwnerOnly => None,
            Self::GroupRead { gid } | Self::GroupReadWrite { gid } => Some(gid),
        }
    }
}

/// Result of physically validating an existing POSIX SHM object.
///
/// This check is deliberately below ring semantics: it verifies that the
/// named object satisfies the access policy and is large enough for the requested mapping,
/// but it does not inspect an owning data structure's magic, version, or
/// geometry header.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ShmValidation {
    /// No object currently exists under the requested name.
    Missing,
    /// The object exists and can safely back at least the requested mapping.
    Valid { actual_size: usize },
}

/// A mapped POSIX SHM region. Drop unmaps; `unlink` removes the
/// underlying name and any companion lock file (only the *creator*
/// should call it on shutdown).
pub struct ShmRegion {
    name: CString,
    lock_path: PathBuf,
    /// Whether this region uses the companion file for ordered writes.
    ///
    /// The descriptor itself is deliberately not retained: every critical
    /// section opens its own file description so a later fork does not inherit
    /// an idle descriptor that can keep a future `flock` alive.
    process_lock: bool,
    ptr: NonNull<u8>,
    len: usize,
    /// True when this handle was the one that *created* the segment
    /// (so it knows to `shm_unlink` if asked). Other attachers see
    /// `false`.
    created: bool,
}

impl ShmRegion {
    /// Validate an existing shared-memory object without creating, mapping,
    /// resetting, or unlinking it.
    ///
    /// Returns [`ShmValidation::Missing`] when the name does not exist. A
    /// present object must be at least `minimum_size` bytes; larger objects
    /// are accepted because some platforms report page-rounded SHM sizes.
    /// The owning ring or table remains responsible for validating its own
    /// persisted ABI header after mapping.
    /// The expected owner is the effective uid and the policy is `OwnerOnly`.
    pub fn validate_existing(
        name: &str,
        minimum_size: usize,
    ) -> io::Result<ShmValidation> {
        Self::validate_existing_with_policy(
            name,
            minimum_size,
            unsafe { libc::geteuid() },
            ShmAccessPolicy::default(),
        )
    }

    /// Validate size, expected owner and maximum permissions without mutation.
    /// `owner_uid` is trusted configuration, not metadata read from the object.
    pub fn validate_existing_with_policy(
        name: &str,
        minimum_size: usize,
        owner_uid: u32,
        policy: ShmAccessPolicy,
    ) -> io::Result<ShmValidation> {
        let cname = CString::new(name).map_err(|_| {
            io::Error::new(io::ErrorKind::InvalidInput, "shm name has nul byte")
        })?;
        let raw_fd = loop {
            // SAFETY: passing a valid C string and well-known POSIX flags.
            // macOS `shm_open` rejects `O_CLOEXEC` with EINVAL. The descriptor
            // is scoped to this validation call and closes before return.
            let fd = unsafe {
                libc::shm_open(cname.as_ptr(), libc::O_RDONLY, 0o600)
            };
            if fd >= 0 {
                break fd;
            }
            let error = io::Error::last_os_error();
            if error.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            if error.raw_os_error() == Some(libc::ENOENT) {
                return Ok(ShmValidation::Missing);
            }
            return Err(error);
        };
        // SAFETY: `raw_fd` was returned by `shm_open` and is now uniquely
        // owned by this scope.
        let fd = unsafe { OwnedFd::from_raw_fd(raw_fd) };
        let actual_size = shm_object_size(&fd, name, owner_uid, policy)?;
        validate_minimum_size(name, actual_size, minimum_size)?;
        Ok(ShmValidation::Valid { actual_size })
    }

    /// Map an existing shared-memory object read-only.
    ///
    /// This path never creates, sizes, locks, resets, or unlinks the object.
    /// It is kept crate-private so callers receive a capability such as a
    /// read-only ring view rather than a [`ShmRegion`] that also exposes
    /// lifecycle and writable-pointer operations.
    pub(crate) fn open_existing_read_only(
        name: &str,
        minimum_size: usize,
        owner_uid: u32,
        policy: ShmAccessPolicy,
    ) -> io::Result<Self> {
        Self::open_existing(name, minimum_size, owner_uid, policy, false)
    }

    /// Attach writable to an existing object, without creating or sizing it.
    /// Semantic readers use this only for existing atomic reader-pin fields.
    pub fn open_existing_read_write(
        name: &str,
        minimum_size: usize,
        owner_uid: u32,
        policy: ShmAccessPolicy,
    ) -> io::Result<Self> {
        Self::open_existing(name, minimum_size, owner_uid, policy, true)
    }

    fn open_existing(
        name: &str,
        minimum_size: usize,
        owner_uid: u32,
        policy: ShmAccessPolicy,
        writable: bool,
    ) -> io::Result<Self> {
        let cname = CString::new(name).map_err(|_| {
            io::Error::new(io::ErrorKind::InvalidInput, "shm name has nul byte")
        })?;
        let raw_fd = loop {
            // SAFETY: passing a valid C string and existing-only POSIX flags.
            // macOS `shm_open` rejects `O_CLOEXEC` with EINVAL. This
            // descriptor closes immediately after `mmap`, before the view is
            // returned, so it cannot leak across a later exec.
            let fd = unsafe {
                libc::shm_open(
                    cname.as_ptr(),
                    if writable { libc::O_RDWR } else { libc::O_RDONLY },
                    0o600,
                )
            };
            if fd >= 0 {
                break fd;
            }
            let error = io::Error::last_os_error();
            if error.kind() == io::ErrorKind::Interrupted {
                continue;
            }
            return Err(error);
        };
        // SAFETY: `raw_fd` was returned by `shm_open` and is now uniquely
        // owned by this scope.
        let fd = unsafe { OwnedFd::from_raw_fd(raw_fd) };
        let actual_size = shm_object_size(&fd, name, owner_uid, policy)?;
        validate_minimum_size(name, actual_size, minimum_size)?;

        // Map the complete object so its persisted header can describe the
        // geometry without the observer reproducing the producer's layout.
        // SAFETY: fd is valid, actual_size is positive after minimum
        // validation. Protection follows the requested attachment capability.
        let ptr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                actual_size,
                libc::PROT_READ | if writable { libc::PROT_WRITE } else { 0 },
                libc::MAP_SHARED,
                fd.as_raw_fd(),
                0,
            )
        };
        if ptr == libc::MAP_FAILED {
            return Err(io::Error::last_os_error());
        }
        // SAFETY: mmap returned a non-null pointer (checked above).
        let ptr = NonNull::new(ptr.cast::<u8>())
            .expect("mmap returned non-null on success");

        Ok(Self {
            lock_path: lock_dir_for_uid(owner_uid)
                .join(format!("{}.lock", name.trim_start_matches('/'))),
            name: cname,
            process_lock: false,
            ptr,
            len: actual_size,
            created: false,
        })
    }

    /// Open or create a shared-memory segment of `size` bytes,
    /// memory-mapped read/write. Idempotent: if the segment already
    /// exists with the same name and enough mapped bytes, it is reused
    /// (`created = false`). First creation does `ftruncate(size)`;
    /// later opens verify the existing object before mapping it. Some
    /// platforms report a page-rounded SHM size, so a larger `st_size`
    /// is valid; the owning data structure must verify its own header.
    /// Existing objects must belong to the effective uid and satisfy `OwnerOnly`.
    pub fn open_or_create(
        name: &str,
        size: usize,
    ) -> io::Result<Self> {
        Self::open_or_create_with_policy(name, size, ShmAccessPolicy::default())
    }

    /// Open or create using an explicit access policy. Existing objects must
    /// belong to the effective uid and pass the policy before being mapped.
    /// Rejection never resizes, chmods, chowns or unlinks an existing object.
    pub fn open_or_create_with_policy(
        name: &str,
        size: usize,
        policy: ShmAccessPolicy,
    ) -> io::Result<Self> {
        let (region, initialization_lock) =
            Self::open_or_create_inner(name, size, false, policy)?;
        debug_assert!(initialization_lock.is_none());
        Ok(region)
    }

    /// Open or create a region while holding its process lock through caller
    /// initialization. This prevents a peer from observing the interval
    /// between `shm_open` and the owning data structure's initialized header.
    pub fn open_or_create_locked(
        name: &str,
        size: usize,
    ) -> io::Result<(Self, ShmRegionLock)> {
        Self::open_or_create_locked_with_policy(
            name,
            size,
            ShmAccessPolicy::default(),
        )
    }

    /// Policy-aware creation with the existing owner-local initialization lock.
    /// Group permissions do not make that lock or fleet membership cross-user.
    pub fn open_or_create_locked_with_policy(
        name: &str,
        size: usize,
        policy: ShmAccessPolicy,
    ) -> io::Result<(Self, ShmRegionLock)> {
        let (region, initialization_lock) =
            Self::open_or_create_inner(name, size, true, policy)?;
        Ok((
            region,
            initialization_lock
                .expect("locked SHM open must return its initialization lock"),
        ))
    }

    fn open_or_create_inner(
        name: &str,
        size: usize,
        process_lock: bool,
        policy: ShmAccessPolicy,
    ) -> io::Result<(Self, Option<ShmRegionLock>)> {
        let cname = CString::new(name).map_err(|_| {
            io::Error::new(io::ErrorKind::InvalidInput, "shm name has nul byte")
        })?;
        let lock_path = lock_file_path(name);
        let initialization_lock = if process_lock {
            Some(lock_path_exclusive(&lock_path)?)
        } else {
            None
        };

        // Darwin cannot fchmod/fchown a POSIX SHM descriptor. Grant only the
        // intended group at creation; attaching a matching existing object is OK.
        #[cfg(target_os = "macos")]
        let can_create =
            policy.gid().is_none_or(|gid| gid == unsafe { libc::getegid() });
        #[cfg(target_os = "macos")]
        let creation_mode = policy.mode();
        // Elsewhere, don't expose the object to its initial group before chown.
        #[cfg(not(target_os = "macos"))]
        let (can_create, creation_mode) = (true, 0o600u32);
        // Apple declares shm_open as variadic, so its mode stays promoted to u32.
        #[cfg(not(target_vendor = "apple"))]
        let creation_mode = creation_mode as libc::mode_t;
        // Try create-exclusive first; if it already exists, open.
        let (raw_fd, created) = unsafe {
            // SAFETY: passing a valid C string and well-known POSIX flags.
            let fd = if can_create {
                libc::shm_open(
                    cname.as_ptr(),
                    libc::O_RDWR | libc::O_CREAT | libc::O_EXCL,
                    creation_mode,
                )
            } else {
                libc::shm_open(cname.as_ptr(), libc::O_RDWR, 0)
            };
            if fd >= 0 {
                (fd, can_create)
            } else {
                // Could be EEXIST (already created by a peer) or another error.
                let err = io::Error::last_os_error();
                if !can_create && err.raw_os_error() == Some(libc::ENOENT) {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "creating group-shared SHM requires the selected effective gid",
                    ));
                }
                if err.raw_os_error() != Some(libc::EEXIST) {
                    return Err(err);
                }
                let fd = libc::shm_open(cname.as_ptr(), libc::O_RDWR, 0o600);
                if fd < 0 {
                    return Err(io::Error::last_os_error());
                }
                (fd, false)
            }
        };
        // SAFETY: `raw_fd` was returned by `shm_open` and is now uniquely
        // owned by this scope.
        let fd = unsafe { OwnedFd::from_raw_fd(raw_fd) };

        #[cfg(not(target_os = "macos"))]
        if created && let Err(error) = configure_new_shm(&fd, policy) {
            let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
            return Err(error);
        }

        // Verify ownership and permissions before ftruncate or mmap. Only a
        // newly created object is ours to unlink on initialization failure.
        let owner_uid = unsafe { libc::geteuid() };
        if let Err(error) = shm_object_size(&fd, name, owner_uid, policy) {
            if created {
                let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
            }
            return Err(error);
        }

        // Size the segment on first creation.
        if created {
            // SAFETY: fd is a valid POSIX fd we just received.
            let rc =
                unsafe { libc::ftruncate(fd.as_raw_fd(), size as libc::off_t) };
            if rc != 0 {
                let err = io::Error::last_os_error();
                let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
                return Err(err);
            }
        }

        // Never mmap beyond the real SHM object: access past it can raise
        // SIGBUS. A larger reported size is valid on platforms (notably
        // macOS) that page-round POSIX SHM objects; callers verify their
        // own ABI metadata after mapping.
        let actual_size = match shm_object_size(&fd, name, owner_uid, policy) {
            Ok(actual_size) => actual_size,
            Err(error) => {
                if created {
                    let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
                }
                return Err(error);
            }
        };
        if let Err(error) = validate_minimum_size(name, actual_size, size) {
            if created {
                let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
            }
            return Err(error);
        }

        // Memory-map the segment.
        // SAFETY: fd valid, size positive, flags well-known.
        let ptr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                size,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED,
                fd.as_raw_fd(),
                0,
            )
        };

        if ptr == libc::MAP_FAILED {
            let err = io::Error::last_os_error();
            if created {
                let _ = unsafe { libc::shm_unlink(cname.as_ptr()) };
            }
            return Err(err);
        }

        // SAFETY: mmap returned a non-null pointer (we just checked).
        let ptr = NonNull::new(ptr.cast::<u8>())
            .expect("mmap returned non-null on success");

        Ok((
            Self {
                name: cname,
                lock_path,
                process_lock,
                ptr,
                len: size,
                created,
            },
            initialization_lock,
        ))
    }

    /// Raw mapped pointer to the start of the region.
    pub fn as_ptr(&self) -> *mut u8 {
        self.ptr.as_ptr()
    }

    /// Length of the mapped region (the `size` passed to `open_or_create`).
    pub fn len(&self) -> usize {
        self.len
    }

    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// True when this handle was the one that created the segment.
    /// Useful for picking which process performs first-time
    /// initialization of the header.
    pub fn created(&self) -> bool {
        self.created
    }

    /// Acquire an exclusive cross-process lock tied to this SHM name.
    ///
    /// `flock` ownership is held by the kernel and is released when a process
    /// exits or the descriptor closes, including abnormal termination.
    pub fn lock_exclusive(&self) -> io::Result<ShmRegionLock> {
        if !self.process_lock {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "SHM region was opened without a process lock",
            ));
        }
        lock_path_exclusive(&self.lock_path)
    }

    #[cfg(test)]
    pub(crate) fn try_lock_exclusive(&self) -> io::Result<ShmRegionLock> {
        if !self.process_lock {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "SHM region was opened without a process lock",
            ));
        }
        let lock_fd = open_lock_file(&self.lock_path)?;
        let rc = unsafe {
            libc::flock(lock_fd.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB)
        };
        if rc == 0 {
            Ok(ShmRegionLock { lock_fd })
        } else {
            Err(io::Error::last_os_error())
        }
    }

    /// Remove the underlying segment name. Existing mappings stay
    /// valid until each process drops its `ShmRegion`. Use only on
    /// shutdown / fleet teardown by the process that owns lifecycle.
    pub fn unlink(&self) -> io::Result<()> {
        // SAFETY: name is a valid C string.
        let rc = unsafe { libc::shm_unlink(self.name.as_ptr()) };
        let shm_error = if rc != 0 {
            let err = io::Error::last_os_error();
            // ENOENT is fine — segment was already unlinked.
            if err.raw_os_error() == Some(libc::ENOENT) {
                None
            } else {
                Some(err)
            }
        } else {
            None
        };
        let mut lock_error = match std::fs::remove_file(&self.lock_path) {
            Ok(()) => None,
            Err(error) if error.kind() == io::ErrorKind::NotFound => None,
            Err(error) => Some(error),
        };
        let name = self.name.to_string_lossy();
        // SAFETY: a plain syscall with no arguments.
        let uid = unsafe { libc::geteuid() };
        match companion_lock_files(&name, uid) {
            Ok(files) => {
                for file in files {
                    if let Err(error) = std::fs::remove_file(&file)
                        && error.kind() != io::ErrorKind::NotFound
                    {
                        lock_error.get_or_insert(error);
                    }
                }
            }
            Err(error) => {
                lock_error.get_or_insert(error);
            }
        }
        if let Some(error) = shm_error.or(lock_error) {
            return Err(error);
        }
        Ok(())
    }
}

#[cfg(not(target_os = "macos"))]
fn configure_new_shm(
    fd: &OwnedFd,
    policy: ShmAccessPolicy,
) -> io::Result<()> {
    let Some(gid) = policy.gid() else {
        return Ok(());
    };
    if gid == !0 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "invalid SHM group id",
        ));
    }
    // SAFETY: this is our new, still owner-only object. Preserve its uid.
    if unsafe { libc::fchown(fd.as_raw_fd(), !0, gid) } != 0 {
        return Err(io::Error::last_os_error());
    }
    // Explicit group-sharing policy, applied only after group ownership is set.
    if unsafe { libc::fchmod(fd.as_raw_fd(), policy.mode() as _) } != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

fn shm_object_size(
    fd: &OwnedFd,
    name: &str,
    owner_uid: u32,
    policy: ShmAccessPolicy,
) -> io::Result<usize> {
    let mut stat = std::mem::MaybeUninit::<libc::stat>::uninit();
    // SAFETY: `fd` is valid and `stat` points to writable storage.
    let stat_rc = unsafe { libc::fstat(fd.as_raw_fd(), stat.as_mut_ptr()) };
    if stat_rc != 0 {
        return Err(io::Error::last_os_error());
    }
    // SAFETY: `fstat` succeeded and initialized the structure.
    let stat = unsafe { stat.assume_init() };
    let mode = u64::from(stat.st_mode) & 0o7777;
    if stat.st_uid != owner_uid
        || policy.gid().is_some_and(|gid| gid != stat.st_gid)
        || mode & !u64::from(policy.mode()) != 0
    {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            format!(
                "SHM segment {name} owner/group/mode do not satisfy {policy:?} for uid {owner_uid}"
            ),
        ));
    }
    let actual_size = stat.st_size;
    usize::try_from(actual_size).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidData,
            format!("SHM segment {name} reported invalid size {actual_size}"),
        )
    })
}

fn validate_minimum_size(
    name: &str,
    actual_size: usize,
    minimum_size: usize,
) -> io::Result<()> {
    if actual_size < minimum_size {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!(
                "SHM segment {name} size {actual_size} is smaller than requested mapping {minimum_size}"
            ),
        ));
    }
    Ok(())
}

/// RAII guard for a [`ShmRegion`]'s process-recoverable exclusive lock.
///
/// Semantic crates use this when a current-state transition must be atomic
/// across fleet processes. Dropping the guard releases the kernel lock.
pub struct ShmRegionLock {
    lock_fd: OwnedFd,
}

impl Drop for ShmRegionLock {
    fn drop(&mut self) {
        let _ = unsafe { libc::flock(self.lock_fd.as_raw_fd(), libc::LOCK_UN) };
    }
}

fn lock_path_exclusive(lock_path: &Path) -> io::Result<ShmRegionLock> {
    lock_fd_exclusive(open_lock_file(lock_path)?)
}

fn open_lock_file(lock_path: &Path) -> io::Result<OwnedFd> {
    use std::os::unix::fs::MetadataExt;

    if let Some(dir) = lock_path.parent() {
        ensure_lock_dir(dir)?;
    }

    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .mode(0o600)
        .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW)
        .open(lock_path)?;

    // The directory check makes this unreachable, which is the reason to make
    // it anyway: it turns a property inferred from the directory's mode into
    // one this function establishes about the descriptor it is about to lock.
    let uid = unsafe { libc::geteuid() };
    let owner = file.metadata()?.uid();
    if owner != uid {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            format!(
                "{} is owned by uid {owner} rather than {uid}",
                lock_path.display()
            ),
        ));
    }

    Ok(file.into())
}

fn lock_fd_exclusive(lock_fd: OwnedFd) -> io::Result<ShmRegionLock> {
    loop {
        let rc = unsafe { libc::flock(lock_fd.as_raw_fd(), libc::LOCK_EX) };
        if rc == 0 {
            return Ok(ShmRegionLock { lock_fd });
        }
        let error = io::Error::last_os_error();
        if error.kind() != io::ErrorKind::Interrupted {
            return Err(error);
        }
    }
}

impl Drop for ShmRegion {
    fn drop(&mut self) {
        // SAFETY: ptr came from mmap of `self.len` bytes; munmap is the inverse.
        unsafe {
            libc::munmap(self.ptr.as_ptr().cast(), self.len);
        }
    }
}

// SAFETY: the underlying region is shared memory and synchronization
// happens at the slot level (atomic seq counters); the handle itself
// is just a pointer + length, safe to send/share.
unsafe impl Send for ShmRegion {}
unsafe impl Sync for ShmRegion {}

/// Build the conventional name for an Orbit ring segment.
pub fn ring_segment_name(
    fleet_name: &str,
    kind: u8,
) -> String {
    // SAFETY: `geteuid` always returns a value; no error path.
    let uid = unsafe { libc::geteuid() };
    ring_segment_name_for_uid(fleet_name, kind, uid)
}

/// Build the conventional name for an Orbit ring segment owned by `uid`.
///
/// This form is intended for inspection and lifecycle tools that need to
/// address a user other than their own effective uid.
pub fn ring_segment_name_for_uid(
    fleet_name: &str,
    kind: u8,
    uid: u32,
) -> String {
    format!("/{SHM_NAMESPACE}-{fleet_name}-{kind}-{uid}")
}

/// Where a fleet's members hold their presence: `<lock dir>/orbit-<fleet>.fleet`.
pub fn fleet_lock_path(fleet_name: &str) -> PathBuf {
    lock_dir().join(format!("{SHM_NAMESPACE}-{fleet_name}.fleet"))
}

/// The same for another user's fleet, for lifecycle tools that address a uid
/// other than their own.
pub fn fleet_lock_path_for_uid(
    fleet_name: &str,
    uid: u32,
) -> PathBuf {
    lock_dir_for_uid(uid).join(format!("{SHM_NAMESPACE}-{fleet_name}.fleet"))
}

/// A process's membership in a fleet: a shared `flock` on the fleet's lock
/// file, held for as long as this lives and released by the kernel when the
/// process dies, however it dies. It is what [`try_lock_fleet_exclusive`]
/// contends with, so a tool that removes the fleet's segments cannot do so
/// while any member is alive.
pub struct FleetMembership {
    lock_fd: OwnedFd,
}

impl Drop for FleetMembership {
    fn drop(&mut self) {
        let _ = unsafe { libc::flock(self.lock_fd.as_raw_fd(), libc::LOCK_UN) };
    }
}

/// Join the fleet's membership. Waits for a lifecycle tool that holds the
/// exclusive lock at that moment; a clear in progress finishes first.
pub fn join_fleet_membership(fleet_name: &str) -> io::Result<FleetMembership> {
    let lock_fd = open_lock_file(&fleet_lock_path(fleet_name))?;
    // SAFETY: `lock_fd` is an open descriptor owned by this call.
    let rc = unsafe { libc::flock(lock_fd.as_raw_fd(), libc::LOCK_SH) };
    if rc != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(FleetMembership { lock_fd })
}

/// Protect an existing fleet against clearing while a diagnostic reader pins
/// its arena. Never creates a membership file or reserves a writer lane.
pub fn hold_existing_fleet(fleet_name: &str) -> io::Result<FleetMembership> {
    use std::os::unix::fs::MetadataExt;
    let path = fleet_lock_path(fleet_name);
    let file = OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW)
        .open(&path)?;
    let metadata = file.metadata()?;
    if !metadata.is_file() || metadata.uid() != unsafe { libc::geteuid() } {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "fleet membership file has an unexpected owner or type",
        ));
    }
    let lock_fd: OwnedFd = file.into();
    loop {
        let rc = unsafe { libc::flock(lock_fd.as_raw_fd(), libc::LOCK_SH) };
        if rc == 0 {
            return Ok(FleetMembership { lock_fd });
        }
        let error = io::Error::last_os_error();
        if error.kind() != io::ErrorKind::Interrupted {
            return Err(error);
        }
    }
}

/// Exclusive hold on a fleet, for the tool that removes its segments.
///
/// `Ok(None)` means a member is alive and the fleet must not be touched;
/// there is deliberately no way to force past it. `Ok(Some(_))` keeps the
/// fleet closed to new members until the guard is dropped, so a removal
/// cannot interleave with a start. Never waits.
pub fn try_lock_fleet_exclusive(
    fleet_name: &str,
    uid: u32,
) -> io::Result<Option<ShmRegionLock>> {
    // Creating the file when no member ever joined is right: the guard then
    // keeps a first member from starting in the middle of a removal.
    let lock_fd = open_lock_file(&fleet_lock_path_for_uid(fleet_name, uid))?;
    // SAFETY: `lock_fd` is an open descriptor owned by this call.
    let rc = unsafe {
        libc::flock(lock_fd.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB)
    };
    if rc == 0 {
        return Ok(Some(ShmRegionLock { lock_fd }));
    }
    let error = io::Error::last_os_error();
    if error.kind() == io::ErrorKind::WouldBlock {
        return Ok(None);
    }
    Err(error)
}

/// One process's hold on one lane of a segment: an exclusive `flock` on the
/// lane's own lock file, kept for as long as this lives and released by the
/// kernel when the process dies, however it dies.
///
/// It is how a lane's owner is known to be gone without a timeout and without
/// a PID, which another PID namespace would read wrongly: a hold that can be
/// taken has nobody behind it.
pub struct LaneHold {
    lock_fd: OwnedFd,
}

/// A lane lock prepared at opening, with no path construction or file open
/// during acquisition. Each opener owns a distinct file description.
/// Open after fork; inherited instances cannot be used by the child.
pub struct PreparedLaneLock {
    fd: OwnedFd,
    busy: AtomicBool,
    pid: u32,
}

impl PreparedLaneLock {
    pub fn open(
        shm_name: &str,
        lane: usize,
    ) -> io::Result<Self> {
        let name = format!("{}.lane{lane}", shm_name.trim_start_matches('/'));
        Ok(Self {
            fd: open_lock_file(&lock_file_path(&name))?,
            busy: AtomicBool::new(false),
            pid: std::process::id(),
        })
    }

    /// One local CAS and at most one LOCK_NB syscall. No owner wait or retry.
    pub fn try_lock(&self) -> io::Result<Option<PreparedLaneGuard<'_>>> {
        if self.pid != std::process::id() {
            return Err(io::Error::from(io::ErrorKind::PermissionDenied));
        }
        if self
            .busy
            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
            .is_err()
        {
            return Ok(None);
        }
        // SAFETY: the local flag excludes overlapping use of this description.
        if unsafe {
            libc::flock(self.fd.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB)
        } == 0
        {
            return Ok(Some(PreparedLaneGuard { lock: self }));
        }
        let error = io::Error::last_os_error();
        self.busy.store(false, Ordering::Release);
        if error.kind() == io::ErrorKind::WouldBlock {
            Ok(None)
        } else {
            Err(error)
        }
    }
}

pub struct PreparedLaneGuard<'a> {
    lock: &'a PreparedLaneLock,
}

impl Drop for PreparedLaneGuard<'_> {
    fn drop(&mut self) {
        if self.lock.pid == std::process::id() {
            // SAFETY: this guard owns this file description's exclusive hold.
            unsafe {
                libc::flock(self.lock.fd.as_raw_fd(), libc::LOCK_UN);
            }
            self.lock.busy.store(false, Ordering::Release);
        }
    }
}

impl Drop for LaneHold {
    fn drop(&mut self) {
        let _ = unsafe { libc::flock(self.lock_fd.as_raw_fd(), libc::LOCK_UN) };
    }
}

/// Take lane `lane` of segment `shm_name`, if no live process holds it.
///
/// `Ok(None)` means a process holding it is alive. Never waits.
pub fn try_hold_lane(
    shm_name: &str,
    lane: usize,
) -> io::Result<Option<LaneHold>> {
    let name = format!("{}.lane{lane}", shm_name.trim_start_matches('/'));
    let lock_fd = open_lock_file(&lock_file_path(&name))?;
    // SAFETY: `lock_fd` is an open descriptor owned by this call.
    let rc = unsafe {
        libc::flock(lock_fd.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB)
    };
    if rc == 0 {
        return Ok(Some(LaneHold { lock_fd }));
    }
    let error = io::Error::last_os_error();
    if error.kind() == io::ErrorKind::WouldBlock {
        return Ok(None);
    }
    Err(error)
}

/// One process's hold on one directory-watcher slot of a segment: the same
/// kernel-released `flock` as a [`LaneHold`], on a lock file of its own.
///
/// A watcher is not a lane. It serves nothing and takes no frames; it only
/// wants to be woken when the lane directory moves. Its slot is still owned
/// the way a lane is, so a slot a dead process left can be taken by the next
/// one without a timeout.
pub struct WatcherHold {
    lock_fd: OwnedFd,
}

impl Drop for WatcherHold {
    fn drop(&mut self) {
        let _ = unsafe { libc::flock(self.lock_fd.as_raw_fd(), libc::LOCK_UN) };
    }
}

/// Take watcher slot `slot` of segment `shm_name`, if no live process holds
/// it.
///
/// `Ok(None)` means a process holding it is alive. Never waits.
pub fn try_hold_watcher(
    shm_name: &str,
    slot: usize,
) -> io::Result<Option<WatcherHold>> {
    let name = format!("{}.watch{slot}", shm_name.trim_start_matches('/'));
    let lock_fd = open_lock_file(&lock_file_path(&name))?;
    // SAFETY: `lock_fd` is an open descriptor owned by this call.
    let rc = unsafe {
        libc::flock(lock_fd.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB)
    };
    if rc == 0 {
        return Ok(Some(WatcherHold { lock_fd }));
    }
    let error = io::Error::last_os_error();
    if error.kind() == io::ErrorKind::WouldBlock {
        return Ok(None);
    }
    Err(error)
}

/// Every lock file beside segment `shm_name` of `uid`: the region's own, one
/// per lane that was ever held ([`try_hold_lane`]) and one per watcher slot
/// ([`try_hold_watcher`]). What removes a segment
/// removes these with it; an unlocked one left behind is harmless, but it is
/// one more file per lane for every segment that ever existed.
pub fn companion_lock_files(
    shm_name: &str,
    uid: u32,
) -> io::Result<Vec<PathBuf>> {
    let base = shm_name.trim_start_matches('/');
    let own = format!("{base}.lock");
    let lane = format!("{base}.lane");
    let watch = format!("{base}.watch");
    let dir = lock_dir_for_uid(uid);
    let entries = match std::fs::read_dir(&dir) {
        Ok(entries) => entries,
        Err(error) if error.kind() == io::ErrorKind::NotFound => {
            return Ok(Vec::new());
        }
        Err(error) => return Err(error),
    };
    let mut found = Vec::new();
    for entry in entries {
        let entry = entry?;
        let Some(file) = entry.file_name().to_str().map(str::to_owned) else {
            continue;
        };
        if file == own
            || ((file.starts_with(&lane) || file.starts_with(&watch))
                && file.ends_with(".lock"))
        {
            found.push(entry.path());
        }
    }
    Ok(found)
}

fn lock_file_path(shm_name: &str) -> PathBuf {
    lock_dir().join(format!("{}.lock", shm_name.trim_start_matches('/')))
}

/// Where the companion lock files live.
///
/// They used to live in `/tmp` directly, as
/// `/tmp/orbit-{fleet}-{kind}-{uid}.lock`, opened `O_CREAT` without `O_EXCL`
/// and without asking who owned what the open found. `/tmp` is world-writable,
/// so any local user could create that file first and then hold `LOCK_EX` on it
/// for as long as they liked: every process in the fleet would sit in `flock` —
/// not fail, block — waiting for a lock it was never going to get. The SHM
/// segment name is uid-scoped and so cannot be squatted this way; the lock path
/// was not. `O_NOFOLLOW` prevented the symlink version of the trick and nothing
/// else.
///
/// They now live in a per-uid directory created `0700`, which a user who is not
/// us cannot put a file into. What such a user can still do is create the
/// directory first, so its owner and mode are checked on every open rather than
/// assumed from having created it: a directory that is not ours, or not
/// private, fails the open with the path in the message instead of parking the
/// process on a lock.
///
/// `XDG_RUNTIME_DIR` is preferred where the session provides one, because it is
/// already per-user and `0700` and so is not inside a world-writable directory
/// at all. macOS has no such variable but gives each user a private `TMPDIR`;
/// `/tmp` is the fallback, with the checks above carrying the weight.
fn lock_dir() -> PathBuf {
    let uid = unsafe { libc::geteuid() };
    lock_dir_for_uid(uid)
}

/// Locate the existing per-user companion lock directory without creating it.
/// Tools can use its filenames as discovery hints, but must verify SHM objects.
pub fn lock_dir_for_uid(uid: u32) -> PathBuf {
    let base = std::env::var_os("XDG_RUNTIME_DIR")
        .map(PathBuf::from)
        .filter(|dir| dir.is_absolute())
        .unwrap_or_else(|| PathBuf::from("/tmp"));

    base.join(format!("{SHM_NAMESPACE}-{uid}"))
}

/// Creates the lock directory if it is missing and refuses it if it is not
/// ours. Called when a lock is actually taken rather than when a region is
/// opened: a region that never locks has nothing to squat, and failing its open
/// on a directory it does not use would hand an attacker a wider outage than
/// the one being closed.
fn ensure_lock_dir(dir: &Path) -> io::Result<()> {
    use std::os::unix::fs::DirBuilderExt;

    let uid = unsafe { libc::geteuid() };
    match std::fs::DirBuilder::new().mode(0o700).create(dir) {
        Ok(()) => {}
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
        Err(error) => return Err(error),
    }

    ensure_private_dir(dir, uid)
}

/// Refuses a lock directory that someone else could write to.
///
/// `symlink_metadata` rather than `metadata`: a symlink pointing at a directory
/// we do own would otherwise pass while the lock files landed somewhere the
/// attacker chose.
fn ensure_private_dir(
    dir: &Path,
    uid: u32,
) -> io::Result<()> {
    use std::os::unix::fs::{MetadataExt, PermissionsExt};

    let metadata = std::fs::symlink_metadata(dir)?;
    if !metadata.is_dir() {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            format!("{} is not a directory", dir.display()),
        ));
    }
    if metadata.uid() != uid {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            format!(
                "{} is owned by uid {} rather than {uid}; refusing to lock in a directory \
                 another user controls",
                dir.display(),
                metadata.uid()
            ),
        ));
    }
    if metadata.permissions().mode() & 0o077 != 0 {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            format!(
                "{} is mode {:o}; refusing to lock in a directory others can write to",
                dir.display(),
                metadata.permissions().mode() & 0o777
            ),
        ));
    }

    Ok(())
}

#[cfg(test)]
mod fleet_lock_tests {
    use super::{join_fleet_membership, try_lock_fleet_exclusive};

    #[test]
    fn prepared_lane_excludes_threads_and_distinct_file_descriptions() {
        let name = format!("prepared-{}", std::process::id());
        let first = super::PreparedLaneLock::open(&name, 0).unwrap();
        let second = super::PreparedLaneLock::open(&name, 0).unwrap();
        let hold = first.try_lock().unwrap().unwrap();
        assert!(first.try_lock().unwrap().is_none());
        std::thread::scope(|scope| {
            assert!(
                scope
                    .spawn(|| first.try_lock().unwrap().is_none())
                    .join()
                    .unwrap()
            );
        });
        assert!(second.try_lock().unwrap().is_none());
        drop(hold);
        let other = second.try_lock().unwrap().unwrap();
        assert!(first.try_lock().unwrap().is_none());
        drop(other);
        drop(first);
        drop(second);
        std::fs::remove_file(super::lock_file_path(&format!("{name}.lane0")))
            .unwrap();
    }

    /// A member alive means the fleet cannot be cleared, and there is no
    /// flag that says otherwise; the member going away is what opens it.
    #[test]
    fn a_member_holds_the_fleet_against_exclusive_takers() {
        let fleet = format!("fl{:x}", std::process::id());
        let uid = unsafe { libc::geteuid() };

        let member = join_fleet_membership(&fleet).expect("join");
        assert!(try_lock_fleet_exclusive(&fleet, uid).expect("try").is_none());

        drop(member);
        let exclusive = try_lock_fleet_exclusive(&fleet, uid).expect("try");
        assert!(exclusive.is_some());
        // And a member cannot join while a removal holds the fleet: the
        // shared lock would block, which is the behaviour, not a test to run.
        drop(exclusive);
        let _ =
            std::fs::remove_file(super::fleet_lock_path_for_uid(&fleet, uid));
    }
}