goosefs-sdk 0.2.1

// Copyright (C) 2026 Tencent. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! `UringPageStore` — io_uring backend for `PageStore`.
//!
//! Implements the same `PageStore` trait as `LocalPageStore` (tokio::fs
//! backend) but routes all hot-path IO through io_uring SQE/CQE, eliminating
//! `spawn_blocking` overhead.
//!
//! References: Lance `reader.rs:97-292` `UringReader` (extended from read-only
//! to full get/put/delete).
//!
//! See `docs/CLIENT_PAGE_CACHE_DESIGN.md` .

use super::driver::{submit_request, try_submit_request};
use super::future::UringOpFuture;
use super::requests::{IoRequest, RequestState, UringOpType};
use super::{hash_file_id, io_error, NUM_BUCKETS};
use crate::cache::metric_name as mn;
use crate::cache::page_id::PageId;
use crate::cache::store::{LocalPageStore, PageStore};
use crate::error::Result;
use crate::metrics::counter;
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use foyer_memory::{Cache, CacheBuilder, S3FifoConfig};
use std::ffi::CString;
use std::fs::File;
use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// Sidecar filename holding a file's `(length, mtime)` identity — identical to
/// `LocalPageStore` so both backends share the same on-disk format.
const IDENTITY_FILE: &str = ".identity";

/// Maximum time to wait for a single io_uring operation before declaring
/// it hung. Under normal operation, NVMe IO completes in ~10 µs. The
/// timeout prevents a lost CQE (kernel bug, driver thread panic) from
/// hanging a tokio task forever (H4 fix).
const URING_OP_TIMEOUT: Duration = Duration::from_secs(30);

/// TTL for cached directory fds. Entries not accessed within this window are
/// eligible for lazy eviction on the next `get_dir_fd` cleanup pass.
const DIR_FD_TTL: Duration = Duration::from_secs(300);

/// Soft cap on the directory fd cache. When exceeded, a cleanup pass removes
/// stale (TTL-expired) entries. The cap is generous because the number of
/// unique files in a typical workload is small (1–100).
const DIR_FD_CACHE_SOFT_CAP: usize = 4096;

/// Idle window after which a cached page fd is considered stale.
/// Mirrors Lance's `HANDLE_CACHE` design (`lance-io/src/uring/reader.rs:60`).
///
/// **Enforced lazily**, on `get_bytes`: foyer has no built-in TTL, unlike the
/// `moka` cache this replaced. See [`PageFdEntry::is_expired`] for the
/// behavioural difference and why it is acceptable here.
const PAGE_FD_CACHE_TTL: Duration = Duration::from_secs(60);

/// Maximum number of cached page fds. Mirrors Lance's `HANDLE_CACHE` design
/// (`lance-io/src/uring/reader.rs:61`).
///
/// Each entry has weight 1, so this is a page *count*, and foyer evicts once
/// it is exceeded.
///
/// 10,000 entries × `RawFd` (4 bytes) + `Arc<File>` overhead ≈ 1-2 MB.
/// **Requires `ulimit -n >= 10240`** to avoid `EMFILE` when the cap is
/// reached. Lance's README notes the same constraint. The lazy TTL does not
/// change this bound — the cap alone is what keeps fd usage in check.
const PAGE_FD_CACHE_MAX_CAPACITY: usize = 10_000;

/// Shard count for the page fd cache. Matches the order of magnitude of the
/// segment count the previous `moka` cache used internally.
const PAGE_FD_CACHE_SHARDS: usize = 64;

/// A cached directory file descriptor, keyed by `file_id`.
///
/// The directory is `<root>/<bucket>/<file_id>/` — the parent of all page
/// files for that `file_id`. Caching the directory fd lets `get()` use
/// `openat(dirfd, "page_index")` (1-level path resolution) instead of
/// `openat(AT_FDCWD, "<root>/<bucket>/<file_id>/<page_index>")` (4-level),
/// dramatically reducing kernel VFS lock contention under concurrency.
///
/// `last_access` is an `AtomicU64` (epoch nanos) so it can be updated through
/// a `DashMap::get()` **read guard** without taking a write lock — this is
/// critical to avoid blocking tokio workers on the hot path.
struct DirFdEntry {
    /// Owns the directory fd.
    ///
    /// This used to be a bare `RawFd` with a `Drop` that called
    /// `libc::close`, on the reasoning that no reader could be holding a
    /// *directory* fd because page fds were opened and closed per `get()`.
    /// The page fd cache invalidated that: `PAGE_FD_CACHE` is process-global
    /// and outlives any individual store, so dropping a store closed its dir
    /// fds while page fds with the same numbers were still cached. The kernel
    /// reuses fd numbers, so a later read could hit a stale entry and read a
    /// different file — and the eventual second close aborted the process.
    ///
    /// `File` closes the fd exactly once, when the last handle goes away.
    ///
    /// Deliberately not accompanied by a cached `RawFd` copy: a second copy of
    /// the fd sitting beside its owner is what made the original bug possible.
    /// Callers take `as_raw_fd()` off the handle they hold.
    dir: Arc<File>,
    last_access: AtomicU64,
}

/// Current time as epoch nanos — used for `DirFdEntry::last_access`.
fn now_nanos() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0)
}

/// Global page fd cache — keyed by `PageId`, holds the open file fd.
///
/// **Mirrors Lance's `HANDLE_CACHE`** (`lance-io/src/uring/reader.rs:55-63`),
/// but backed by `foyer-memory` rather than `moka`. Entries are evicted once
/// [`PAGE_FD_CACHE_MAX_CAPACITY`] is exceeded, and stale entries are dropped
/// lazily on access ([`PageFdEntry::is_expired`]).
///
/// **Why process-global?** A file's fd is valid for the entire process —
/// we don't need per-`UringPageStore` caches. This matches Lance's design
/// and lets caches survive `UringPageStore` re-creation (e.g. when the
/// cache directory rotates).
///
/// **Why S3-FIFO?** It is scan-resistant: a one-shot sequential scan over many
/// distinct pages will not flush the hot random-read working set, which plain
/// LRU (moka's behaviour here) does. It is also the cheapest policy on the read
/// path — its `acquire()` is `Op::immutable`, so a cache hit takes a shard
/// *read* lock, whereas LRU/LFU take a write lock
/// (foyer-memory-0.22.3 `src/eviction/s3fifo.rs:309` vs `lru.rs:241`).
///
/// **Concurrency**: `get` is a sharded hash lookup under a read lock. The cold
/// path (miss → `OP_OPENAT` → insert) is bounded by the rate of distinct page
/// accesses.
static PAGE_FD_CACHE: LazyLock<Cache<PageId, Arc<PageFdEntry>>> = LazyLock::new(|| {
    CacheBuilder::new(PAGE_FD_CACHE_MAX_CAPACITY)
        .with_name("goosefs-page-fd")
        .with_shards(PAGE_FD_CACHE_SHARDS)
        // Weight 1 per entry, so the capacity above is an entry count.
        .with_weighter(|_: &PageId, _: &Arc<PageFdEntry>| 1)
        .with_eviction_config(S3FifoConfig::default())
        .build()
});

/// A cached page file descriptor, keyed by `PageId`.
///
/// **Method E** (mirrors Lance's `UringFileHandle` at
/// `lance-io/src/uring/reader.rs:68-90`): on a cache hit, `get()` uses the
/// cached `RawFd` directly for `OP_READ`, skipping `OP_OPENAT` and
/// `OP_CLOSE` entirely (1 SQE instead of 3). This eliminates the VFS
/// `inode_lock` contention that dominates the miss path under concurrency
/// (result_3: fd_open = 61.3% of total latency).
///
/// `file: Arc<File>` keeps the file alive while the cache holds the entry;
/// when the entry is evicted, `Arc::drop` closes the fd automatically —
/// identical to Lance's pattern, and unchanged by the move to foyer (both
/// caches are reference-counted).
struct PageFdEntry {
    /// `RawFd` cached at `new()` time to avoid repeated `as_raw_fd()` calls
    /// on the hot path. Valid as long as `file` is alive.
    fd: RawFd,
    /// The open file handle. Holding `Arc<File>` (not just `RawFd`) ensures
    /// the fd is closed only when the cache evicts the entry — not when a
    /// concurrent reader is still using it.
    #[allow(dead_code)]
    file: Arc<File>,
    /// Insertion instant, for the lazy TTL check.
    ///
    /// `Instant` is monotonic, so a wall-clock jump cannot resurrect an
    /// expired fd (`SystemTime` would allow that).
    inserted_at: Instant,
}

impl PageFdEntry {
    /// Wrap a freshly-opened `File` for the cache. The fd is captured once
    /// and reused for all subsequent `OP_READ` calls.
    fn new(file: File) -> Self {
        let fd = file.as_raw_fd();
        Self {
            fd,
            file: Arc::new(file),
            inserted_at: Instant::now(),
        }
    }

    /// Whether this entry has outlived [`PAGE_FD_CACHE_TTL`].
    ///
    /// foyer has no built-in TTL, so expiry is checked here and the entry is
    /// removed on the next access. Compared to moka — which also expired
    /// lazily but additionally swept in the background — an idle expired fd now
    /// lingers until it is either touched again or pushed out by capacity.
    ///
    /// That is acceptable: the fd count is still bounded by
    /// [`PAGE_FD_CACHE_MAX_CAPACITY`], so the `ulimit -n >= 10240` requirement
    /// is unchanged. Correctness never depended on the TTL either — `put` and
    /// `delete` remove the entry explicitly when a page file is replaced or
    /// unlinked. The TTL only reclaims fds that stopped being useful.
    #[inline]
    fn is_expired(&self) -> bool {
        self.inserted_at.elapsed() > PAGE_FD_CACHE_TTL
    }
}

