uqa-storage 0.5.0

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Native byte locks and positioned I/O shared by physical coordination adapters. On POSIX, each adapter must keep one process-wide descriptor per coordination file: closing any other descriptor would release that process's record locks. Callers own local arbitration, lock ordering and cancellation.

mod path_watch;
pub use path_watch::PathChangeWatch;

#[cfg(any(target_os = "linux", target_os = "macos"))]
mod shared_mapping;
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub use shared_mapping::SharedFileMapping;

#[cfg(windows)]
use std::os::windows::{fs::FileExt, io::AsRawHandle};
#[cfg(unix)]
use std::os::{fd::AsRawFd, unix::fs::FileExt};

#[cfg(unix)]
pub fn try_lock_byte(file: &std::fs::File, offset: u64, write: bool) -> std::io::Result<()> {
    unix_byte_mode(
        file,
        offset,
        if write {
            libc::F_WRLCK as libc::c_short
        } else {
            libc::F_RDLCK as libc::c_short
        },
    )
}

#[cfg(unix)]
pub fn unlock_byte(file: &std::fs::File, offset: u64) -> std::io::Result<()> {
    unix_byte_mode(file, offset, libc::F_UNLCK as libc::c_short)
}

/// Lock one byte, waiting for conflicting holders in other processes. The caller must never wait while holding a byte that a holder of this one may wait for.
#[cfg(unix)]
pub fn lock_byte(file: &std::fs::File, offset: u64, write: bool) -> std::io::Result<()> {
    let mode = if write {
        libc::F_WRLCK as libc::c_short
    } else {
        libc::F_RDLCK as libc::c_short
    };
    loop {
        match unix_byte_command(file, offset, mode, libc::F_SETLKW) {
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
            result => return result,
        }
    }
}

#[cfg(unix)]
fn unix_byte_mode(file: &std::fs::File, offset: u64, mode: libc::c_short) -> std::io::Result<()> {
    unix_byte_command(file, offset, mode, libc::F_SETLK)
}

#[cfg(unix)]
fn unix_byte_command(
    file: &std::fs::File,
    offset: u64,
    mode: libc::c_short,
    command: libc::c_int,
) -> std::io::Result<()> {
    // SAFETY: flock is initialized before fcntl borrows it, and File owns the descriptor for the call.
    let mut flock: libc::flock = unsafe { std::mem::zeroed() };
    flock.l_type = mode;
    flock.l_whence = libc::SEEK_SET as libc::c_short;
    flock.l_start = libc::off_t::try_from(offset).map_err(|_| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "record-lock offset exceeds the platform off_t range",
        )
    })?;
    flock.l_len = 1;
    let result = unsafe { libc::fcntl(file.as_raw_fd(), command, &flock) };
    if result == -1 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

#[cfg(unix)]
pub fn read_exact_at(file: &std::fs::File, bytes: &mut [u8], offset: u64) -> std::io::Result<()> {
    file.read_exact_at(bytes, offset)
}

#[cfg(unix)]
pub fn write_all_at(file: &std::fs::File, bytes: &[u8], offset: u64) -> std::io::Result<()> {
    file.write_all_at(bytes, offset)
}

/// Fill `bytes` from `offset`, reading as zero whatever lies after the end of the file.
pub fn read_zero_extended_at(
    file: &std::fs::File,
    bytes: &mut [u8],
    offset: u64,
) -> std::io::Result<()> {
    let mut consumed = 0usize;
    while consumed < bytes.len() {
        #[cfg(unix)]
        let read = file.read_at(
            &mut bytes[consumed..],
            offset.saturating_add(consumed as u64),
        );
        #[cfg(windows)]
        let read = file.seek_read(
            &mut bytes[consumed..],
            offset.saturating_add(consumed as u64),
        );
        match read {
            Ok(0) => break,
            Ok(read) => consumed += read,
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
            Err(error) => return Err(error),
        }
    }
    bytes[consumed..].fill(0);
    Ok(())
}

#[cfg(windows)]
pub fn read_exact_at(file: &std::fs::File, bytes: &mut [u8], offset: u64) -> std::io::Result<()> {
    let mut consumed = 0usize;
    while consumed < bytes.len() {
        let read = file.seek_read(
            &mut bytes[consumed..],
            offset.saturating_add(consumed as u64),
        )?;
        if read == 0 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                "positioned file read reached end of file",
            ));
        }
        consumed += read;
    }
    Ok(())
}

#[cfg(windows)]
pub fn write_all_at(file: &std::fs::File, bytes: &[u8], offset: u64) -> std::io::Result<()> {
    let mut consumed = 0usize;
    while consumed < bytes.len() {
        let written =
            file.seek_write(&bytes[consumed..], offset.saturating_add(consumed as u64))?;
        if written == 0 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::WriteZero,
                "positioned file write returned zero bytes",
            ));
        }
        consumed += written;
    }
    Ok(())
}

#[cfg(unix)]
pub fn lock_would_block(error: &std::io::Error) -> bool {
    matches!(error.raw_os_error(), Some(libc::EAGAIN | libc::EACCES))
}

#[cfg(windows)]
pub fn lock_would_block(error: &std::io::Error) -> bool {
    error.raw_os_error() == Some(windows_sys::Win32::Foundation::ERROR_LOCK_VIOLATION as i32)
}

#[cfg(windows)]
fn windows_overlapped(offset: u64) -> windows_sys::Win32::System::IO::OVERLAPPED {
    windows_sys::Win32::System::IO::OVERLAPPED {
        Anonymous: windows_sys::Win32::System::IO::OVERLAPPED_0 {
            Anonymous: windows_sys::Win32::System::IO::OVERLAPPED_0_0 {
                Offset: offset as u32,
                OffsetHigh: (offset >> 32) as u32,
            },
        },
        ..Default::default()
    }
}

#[cfg(windows)]
pub fn try_lock_byte(file: &std::fs::File, offset: u64, write: bool) -> std::io::Result<()> {
    windows_lock_byte(
        file,
        offset,
        write,
        windows_sys::Win32::Storage::FileSystem::LOCKFILE_FAIL_IMMEDIATELY,
    )
}

/// Lock one byte, waiting for conflicting holders in other processes. The caller must never wait while holding a byte that a holder of this one may wait for.
#[cfg(windows)]
pub fn lock_byte(file: &std::fs::File, offset: u64, write: bool) -> std::io::Result<()> {
    windows_lock_byte(file, offset, write, 0)
}

#[cfg(windows)]
fn windows_lock_byte(
    file: &std::fs::File,
    offset: u64,
    write: bool,
    wait: u32,
) -> std::io::Result<()> {
    use windows_sys::Win32::Storage::FileSystem::{LockFileEx, LOCKFILE_EXCLUSIVE_LOCK};
    let mut overlapped = windows_overlapped(offset);
    let flags = wait | if write { LOCKFILE_EXCLUSIVE_LOCK } else { 0 };
    let result = unsafe { LockFileEx(file.as_raw_handle(), flags, 0, 1, 0, &raw mut overlapped) };
    if result == 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

#[cfg(windows)]
pub fn unlock_byte(file: &std::fs::File, offset: u64) -> std::io::Result<()> {
    let mut overlapped = windows_overlapped(offset);
    let result = unsafe {
        windows_sys::Win32::Storage::FileSystem::UnlockFileEx(
            file.as_raw_handle(),
            0,
            1,
            0,
            &raw mut overlapped,
        )
    };
    if result == 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}