/// io_uring-backed `PageStore`.
///
/// Disk layout is identical to `LocalPageStore`:
/// `<dir>/<page_size>/<bucket>/<file_id>/<page_index>`
///
/// so both backends can read each other's files (cross-backend compatibility).
///
/// # Read path (page fd cache + dir fd cache)
///
/// Each page is a separate file on disk. Two layers of caching are used:
///
/// 1. **Page fd cache** (Method E, keyed by `PageId`): on a hit, `get()` uses
///    the cached fd directly for `OP_READ` — **1 SQE, no openat**. This is the
///    hot path that eliminates VFS inode_lock contention entirely.
///
/// 2. **Dir fd cache** (Method A, keyed by `file_id`): on a page fd cache
///    miss, the directory fd cache provides `openat(dirfd, "page_index")`
///    (1-level path resolution) instead of `openat(AT_FDCWD, full_path)`
///    (4-level).
///
/// The previous per-page fd cache (old P4) used a `Mutex<LruCache>` with only
/// 1024 entries and achieved ~15% hit rate — it was removed because the
/// `Mutex` blocked tokio workers. Method E uses `DashMap` (lock-free reads)
/// with lazy TTL-based eviction, avoiding both issues.
pub struct UringPageStore {
    /// Root directory: `<dir>/<page_size>`.
    root: PathBuf,
    #[allow(dead_code)]
    page_size: u64,
    /// Directory fd cache: `file_id → DirFdEntry`.
    /// DashMap provides lock-free reads (shard-level RwLock, read-optimised).
    dir_fd_cache: DashMap<Arc<str>, DirFdEntry>,
    /// When `true`, the read path (dir open / page openat / read) uses
    /// synchronous libc syscalls (`open`/`openat`/`pread`) on the calling
    /// thread instead of io_uring SQE/CQE.
    ///
    /// Intended for complex analytical workloads where io_uring
    /// underperforms plain `pread`. The calling tokio worker is blocked
    /// for the duration of the local disk read, so this is opt-in
    /// (`client_cache_sync_read_enabled`, default `false`) and only
    /// recommended when the cache directory sits on local NVMe and the
    /// working set mostly fits the OS page cache. Write/delete paths
    /// always stay on io_uring.
    pread_enabled: bool,
}

impl UringPageStore {
    /// Create the store and its root directory (io_uring read mode).
    ///
    /// Directory creation uses `tokio::fs` (not on the hot path).
    pub async fn create(dir: &Path, page_size: u64) -> Result<Self> {
        Self::create_with_pread(dir, page_size, false).await
    }

    /// Create the store, selecting the read-mode: `pread_enabled = true`
    /// routes reads through synchronous `pread` on the calling thread;
    /// `false` keeps the io_uring SQE/CQE path.
    pub async fn create_with_pread(
        dir: &Path,
        page_size: u64,
        pread_enabled: bool,
    ) -> Result<Self> {
        let root = dir.join(page_size.to_string());
        tokio::fs::create_dir_all(&root)
            .await
            .map_err(|e| io_error(format!("create uring cache dir {}", root.display()), e))?;
        Ok(Self {
            root,
            page_size,
            dir_fd_cache: DashMap::new(),
            pread_enabled,
        })
    }

    /// Page file path: `<root>/<bucket>/<file_id>/<page_index>`.
    /// Identical to `LocalPageStore::page_path`.
    fn page_path(&self, page_id: &PageId) -> PathBuf {
        let bucket = hash_file_id(&page_id.file_id) % NUM_BUCKETS;
        self.root
            .join(bucket.to_string())
            .join(page_id.file_id.as_ref())
            .join(page_id.page_index.to_string())
    }

    /// Parent directory path for a file's pages: `<root>/<bucket>/<file_id>/`.
    fn file_dir_path(&self, file_id: &str) -> PathBuf {
        let bucket = hash_file_id(file_id) % NUM_BUCKETS;
        self.root.join(bucket.to_string()).join(file_id)
    }

    /// Identity sidecar path: `<root>/<bucket>/<file_id>/.identity`.
    /// Identical to `LocalPageStore::identity_path`.
    fn identity_path(&self, file_id: &str) -> PathBuf {
        let bucket = hash_file_id(file_id) % NUM_BUCKETS;
        self.root
            .join(bucket.to_string())
            .join(file_id)
            .join(IDENTITY_FILE)
    }

    // ── Directory fd cache ────────────────────────────────────

    /// Get the parent directory fd for `file_id`, caching it in the lock-free
    /// `DashMap` for subsequent calls.
    ///
    /// **Hot path**: on a cache hit this is a single `DashMap::get` (shard
    /// read-lock, ~10 ns). On a miss it pays one `OP_OPENAT` for the
    /// directory (3-level path), but all subsequent pages of the same file
    /// reuse the cached fd.
    ///
    /// Returns `Ok(dirfd)` on success, or `Err(NotFound)` if the directory
    /// does not exist (meaning no pages are cached for this file).
    /// Returns the owning handle rather than a bare `RawFd`.
    ///
    /// The `DashMap` read guard is released when this function returns, so a
    /// concurrent `maybe_cleanup_dir_fd_cache` may evict the entry immediately
    /// afterwards. Since the entry owns the fd, that would close it while the
    /// caller was still using it. Handing back the `Arc` keeps it alive for as
    /// long as the caller needs it.
    async fn get_dir_fd(&self, file_id: &Arc<str>) -> std::io::Result<Arc<File>> {
        // 1) Lock-free read — the common case (cache hit).
        //    `DashMap::get` takes a per-shard read guard (~10 ns).
        //    The `AtomicU64::store` updates `last_access` without upgrading
        //    to a write guard — this is the key to not blocking tokio workers.
        if let Some(entry) = self.dir_fd_cache.get(file_id) {
            entry.last_access.store(now_nanos(), Ordering::Relaxed);
            return Ok(Arc::clone(&entry.dir));
        }

        // 2) Cache miss — open the directory via io_uring OP_OPENAT.
        let dir_path = self.file_dir_path(file_id);
        let dirfd = self
            .open_fd(&dir_path, libc::O_RDONLY | libc::O_DIRECTORY)
            .await?;
        // OwnedFd -> File is a safe, infallible conversion; no `unsafe` and no
        // window where the fd has no owner.
        let dir_file = File::from(dirfd);

        // 3) Insert. If another thread won the race, close our redundant fd.
        //    `entry()` takes a per-shard write guard only briefly — this is
        //    the cold path (first access to a file), so it does not contend.
        match self.dir_fd_cache.entry(file_id.clone()) {
            dashmap::mapref::entry::Entry::Occupied(existing) => {
                // Another thread won the race. Dropping our `File` closes our
                // redundant fd — no manual close, so no way to close the wrong
                // one.
                drop(dir_file);
                Ok(Arc::clone(&existing.get().dir))
            }
            dashmap::mapref::entry::Entry::Vacant(vacant) => {
                let dir = Arc::new(dir_file);
                let handle = Arc::clone(&dir);
                vacant.insert(DirFdEntry {
                    dir,
                    last_access: AtomicU64::new(now_nanos()),
                });
                // Best-effort cleanup if the cache has grown large. Safe to run
                // here: `handle` keeps our own fd alive even if this very entry
                // is chosen for eviction.
                self.maybe_cleanup_dir_fd_cache();
                Ok(handle)
            }
        }
    }

    /// Best-effort cleanup of stale directory fd cache entries.
    ///
    /// Called lazily when the cache exceeds `DIR_FD_CACHE_SOFT_CAP`. Removes
    /// entries whose `last_access` is older than `DIR_FD_TTL`. This prevents
    /// unbounded growth from many distinct files without blocking the hot
    /// path (cleanup runs only on the rare size-threshold breach).
    fn maybe_cleanup_dir_fd_cache(&self) {
        if self.dir_fd_cache.len() <= DIR_FD_CACHE_SOFT_CAP {
            return;
        }
        let now = now_nanos();
        let ttl_nanos = DIR_FD_TTL.as_nanos() as u64;
        // Collect stale keys first, then remove — avoids holding DashMap
        // write guards during iteration.
        let stale: Vec<Arc<str>> = self
            .dir_fd_cache
            .iter()
            .filter(|entry| {
                now.saturating_sub(entry.last_access.load(Ordering::Relaxed)) > ttl_nanos
            })
            .map(|entry| entry.key().clone())
            .collect();
        for key in stale {
            self.dir_fd_cache.remove(&key);
        }
    }

    // ── io_uring operation helpers ───────────────────────────

    /// Build a NUL-terminated path buffer for `OP_OPENAT` / `OP_UNLINKAT`.
    ///
    /// io_uring's `openat`/`unlinkat` SQEs take a C string pointer; without the
    /// trailing `\0` the kernel reads past the buffer (undefined behavior /
    /// open failures). Always use [`CString::to_bytes_with_nul`].
    fn path_buffer_with_nul(path: &str) -> std::io::Result<BytesMut> {
        let cstring = CString::new(path)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
        Ok(BytesMut::from(cstring.to_bytes_with_nul()))
    }

    /// Asynchronously open a file via `OP_OPENAT` — zero `spawn_blocking`.
    ///
    /// `flags` should include `O_RDONLY` / `O_WRONLY` etc. `O_CLOEXEC` is
    /// added automatically. Returns the raw fd.
    ///
    /// When `pread_enabled` is set, opens synchronously via `open(2)` on
    /// the calling thread instead (no SQE/CQE round-trip).
    /// Open `path`, returning an owning handle.
    ///
    /// Returns `OwnedFd` rather than `RawFd` deliberately: with a bare fd, the
    /// obligation to close it exactly once lives only in comments, and has to be
    /// re-established at every early return. `OwnedFd` moves that into the type
    /// system.
    async fn open_fd(&self, path: &Path, flags: i32) -> std::io::Result<OwnedFd> {
        let buffer = Self::path_buffer_with_nul(&path.to_string_lossy())?;

        if self.pread_enabled {
            // SAFETY: `buffer` is NUL-terminated by `path_buffer_with_nul`.
            // O_CLOEXEC matches the io_uring open path (driver.rs). No
            // `mode` argument is passed because the read path never uses
            // O_CREAT here.
            let fd = unsafe {
                libc::open(
                    buffer.as_ptr() as *const libc::c_char,
                    flags | libc::O_CLOEXEC,
                )
            };
            return if fd < 0 {
                Err(std::io::Error::last_os_error())
            } else {
                // SAFETY: `fd` was just returned by a successful `open`/`openat`
                // and is not owned anywhere else.
                Ok(unsafe { OwnedFd::from_raw_fd(fd) })
            };
        }

        let request = Arc::new(IoRequest {
            fd: libc::AT_FDCWD,
            offset: 0,
            length: 0,
            op_type: UringOpType::OpenAt,
            open_flags: flags,
            state: std::sync::Mutex::new(RequestState {
                completed: false,
                waker: None,
                err: None,
                buffer,
                bytes_transferred: 0,
                consumed: false,
                result_code: 0,
            }),
        });

        submit_request(Arc::clone(&request));
        let (result, _bytes) =
            match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                Ok(res) => res,
                Err(_) => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "io_uring open timed out",
                    ));
                }
            };

        if result < 0 {
            Err(std::io::Error::from_raw_os_error(-result))
        } else {
            // A successful `OP_OPENAT` cannot yield 0/1/2 while the standard
            // streams are open, so such a value means an error code leaked
            // into the success path.
            //
            // Checked in release builds too, on purpose: taking ownership of a
            // standard stream closes it, and the process then aborts far away
            // from here with *no* diagnostic at all — the abort message is
            // written to the very fd that was destroyed. Failing loudly right
            // at the source is worth one comparison per cache miss.
            assert!(
                result > 2,
                "OP_OPENAT returned fd={result}: an error code leaked into the success path"
            );
            // SAFETY: a non-negative CQE result from OP_OPENAT is a fresh fd
            // owned by nothing else.
            Ok(unsafe { OwnedFd::from_raw_fd(result as RawFd) })
        }
    }

    /// Asynchronously open a file relative to a directory fd via `OP_OPENAT`.
    ///
    /// Unlike `open_fd` (which uses `AT_FDCWD` + full path), this uses a
    /// cached directory fd (`dirfd`) and a relative name (e.g. `"42"` for
    /// page index 42). The kernel resolves only 1 path component instead of
    /// 4, avoiding 3 levels of dcache/inode lock contention under concurrency.
    ///
    /// `flags` should include `O_RDONLY` etc. `O_CLOEXEC` is added automatically.
    ///
    /// When `pread_enabled` is set, opens synchronously via `openat(2)` on
    /// the calling thread instead (no SQE/CQE round-trip).
    /// `openat(dirfd, name)`, returning an owning handle. See [`Self::open_fd`]
    /// for why this is `OwnedFd`.
    async fn openat_relative(
        &self,
        dirfd: RawFd,
        name: &str,
        flags: i32,
    ) -> std::io::Result<OwnedFd> {
        let buffer = Self::path_buffer_with_nul(name)?;

        if self.pread_enabled {
            // SAFETY: `buffer` is NUL-terminated by `path_buffer_with_nul`;
            // `dirfd` is a valid directory fd from the dir fd cache.
            // O_CLOEXEC matches the io_uring open path (driver.rs). No
            // `mode` argument is passed because the read path never uses
            // O_CREAT here.
            let fd = unsafe {
                libc::openat(
                    dirfd,
                    buffer.as_ptr() as *const libc::c_char,
                    flags | libc::O_CLOEXEC,
                )
            };
            return if fd < 0 {
                Err(std::io::Error::last_os_error())
            } else {
                // SAFETY: `fd` was just returned by a successful `open`/`openat`
                // and is not owned anywhere else.
                Ok(unsafe { OwnedFd::from_raw_fd(fd) })
            };
        }

        let request = Arc::new(IoRequest {
            fd: dirfd,
            offset: 0,
            length: 0,
            op_type: UringOpType::OpenAt,
            open_flags: flags,
            state: std::sync::Mutex::new(RequestState {
                completed: false,
                waker: None,
                err: None,
                buffer,
                bytes_transferred: 0,
                consumed: false,
                result_code: 0,
            }),
        });

        submit_request(Arc::clone(&request));
        let (result, _bytes) =
            match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                Ok(res) => res,
                Err(_) => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "io_uring openat_relative timed out",
                    ));
                }
            };

        if result < 0 {
            Err(std::io::Error::from_raw_os_error(-result))
        } else {
            // A successful `OP_OPENAT` cannot yield 0/1/2 while the standard
            // streams are open, so such a value means an error code leaked
            // into the success path.
            //
            // Checked in release builds too, on purpose: taking ownership of a
            // standard stream closes it, and the process then aborts far away
            // from here with *no* diagnostic at all — the abort message is
            // written to the very fd that was destroyed. Failing loudly right
            // at the source is worth one comparison per cache miss.
            assert!(
                result > 2,
                "OP_OPENAT returned fd={result}: an error code leaked into the success path"
            );
            // SAFETY: a non-negative CQE result from OP_OPENAT is a fresh fd
            // owned by nothing else.
            Ok(unsafe { OwnedFd::from_raw_fd(result as RawFd) })
        }
    }

    /// Close an fd via `OP_CLOSE` — fire-and-forget (no await).
    ///
    /// The SQE is submitted to the io_uring ring; the kernel closes the fd
    /// asynchronously. This eliminates the third round-trip in `get()`
    /// (H3 fix). If submission fails (channel full or disconnected),
    /// falls back to synchronous `libc::close` to prevent fd leaks (H5 fix).
    ///
    /// Takes `OwnedFd`, not `RawFd`. With a raw fd this function could be called
    /// for an fd the caller did not own, or twice for the same one, and neither
    /// mistake is visible at the call site — the fd number stays valid because
    /// the kernel has already reassigned it to something else, so the close
    /// "succeeds" while destroying an unrelated file handle. Requiring ownership
    /// makes both mistakes fail to compile.
    fn close_fd_background(&self, fd: OwnedFd) {
        // Consume ownership before releasing the raw number: from here on this
        // fd belongs to the kernel (via the SQE) or to the fallback close
        // below, and no caller can reach it again.
        let fd = fd.into_raw_fd();
        let request = Arc::new(IoRequest {
            fd,
            offset: 0,
            length: 0,
            op_type: UringOpType::Close,
            open_flags: 0,
            state: std::sync::Mutex::new(RequestState {
                completed: false,
                waker: None,
                err: None,
                buffer: BytesMut::new(),
                bytes_transferred: 0,
                consumed: false,
                result_code: 0,
            }),
        });
        if !try_submit_request(request) {
            // Channel full/disconnected — close synchronously to prevent fd leak.
            unsafe { libc::close(fd) };
        }
    }

    /// Asynchronously unlink a file via `OP_UNLINKAT`.
    /// Returns `Ok(())` if the file does not exist (idempotent).
    async fn unlink_path(&self, path: &Path) -> std::io::Result<()> {
        let buffer = Self::path_buffer_with_nul(&path.to_string_lossy())?;

        let request = Arc::new(IoRequest {
            fd: libc::AT_FDCWD,
            offset: 0,
            length: 0,
            op_type: UringOpType::UnlinkAt,
            open_flags: 0,
            state: std::sync::Mutex::new(RequestState {
                completed: false,
                waker: None,
                err: None,
                buffer,
                bytes_transferred: 0,
                consumed: false,
                result_code: 0,
            }),
        });

        submit_request(Arc::clone(&request));
        let (result, _) =
            match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                Ok(res) => res,
                Err(_) => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "io_uring unlink timed out",
                    ));
                }
            };

        if result < 0 {
            let e = std::io::Error::from_raw_os_error(-result);
            if e.kind() == std::io::ErrorKind::NotFound {
                return Ok(()); // idempotent
            }
            return Err(e);
        }
        Ok(())
    }

    /// Build a read request for an already-opened fd.
    fn new_read_request(fd: RawFd, offset: usize, len: usize) -> Arc<IoRequest> {
        let mut buffer = BytesMut::with_capacity(len);
        // SAFETY: buffer has capacity for `len` bytes; io_uring writes into it
        // before the future exposes it as `Bytes`.
        unsafe {
            buffer.set_len(len);
        }

        Arc::new(IoRequest {
            fd,
            offset: offset as u64,
            length: len,
            op_type: UringOpType::Read,
            open_flags: 0,
            state: std::sync::Mutex::new(RequestState {
                completed: false,
                waker: None,
                err: None,
                buffer,
                bytes_transferred: 0,
                consumed: false,
                result_code: 0,
            }),
        })
    }

    /// Await a submitted read request and return the kernel-filled buffer.
    async fn wait_read_request(request: Arc<IoRequest>) -> std::io::Result<Bytes> {
        let (result, read_bytes) =
            match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                Ok(res) => res,
                Err(_) => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "io_uring read timed out",
                    ));
                }
            };

        if result < 0 {
            Err(std::io::Error::from_raw_os_error(-result))
        } else {
            Ok(read_bytes)
        }
    }

    /// Read from an already-opened fd via io_uring `OP_READ` and return the
    /// kernel-filled buffer directly.
    ///
    /// When `pread_enabled` is set, reads synchronously via `pread(2)` on
    /// the calling thread instead (no SQE/CQE round-trip).
    async fn read_with_fd(&self, fd: RawFd, offset: usize, len: usize) -> std::io::Result<Bytes> {
        if self.pread_enabled {
            return Self::pread_read_sync(fd, offset, len);
        }
        let request = Self::new_read_request(fd, offset, len);
        submit_request(Arc::clone(&request));
        Self::wait_read_request(request).await
    }

    /// Synchronously read up to `len` bytes from `fd` at `offset` via
    /// `pread(2)`, looping over short reads and `EINTR` until the buffer
    /// is full or EOF. Returns the buffer truncated to the bytes actually
    /// read (empty at EOF) — the same contract as the io_uring read path
    /// (short-read retry + EOF truncate in `driver.rs`).
    ///
    /// Runs on the calling thread: the caller (a tokio worker) is blocked
    /// for the duration of the syscall. Only used when `pread_enabled`
    /// is set.
    ///
    /// `pread` uses an explicit offset and never mutates the fd's file
    /// position, so concurrent `pread` calls on the same fd from multiple
    /// threads are safe (POSIX).
    fn pread_read_sync(fd: RawFd, offset: usize, len: usize) -> std::io::Result<Bytes> {
        // Keep the buffer's length at zero and write through the spare
        // capacity so we never expose uninitialised bytes as an initialised
        // `&[u8]`. `set_len` is advanced to `filled` only after each `pread`
        // reports how many bytes it initialised, so the value frozen as
        // `Bytes` covers exactly the initialised prefix (no `zeroed`/`memset`
        // and no `truncate` needed).
        let mut buffer = BytesMut::with_capacity(len);

        let mut filled = 0usize;
        while filled < len {
            // `spare_capacity_mut()` returns the uninitialised tail starting at
            // the current length (`filled`), typed as `[MaybeUninit<u8>]`, so
            // no reference to uninitialised memory is ever created.
            let spare = buffer.spare_capacity_mut();
            // SAFETY: `spare` points to at least `len - filled` writable bytes
            // (capacity is `len`, current length is `filled`); we ask `pread`
            // for at most `len - filled` bytes so the write stays in bounds.
            // `fd` is a valid open fd (kept alive by the caller's
            // `Arc<PageFdEntry>` on the hot path, or freshly opened on the
            // cold path).
            let n = unsafe {
                libc::pread(
                    fd,
                    spare.as_mut_ptr() as *mut libc::c_void,
                    len - filled,
                    (offset + filled) as libc::off_t,
                )
            };
            if n < 0 {
                let e = std::io::Error::last_os_error();
                if e.kind() == std::io::ErrorKind::Interrupted {
                    continue; // EINTR — retry
                }
                return Err(e);
            }
            if n == 0 {
                break; // EOF — partial read complete
            }
            filled += n as usize;
            // SAFETY: `pread` initialised the `[prev_len, filled)` bytes and the
            // `[0, prev_len)` prefix was initialised by earlier iterations, so
            // the whole `[0, filled)` range is initialised. `filled <= len`
            // (loop guard + `pread` never returns more than requested), so
            // `filled <= capacity`, satisfying `set_len`'s contract.
            unsafe {
                buffer.set_len(filled);
            }
        }
        Ok(buffer.freeze())
    }

    /// Compatibility wrapper for existing tests/callers that provide output
    /// buffers. The cache hot path now gets concurrency from
    /// `LocalCacheManager::get_batch_bytes` + `join_all`, so we keep this
    /// method small and avoid a second, unused batch API layer.
    pub async fn get_batch(
        &self,
        requests: Vec<(PageId, usize, usize)>,
        results: Vec<&mut [u8]>,
    ) -> Result<()> {
        assert_eq!(
            requests.len(),
            results.len(),
            "requests and results must have the same length"
        );
        for ((page_id, offset, len), dst) in requests.into_iter().zip(results.into_iter()) {
            let bytes = self.get_bytes(&page_id, offset, len).await?;
            let n = bytes.len().min(dst.len());
            if n > 0 {
                dst[..n].copy_from_slice(&bytes[..n]);
            }
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl PageStore for UringPageStore {
    /// Read a page via directory fd cache + `OP_OPENAT` + `OP_READ` + `OP_CLOSE`.
    ///
    /// The directory fd cache (keyed by `file_id`) eliminates 3 of 4 levels
    /// of kernel VFS path resolution on every `get`. On a cache hit:
    ///
    /// 1. `DashMap::get(file_id)` → returns cached `dirfd` (lock-free read,
    ///    ~10 ns, no tokio worker blocking).
    /// 2. `OP_OPENAT(dirfd, "page_index")` → 1-level path resolution.
    /// 3. `OP_READ(fd, offset, len)` → pread via io_uring.
    /// 4. `OP_CLOSE(fd)` → fire-and-forget.
    ///
    /// On a dir fd cache miss (first read of a new file), step 1 falls back
    /// to `OP_OPENAT(AT_FDCWD, "<root>/<bucket>/<file_id>")` (3-level), but
    /// all subsequent reads of any page in the same file reuse the cached
    /// `dirfd` — giving ~100% hit rate for workloads that read multiple pages
    /// per file (which is always the case for Lance columnar reads).
    ///
    /// **Concurrency**: `DashMap` uses per-shard RwLocks. The hot path
    /// (`get` → `get_dir_fd` cache hit) takes only a **read** guard and
    /// updates `last_access` via `AtomicU64` — no write lock, no tokio
    /// worker blocking. The cold path (cache miss → `entry()`) takes a brief
    /// write guard but only on the first access to each file.
    ///
    /// See the concurrent uring analysis for design context.
    /// for the concurrency analysis that motivated this design.
    /// Read a page via **page fd cache** (Method E) → dir fd cache fallback.
    ///
    /// **Hot path (page fd cache hit)** — 1 SQE:
    /// 1. `DashMap::get(page_id)` → returns cached `fd` (lock-free read).
    /// 2. `OP_READ(fd, offset, len)` → pread via io_uring.
    ///
    /// No `OP_OPENAT`, no `OP_CLOSE` — the fd is reused across reads. This
    /// eliminates VFS inode_lock contention entirely on the hot path.
    ///
    /// **Cold path (page fd cache miss)** — 3 SQEs (same as Method A):
    /// 1. `get_dir_fd(file_id)` → cached dirfd (1-level openat on miss).
    /// 2. `OP_OPENAT(dirfd, "page_index")` → 1-level path resolution.
    /// 3. `OP_READ(fd, offset, len)` → pread via io_uring.
    /// 4. Insert into `page_fd_cache` for future hits.
    /// 5. `OP_CLOSE` is **not** done — the fd is kept in the cache.
    ///
    /// **Concurrency**: same lock-free `DashMap` read guard pattern as
    /// `get_dir_fd`. `last_access` updated via `AtomicU64` — no write lock.
    ///
    /// See the concurrent uring analysis for design context.
    ///  and Method E for the analysis that motivated this design.
    async fn get(&self, page_id: &PageId, offset: usize, dst: &mut [u8]) -> Result<usize> {
        let bytes = self.get_bytes(page_id, offset, dst.len()).await?;
        let n = bytes.len().min(dst.len());
        if n > 0 {
            dst[..n].copy_from_slice(&bytes[..n]);
        }
        Ok(n)
    }

    async fn get_bytes(&self, page_id: &PageId, offset: usize, len: usize) -> Result<Bytes> {
        if len == 0 {
            return Ok(Bytes::new());
        }

        // ── Hot path: page fd cache hit → 1 SQE (OP_READ only) ───────
        //
        // No `.await` here: foyer's in-memory cache is synchronous, so the hit
        // path no longer builds a future state machine just to look up an fd.
        //
        // NOTE: this must be `get`, never `touch`. `touch` calls
        // `Eviction::acquire` without the paired `release`, which under
        // `LruConfig` pins the record forever
        // (foyer-memory-0.22.3 `src/raw.rs:836-851` is the only `release` site).
        // S3-FIFO is not affected, but using `get` keeps this correct for any
        // policy — and we need the value anyway.
        if let Some(cached) = PAGE_FD_CACHE.get(page_id) {
            // Clone the `Arc` out of the entry so the `CacheEntry` guard can be
            // dropped before the read: holding it across `.await` would pin the
            // record under a pinning policy.
            let entry = cached.value().clone();
            drop(cached);

            if entry.is_expired() {
                // Lazy TTL: drop the stale entry and fall through to the miss
                // path, which re-opens the file and re-inserts a fresh entry.
                PAGE_FD_CACHE.remove(page_id);
                counter(mn::CLIENT_CACHE_PAGE_FD_TTL_EXPIRED).inc(1);
            } else {
                let fd = entry.fd;
                // `entry: Arc<PageFdEntry>` keeps the underlying `Arc<File>`
                // alive for the duration of the read, so the fd stays valid
                // even if the cache evicts the entry concurrently.
                let _entry = entry;

                return match self.read_with_fd(fd, offset, len).await {
                    Ok(bytes) => Ok(bytes),
                    Err(e) => {
                        PAGE_FD_CACHE.remove(page_id);
                        if e.kind() == std::io::ErrorKind::NotFound {
                            Ok(Bytes::new())
                        } else {
                            Err(io_error("uring read (page fd cache hit)", e))
                        }
                    }
                };
            }
        }

        // ── Cold path: page fd cache miss → dir fd cache + openat + read ─
        //
        // `dir_handle` is held until after the `openat` below completes: it owns
        // the directory fd, so letting it drop earlier would allow a concurrent
        // cleanup to close the fd we are resolving against.
        let dir_handle = match self.get_dir_fd(&page_id.file_id).await {
            Ok(h) => h,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Bytes::new()),
            Err(e) => return Err(io_error("uring open dir", e)),
        };
        let dirfd = dir_handle.as_raw_fd();

        let page_name = page_id.page_index.to_string();
        let fd = match self
            .openat_relative(dirfd, &page_name, libc::O_RDONLY)
            .await
        {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Bytes::new()),
            Err(e) => return Err(io_error("uring open page", e)),
        };
        // The page fd is open now, so the directory fd is no longer needed.
        drop(dir_handle);

        let read_bytes = match self.read_with_fd(fd.as_raw_fd(), offset, len).await {
            Ok(bytes) => bytes,
            Err(e) => {
                // Consumes the fd; the success path below cannot reach it.
                self.close_fd_background(fd);
                if e.kind() == std::io::ErrorKind::NotFound {
                    return Ok(Bytes::new());
                }
                return Err(io_error("uring read", e));
            }
        };

        // OwnedFd -> File is safe and infallible. The fd is closed when the last
        // `Arc<File>` goes away, i.e. once the cache entry is evicted and no
        // reader still holds it.
        let file = File::from(fd);
        // The returned `CacheEntry` is dropped immediately — we hold no
        // reference to it, so foyer is free to reclaim the record.
        drop(PAGE_FD_CACHE.insert(page_id.clone(), Arc::new(PageFdEntry::new(file))));

        Ok(read_bytes)
    }

    /// Write a page — `OP_OPENAT` + `OP_WRITE` + `OP_CLOSE` + `rename`.
    ///
    /// Uses the atomic `tmp + rename` pattern identical to `LocalPageStore`.
    /// `rename` uses `std::fs::rename` (sync) because it is NOT on the cache
    /// hit hot path — it only runs on cache miss fill.
    ///
    /// See design .
    async fn put(&self, page_id: &PageId, page: &[u8]) -> Result<()> {
        let final_path = self.page_path(page_id);
        let parent = final_path
            .parent()
            .expect("page path always has a parent")
            .to_path_buf();

        // Directory creation is not on the hot path — use tokio::fs.
        tokio::fs::create_dir_all(&parent)
            .await
            .map_err(|e| io_error("create page dir", e))?;

        let tmp_path = parent.join(format!(
            "{}.tmp-{}",
            page_id.page_index,
            uuid::Uuid::new_v4()
        ));
        let tmp_buffer = Self::path_buffer_with_nul(&tmp_path.to_string_lossy())
            .map_err(|e| io_error("cstring", e))?;

        // 1) OP_OPENAT (O_WRONLY | O_CREAT | O_TRUNC)
        let fd = {
            let request = Arc::new(IoRequest {
                fd: libc::AT_FDCWD,
                offset: 0,
                length: 0,
                op_type: UringOpType::OpenAt,
                open_flags: libc::O_WRONLY | libc::O_CREAT | libc::O_TRUNC,
                state: std::sync::Mutex::new(RequestState {
                    completed: false,
                    waker: None,
                    err: None,
                    buffer: tmp_buffer,
                    bytes_transferred: 0,
                    consumed: false,
                    result_code: 0,
                }),
            });
            submit_request(Arc::clone(&request));
            let (result, _) =
                match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                    Ok(res) => res,
                    Err(_) => {
                        return Err(io_error(
                            "uring open tmp timeout",
                            std::io::Error::new(
                                std::io::ErrorKind::TimedOut,
                                "io_uring open timed out",
                            ),
                        ));
                    }
                };
            if result < 0 {
                let e = std::io::Error::from_raw_os_error(-result);
                return Err(io_error("uring open tmp", e));
            }
            // See `openat_relative` for why 0/1/2 cannot be a real fd here.
            assert!(
                result > 2,
                "OP_OPENAT returned fd={result}: an error code leaked into the success path"
            );
            // SAFETY: a non-negative CQE result from OP_OPENAT is a fresh fd
            // owned by nothing else. Wrapping it here means each of the three
            // exits below (write timeout, write error, success) must *move* it,
            // so exactly one close can happen.
            unsafe { OwnedFd::from_raw_fd(result as RawFd) }
        };

        // 2) OP_WRITE (entire page)
        {
            let request = Arc::new(IoRequest {
                fd: fd.as_raw_fd(),
                offset: 0,
                length: page.len(),
                op_type: UringOpType::Write,
                open_flags: 0,
                state: std::sync::Mutex::new(RequestState {
                    completed: false,
                    waker: None,
                    err: None,
                    buffer: BytesMut::from(page),
                    bytes_transferred: 0,
                    consumed: false,
                    result_code: 0,
                }),
            });
            submit_request(Arc::clone(&request));
            let (result, _) =
                match tokio::time::timeout(URING_OP_TIMEOUT, UringOpFuture { request }).await {
                    Ok(res) => res,
                    Err(_) => {
                        self.close_fd_background(fd);
                        return Err(io_error(
                            "uring write timeout",
                            std::io::Error::new(
                                std::io::ErrorKind::TimedOut,
                                "io_uring write timed out",
                            ),
                        ));
                    }
                };
            if result < 0 {
                self.close_fd_background(fd);
                let e = std::io::Error::from_raw_os_error(-result);
                return Err(io_error("uring write", e));
            }
        }

        // 3) OP_CLOSE — fire-and-forget (H3 fix).
        self.close_fd_background(fd);

        // 4) rename — async to avoid blocking the tokio worker (H2 fix).
        //    POSIX atomic rename on NVMe is ~5 µs.
        let rename_result = tokio::fs::rename(&tmp_path, &final_path).await;
        if rename_result.is_err() {
            // Best-effort cleanup of the temp file.
            let _ = tokio::fs::remove_file(&tmp_path).await;
        }
        rename_result.map_err(|e| io_error("rename temp page file", e))?;

        // Invalidate the global page fd cache entry — the atomic rename
        // replaced the page file, so any cached fd points to the old
        // inode. Must be removed so the next get() re-opens the new file.
        //
        // This, not the TTL, is what guarantees correctness after a rewrite.
        PAGE_FD_CACHE.remove(page_id);

        Ok(())
    }

    /// Delete a page — `OP_UNLINKAT`.
    ///
    /// The page file is unlinked relative to the full path (via `AT_FDCWD`).
    /// The directory fd cache entry for the file is **not** invalidated here
    /// because the directory itself still exists — only the page file inside
    /// it was removed. The dir fd remains valid for future `get` calls that
    /// read other pages of the same file.
    ///
    /// On Unix, unlinking a file with open fds is safe — the inode survives
    /// until all fds close, so any concurrent read completes normally.
    ///
    /// Compared to `LocalPageStore::delete` (1 × `spawn_blocking`), this uses
    /// 1 io_uring SQE. Returns `Ok(())` if the file does not exist.
    async fn delete(&self, page_id: &PageId) -> Result<()> {
        // Invalidate the global page fd cache entry BEFORE unlinking. The
        // fd in the cache points to the inode that will be unlinked — we
        // must close it so the inode can be freed. Concurrent reads using
        // the old fd are safe on Unix (inode survives until last close),
        // but we remove the cache entry so future reads go through the
        // miss path.
        PAGE_FD_CACHE.remove(page_id);

        let path = self.page_path(page_id);
        self.unlink_path(&path)
            .await
            .map_err(|e| io_error("uring unlink", e))?;
        Ok(())
    }

    // ── Identity sidecar (not on the hot path — uses tokio::fs) ──

    fn root_dir(&self) -> &Path {
        &self.root
    }

    async fn write_identity(&self, file_id: &str, length: i64, mtime: i64) -> Result<()> {
        let final_path = self.identity_path(file_id);
        let parent = final_path
            .parent()
            .expect("identity path always has a parent")
            .to_path_buf();
        tokio::fs::create_dir_all(&parent)
            .await
            .map_err(|e| io_error(format!("create identity dir {}", parent.display()), e))?;
        let tmp_path = parent.join(format!("{}.tmp-{}", IDENTITY_FILE, uuid::Uuid::new_v4()));
        let contents = format!("{length},{mtime}");
        let write_result = async {
            tokio::fs::write(&tmp_path, contents.as_bytes())
                .await
                .map_err(|e| io_error("write temp identity file", e))?;
            tokio::fs::rename(&tmp_path, &final_path)
                .await
                .map_err(|e| io_error("rename temp identity file", e))?;
            Ok::<(), crate::error::Error>(())
        }
        .await;
        if write_result.is_err() {
            let _ = tokio::fs::remove_file(&tmp_path).await;
        }
        write_result
    }

    async fn read_identity(&self, file_id: &str) -> Option<(i64, i64)> {
        let path = self.identity_path(file_id);
        let contents = tokio::fs::read_to_string(&path).await.ok()?;
        // Shared parser with LocalPageStore so both backends accept the same
        // sidecar format (and share unit-test coverage).
        LocalPageStore::parse_identity(&contents)
    }

    async fn delete_identity(&self, file_id: &str) -> Result<()> {
        let path = self.identity_path(file_id);
        match tokio::fs::remove_file(&path).await {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(io_error("delete identity file", e)),
        }
    }
}

#[cfg(test)]
mod tests {
    //! # No CI coverage
    //!
    //! Every test below is `#[ignore]`d, and **no CI job runs them**. Two
    //! separate reasons, both easy to mistake for coverage:
    //!
    //! * `ci.yml` runs `cargo nextest run --workspace --lib --tests`, which
    //!   skips `#[ignore]` by default.
    //! * `scripts/ci/run_rust_integration.sh` does pass `--ignored`, but only
    //!   to specific `--test <file>` targets, i.e. integration tests under
    //!   `tests/`. These live in `src/`, so `--lib` is never named. Nothing
    //!   under `tests/` touches the uring store either.
    //!
    //! The gap is not theoretical. A bug that turned an `ENOENT` errno into
    //! "fd 2", closed stderr, and aborted the process mid-suite was invisible
    //! to CI and only surfaced when the suite was run by hand on a host with
    //! io_uring. See the fix in `future.rs::poll`.
    //!
    //! Adding `cargo test --lib uring -- --ignored` to the integration script
    //! looks like the obvious closure, but it is **unverified**: the runner may
    //! not permit `OP_OPENAT` at all (hence the `#[ignore]` reason on each
    //! test). `temp_store` probes and returns `None` in that case, so the tests
    //! would skip silently and the job would stay green while covering
    //! nothing — worse than a known gap. Confirm the runner can actually do an
    //! io_uring OPENAT before relying on it.
    //!
    //! Until then, treat "these pass" as a claim that only holds where they
    //! were actually executed.

    use super::*;
    use crate::cache::store::uring::is_uring_available;

    /// Build a temp store, or skip when io_uring create-OPENAT is not usable.
    ///
    /// GitHub Actions often allows `IoUring::new` (and even a sync OPENAT
    /// probe) but rejects the same opcode on the background uring worker
    /// used by `put`. So we warm-up with a real `put` through the store
    /// and treat `PermissionDenied` as "skip this host".
    async fn temp_store(page_size: u64) -> Option<(UringPageStore, PathBuf)> {
        if !is_uring_available() {
            eprintln!(
                "skipping uring test: io_uring OPENAT not usable on this host \
                 (ring setup may still succeed)"
            );
            return None;
        }
        let base = std::env::temp_dir().join(format!("gfs_uring_test_{}", uuid::Uuid::new_v4()));
        let store = UringPageStore::create(&base, page_size).await.ok()?;

        // End-to-end gate: same code path as every failing CI test.
        let probe_id = PageId::new("__uring_probe__", 0);
        match store.put(&probe_id, b"probe").await {
            Ok(()) => {
                let _ = store.delete(&probe_id).await;
                Some((store, base))
            }
            Err(e) => {
                let msg = e.to_string();
                let is_perm = msg.contains("Operation not permitted")
                    || msg.contains("PermissionDenied")
                    || msg.contains("uring open tmp")
                    || msg.contains("uring open dir");
                if is_perm {
                    eprintln!(
                        "skipping uring test: put via io_uring returned EPERM \
                         (common on GitHub Actions / restricted seccomp): {msg}"
                    );
                } else {
                    eprintln!("skipping uring test: warm-up put failed: {msg}");
                }
                let _ = tokio::fs::remove_dir_all(&base).await;
                None
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_put_get_roundtrip() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("roundtrip", 0);
        let data = b"hello uring page cache".to_vec();

        store.put(&id, &data).await.unwrap();

        let mut dst = vec![0u8; data.len()];
        let n = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n, data.len());
        assert_eq!(&dst, &data);

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_with_offset() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("with-offset", 0);
        store.put(&id, b"0123456789").await.unwrap();

        let mut dst = vec![0u8; 4];
        let n = store.get(&id, 3, &mut dst).await.unwrap();
        assert_eq!(n, 4);
        assert_eq!(&dst, b"3456");

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_missing_returns_zero() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("missing", 0);
        let mut dst = vec![0u8; 8];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), 0);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_short_read_at_tail() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("short-read", 0);
        store.put(&id, b"abc").await.unwrap();

        // Ask for more than the page holds → fills only the available bytes.
        let mut dst = vec![0u8; 16];
        let n = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n, 3);
        assert_eq!(&dst[..3], b"abc");
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_delete_then_miss() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("delete-miss", 1);
        store.put(&id, b"data").await.unwrap();
        store.delete(&id).await.unwrap();

        let mut dst = vec![0u8; 4];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), 0);
        // Deleting again is a no-op.
        store.delete(&id).await.unwrap();
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_concurrent_get_same_page() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("conc-same-page", 0);
        let data = vec![0x42u8; 64];
        store.put(&id, &data).await.unwrap();

        // 32 concurrent reads of the same page.
        let store = Arc::new(store);
        let mut handles = Vec::new();
        for _ in 0..32 {
            let store = Arc::clone(&store);
            let id = id.clone();
            handles.push(tokio::spawn(async move {
                let mut dst = vec![0u8; 64];
                let n = store.get(&id, 0, &mut dst).await.unwrap();
                assert_eq!(n, 64);
                assert_eq!(dst, vec![0x42u8; 64]);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_repeated_reads() {
        // Repeated reads of the same page should all succeed. The directory
        // fd cache means the first read opens the directory, subsequent reads
        // reuse the cached dirfd (lock-free DashMap lookup) + 1-level
        // openat for the page file.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("repeat", 0);
        store.put(&id, b"repeated-read-data").await.unwrap();

        for _ in 0..10 {
            let mut dst = vec![0u8; 18];
            let n = store.get(&id, 0, &mut dst).await.unwrap();
            assert_eq!(n, 18);
            assert_eq!(&dst, b"repeated-read-data");
        }
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_dir_fd_cache_reused_across_pages() {
        // Multiple pages of the same file share the cached directory fd.
        // After the first get, subsequent gets for different page indices
        // reuse the cached dirfd — only 1-level openat is needed.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id0 = unique_id("batch-multi", 0);
        let id1 = unique_id("batch-multi", 1);
        let id2 = unique_id("batch-multi", 2);

        store.put(&id0, b"page-zero-data!!").await.unwrap();
        store.put(&id1, b"page-one-data!!!").await.unwrap();
        store.put(&id2, b"page-two-data!!!").await.unwrap();

        // Read all three pages — the dir fd cache should be populated on
        // the first get and reused for the other two.
        let mut dst = vec![0u8; 16];
        assert_eq!(store.get(&id0, 0, &mut dst).await.unwrap(), 16);
        assert_eq!(&dst, b"page-zero-data!!");
        assert_eq!(store.get(&id1, 0, &mut dst).await.unwrap(), 16);
        assert_eq!(&dst, b"page-one-data!!!");
        assert_eq!(store.get(&id2, 0, &mut dst).await.unwrap(), 16);
        assert_eq!(&dst, b"page-two-data!!!");

        // Verify the dir fd cache has exactly one entry for this file.
        assert_eq!(store.dir_fd_cache.len(), 1);

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_dir_fd_cache_concurrent_different_pages() {
        // 32 concurrent reads of different pages from the same file.
        // This tests the DashMap's concurrency: all tasks share one dirfd
        // entry (lock-free reads), each does its own 1-level openat + read
        // + close. No tokio worker should be blocked.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let file_id: Arc<str> = Arc::from("file-conc-multi");

        // Write 32 pages.
        for i in 0..32u64 {
            let id = PageId::new(file_id.clone(), i);
            store.put(&id, &[i as u8; 8]).await.unwrap();
        }

        let store = Arc::new(store);
        let mut handles = Vec::new();
        for i in 0..32u64 {
            let store = Arc::clone(&store);
            let file_id = file_id.clone();
            handles.push(tokio::spawn(async move {
                let id = PageId::new(file_id, i);
                let mut dst = vec![0u8; 8];
                let n = store.get(&id, 0, &mut dst).await.unwrap();
                assert_eq!(n, 8);
                assert_eq!(dst, vec![i as u8; 8]);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }

        // After 32 concurrent reads, the dir fd cache should still have
        // exactly one entry for this file (all tasks shared it).
        assert_eq!(store.dir_fd_cache.len(), 1);

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_batch_concurrent() {
        // Phase D: verify get_batch reads multiple pages concurrently and
        // returns the correct data for each.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let file_id: Arc<str> = Arc::from("file-batch");

        // Write 8 pages of 16 bytes each with distinct content.
        let n_pages = 8u64;
        for i in 0..n_pages {
            let id = PageId::new(file_id.clone(), i);
            let data: Vec<u8> = (0..16u8).map(|b| b.wrapping_add(i as u8)).collect();
            store.put(&id, &data).await.unwrap();
        }

        // Batch read all 8 pages.
        let requests: Vec<(PageId, usize, usize)> = (0..n_pages)
            .map(|i| (PageId::new(file_id.clone(), i), 0, 16))
            .collect();
        let mut bufs: Vec<Vec<u8>> = (0..n_pages).map(|_| vec![0u8; 16]).collect();
        let results: Vec<&mut [u8]> = bufs.iter_mut().map(|b| b.as_mut_slice()).collect();

        store
            .get_batch(requests, results)
            .await
            .expect("batch read should succeed");

        // Verify each result matches the expected data.
        for (i, buf) in bufs.iter().enumerate() {
            let expected: Vec<u8> = (0..16u8).map(|b| b.wrapping_add(i as u8)).collect();
            assert_eq!(buf, &expected, "page {i} data mismatch");
        }

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_batch_multi_file() {
        // Phase D: verify get_batch groups requests by file_id and uses the
        // per-file dirfd cache.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let file_a: Arc<str> = Arc::from("file-a");
        let file_b: Arc<str> = Arc::from("file-b");

        for i in 0..3u64 {
            store
                .put(&PageId::new(file_a.clone(), i), &[0xA0 + i as u8; 8])
                .await
                .unwrap();
            store
                .put(&PageId::new(file_b.clone(), i), &[0xB0 + i as u8; 8])
                .await
                .unwrap();
        }

        // Mixed batch: 2 pages from file_a, 2 from file_b.
        let requests = vec![
            (PageId::new(file_a.clone(), 0), 0, 8),
            (PageId::new(file_b.clone(), 0), 0, 8),
            (PageId::new(file_a.clone(), 1), 0, 8),
            (PageId::new(file_b.clone(), 1), 0, 8),
        ];
        let mut bufs: Vec<Vec<u8>> = (0..4).map(|_| vec![0u8; 8]).collect();
        let results: Vec<&mut [u8]> = bufs.iter_mut().map(|b| b.as_mut_slice()).collect();

        store.get_batch(requests, results).await.unwrap();

        assert_eq!(
            bufs[0],
            vec![0xA0, 0xA0, 0xA0, 0xA0, 0xA0, 0xA0, 0xA0, 0xA0]
        );
        assert_eq!(
            bufs[1],
            vec![0xB0, 0xB0, 0xB0, 0xB0, 0xB0, 0xB0, 0xB0, 0xB0]
        );
        assert_eq!(
            bufs[2],
            vec![0xA1, 0xA1, 0xA1, 0xA1, 0xA1, 0xA1, 0xA1, 0xA1]
        );
        assert_eq!(
            bufs[3],
            vec![0xB1, 0xB1, 0xB1, 0xB1, 0xB1, 0xB1, 0xB1, 0xB1]
        );

        // Both files should have a dirfd cache entry.
        assert_eq!(store.dir_fd_cache.len(), 2);

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_identity_roundtrip() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        store
            .write_identity("file-id-1", 4096, 1_700_000_000_000)
            .await
            .unwrap();
        let identity = store.read_identity("file-id-1").await;
        assert_eq!(identity, Some((4096, 1_700_000_000_000)));

        store.delete_identity("file-id-1").await.unwrap();
        assert_eq!(store.read_identity("file-id-1").await, None);

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    // ── Method E: page fd cache tests (foyer-backed) ─────────

    /// A `PageId` whose file id is unique to `label`, with `idx` as the page
    /// index within that file.
    ///
    /// Both fd caches are keyed by identity that outlives a single test:
    /// `PAGE_FD_CACHE` is process-global and keyed by `PageId`, and
    /// `dir_fd_cache` is keyed by `file_id`. Two tests sharing either key read
    /// each other's files — silently, because a cached entry keeps its file alive
    /// through `Arc<File>`. So every test needs its own `label`.
    ///
    /// `idx` deliberately does NOT affect the file id. An earlier version folded
    /// it in, which made `unique_id(l, 0..3)` three separate *files* instead of
    /// three pages of one file, and broke every test asserting that pages of a
    /// file share one directory fd.
    fn unique_id(label: &str, idx: u64) -> PageId {
        PageId::new(format!("file-{label}-test"), idx)
    }

    /// Helper: check if a `PageId` is in the page fd cache.
    async fn is_in_page_cache(id: &PageId) -> bool {
        PAGE_FD_CACHE.get(id).is_some()
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_page_fd_cache_hit_after_first_read() {
        // First get() opens the file (miss) → inserts into PAGE_FD_CACHE.
        // Second get() should hit (1 SQE, no openat).
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("pgcache-hit", 0);
        let data = b"page fd cache test data".to_vec();

        store.put(&id, &data).await.unwrap();

        // First read — should populate the page fd cache.
        let mut dst = vec![0u8; data.len()];
        let n1 = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n1, data.len());
        assert_eq!(&dst, &data);
        assert!(
            is_in_page_cache(&id).await,
            "page fd cache should contain entry after first read"
        );

        // Second read — should hit the page fd cache.
        let mut dst2 = vec![0u8; data.len()];
        let n2 = store.get(&id, 0, &mut dst2).await.unwrap();
        assert_eq!(n2, data.len());
        assert_eq!(&dst2, &data);
        assert!(
            is_in_page_cache(&id).await,
            "page fd cache should still contain entry (reuse)"
        );

        // Cleanup: invalidate cache entry so other tests don't see it.
        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_page_fd_cache_different_pages() {
        // Each page gets its own page fd cache entry.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id0 = unique_id("multi-pg", 0);
        let id1 = unique_id("multi-pg", 1);
        let id2 = unique_id("multi-pg", 2);

        // All three payloads are the same length so one `dst` fits each exactly;
        // a mismatch here is what made this test fail before.
        store.put(&id0, b"page-zero!").await.unwrap();
        store.put(&id1, b"page-one!!").await.unwrap();
        store.put(&id2, b"page-two!!").await.unwrap();

        // Read all three pages — each should populate the page fd cache.
        let mut dst = vec![0u8; 10];
        assert_eq!(store.get(&id0, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-zero!");
        assert_eq!(store.get(&id1, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-one!!");
        assert_eq!(store.get(&id2, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-two!!");

        // 3 page fd entries + 1 dir fd entry.
        assert!(is_in_page_cache(&id0).await);
        assert!(is_in_page_cache(&id1).await);
        assert!(is_in_page_cache(&id2).await);
        assert_eq!(store.dir_fd_cache.len(), 1);

        // Re-read all three — should all be page fd cache hits.
        assert_eq!(store.get(&id0, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-zero!");
        assert_eq!(store.get(&id1, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-one!!");
        assert_eq!(store.get(&id2, 0, &mut dst).await.unwrap(), 10);
        assert_eq!(&dst, b"page-two!!");

        // Cleanup.
        for id in [&id0, &id1, &id2] {
            PAGE_FD_CACHE.remove(id);
        }
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_page_fd_cache_concurrent_same_page() {
        // 32 concurrent reads of the same page — first one populates the
        // cache, the rest should hit (or race with the miss path).
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("conc-pgcache", 0);
        let data = vec![0xABu8; 64];
        store.put(&id, &data).await.unwrap();

        let store = Arc::new(store);
        let mut handles = Vec::new();
        for _ in 0..32 {
            let store = Arc::clone(&store);
            let id = id.clone();
            handles.push(tokio::spawn(async move {
                let mut dst = vec![0u8; 64];
                let n = store.get(&id, 0, &mut dst).await.unwrap();
                assert_eq!(n, 64);
                assert_eq!(dst, vec![0xABu8; 64]);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }

        // After 32 concurrent reads, the page fd cache should have an entry.
        assert!(
            is_in_page_cache(&id).await,
            "page fd cache should contain entry after concurrent reads"
        );

        // Cleanup.
        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_page_fd_cache_invalidation_on_delete() {
        // After delete(), the page fd cache entry should be removed.
        // A subsequent get() should return 0 (miss).
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("invalidate-del", 0);
        store.put(&id, b"will-be-deleted").await.unwrap();

        // Populate the page fd cache. 15 bytes, not 16 — the payload length and
        // the assertion disagreed before.
        let mut dst = vec![0u8; 15];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), 15);
        assert!(is_in_page_cache(&id).await);

        // Delete should invalidate the cache entry.
        store.delete(&id).await.unwrap();
        assert!(
            !is_in_page_cache(&id).await,
            "page fd cache should be empty after delete"
        );

        // Subsequent get should miss (return 0).
        let n = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n, 0, "get should return 0 after delete");

        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_page_fd_cache_invalidation_on_put_overwrite() {
        // After put() overwrites a page, the page fd cache entry should be
        // invalidated so the next get() reads the new data.
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("overwrite", 0);

        store.put(&id, b"old-data-here!").await.unwrap();
        let mut dst = vec![0u8; 14];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), 14);
        assert_eq!(&dst, b"old-data-here!");
        assert!(is_in_page_cache(&id).await);

        // Overwrite — put() should invalidate the cache entry.
        store.put(&id, b"new-data-here!!").await.unwrap();
        assert!(
            !is_in_page_cache(&id).await,
            "page fd cache should be empty after overwrite"
        );

        // Next get() should read the new data (re-open the file).
        let mut dst2 = vec![0u8; 15];
        assert_eq!(store.get(&id, 0, &mut dst2).await.unwrap(), 15);
        assert_eq!(&dst2, b"new-data-here!!");

        // Cleanup.
        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[test]
    fn path_buffer_with_nul_terminates_and_rejects_interior_nul() {
        let buf = UringPageStore::path_buffer_with_nul("/tmp/page-42").unwrap();
        assert_eq!(
            *buf.last().unwrap(),
            0,
            "OP_OPENAT/OP_UNLINKAT path buffer must end with NUL"
        );
        assert_eq!(&buf[..buf.len() - 1], b"/tmp/page-42");

        let err = UringPageStore::path_buffer_with_nul("bad\0name").unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
    }

    #[tokio::test]
    #[ignore = "requires usable io_uring OPENAT; denied (EPERM) on GitHub Actions"]
    async fn uring_get_bytes_returns_page_slice() {
        let Some((store, base)) = temp_store(1024).await else {
            return;
        };
        let id = unique_id("get-bytes", 0);
        store.put(&id, b"0123456789").await.unwrap();

        let bytes = store.get_bytes(&id, 2, 5).await.unwrap();
        assert_eq!(&bytes[..], b"23456");

        let missing = store
            .get_bytes(&unique_id("missing-bytes", 0), 0, 8)
            .await
            .unwrap();
        assert!(missing.is_empty());

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    // ── Sync pread mode tests (`pread_enabled = true`) ─────────
    //
    // These exercise the sync read path (`open`/`openat`/`pread` on the
    // calling thread). Page files are written directly with `std::fs`
    // instead of `put()` so the tests run without an io_uring-capable
    // kernel — `put`/`delete` always use io_uring regardless of the
    // switch, and their cache-invalidation semantics are mode-independent
    // (covered by the io_uring-mode tests above).

    /// Helper: create a store in sync pread read mode.
    async fn temp_pread_store(page_size: u64) -> (UringPageStore, PathBuf) {
        let base = std::env::temp_dir().join(format!("gfs_pread_test_{}", uuid::Uuid::new_v4()));
        let store = UringPageStore::create_with_pread(&base, page_size, true)
            .await
            .unwrap();
        (store, base)
    }

    /// Helper: write a page file directly, bypassing `put()` (io_uring).
    /// Uses the store's own `page_path` so the bucket layout matches.
    fn write_page_direct(store: &UringPageStore, id: &PageId, data: &[u8]) {
        let path = store.page_path(id);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(path, data).unwrap();
    }

    #[tokio::test]
    async fn pread_get_roundtrip() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-roundtrip", 0);
        let data = b"hello sync pread page cache".to_vec();
        write_page_direct(&store, &id, &data);

        let mut dst = vec![0u8; data.len()];
        let n = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n, data.len());
        assert_eq!(&dst, &data);

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_get_with_offset() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-offset", 0);
        write_page_direct(&store, &id, b"0123456789");

        let mut dst = vec![0u8; 4];
        let n = store.get(&id, 3, &mut dst).await.unwrap();
        assert_eq!(n, 4);
        assert_eq!(&dst, b"3456");

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_get_missing_returns_zero() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-missing", 0);
        let mut dst = vec![0u8; 8];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), 0);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_get_short_read_at_tail() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-tail", 0);
        write_page_direct(&store, &id, b"abc");

        // Ask for more than the page holds → fills only the available bytes.
        let mut dst = vec![0u8; 16];
        let n = store.get(&id, 0, &mut dst).await.unwrap();
        assert_eq!(n, 3);
        assert_eq!(&dst[..3], b"abc");

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_get_bytes_returns_page_slice() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-get-bytes", 0);
        write_page_direct(&store, &id, b"0123456789");

        let bytes = store.get_bytes(&id, 2, 5).await.unwrap();
        assert_eq!(&bytes[..], b"23456");

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_concurrent_get_same_page() {
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-conc", 0);
        write_page_direct(&store, &id, &vec![0x7Eu8; 64]);

        // 32 concurrent sync preads of the same page (shared cached fd —
        // pread is position-independent and thread-safe).
        let store = Arc::new(store);
        let mut handles = Vec::new();
        for _ in 0..32 {
            let store = Arc::clone(&store);
            let id = id.clone();
            handles.push(tokio::spawn(async move {
                let mut dst = vec![0u8; 64];
                let n = store.get(&id, 0, &mut dst).await.unwrap();
                assert_eq!(n, 64);
                assert_eq!(dst, vec![0x7Eu8; 64]);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_page_fd_cache_hit_after_first_read() {
        // First get() opens the file via sync openat (cold) → inserts into
        // PAGE_FD_CACHE; second get() hits the cache and preads the cached fd.
        let (store, base) = temp_pread_store(1024).await;
        let id = unique_id("pread-pgcache", 0);
        let data = b"pread fd cache test".to_vec();
        write_page_direct(&store, &id, &data);

        let mut dst = vec![0u8; data.len()];
        assert_eq!(store.get(&id, 0, &mut dst).await.unwrap(), data.len());
        assert_eq!(&dst, &data);
        assert!(
            is_in_page_cache(&id).await,
            "page fd cache should contain entry after first read"
        );

        let mut dst2 = vec![0u8; data.len()];
        assert_eq!(store.get(&id, 0, &mut dst2).await.unwrap(), data.len());
        assert_eq!(&dst2, &data);
        assert!(
            is_in_page_cache(&id).await,
            "page fd cache should still contain entry (reuse)"
        );

        PAGE_FD_CACHE.remove(&id);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[tokio::test]
    async fn pread_dir_fd_cache_reused_across_pages() {
        // Multiple pages of the same file share one cached directory fd in
        // pread mode too (DashMap logic is mode-independent).
        let (store, base) = temp_pread_store(1024).await;
        let file_id: Arc<str> = Arc::from("pread-file-multi");
        let id0 = PageId::new(file_id.clone(), 0);
        let id1 = PageId::new(file_id.clone(), 1);

        write_page_direct(&store, &id0, b"page-zero-data!!");
        write_page_direct(&store, &id1, b"page-one-data!!!");

        let mut dst = vec![0u8; 16];
        assert_eq!(store.get(&id0, 0, &mut dst).await.unwrap(), 16);
        assert_eq!(&dst, b"page-zero-data!!");
        assert_eq!(store.get(&id1, 0, &mut dst).await.unwrap(), 16);
        assert_eq!(&dst, b"page-one-data!!!");
        assert_eq!(store.dir_fd_cache.len(), 1);

        PAGE_FD_CACHE.remove(&id0);
        PAGE_FD_CACHE.remove(&id1);
        let _ = tokio::fs::remove_dir_all(&base).await;
    }

    #[test]
    fn pread_read_sync_eof_truncates_to_available_bytes() {
        // Direct unit coverage of `pread_read_sync`: a read longer than the
        // file must truncate at EOF, and a read at/past EOF must be empty.
        let dir = std::env::temp_dir().join(format!("gfs_pread_unit_{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("page");
        std::fs::write(&path, b"abc").unwrap();
        let file = std::fs::File::open(&path).unwrap();
        let fd = file.as_raw_fd();

        let bytes = UringPageStore::pread_read_sync(fd, 0, 16).unwrap();
        assert_eq!(&bytes[..], b"abc");

        let at_eof = UringPageStore::pread_read_sync(fd, 3, 4).unwrap();
        assert!(at_eof.is_empty());

        let past_eof = UringPageStore::pread_read_sync(fd, 100, 4).unwrap();
        assert!(past_eof.is_empty());

        let offset = UringPageStore::pread_read_sync(fd, 1, 2).unwrap();
        assert_eq!(&offset[..], b"bc");

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

    /// Every uring test must build its `PageId` through [`unique_id`].
    ///
    /// `PAGE_FD_CACHE` is process-global and keyed by `PageId` alone — the store
    /// root is not part of the key. Each `temp_store` gets its own uuid
    /// directory, so page *files* never clash, but two tests sharing a `PageId`
    /// share a cache entry. A cached entry keeps its file alive through
    /// `Arc<File>` (see the `PAGE_FD_CACHE.get` hit path), so the second test
    /// reads the *first* test's file and gets no error — just the wrong bytes.
    ///
    /// Five tests failed the first time this suite ran on a real Linux host, and
    /// three of them for this reason. The symptom is silent and only appears
    /// under parallel execution on a host where io_uring actually works, which
    /// is a poor combination to rely on someone noticing. Hence a source check.
    ///
    /// `include_str!` binds the file at compile time, so this does not depend on
    /// the working directory when the test runs.
    /// Every uring test must build its `PageId` through [`unique_id`].
    ///
    /// `PAGE_FD_CACHE` is process-global and keyed by `PageId` alone — the store
    /// root is not part of the key. Each `temp_store` gets its own uuid
    /// directory, so page *files* never clash, but two tests sharing a `PageId`
    /// share a cache entry, and a cached entry keeps its file alive through
    /// `Arc<File>` (see the `PAGE_FD_CACHE.get` hit path). The second test then
    /// reads the *first* test's file and gets no error at all — just the wrong
    /// bytes.
    ///
    /// Three tests failed exactly this way the first time the suite ran on a real
    /// Linux host. The symptom is silent and needs parallel execution on a host
    /// where io_uring works, so it is not something to leave to review.
    ///
    /// Only a **string literal** first argument is reported.
    /// `PageId::new(file_id.clone(), i)` is fine: the local carries the
    /// uniqueness. An earlier version of this guard checked for
    /// `contains("PageId::new(")` and flagged 19 lines, 14 of them wrongly —
    /// including its own matching string.
    ///
    /// `include_str!` binds the file at compile time, so this does not depend on
    /// the working directory.
    #[test]
    fn uring_tests_use_unique_page_ids() {
        let src = include_str!("store.rs");
        let Some(tests_start) = src.find("\nmod tests {") else {
            panic!("could not locate `mod tests` in store.rs");
        };
        let prefix_lines = src[..tests_start].lines().count();

        // Built by concatenation so this line does not match its own pattern —
        // otherwise the guard reports itself and can never pass.
        let needle = concat!("PageId", "::new(");

        let mut offenders = Vec::new();
        for (offset, line) in src[tests_start..].lines().enumerate() {
            let trimmed = line.trim();
            if trimmed.starts_with("//") {
                continue;
            }
            // Extract a literal first argument, if there is one.
            let Some(after) = trimmed.split(needle).nth(1) else {
                continue;
            };
            let Some(inner) = after.trim_start().strip_prefix('"') else {
                continue; // an expression, not a literal — fine
            };
            let Some(file_id) = inner.split('"').next() else {
                continue;
            };
            // `temp_store`'s probe page is written and deleted before the store
            // reaches a test, under a name no test uses.
            if file_id == "__uring_probe__" {
                continue;
            }
            offenders.push(format!("  line {}: {}", prefix_lines + offset + 1, trimmed));
        }

        assert!(
            offenders.is_empty(),
            "these uring tests build a PageId from a hardcoded file id, which \
             collides in the process-global PAGE_FD_CACHE and makes tests read \
             each other's files:\n{}\n\nUse `unique_id(\"label\", idx)` instead.",
            offenders.join("\n")
        );
    }
}