fxcp_core/operations/

mod.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
//
//! SmartCopier, io_uring pipeline, reflink/CoW.

/// SIMD zero-block detection (AVX-512, AVX2, NEON, scalar fallback).
pub mod simd;
/// Filesystem capability probing and copy tier selection.
pub mod capabilities;
/// Container and device-mapper storage stack detection.
pub mod container;
/// io_uring NAPI busy-poll registration for NFS-target rings.
pub mod napi;

pub(crate) use simd::*;
pub use capabilities::*;
pub use container::*;
pub use napi::{register_napi_with_ring, unregister_napi_from_ring, try_register_napi, DEFAULT_NAPI_BUSY_POLL_TO_US};

use std::path::{Path, PathBuf};
use std::os::unix::io::{AsRawFd, RawFd, FromRawFd, IntoRawFd, BorrowedFd};
use std::sync::atomic::Ordering;
use std::io;
use std::ffi::CString;
use tokio::task::spawn_blocking;
use io_uring::{opcode, types, IoUring};
use uuid::Uuid;
use libc;
use nix::sys::statfs;
use std::time::{Duration, Instant};
use tracing::{warn, debug, error, trace, info};
use crate::buffer::{BufferPool};
use crate::error::{FxcpError, Result};
use crate::security;
use crate::metrics;
use std::os::unix::fs::MetadataExt;
use std::os::unix::ffi::OsStrExt;
use std::sync::Arc;
use dashmap::DashMap;
use crate::governor::Governor;
use tokio::fs::File;
use tokio::io::unix::AsyncFd;

const RWF_UNCACHED: i32 = 0x00000040;
const RWF_ATOMIC: i32 = 0x00000080;
const OP_TYPE_MASK: u64 = 0xFFFF_0000_0000_0000;
const INDEX_MASK: u64 = 0x0000_0000_0000_FFFF;
const READ_OP: u64 = 1 << 48;
const WRITE_OP: u64 = 2 << 48;
const FALLOC_OP: u64 = 3 << 48;
const HOLE_OP: u64 = 4 << 48;
const POSTCOPY_FSYNC_OP: u64 = 5 << 48;
const POSTCOPY_RENAME_OP: u64 = 6 << 48;
const UNLINK_OP: u64 = 8 << 48;
const LINK_OP: u64 = 9 << 48;
const SYMLINK_OP: u64 = 10 << 48;
const TRUNCATE_OP: u64 = 11 << 48;
const STATX_OP: u64 = 12 << 48;
const FADVISE_OP: u64 = 13 << 48;
const SRC_FIXED_SLOT: u32 = 0;
const DST_FIXED_SLOT: u32 = 1;
const FICLONE: u64 = crate::constants::FICLONE_IOCTL;
const FICLONERANGE: u64 = 0x4020940D;
const F2FS_IOC_START_ATOMIC_WRITE: u64 = 0xF501;
const F2FS_IOC_COMMIT_ATOMIC_WRITE: u64 = 0xF502;
const F2FS_IOC_ABORT_ATOMIC_WRITE: u64 = 0xF505;
const F2FS_IOC_SET_PIN_FILE: u64 = 0xF50D;
// Planned btrfs snapshot integration
#[allow(dead_code)]
const BTRFS_IOC_SNAP_CREATE_V2: u64 = 0x50009417;
#[allow(dead_code)]
const BTRFS_IOC_SUBVOL_CREATE_V2: u64 = 0x50009418;
#[allow(dead_code)]
const BTRFS_IOC_SCRUB: u64 = 0xC400941B;
#[allow(dead_code)]
const BTRFS_IOC_SEND: u64 = 0x40489426;
const BTRFS_IOC_INO_LOOKUP: u64 = 0xD0009412;

#[repr(C)]
struct FileCloneRange {
    src_fd: i64,
    src_offset: u64,
    src_length: u64,
    dest_offset: u64,
}

/// Tracks fsync latency with exponential moving average for adaptive timeout calculation
#[derive(Debug, Clone)]
pub struct FsyncLatencyTracker {
    avg_us: u64,
    deviation_us: u64,
}

impl Default for FsyncLatencyTracker {
    fn default() -> Self {
        Self {
            avg_us: crate::constants::FSYNC_LATENCY_DEFAULT_AVG_US,
            deviation_us: crate::constants::FSYNC_LATENCY_DEFAULT_DEVIATION_US,
        }
    }
}

impl FsyncLatencyTracker {
    /// Record a successful fsync and update the moving average
    pub fn record_success(&mut self, duration: Duration) {
        let sample = duration.as_micros() as u64;
        let diff = sample.abs_diff(self.avg_us);
        self.deviation_us = (self.deviation_us * 3 + diff) / 4;
        self.avg_us = (self.avg_us * 7 + sample) / 8;
    }

    /// Record a timeout and double the average estimate
    pub fn record_timeout(&mut self) {
        self.avg_us = self.avg_us.saturating_mul(2).min(crate::constants::FSYNC_LATENCY_MAX_US);
        self.deviation_us = self.deviation_us.saturating_mul(2);
    }

    /// Calculate an adaptive timeout based on observed latency (avg + 4*deviation, clamped 5s..60s)
    pub fn get_timeout(&self) -> Duration {
        let timeout_us = self.avg_us + (4 * self.deviation_us);
        Duration::from_micros(timeout_us.clamp(crate::constants::FSYNC_TIMEOUT_MIN_US, crate::constants::FSYNC_TIMEOUT_MAX_US))
    }
}

// Planned btrfs snapshot integration
#[allow(dead_code)]
#[repr(C)]
pub(crate) struct FileHandle {
    pub(crate) handle_bytes: u32,
    pub(crate) handle_type: i32,
    pub(crate) f_handle: [u8; 128],
}

#[allow(dead_code)]
#[repr(C)]
struct BtrfsIoctlVolArgsV2 {
    fd: i64,
    transid: u64,
    flags: u64,
    union_reserved: [u64; 4],
    name: [i8; 4040],
}

#[allow(dead_code)]
#[repr(C)]
struct BtrfsScrubArgs {
    devid: u64,
    start: u64,
    end: u64,
    flags: u64,
    progress: [u64; 16],
}

#[allow(dead_code)]
#[repr(C)]
struct BtrfsIoctlSendArgs {
    send_fd: i64,
    clone_sources_count: u64,
    clone_sources: u64,
    parent_root: u64,
    flags: u64,
    reserved: [u64; 4],
}

#[repr(C)]
struct BtrfsIoctlInoLookupArgs {
    treeid: u64,
    objectid: u64,
    name: [u8; 4080],
}

/// Statistics collected during a copy operation
#[derive(Debug, Default, Clone, Copy)]
pub struct CopyStats {
    /// Total bytes read and written
    pub bytes_processed: u64,
    /// Bytes detected as all-zero (hole-punched instead of written)
    pub bytes_zeros: u64,
    /// Wall-clock time spent on I/O
    pub io_duration: Duration,
    /// Number of io_uring operations submitted
    pub ops_count: u64,
}

#[derive(Debug, Clone)]
enum FileSegment {
    Data { offset: u64, len: u64 },
    Hole { offset: u64, len: u64 },
}

pub(crate) enum Operation {
    CopyFile { src: PathBuf, dst: PathBuf, src_file_size: u64, target_label: String, is_sparse: bool },
    CopyRange { src: PathBuf, dst: PathBuf, offset: u64, length: u64, src_file_size: u64, target_label: String },
    Truncate { dst: PathBuf, size: u64 },
    Rename { src: PathBuf, dst: PathBuf, flags: u32 },
    Fallocate { dst: PathBuf, mode: i32, offset: u64, length: u64 },
    Unlink { path: PathBuf, is_dir: bool },
    Link { existing: PathBuf, new: PathBuf },
    Symlink { target: PathBuf, linkpath: PathBuf },
}

/// Copy src to dst using reflink (FICLONE) if supported, falling back to
/// sendfile and finally a read+write loop. Returns bytes copied.
///
/// This is the synchronous small-file helper for code paths that cannot use
/// the full async SmartCopier (e.g. snapshot creation, tree metadata copies).
/// Never use `std::fs::copy()`  --  it bypasses CoW and reflink entirely.
pub fn reflink_or_copy(src: &Path, dst: &Path) -> io::Result<u64> {
    use std::os::unix::fs::OpenOptionsExt;

    let src_file = std::fs::File::open(src)?;
    let dst_file = std::fs::OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .mode(0o644)
        .open(dst)?;

    let sfd = src_file.as_raw_fd();
    let dfd = dst_file.as_raw_fd();
    let size = src_file.metadata()?.len();

    // Tier 1: Try FICLONE (instant CoW reflink)
    // SAFETY: sfd and dfd are valid open file descriptors from File::open above.
    let ret = unsafe { libc::ioctl(dfd, FICLONE as _, sfd) };
    if ret == 0 {
        return Ok(size);
    }

    // Tier 2: sendfile (kernel-space copy, no userspace buffer)
    let mut total: u64 = 0;
    while total < size {
        let remaining = size - total;
        // SAFETY: sfd and dfd are valid open file descriptors.
        let n = unsafe {
            libc::sendfile(dfd, sfd, std::ptr::null_mut(), remaining.min(crate::constants::SENDFILE_CHUNK_SIZE) as usize)
        };
        match n {
            -1 => {
                let err = io::Error::last_os_error();
                match err.raw_os_error() {
                    Some(libc::EINVAL) | Some(libc::ENOSYS) => break,
                    _ => return Err(err),
                }
            }
            0 => break,
            n => total += n as u64,
        }
    }
    if total == size {
        return Ok(size);
    }

    // Tier 3 fallback: read/write via std::io::copy (rare  --  procfs, FUSE, etc.)
    drop(src_file);
    drop(dst_file);
    let mut r = std::fs::File::open(src)?;
    let mut w = std::fs::OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .open(dst)?;
    io::copy(&mut r, &mut w)
}

/// Trait for filesystem operations that auto-select the optimal I/O path
pub trait OptimizedFs {
    /// Copy an entire file, choosing reflink/sendfile/io_uring as appropriate
    fn optimized_copy(&mut self, src: PathBuf, dst: PathBuf, src_file_size: u64, target_label: String, buffer_limit: Option<usize>, skip_fsync: bool, is_sparse: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Copy a byte range within a file (used for Merkle delta resync)
    fn optimized_copy_range(&mut self, src: PathBuf, dst: PathBuf, offset: u64, length: u64, src_file_size: u64, target_label: String, buffer_limit: Option<usize>, skip_fsync: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Truncate a file to the given size
    fn optimized_truncate(&mut self, dst: PathBuf, size: u64) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Rename a file with optional flags (e.g. RENAME_EXCHANGE)
    fn optimized_rename(&mut self, src: PathBuf, dst: PathBuf, flags: u32) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Allocate or punch holes in a file
    fn optimized_fallocate(&mut self, dst: PathBuf, mode: i32, offset: u64, length: u64) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Unlink a file or remove a directory via io_uring UnlinkAt (AT_REMOVEDIR for dirs).
    /// Falls back to spawn_blocking on EOPNOTSUPP.
    fn optimized_unlink(&mut self, path: PathBuf, is_dir: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Create a hard link via io_uring LinkAt.
    /// Falls back to spawn_blocking on EOPNOTSUPP.
    fn optimized_link(&mut self, existing: PathBuf, new: PathBuf) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
    /// Create a symbolic link via io_uring SymlinkAt.
    /// Falls back to spawn_blocking on EOPNOTSUPP.
    fn optimized_symlink(&mut self, target: PathBuf, linkpath: PathBuf) -> impl std::future::Future<Output = Result<CopyStats>> + Send;
}

pub(crate) struct CleanupGuard {
    files: Vec<PathBuf>,
}

impl CleanupGuard {
    pub(crate) fn new(files: Vec<PathBuf>) -> Self {
        Self { files }
    }

    #[allow(dead_code)]
    pub(crate) fn empty() -> Self {
        Self { files: Vec::new() }
    }

    #[allow(dead_code)]
    pub fn register(&mut self, path: PathBuf) {
        self.files.push(path);
    }

    #[allow(dead_code)]
    pub fn disarm(&mut self, path: &Path) {
        self.files.retain(|p| p != path);
    }
}

impl Drop for CleanupGuard {
    fn drop(&mut self) {
        for p in &self.files {
            let _ = std::fs::remove_file(p);
        }
    }
}

/// Resolve a btrfs inode number to its filesystem path via BTRFS_IOC_INO_LOOKUP
pub fn btrfs_resolve_inode(fd: RawFd, inode: u64) -> Result<PathBuf> {
    let mut args = BtrfsIoctlInoLookupArgs { treeid: 0, objectid: inode, name: [0; 4080] };
    // SAFETY: fd is a valid open file descriptor, args is a properly initialized
    // repr(C) struct matching the kernel's expected layout for BTRFS_IOC_INO_LOOKUP.
    let ret = unsafe { libc::ioctl(fd, BTRFS_IOC_INO_LOOKUP, &mut args) };
    if ret < 0 { return Err(FxcpError::Io(std::io::Error::last_os_error())); }
    
    let name_slice = match args.name.iter().position(|&c| c == 0) {
        Some(pos) => &args.name[..pos],
        None => &args.name[..],
    };
    
    Ok(PathBuf::from(std::ffi::OsStr::from_bytes(name_slice)))
}

#[allow(dead_code)]
pub(crate) fn btrfs_create_snapshot(src_fd: RawFd, dest_dir: &Path, name: &str) -> Result<()> {
    let dest_dir_file = std::fs::File::open(dest_dir).map_err(FxcpError::Io)?;
    let cname = CString::new(name).map_err(|_| FxcpError::Config("Invalid snapshot name".into()))?;
    
    if cname.as_bytes().len() > 4039 {
        return Err(FxcpError::Config("Snapshot name too long".into()));
    }

    let mut args = BtrfsIoctlVolArgsV2 {
        fd: src_fd as i64,
        transid: 0,
        flags: 0,
        union_reserved: [0; 4],
        name: [0; 4040]
    };

    // SAFETY: cname length is checked <= 4039 above, fitting within the
    // 4040-byte name field. Both pointers are valid and non-overlapping.
    unsafe {
        std::ptr::copy_nonoverlapping(
            cname.as_bytes_with_nul().as_ptr(),
            args.name.as_mut_ptr().cast::<u8>(),
            cname.as_bytes_with_nul().len()
        );
    }

    // SAFETY: dest_dir_file is a valid fd, args is a repr(C) struct with
    // the snapshot name copied in and fd set to the source subvolume.
    let ret = unsafe { libc::ioctl(dest_dir_file.as_raw_fd(), BTRFS_IOC_SNAP_CREATE_V2, &args) };
    if ret < 0 {
        return Err(FxcpError::Io(std::io::Error::last_os_error()));
    }
    Ok(())
}

#[allow(dead_code)]
pub(crate) fn btrfs_create_subvol(dest_dir: &Path, name: &str) -> Result<()> {
    let dir_file = std::fs::File::open(dest_dir).map_err(FxcpError::Io)?;
    let cname = CString::new(name).map_err(|_| FxcpError::Config("Invalid subvol name".into()))?;
    
    if cname.as_bytes().len() > 4039 {
        return Err(FxcpError::Config("Subvolume name too long".into()));
    }

    let mut args = BtrfsIoctlVolArgsV2 {
        fd: 0,
        transid: 0,
        flags: 0,
        union_reserved: [0; 4],
        name: [0; 4040]
    };

    // SAFETY: cname length is checked <= 4039 above, fitting within the
    // 4040-byte name field. Both pointers are valid and non-overlapping.
    unsafe {
        std::ptr::copy_nonoverlapping(
            cname.as_bytes_with_nul().as_ptr(),
            args.name.as_mut_ptr().cast::<u8>(),
            cname.as_bytes_with_nul().len()
        );
    }

    // SAFETY: dir_file is a valid fd, args is a repr(C) struct with the
    // subvolume name written into the name field.
    let ret = unsafe { libc::ioctl(dir_file.as_raw_fd(), BTRFS_IOC_SUBVOL_CREATE_V2, &args) };
    if ret < 0 {
        return Err(FxcpError::Io(std::io::Error::last_os_error()));
    }
    Ok(())
}

#[allow(dead_code)]
pub(crate) fn btrfs_scrub_start(mount_point: &Path) -> Result<()> {
    let f = std::fs::File::open(mount_point).map_err(FxcpError::Io)?;
    // SAFETY: BtrfsScrubArgs is repr(C) with no padding requirements
    // beyond zero-initialization. All-zeros is a valid state.
    let mut args: BtrfsScrubArgs = unsafe { std::mem::zeroed() };
    args.devid = 0;
    
    // SAFETY: f is a valid fd for the mount point, args is properly initialized.
    let ret = unsafe { libc::ioctl(f.as_raw_fd(), BTRFS_IOC_SCRUB, &mut args) };
    if ret < 0 {
        let err = std::io::Error::last_os_error();
        if err.raw_os_error() == Some(libc::EINPROGRESS) {
            return Ok(());
        }
        return Err(FxcpError::Io(err));
    }
    Ok(())
}

#[allow(dead_code)]
pub(crate) fn btrfs_send_stream(
    subvol_fd: RawFd, 
    parent_root_id: u64, 
    clone_sources: &[u64]
) -> Result<std::fs::File> {
    let (pipe_r, pipe_w) = nix::unistd::pipe().map_err(FxcpError::System)?;
    let clone_sources_vec = clone_sources.to_vec();
    
    // SAFETY: subvol_fd is a valid open fd passed by the caller.
    let fd_dup = nix::unistd::dup(unsafe { BorrowedFd::borrow_raw(subvol_fd) }).map_err(FxcpError::System)?;
    let pipe_w_fd = pipe_w.into_raw_fd();

    std::thread::spawn(move || {
        let ptr = if !clone_sources_vec.is_empty() {
            clone_sources_vec.as_ptr() as u64
        } else {
            0
        };
        
        let mut args = BtrfsIoctlSendArgs {
            send_fd: pipe_w_fd as i64,
            clone_sources_count: clone_sources_vec.len() as u64,
            clone_sources: ptr,
            parent_root: parent_root_id,
            flags: 0,
            reserved: [0; 4]
        };

        // SAFETY: fd_dup is a valid duplicated fd, args is a repr(C) struct
        // with clone_sources pointing to a live Vec (kept alive by the closure).
        let ret = unsafe { libc::ioctl(fd_dup.as_raw_fd(), BTRFS_IOC_SEND, &mut args) };
        drop(fd_dup);
        // SAFETY: pipe_w_fd is a valid raw fd from into_raw_fd().
        unsafe { libc::close(pipe_w_fd); }
        
        if ret < 0 {
            error!("Btrfs Send Failed: {}", std::io::Error::last_os_error());
        } else {
            debug!("Btrfs Send Completed successfully.");
        }
    });

    // SAFETY: pipe_r is a valid fd from nix::unistd::pipe(). Ownership
    // transfers to File, which will close it on drop.
    let file = unsafe { std::fs::File::from_raw_fd(pipe_r.into_raw_fd()) };
    Ok(file)
}

#[allow(dead_code)]
pub(crate) fn open_by_handle_at(mount_fd: RawFd, handle: &FileHandle) -> Result<std::fs::File> {
    // SAFETY: mount_fd is a valid fd, handle points to a valid FileHandle
    // struct obtained from name_to_handle_at. The kernel validates the handle.
    let fd = unsafe {
        libc::syscall(
            libc::SYS_open_by_handle_at,
            mount_fd,
            handle as *const _ as *const libc::c_void,
            libc::O_RDONLY | libc::O_NOATIME
        )
    };
    if fd < 0 {
        return Err(FxcpError::Io(std::io::Error::last_os_error()));
    }
    // SAFETY: fd is a valid file descriptor (checked >= 0 above).
    // Ownership transfers to File.
    Ok(unsafe { std::fs::File::from_raw_fd(fd as i32) })
}

/// Async copy engine backed by io_uring with auto-adaptive tier selection
///
/// Probes source/target capabilities and selects the fastest copy path:
/// reflink (FICLONE), copy_file_range, sendfile, or pipelined io_uring.
pub struct SmartCopier {
    /// io_uring instance for async I/O submission
    pub ring: IoUring,
    /// Registered buffer pool for io_uring fixed-buffer reads/writes
    pub buffer_pool: BufferPool,
    /// Separate aligned buffer pool for hardware atomic writes (RWF_ATOMIC)
    pub atomic_buffer_pool: Option<BufferPool>,
    /// Async wrapper around the io_uring eventfd for tokio integration
    pub async_fd: Arc<AsyncFd<RawFd>>,
    /// Enable VDO zero-block optimization (punch holes for all-zero blocks)
    pub vdo_opt: bool,
    /// Whether O_DIRECT is safe on the target filesystem
    pub direct_io_ok: bool,
    /// Detected capabilities of the source filesystem
    pub source_caps: Arc<Capabilities>,
    /// Detected capabilities of the target filesystem
    pub target_caps: Arc<Capabilities>,
    /// Number of consecutive zero blocks before triggering VDO stall warning
    pub vdo_stall_threshold: u32,
    /// Optional callback invoked after each file copy with the target inode
    pub barrier_callback: Option<Box<dyn Fn(u64) + Send + Sync>>,
    /// Use RWF_UNCACHED for source reads
    pub source_uncached: bool,
    /// Use RWF_UNCACHED for target writes
    pub target_uncached: bool,
    /// PSI-based system stress governor
    pub governor: Option<Arc<Governor>>,
    /// Adaptive fsync timeout tracker
    pub fsync_tracker: FsyncLatencyTracker,
    /// Skip fsync after writes (for batch operations that fsync at the end)
    pub skip_fsync: bool,
    /// Seconds without progress before declaring a segment stall
    pub segment_stall_timeout_secs: u64,
    /// Maximum seconds for an entire segment copy before timeout
    pub segment_overall_timeout_secs: u64,
    /// Separate io_uring ring for postcopy operations (rename, fsync, unlink, link, symlink, truncate, statx).
    /// Uses IORING_SETUP_ATTACH_WQ to share the kernel async worker pool with `ring`.
    pub postcopy_ring: IoUring,
    /// Async eventfd wrapper for the postcopy ring.
    pub postcopy_async_fd: Arc<AsyncFd<RawFd>>,
}

impl OptimizedFs for SmartCopier {
    fn optimized_copy(&mut self, src: PathBuf, dst: PathBuf, src_file_size: u64, target_label: String, buffer_limit: Option<usize>, skip_fsync: bool, is_sparse: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::CopyFile { src, dst, src_file_size, target_label, is_sparse };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, buffer_limit, skip_fsync, self.segment_stall_timeout_secs, self.segment_overall_timeout_secs, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_rename(&mut self, src: PathBuf, dst: PathBuf, flags: u32) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Rename { src, dst, flags };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_copy_range(&mut self, src: PathBuf, dst: PathBuf, offset: u64, length: u64, src_file_size: u64, target_label: String, buffer_limit: Option<usize>, skip_fsync: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::CopyRange { src, dst, offset, length, src_file_size, target_label };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, buffer_limit, skip_fsync, self.segment_stall_timeout_secs, self.segment_overall_timeout_secs, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_truncate(&mut self, dst: PathBuf, size: u64) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Truncate { dst, size };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_fallocate(&mut self, dst: PathBuf, mode: i32, offset: u64, length: u64) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Fallocate { dst, mode, offset, length };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_unlink(&mut self, path: PathBuf, is_dir: bool) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Unlink { path, is_dir };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_link(&mut self, existing: PathBuf, new: PathBuf) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Link { existing, new };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }

    fn optimized_symlink(&mut self, target: PathBuf, linkpath: PathBuf) -> impl std::future::Future<Output = Result<CopyStats>> + Send {
        async move {
            let op = Operation::Symlink { target, linkpath };
            Self::dispatch(op, &mut self.ring, &mut self.buffer_pool, self.atomic_buffer_pool.as_mut(), self.async_fd.clone(), self.vdo_opt, self.direct_io_ok, &self.source_caps, &self.target_caps, self.vdo_stall_threshold, &self.barrier_callback, self.source_uncached, self.target_uncached, self.governor.clone(), &mut self.fsync_tracker, None, false, 60, 600, &mut self.postcopy_ring, &self.postcopy_async_fd).await
        }
    }
}

impl SmartCopier {
    /// Stat a path via io_uring `Statx` requesting only `STATX_NLINK`, returning the
    /// hard-link count.  Falls back to `spawn_blocking(std::fs::metadata)` when the
    /// kernel returns `EOPNOTSUPP` or `EINVAL` (e.g. older kernels, NFS).
    pub async fn optimized_statx_nlink(&mut self, path: PathBuf) -> Result<u64> {
        let path_cstr = CString::new(path.as_os_str().as_bytes())
            .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;

        // SAFETY: zero-initialised statx is valid  --  kernel fills requested fields.
        let mut statx_buf: libc::statx = unsafe { std::mem::zeroed() };

        let sqe = opcode::Statx::new(
            types::Fd(libc::AT_FDCWD),
            path_cstr.as_ptr(),
            &mut statx_buf as *mut libc::statx as *mut _,
        )
        .flags(0)
        .mask(libc::STATX_NLINK)
        .build()
        .user_data(STATX_OP);

        match Self::submit_and_wait_single(&mut self.postcopy_ring, sqe, &self.postcopy_async_fd).await {
            Ok(_) => Ok(statx_buf.stx_nlink as u64),
            Err(FxcpError::Io(ref e))
                if e.raw_os_error() == Some(libc::EOPNOTSUPP)
                    || e.raw_os_error() == Some(libc::EINVAL) =>
            {
                debug!("io_uring Statx unsupported for {}, falling back to spawn_blocking", path.display());
                let meta = spawn_blocking(move || std::fs::metadata(&path))
                    .await
                    .map_err(FxcpError::Join)?
                    .map_err(FxcpError::Io)?;
                Ok(meta.nlink())
            }
            Err(e) => Err(e),
        }
    }
}

/// Ensure the parent directory of a target path exists
pub(crate) fn prepare_target_parent(target: &Path) -> std::io::Result<()> {
    if let Some(parent) = target.parent()
        && !parent.exists() {
            std::fs::create_dir_all(parent)?;
        }
    Ok(())
}

/// Full-file sendfile(2) transfer. Returns `CopyStats` on success.
/// Called from within `spawn_blocking`  --  safe to block.
///
/// # Safety
/// `sfd` and `dfd` must be valid open file descriptors held alive by the caller.
fn sendfile_full(sfd: RawFd, dfd: RawFd, size: u64, label: &str) -> Result<CopyStats> {
    let start = Instant::now();
    let mut total: u64 = 0;
    while total < size {
        let remaining = size - total;
        // SAFETY: sfd and dfd are valid open file descriptors held alive by the caller.
        let ret = unsafe {
            libc::sendfile(dfd, sfd, std::ptr::null_mut(), remaining.min(crate::constants::SENDFILE_CHUNK_SIZE) as usize)
        };
        if ret < 0 {
            return Err(FxcpError::Io(std::io::Error::last_os_error()));
        }
        if ret == 0 { break; }
        total += ret as u64;
    }
    if total == size {
        Ok(CopyStats {
            bytes_processed: size,
            bytes_zeros: 0,
            io_duration: start.elapsed(),
            ops_count: 1,
        })
    } else {
        Err(FxcpError::Io(io::Error::new(
            io::ErrorKind::UnexpectedEof,
            format!("{}: {} of {} bytes", label, total, size),
        )))
    }
}

/// Build a Write SQE targeting either a fixed file slot (`WriteFixed`) or raw fd (`Write`).
///
/// When `use_fixed` is true, uses `WriteFixed` with a pre-registered file slot.
/// When false, falls back to `Write` with a raw fd (no registered buffer index needed).
///
/// # Safety
/// Caller must ensure `ptr` is valid for `len` bytes and the fd/slot is registered
/// with the ring for the lifetime of the submitted SQE.
unsafe fn build_write_sqe(
    use_fixed: bool,
    dfd: RawFd,
    ptr: *const u8,
    len: u32,
    buf_index: u16,
    offset: u64,
    rw_flags: i32,
    user_data: u64,
) -> io_uring::squeue::Entry {
    if use_fixed {
        opcode::WriteFixed::new(types::Fixed(DST_FIXED_SLOT), ptr, len, buf_index)
            .offset(offset).rw_flags(rw_flags).build().user_data(user_data)
    } else {
        opcode::Write::new(types::Fd(dfd), ptr, len)
            .offset(offset).rw_flags(rw_flags).build().user_data(user_data)
    }
}

/// Attempt NFS sendfile fast-path for large files on NFS targets.
/// Returns `Ok(Some(stats))` if sendfile succeeded, `Ok(None)` to fall through.
async fn try_nfs_sendfile(
    sfd: RawFd,
    dfd: RawFd,
    src_file_size: u64,
    is_sparse: bool,
    target_is_nfs: bool,
    target_label: &str,
) -> Result<Option<CopyStats>> {
    if is_sparse || !target_is_nfs || src_file_size < crate::constants::NFS_SENDFILE_MIN_SIZE {
        return Ok(None);
    }

    let sf_size = src_file_size;
    let sendfile_res = spawn_blocking(move || {
        sendfile_full(sfd, dfd, sf_size, "NFS sendfile")
    }).await.map_err(FxcpError::Join)?;

    match sendfile_res {
        Ok(sf_stats) => {
            metrics::COPY_METHOD_OFFLOAD.with_label_values(&[target_label]).inc();
            debug!("execute_full_copy_logic: NFS sendfile fast-path ({} bytes)", src_file_size);
            Ok(Some(sf_stats))
        }
        Err(e) => {
            debug!("execute_full_copy_logic: NFS sendfile failed ({}), io_uring fallback", e);
            // SAFETY: sfd and dfd are valid open fds. Reset positions for fallback.
            unsafe {
                libc::lseek(sfd, 0, libc::SEEK_SET);
                libc::lseek(dfd, 0, libc::SEEK_SET);
            };
            Ok(None)
        }
    }
}

/// Attempt cachestat-guided sendfile fast-path for cached files in the 64KB-1MB range.
/// Returns `Ok(Some(stats))` if sendfile succeeded, `Ok(None)` to fall through.
async fn try_cachestat_sendfile(
    sfd: RawFd,
    dfd: RawFd,
    src_file_size: u64,
    is_sparse: bool,
    target_label: &str,
) -> Result<Option<CopyStats>> {
    const CACHESTAT_SIZE_HIGH: u64 = 1_048_576;
    if is_sparse
        || !(crate::constants::NFS_SENDFILE_MIN_SIZE..=CACHESTAT_SIZE_HIGH).contains(&src_file_size)
    {
        return Ok(None);
    }

    let residency = match query_cache_residency(sfd, src_file_size) {
        Some(r) if r > crate::constants::CACHE_RESIDENCY_SENDFILE_THRESHOLD => r,
        Some(r) => {
            debug!("execute_full_copy_logic: cachestat {:.0}% cached  --  preferring io_uring", r * 100.0);
            return Ok(None);
        }
        None => return Ok(None),
    };

    debug!("execute_full_copy_logic: cachestat {:.0}% cached  --  sendfile fast-path", residency * 100.0);
    let sf_size = src_file_size;
    let sendfile_res = spawn_blocking(move || {
        sendfile_full(sfd, dfd, sf_size, "sendfile")
    }).await.map_err(FxcpError::Join)?;

    match sendfile_res {
        Ok(sf_stats) => {
            metrics::COPY_METHOD_OFFLOAD.with_label_values(&[target_label]).inc();
            Ok(Some(sf_stats))
        }
        Err(e) => {
            debug!("execute_full_copy_logic: sendfile failed ({}), io_uring fallback", e);
            // SAFETY: resetting file positions for io_uring fallback.
            unsafe { libc::lseek(sfd, 0, libc::SEEK_SET); libc::lseek(dfd, 0, libc::SEEK_SET); };
            Ok(None)
        }
    }
}

/// Submit an io_uring fsync with adaptive timeout and blocking fallback.
/// When `skip_fsync` is true, only flushes pending submissions without fsync.
async fn uring_fsync_with_timeout(
    ring: &mut IoUring,
    dfd: RawFd,
    use_fixed: bool,
    async_fd: &AsyncFd<RawFd>,
    fsync_tracker: &mut FsyncLatencyTracker,
    target_path: &Path,
    skip_fsync: bool,
) -> Result<()> {
    if skip_fsync {
        ring.submit()?;
        return Ok(());
    }

    let fsync_op = if use_fixed {
        opcode::Fsync::new(types::Fixed(DST_FIXED_SLOT)).build().user_data(0)
    } else {
        opcode::Fsync::new(types::Fd(dfd)).build().user_data(0)
    };
    let mut pushed = false;

    for _ in 0..10 {
        // SAFETY: the SQE is fully built with valid fd references.
        if unsafe { ring.submission().push(&fsync_op) }.is_ok() {
            pushed = true;
            break;
        }
        let _ = ring.submit();
        if ring.submission().is_full() {
            match tokio::time::timeout(Duration::from_millis(crate::constants::IO_URING_READABLE_TIMEOUT_MS), async_fd.readable()).await {
                Ok(Ok(mut guard)) => {
                    let mut buf = [0u8; 8];
                    // SAFETY: reading 8 bytes from the eventfd to drain readiness notification.
                    let _ = unsafe { libc::read(async_fd.get_ref().as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, 8) };
                    guard.clear_ready();
                },
                Ok(Err(e)) => return Err(FxcpError::Io(e)),
                Err(_) => {
                    if let Some(_) = ring.completion().next() {
                        // drain
                    }
                }
            }
        }
    }

    if !pushed {
         let _ = ring.submit();
         // SAFETY: the SQE is fully built with valid fd references.
         unsafe { ring.submission().push(&fsync_op) }.map_err(|e| FxcpError::Io(io::Error::other(e.to_string())))?;
    }

    ring.submit()?;

    let mut fsync_found = false;
    let mut last_log = Instant::now();
    let fsync_start = Instant::now();
    let timeout_duration = fsync_tracker.get_timeout();
    let fsync_deadline = Instant::now() + timeout_duration;

    while !fsync_found {
         if let Some(c) = ring.completion().next() {
             if c.user_data() == 0 {
                 if c.result() < 0 {
                     return Err(FxcpError::Io(io::Error::from_raw_os_error(-c.result())));
                 }
                 fsync_found = true;
             }
             continue;
         }

         if Instant::now() > fsync_deadline {
             warn!("io_uring fsync timed out on {:?} (> {:?}). Ramp-up triggered. Falling back to blocking fsync.", target_path, timeout_duration);
             fsync_tracker.record_timeout();
              // SAFETY: dfd is a valid open file descriptor.
             unsafe { libc::fsync(dfd) };
             break;
         }

         match tokio::time::timeout(Duration::from_millis(crate::constants::IO_URING_READABLE_TIMEOUT_MS), async_fd.readable()).await {
            Ok(Ok(mut guard)) => {
                let mut buf = [0u8; 8];
                // SAFETY: reading 8 bytes from the eventfd to drain readiness notification.
                let _ = unsafe { libc::read(async_fd.get_ref().as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, 8) };
                guard.clear_ready();
            },
            Ok(Err(e)) => return Err(FxcpError::Io(e)),
            Err(_) => {
                let _ = ring.submit();
                if let Some(_) = ring.completion().next() {
                    // drain
                }
            }
         }

         if last_log.elapsed() > Duration::from_secs(crate::constants::LOG_THROTTLE_INTERVAL_SECS) {
             info!("Still waiting for fsync completion on {:?} (Elapsed: {:?}, Timeout: {:?})...", target_path, fsync_start.elapsed(), timeout_duration);
             last_log = Instant::now();
         }
    }
    if fsync_found {
        fsync_tracker.record_success(fsync_start.elapsed());
    }

    Ok(())
}

/// Handle a completed READ_OP CQE: detect zero blocks and emit the appropriate
/// write or hole-punch SQE. Returns `(bytes_zeros_delta, bytes_processed_delta)` on
/// success (a SQE was pushed). On error the buffer is released internally.
fn handle_read_completion(
    ring: &mut IoUring,
    active_pool: &mut BufferPool,
    index: u16,
    bytes_transferred: u64,
    original_offset: u64,
    dfd: RawFd,
    use_fixed: bool,
    vdo_opt: bool,
    write_flags: i32,
) -> Result<(u64, u64)> {
    if let Err(e) = active_pool.set_len(index, bytes_transferred as usize) {
        error!("io_uring buffer set_len failed: {}", e);
        active_pool.release(index);
        return Err(e);
    }

    let ptr = match active_pool.get_ptr(index) {
        Some(ptr) => ptr,
        None => {
            error!("io_uring buffer pool index out of range (write dispatch): index={}", index);
            active_pool.release(index);
            return Err(FxcpError::Io(io::Error::other("io_uring buffer pool index out of range (write dispatch)")));
        }
    };

    let is_zero_block = if vdo_opt && bytes_transferred > 0 {
        // SAFETY: ptr is from buffer_pool.get_ptr() (valid allocation),
        // and bytes_transferred <= buffer capacity (set by io_uring completion).
        let s = unsafe { std::slice::from_raw_parts(ptr, bytes_transferred as usize) };
        is_zero_block(s)
    } else { false };

    if is_zero_block && bytes_transferred >= crate::constants::MIN_HOLE_PUNCH_SIZE {
        let user_data_falloc = FALLOC_OP | (index as u64);
        let falloc_op = if use_fixed {
            opcode::Fallocate::new(types::Fixed(DST_FIXED_SLOT), bytes_transferred)
                .offset(original_offset)
                .mode(libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE)
                .build().user_data(user_data_falloc)
        } else {
            opcode::Fallocate::new(types::Fd(dfd), bytes_transferred)
                .offset(original_offset)
                .mode(libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE)
                .build().user_data(user_data_falloc)
        };

        // SAFETY: the SQE is fully built with valid fd references.
        if unsafe { ring.submission().push(&falloc_op) }.is_ok() {
            return Ok((bytes_transferred, bytes_transferred));
        } else {
            active_pool.release(index);
            return Err(FxcpError::Io(io::Error::other("Submission queue full during zero-block fallocate")));
        }
    }

    let user_data_write = WRITE_OP | (index as u64);
    // SAFETY: ptr is valid for bytes_transferred from the pool; fd/slot registered with ring.
    let write_op = unsafe { build_write_sqe(use_fixed, dfd, ptr, bytes_transferred as u32, index, original_offset, write_flags, user_data_write) };

    // SAFETY: the SQE is fully built with valid fd and buffer references.
    if unsafe { ring.submission().push(&write_op) }.is_ok() {
        Ok((0, 0))
    } else {
        active_pool.release(index);
        Err(FxcpError::Io(io::Error::other("Submission queue full during write")))
    }
}

impl SmartCopier {
    async fn open_source_noatime(path: &Path) -> Result<File> {
        let mut opts = tokio::fs::OpenOptions::new();
        opts.read(true);
        opts.custom_flags(libc::O_NOATIME);
        let file = match opts.open(path).await {
            Ok(f) => f,
            Err(e) => {
                if let Some(raw) = e.raw_os_error()
                    && (raw == libc::EPERM || raw == libc::EACCES) {
                        return File::open(path).await.map_err(FxcpError::Io);
                    }
                return Err(FxcpError::Io(e));
            }
        };
        // SAFETY: fd is a valid open file descriptor. POSIX_FADV_SEQUENTIAL
        // doubles the kernel readahead window; no-op on O_DIRECT or unsupported FS.
        // posix_fadvise returns error code directly (not via errno).
        let ret = unsafe {
            libc::posix_fadvise(file.as_raw_fd(), 0, 0, libc::POSIX_FADV_SEQUENTIAL)
        };
        if ret != 0 {
            debug!("posix_fadvise(SEQUENTIAL) failed: {}", std::io::Error::from_raw_os_error(ret));
        }
        Ok(file)
    }

    fn set_f2fs_pinning(fd: RawFd, enable: bool) {
        let val: u32 = if enable { 1 } else { 0 };
        // SAFETY: fd is a valid open file descriptor. The ioctl sets/clears
        // F2FS pinning; failure is non-fatal (ignored with let _).
        let _ = unsafe { libc::ioctl(fd, F2FS_IOC_SET_PIN_FILE, &val) };
    }

    /// Copy a file (or range) from src to dst using the full io_uring pipeline
    pub async fn copy(
        src: &Path,
        dst: &Path,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        offset: u64,
        length: u64,
        direct_io_ok: bool,
        src_file_size: u64,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        vdo_stall_threshold: u32,
        source_uncached: bool,
        target_uncached: bool,
        barrier_callback: &Option<Box<dyn Fn(u64) + Send + Sync>>,
        governor: Option<Arc<Governor>>,
        target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        is_sparse: bool,
        postcopy_ring: &mut IoUring,
        postcopy_async_fd: &Arc<AsyncFd<RawFd>>,
    ) -> Result<CopyStats> {
        Self::copy_with_limit(src, dst, ring, buffer_pool, atomic_pool, async_fd, vdo_opt, offset, length, direct_io_ok, src_file_size, source_caps, target_caps, vdo_stall_threshold, source_uncached, target_uncached, barrier_callback, governor, target_label, fsync_tracker, None, skip_fsync, segment_stall_timeout_secs, segment_overall_timeout_secs, is_sparse, postcopy_ring, postcopy_async_fd).await
    }

    /// Copy with an optional buffer count limit for backpressure control
    pub async fn copy_with_limit(
        src: &Path,
        dst: &Path,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        offset: u64,
        length: u64,
        direct_io_ok: bool,
        src_file_size: u64,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        vdo_stall_threshold: u32,
        source_uncached: bool,
        target_uncached: bool,
        barrier_callback: &Option<Box<dyn Fn(u64) + Send + Sync>>,
        governor: Option<Arc<Governor>>,
        target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        is_sparse: bool,
        postcopy_ring: &mut IoUring,
        postcopy_async_fd: &Arc<AsyncFd<RawFd>>,
    ) -> Result<CopyStats> {
        if !src.exists() {
            return Err(FxcpError::Io(std::io::Error::new(
                std::io::ErrorKind::NotFound,
                format!("Source file no longer exists: {:?}", src)
            )));
        }

        if offset == 0 && length == src_file_size {
             Self::execute_copy_file(
                src, dst, ring, buffer_pool, atomic_pool, async_fd, vdo_opt, direct_io_ok,
                source_caps, target_caps, src_file_size, vdo_stall_threshold,
                barrier_callback, source_uncached, target_uncached, governor,
                target_label, fsync_tracker, buffer_limit, skip_fsync,
                segment_stall_timeout_secs, segment_overall_timeout_secs,
                is_sparse,
                postcopy_ring, postcopy_async_fd,
             ).await
        } else {
             Self::execute_copy_range(
                src, dst, ring, buffer_pool, atomic_pool, async_fd, vdo_opt, offset, length,
                direct_io_ok, src_file_size, source_caps, target_caps,
                vdo_stall_threshold, source_uncached, target_uncached, governor,
                target_label, fsync_tracker, buffer_limit, skip_fsync,
                segment_stall_timeout_secs, segment_overall_timeout_secs
             ).await
        }
    }

    /// Dispatch a filesystem operation (copy, rename, truncate, fallocate)
    pub(crate) async fn dispatch(
        op: Operation,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        direct_io_ok: bool,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        vdo_stall_threshold: u32,
        barrier_callback: &Option<Box<dyn Fn(u64) + Send + Sync>>,
        source_uncached: bool,
        target_uncached: bool,
        governor: Option<Arc<Governor>>,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        postcopy_ring: &mut IoUring,
        postcopy_async_fd: &Arc<AsyncFd<RawFd>>,
    ) -> Result<CopyStats> {
        match op {
            Operation::CopyFile { src, dst, src_file_size, target_label, is_sparse } => {
                Self::execute_copy_file(
                    &src, &dst, ring, buffer_pool, atomic_pool, async_fd, vdo_opt,
                    direct_io_ok, source_caps, target_caps,
                    src_file_size, vdo_stall_threshold, barrier_callback,
                    source_uncached, target_uncached, governor,
                    target_label, fsync_tracker, buffer_limit, skip_fsync,
                    segment_stall_timeout_secs, segment_overall_timeout_secs,
                    is_sparse,
                    postcopy_ring, postcopy_async_fd,
                ).await
            },
            Operation::CopyRange { src, dst, offset, length, src_file_size, target_label } => {
                Self::execute_copy_range(
                    &src, &dst, ring, buffer_pool, atomic_pool, async_fd, vdo_opt,
                    offset, length, direct_io_ok, src_file_size,
                    source_caps, target_caps, vdo_stall_threshold,
                    source_uncached, target_uncached, governor,
                    target_label, fsync_tracker, buffer_limit, skip_fsync,
                    segment_stall_timeout_secs, segment_overall_timeout_secs
                ).await
            },
            Operation::Truncate { dst, size } => {
                let f = std::fs::OpenOptions::new()
                    .write(true)
                    .open(&dst)
                    .map_err(FxcpError::Io)?;
                let sqe = opcode::Ftruncate::new(types::Fd(f.as_raw_fd()), size)
                    .build()
                    .user_data(TRUNCATE_OP);
                match Self::submit_and_wait_single(postcopy_ring, sqe, postcopy_async_fd).await {
                    Ok(_) => Ok(CopyStats::default()),
                    Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EOPNOTSUPP) => {
                        drop(f);
                        spawn_blocking(move || {
                            security::truncate_file(&dst, size).map(|_| CopyStats::default())
                        }).await.map_err(FxcpError::Join)?
                    }
                    Err(e) => Err(e),
                }
            },
            Operation::Rename { src, dst, flags } => {
                spawn_blocking(move || {
                    crate::consistency::atomic_rename(&src, &dst, Some(flags)).map(|_| CopyStats::default()).map_err(FxcpError::Io)
                }).await.map_err(FxcpError::Join)?},
            Operation::Fallocate { dst, mode, offset, length } => {
                spawn_blocking(move || {
                    security::do_fallocate(&dst, offset, length, mode).map(|_| CopyStats::default())
                }).await.map_err(FxcpError::Join)?
            },
            Operation::Unlink { path, is_dir } => {
                let c_path = CString::new(path.as_os_str().as_bytes())
                    .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                let flags = if is_dir { libc::AT_REMOVEDIR } else { 0 };
                let sqe = opcode::UnlinkAt::new(types::Fd(libc::AT_FDCWD), c_path.as_ptr())
                    .flags(flags)
                    .build()
                    .user_data(UNLINK_OP);
                match Self::submit_and_wait_single(postcopy_ring, sqe, postcopy_async_fd).await {
                    Ok(_) => Ok(CopyStats::default()),
                    Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EOPNOTSUPP) => {
                        spawn_blocking(move || {
                            if is_dir { std::fs::remove_dir(&path) } else { std::fs::remove_file(&path) }
                                .map(|_| CopyStats::default())
                                .map_err(FxcpError::Io)
                        }).await.map_err(FxcpError::Join)?
                    }
                    Err(e) => Err(e),
                }
            },
            Operation::Link { existing, new } => {
                let old_cstr = CString::new(existing.as_os_str().as_bytes())
                    .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                let new_cstr = CString::new(new.as_os_str().as_bytes())
                    .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                let sqe = opcode::LinkAt::new(
                    types::Fd(libc::AT_FDCWD), old_cstr.as_ptr(),
                    types::Fd(libc::AT_FDCWD), new_cstr.as_ptr(),
                ).build().user_data(LINK_OP);
                match Self::submit_and_wait_single(postcopy_ring, sqe, postcopy_async_fd).await {
                    Ok(_) => Ok(CopyStats::default()),
                    Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EOPNOTSUPP) => {
                        spawn_blocking(move || {
                            std::fs::hard_link(&existing, &new)
                                .map(|_| CopyStats::default())
                                .map_err(FxcpError::Io)
                        }).await.map_err(FxcpError::Join)?
                    }
                    Err(e) => Err(e),
                }
            },
            Operation::Symlink { target, linkpath } => {
                let target_cstr = CString::new(target.as_os_str().as_bytes())
                    .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                let link_cstr = CString::new(linkpath.as_os_str().as_bytes())
                    .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                let sqe = opcode::SymlinkAt::new(
                    types::Fd(libc::AT_FDCWD),
                    target_cstr.as_ptr(),
                    link_cstr.as_ptr(),
                ).build().user_data(SYMLINK_OP);
                match Self::submit_and_wait_single(postcopy_ring, sqe, postcopy_async_fd).await {
                    Ok(_) => Ok(CopyStats::default()),
                    Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EOPNOTSUPP) => {
                        spawn_blocking(move || {
                            std::os::unix::fs::symlink(&target, &linkpath)
                                .map(|_| CopyStats::default())
                                .map_err(FxcpError::Io)
                        }).await.map_err(FxcpError::Join)?
                    }
                    Err(e) => Err(e),
                }
            },
        }
    }

    /// Probe the source file's sparse layout using `SEEK_HOLE` / `SEEK_DATA`.
    /// Returns the hole ratio (0.0 = dense, 1.0 = fully sparse) AND the full
    /// segment map, so callers can avoid a second walk via `map_sparse_segments`.
    /// Returns `None` if the filesystem doesn't support `SEEK_HOLE` or probing fails.
    fn probe_sparse_with_segments(src_fd: RawFd, file_size: u64) -> Option<(f64, Vec<FileSegment>)> {
        if file_size == 0 {
            return None;
        }
        let mut segments: Vec<FileSegment> = Vec::new();
        let mut hole_bytes: u64 = 0;
        let mut current_offset: i64 = 0;
        let end_offset = file_size as i64;

        loop {
            // SAFETY: src_fd is a valid open file descriptor; SEEK_DATA is a standard lseek whence.
            let data_offset_res = unsafe { libc::lseek(src_fd, current_offset, libc::SEEK_DATA) };
            if data_offset_res < 0 {
                let err = std::io::Error::last_os_error();
                match err.raw_os_error() {
                    Some(libc::ENXIO) => {
                        // Rest of file is a hole
                        let hole_len = (end_offset - current_offset) as u64;
                        if hole_len > 0 {
                            hole_bytes += hole_len;
                            segments.push(FileSegment::Hole { offset: current_offset as u64, len: hole_len });
                        }
                        break;
                    }
                    Some(libc::EOPNOTSUPP) | Some(libc::EINVAL) => return None,
                    _ => break,
                }
            }
            let data_offset = data_offset_res.min(end_offset);
            if data_offset > current_offset {
                let hole_len = (data_offset - current_offset) as u64;
                hole_bytes += hole_len;
                segments.push(FileSegment::Hole { offset: current_offset as u64, len: hole_len });
                current_offset = data_offset;
            }
            if current_offset >= end_offset {
                break;
            }

            // SAFETY: src_fd is a valid open file descriptor; SEEK_HOLE is a standard lseek whence.
            let hole_offset_res = unsafe { libc::lseek(src_fd, current_offset, libc::SEEK_HOLE) };
            let hole_offset = if hole_offset_res < 0 {
                end_offset
            } else {
                hole_offset_res.min(end_offset)
            };

            let data_len = (hole_offset - current_offset) as u64;
            if data_len > 0 {
                segments.push(FileSegment::Data { offset: current_offset as u64, len: data_len });
            }
            current_offset = hole_offset;
            if current_offset >= end_offset {
                break;
            }
        }

        // SAFETY: restoring seek position to start after probing.
        unsafe { libc::lseek(src_fd, 0, libc::SEEK_SET) };

        let ratio = hole_bytes as f64 / file_size as f64;
        Some((ratio, segments))
    }

    fn prepare_destination_file(
        path: PathBuf,
        direct_io: bool,
        f2fs_atomic: bool,
        size: u64,
        is_sparse: bool,
        src_fd: Option<RawFd>,
        is_nfs: bool,
    ) -> Result<(std::fs::File, RawFd, Option<Vec<FileSegment>>)> {
        debug!("prepare_destination_file: {:?}", path);
        prepare_target_parent(&path)?;
        let mut open_opts = std::fs::OpenOptions::new();
        open_opts.read(true).write(true).create(true).truncate(true);
        use std::os::unix::fs::OpenOptionsExt;
        open_opts.custom_flags(libc::O_NOFOLLOW);
        if direct_io {
            open_opts.custom_flags(libc::O_DIRECT | libc::O_NOFOLLOW);
        }
        
        let file = match open_opts.open(&path) {
            Ok(f) => f,
            Err(e) => {
                error!("prepare_destination_file: Failed to open {:?}: {}", path, e);
                return Err(FxcpError::Io(e));
            }
        };
        let fd = file.as_raw_fd();

        if f2fs_atomic {
            Self::set_f2fs_pinning(fd, true);
            // SAFETY: fd is a valid open file descriptor on an F2FS filesystem.
            if unsafe { libc::ioctl(fd, F2FS_IOC_START_ATOMIC_WRITE) } < 0 {
                warn!("F2FS: Failed to start atomic write transaction. Proceeding non-atomically.");
            }
        }

        // Skip SEEK_HOLE/SEEK_DATA probe for NFS targets  --  each lseek is a
        // synchronous RPC and NFS doesn't track holes locally.
        let cached_segments = if is_sparse || is_nfs {
            None
        } else {
            src_fd.and_then(|fd| Self::probe_sparse_with_segments(fd, size))
        };
        // Skip fallocate(mode=0) for NFS targets  --  sendfile extends the file
        // naturally, and fallocate is a synchronous ALLOCATE RPC on NFS that
        // pre-allocates blocks wastefully.
        let skip_fallocate = is_sparse || is_nfs
            || cached_segments.as_ref().is_some_and(|(ratio, _)| *ratio > crate::constants::SPARSE_HOLE_RATIO_SKIP_THRESHOLD);
        let cached_segment_vec = cached_segments.map(|(_, segs)| segs);

        if skip_fallocate && size > 0 {
            debug!("prepare_destination_file: Skipping pre-allocation (sparse or high hole ratio, {} bytes)", size);
        } else if size > 0 {
            debug!("prepare_destination_file: Fallocating {} bytes for {:?}", size, path);
            // SAFETY: fd is a valid open file descriptor. fallocate with
            // mode 0 pre-allocates space without modifying file contents.
            let ret = unsafe { libc::fallocate(fd, 0, 0, size as i64) };
            if ret < 0 {
                let err = std::io::Error::last_os_error();
                if err.raw_os_error() != Some(libc::EOPNOTSUPP) {
                    warn!("Prepare: fallocate failed: {}. Performance may degrade.", err);
                }
            }
        }
        debug!("prepare_destination_file: Ready {:?}", path);
        Ok((file, fd, cached_segment_vec))
    }

    /// Submit a single SQE to the io_uring and wait for its CQE via the ring's async eventfd.
    /// Returns the CQE result value (>= 0 on success). Errors on negative (kernel errno).
    /// Used for post-copy operations (fsync, rename) that don't fit the pipelined model.
    async fn submit_and_wait_single(
        ring: &mut IoUring,
        sqe: io_uring::squeue::Entry,
        async_fd: &AsyncFd<RawFd>,
    ) -> Result<i32> {
        // SAFETY: the SQE is fully built by the caller with valid fd and buffer references.
        unsafe { ring.submission().push(&sqe) }
            .map_err(|_| FxcpError::Io(io::Error::other("io_uring submission queue full")))?;

        ring.submit().map_err(FxcpError::Io)?;

        loop {
            if let Some(cqe) = ring.completion().next() {
                let res = cqe.result();
                if res < 0 {
                    return Err(FxcpError::Io(io::Error::from_raw_os_error(-res)));
                }
                return Ok(res);
            }

            match tokio::time::timeout(
                Duration::from_millis(crate::constants::IO_URING_READABLE_TIMEOUT_MS),
                async_fd.readable()
            ).await {
                Ok(Ok(mut guard)) => {
                    let mut buf = [0u8; 8];
                    // SAFETY: reading 8 bytes from the eventfd to drain readiness notification.
                    let _ = unsafe { libc::read(async_fd.get_ref().as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, 8) };
                    guard.clear_ready();
                }
                Ok(Err(e)) => return Err(FxcpError::Io(e)),
                Err(_) => {
                    // Timeout  --  resubmit and check completion again
                    let _ = ring.submit();
                }
            }
        }
    }

    async fn execute_copy_file(
        src: &Path,
        dst: &Path,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        direct_io_ok: bool,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        src_file_size: u64,
        vdo_stall_threshold: u32,
        barrier_callback: &Option<Box<dyn Fn(u64) + Send + Sync>>,
        source_uncached: bool,
        target_uncached: bool,
        governor: Option<Arc<Governor>>,
        target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        is_sparse: bool,
        postcopy_ring: &mut IoUring,
        postcopy_async_fd: &Arc<AsyncFd<RawFd>>,
    ) -> Result<CopyStats> {
        let target_path = dst.with_extension(format!("tmp.{}", Uuid::new_v4()));
        debug!("execute_copy_file: {:?} -> {:?}", src, target_path);

        let res = Self::execute_full_copy_logic(
            src, &target_path, ring, buffer_pool, atomic_pool, async_fd, vdo_opt, direct_io_ok,
            source_caps, target_caps, false, src_file_size, vdo_stall_threshold,
            barrier_callback, source_uncached, target_uncached, governor,
            target_label, fsync_tracker, buffer_limit, skip_fsync,
            segment_stall_timeout_secs, segment_overall_timeout_secs,
            is_sparse,
            postcopy_ring, postcopy_async_fd,
        ).await?;

        match res {
            Ok(stats) => {

                // Atomic rename: exchange path stays spawn_blocking, simple path uses
                // io_uring with EBADF fallback to spawn_blocking
                let can_exchange = target_caps.exchange_range.load(Ordering::Relaxed);
                let dst_exists = dst.exists();

                if can_exchange && dst_exists {
                    let tp2 = target_path.clone();
                    let dst_owned = dst.to_path_buf();
                    let target_caps_clone = target_caps.clone();

                    spawn_blocking(move || {
                        let is_btrfs = if let Ok(s) = statfs::statfs(&tp2) {
                            s.filesystem_type().0 as u64 == BTRFS_SUPER_MAGIC
                        } else {
                            false
                        };

                        if is_btrfs {
                            crate::consistency::atomic_rename(&tp2, &dst_owned, Some(libc::RENAME_EXCHANGE))
                                .or_else(|_| crate::consistency::atomic_rename(&tp2, &dst_owned, None))
                                .map_err(FxcpError::Io)
                        } else {
                            match crate::consistency::exchange::atomic_exchange(&tp2, &dst_owned) {
                                Ok(_) => Ok(()),
                                Err(e) => {
                                    let raw_err = e.raw_os_error();
                                    if raw_err == Some(libc::EOPNOTSUPP) || raw_err == Some(libc::ENOTTY) {
                                        debug!("Atomic Exchange unsupported on target ({}). Disabling optimization.", e);
                                        target_caps_clone.exchange_range.store(false, Ordering::Relaxed);
                                    } else {
                                        warn!("Atomic Exchange failed: {}. Falling back to rename.", e);
                                    }
                                    crate::consistency::atomic_rename(&tp2, &dst_owned, None).map_err(FxcpError::Io)
                                }
                            }
                        }
                    }).await.map_err(FxcpError::Join)??;
                } else {
                    let old_cstr = CString::new(target_path.as_os_str().as_bytes())
                        .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                    let new_cstr = CString::new(dst.as_os_str().as_bytes())
                        .map_err(|_| FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "path contains null byte")))?;
                    let rename_sqe = opcode::RenameAt::new(
                        types::Fd(libc::AT_FDCWD), old_cstr.as_ptr(),
                        types::Fd(libc::AT_FDCWD), new_cstr.as_ptr(),
                    ).build().user_data(POSTCOPY_RENAME_OP);
                    match Self::submit_and_wait_single(postcopy_ring, rename_sqe, postcopy_async_fd).await {
                        Ok(_) => {},
                        Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EBADF)
                            || e.raw_os_error() == Some(libc::EOPNOTSUPP) => {
                            debug!("io_uring RenameAt failed ({}), falling back to blocking rename", e);
                            let tp_clone = target_path.clone();
                            let dst_clone = dst.to_path_buf();
                            spawn_blocking(move || {
                                crate::consistency::atomic_rename(&tp_clone, &dst_clone, None)
                                    .map_err(FxcpError::Io)
                            }).await.map_err(FxcpError::Join)??;
                        }
                        Err(e) => return Err(e),
                    }
                }

                if let Some(cb) = barrier_callback
                    && let Ok(meta) = std::fs::metadata(dst) { cb(meta.ino()); }
                debug!("execute_copy_file: Success {:?}", dst);
                Ok(stats)
            },
            Err(e) => {
                error!("execute_copy_file: Failed {:?} - {}", target_path, e);
                let _ = std::fs::remove_file(&target_path);
                Err(e)
            }
        }
    }

    async fn execute_copy_range(
        src: &Path,
        dst: &Path,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        offset: u64,
        length: u64,
        direct_io_ok: bool,
        _src_file_size: u64,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        vdo_stall_threshold: u32,
        source_uncached: bool,
        target_uncached: bool,
        governor: Option<Arc<Governor>>,
        target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
    ) -> Result<CopyStats> {
        let src_meta = std::fs::metadata(src).map_err(FxcpError::Io)?;
        let strategy = determine_copy_strategy(src, dst, &src_meta, target_caps);

        let sf = Self::open_source_noatime(src).await?;
        let sfd = sf.as_raw_fd();

        let mut open_opts = tokio::fs::OpenOptions::new();
        open_opts.read(true).write(true).create(false);

        let mut use_direct_io = direct_io_ok;
        if use_direct_io && (!length.is_multiple_of(crate::constants::MINIMUM_ALIGNMENT_BYTES as u64) || !offset.is_multiple_of(crate::constants::MINIMUM_ALIGNMENT_BYTES as u64)) {
            use_direct_io = false;
        }
        if use_direct_io { open_opts.custom_flags(libc::O_DIRECT); }

        let df = open_opts.open(dst).await.map_err(FxcpError::Io)?;
        let dfd = df.as_raw_fd();

        let use_fixed = ring.submitter().register_files(&[sfd, dfd]).is_ok();
        if !use_fixed {
            warn!("io_uring: register_files failed in execute_copy_range, falling back to non-fixed opcodes");
        }

        let mut reflink_done = false;
        let mut stats = CopyStats::default();

        if strategy == CopyStrategy::Reflink {
             let reflink_res = Self::try_reflink_range(sfd, dfd, offset, length, offset, target_label.clone()).await;
             if let Ok(s) = reflink_res { stats = s; reflink_done = true; }
        }

        let target_is_nfs = target_caps.is_nfs.load(Ordering::Relaxed);

        let res = if reflink_done { Ok(stats) } else if !target_is_nfs {
            // register_files failed on a non-NFS local target — skip io_uring to avoid
            // per-segment stalls (up to PROCESS_SEGMENT_STALL_SECS each).
            debug!("io_uring: register_files failed on non-NFS target in copy_range, using copy_file_range/sendfile directly");
            let range_offset = offset;
            let range_len = length;
            let tl = target_label.clone();
            let fallback_res = spawn_blocking(move || {
                let start = Instant::now();
                let mut off_in = range_offset as i64;
                let mut off_out = range_offset as i64;
                let mut remaining = range_len;
                while remaining > 0 {
                    let chunk = remaining.min(crate::constants::COPY_FILE_RANGE_CHUNK_SIZE) as usize;
                    // SAFETY: sfd and dfd are valid open file descriptors.
                    let ret = unsafe {
                        libc::copy_file_range(sfd, &mut off_in, dfd, &mut off_out, chunk, 0)
                    };
                    if ret < 0 {
                        let err = std::io::Error::last_os_error();
                        if err.raw_os_error() == Some(libc::EXDEV) || err.raw_os_error() == Some(libc::EOPNOTSUPP) {
                            break;
                        }
                        return Err(FxcpError::Io(err));
                    }
                    if ret == 0 { break; }
                    remaining -= ret as u64;
                }
                if remaining == 0 {
                    return Ok(CopyStats {
                        bytes_processed: range_len,
                        bytes_zeros: 0,
                        io_duration: start.elapsed(),
                        ops_count: 1,
                    });
                }
                // SAFETY: sfd and dfd are valid. Seek to range start for sendfile fallback.
                unsafe { libc::lseek(sfd, range_offset as i64, libc::SEEK_SET) };
                unsafe { libc::lseek(dfd, range_offset as i64, libc::SEEK_SET) };
                let mut total: u64 = 0;
                let target = range_len - (range_len - remaining);
                while total < target {
                    let rem = target - total;
                    // SAFETY: sfd and dfd are valid open file descriptors.
                    let ret = unsafe {
                        libc::sendfile(dfd, sfd, std::ptr::null_mut(), rem as usize)
                    };
                    if ret < 0 { return Err(FxcpError::Io(std::io::Error::last_os_error())); }
                    if ret == 0 { break; }
                    total += ret as u64;
                }
                Ok(CopyStats {
                    bytes_processed: range_len,
                    bytes_zeros: 0,
                    io_duration: start.elapsed(),
                    ops_count: 1,
                })
            }).await.map_err(FxcpError::Join)?;
            match fallback_res {
                Ok(sf_stats) => {
                    metrics::COPY_METHOD_SENDFILE_FALLBACK.with_label_values(&[&tl]).inc();
                    Ok(sf_stats)
                }
                Err(e) => Err(e),
            }
        } else {
            let target_label_uring = target_label.clone();
            let uring_res = Self::perform_delta_uring_pipelined(
                ring, sfd, dfd, offset, length, vdo_opt, buffer_pool, atomic_pool, async_fd,
                dst.to_path_buf(), source_caps, target_caps, false, vdo_stall_threshold,
                source_uncached, target_uncached, governor, target_label_uring, fsync_tracker, buffer_limit, skip_fsync,
                segment_stall_timeout_secs, segment_overall_timeout_secs,
                use_fixed, None,
            ).await;

            match uring_res {
                Ok(s) => Ok(s),
                Err(uring_err) => {
                    let is_recoverable = matches!(&uring_err, FxcpError::Io(e)
                        if e.raw_os_error() == Some(libc::EBADF)
                        || e.raw_os_error() == Some(libc::EFAULT)
                        || e.kind() == io::ErrorKind::TimedOut);

                    if is_recoverable {
                        warn!("io_uring range copy failed ({}), falling back to copy_file_range/sendfile", uring_err);
                        let range_offset = offset;
                        let range_len = length;
                        let tl = target_label.clone();
                        let sendfile_res = spawn_blocking(move || {
                            let start = Instant::now();
                            let mut off_in = range_offset as i64;
                            let mut off_out = range_offset as i64;
                            let mut remaining = range_len;
                            while remaining > 0 {
                                let chunk = remaining.min(crate::constants::COPY_FILE_RANGE_CHUNK_SIZE) as usize;
                                // SAFETY: sfd and dfd are valid open file descriptors.
                                let ret = unsafe {
                                    libc::copy_file_range(sfd, &mut off_in, dfd, &mut off_out, chunk, 0)
                                };
                                if ret < 0 {
                                    let err = std::io::Error::last_os_error();
                                    if err.raw_os_error() == Some(libc::EXDEV) || err.raw_os_error() == Some(libc::EOPNOTSUPP) {
                                        break;
                                    }
                                    return Err(FxcpError::Io(err));
                                }
                                if ret == 0 { break; }
                                remaining -= ret as u64;
                            }
                            if remaining == 0 {
                                return Ok(CopyStats {
                                    bytes_processed: range_len,
                                    bytes_zeros: 0,
                                    io_duration: start.elapsed(),
                                    ops_count: 1,
                                });
                            }
                            // SAFETY: sfd is valid. Seek to range start for sendfile fallback.
                            unsafe { libc::lseek(sfd, range_offset as i64, libc::SEEK_SET) };
                            unsafe { libc::lseek(dfd, range_offset as i64, libc::SEEK_SET) };
                            let mut total: u64 = 0;
                            let target = range_len - (range_len - remaining);
                            while total < target {
                                let rem = target - total;
                                // SAFETY: sfd and dfd are valid open file descriptors.
                                let ret = unsafe {
                                    libc::sendfile(dfd, sfd, std::ptr::null_mut(), rem as usize)
                                };
                                if ret < 0 { return Err(FxcpError::Io(std::io::Error::last_os_error())); }
                                if ret == 0 { break; }
                                total += ret as u64;
                            }
                            Ok(CopyStats {
                                bytes_processed: range_len,
                                bytes_zeros: 0,
                                io_duration: start.elapsed(),
                                ops_count: 1,
                            })
                        }).await.map_err(FxcpError::Join)?;

                        match sendfile_res {
                            Ok(sf_stats) => {
                                metrics::COPY_METHOD_SENDFILE_FALLBACK.with_label_values(&[&tl]).inc();
                                Ok(sf_stats)
                            }
                            Err(_) => Err(uring_err),
                        }
                    } else {
                        Err(uring_err)
                    }
                }
            }
        };
        
        if use_fixed { let _ = ring.submitter().unregister_files(); }
        drop(sf); drop(df);
        res
    }

    async fn execute_full_copy_logic(
        src: &Path,
        target_path: &Path,
        ring: &mut IoUring,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        vdo_opt: bool,
        direct_io_ok: bool,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        use_atomic: bool,
        src_file_size: u64,
        vdo_stall_threshold: u32,
        _barrier_callback: &Option<Box<dyn Fn(u64) + Send + Sync>>,
        source_uncached: bool,
        target_uncached: bool,
        governor: Option<Arc<Governor>>,
        target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        is_sparse: bool,
        postcopy_ring: &mut IoUring,
        postcopy_async_fd: &Arc<AsyncFd<RawFd>>,
    ) -> Result<std::result::Result<CopyStats, FxcpError>> {
        debug!("execute_full_copy_logic: Start {:?} -> {:?}", src, target_path);

        if !src.exists() {
            return Err(FxcpError::Io(std::io::Error::new(
                std::io::ErrorKind::NotFound,
                format!("Source file no longer exists: {:?}", src)
            )));
        }

        let sf = Self::open_source_noatime(src).await?;
        let sfd = sf.as_raw_fd();
        
        let mut use_direct_io = direct_io_ok;
        if use_direct_io && !src_file_size.is_multiple_of(crate::constants::MINIMUM_ALIGNMENT_BYTES as u64) {
            use_direct_io = false;
        }

        let use_f2fs = target_caps.f2fs_atomic_legacy.load(Ordering::Relaxed);
        let path_clone = target_path.to_path_buf();
        
        let target_is_nfs = target_caps.is_nfs.load(Ordering::Relaxed);

        debug!("execute_full_copy_logic: Preparing destination...");
        let (_file_handle, dfd, cached_segments) = spawn_blocking(move || {
            Self::prepare_destination_file(path_clone, use_direct_io, use_f2fs, src_file_size, is_sparse, Some(sfd), target_is_nfs)
        }).await.map_err(FxcpError::Join)??;

        let use_fixed = ring.submitter().register_files(&[sfd, dfd]).is_ok();
        if use_fixed {
            debug!("execute_full_copy_logic: Registered fixed files (src={}, dst={})", sfd, dfd);
        } else {
            warn!("io_uring: register_files failed, falling back to non-fixed opcodes");
        }

        let mut transfer_done = false;
        let mut stats = CopyStats::default();
        let src_meta = std::fs::metadata(src).map_err(FxcpError::Io)?;
        let strategy = determine_copy_strategy(src, target_path, &src_meta, target_caps);

        if !use_atomic && !use_f2fs && !target_is_nfs && strategy == CopyStrategy::Reflink {
            debug!("execute_full_copy_logic: Attempting reflink...");
            let src_path_owned = src.to_path_buf();
            let reflink_res = Self::try_reflink(sfd, dfd, src_file_size, src_path_owned, target_label.clone()).await;
            if let Ok(reflink_stats) = reflink_res {
                stats = reflink_stats;
                transfer_done = true;
                debug!("execute_full_copy_logic: Reflink success");
            } 
            else {
                debug!("execute_full_copy_logic: Reflink failed/partial, falling back. Error: {:?}", reflink_res.err());
                // SAFETY: sfd and dfd are valid open file descriptors.
                // Resetting file positions to 0 before fallback copy.
                unsafe { libc::lseek(dfd, 0, libc::SEEK_SET); libc::lseek(sfd, 0, libc::SEEK_SET); };
            }
        }

        let mut result_val = Ok(stats);

        if !transfer_done
            && let Some(sf_stats) = try_nfs_sendfile(sfd, dfd, src_file_size, is_sparse, target_is_nfs, &target_label).await? {
                result_val = Ok(sf_stats);
                transfer_done = true;
            }

        if !transfer_done
            && let Some(sf_stats) = try_cachestat_sendfile(sfd, dfd, src_file_size, is_sparse, &target_label).await? {
                result_val = Ok(sf_stats);
                transfer_done = true;
            }

        // Local sendfile fast-path: for non-NFS, non-sparse targets, sendfile is faster
        // than io_uring (avoids 4KB buffer chunking overhead) and avoids EBADF on kernels
        // where register_files succeeds but fixed-fd operations fail (e.g. btrfs 7.3.0-rc0).
        // io_uring is only used for NFS (async pipelining) and sparse files (hole-aware I/O).
        if !transfer_done && !target_is_nfs && !is_sparse {
            debug!("Local non-sparse target: using sendfile (skipping io_uring)");
            let sf_size = src_file_size;
            let tl = target_label.clone();
            let sendfile_res = spawn_blocking(move || {
                sendfile_full(sfd, dfd, sf_size, "sendfile local")
            }).await.map_err(FxcpError::Join)?;
            match sendfile_res {
                Ok(sf_stats) => {
                    metrics::COPY_METHOD_SENDFILE_FALLBACK.with_label_values(&[&tl]).inc();
                    result_val = Ok(sf_stats);
                    transfer_done = true;
                }
                Err(e) => { result_val = Err(e); }
            }
        }

        if !transfer_done {
            debug!("execute_full_copy_logic: Starting io_uring pipeline...");
            let target_label_uring = target_label.clone();
            let uring_res = Self::perform_delta_uring_pipelined(
                ring, sfd, dfd, 0, src_file_size, vdo_opt, buffer_pool, atomic_pool, async_fd,
                target_path.to_path_buf(), source_caps, target_caps, use_atomic,
                vdo_stall_threshold, source_uncached, target_uncached, governor,
                target_label_uring, fsync_tracker, buffer_limit, skip_fsync,
                segment_stall_timeout_secs, segment_overall_timeout_secs,
                use_fixed, cached_segments,
            ).await;

            match uring_res {
                Ok(stats) => { result_val = Ok(stats); }
                Err(uring_err) => {
                    let is_recoverable = matches!(&uring_err, FxcpError::Io(e)
                        if e.raw_os_error() == Some(libc::EBADF)
                        || e.raw_os_error() == Some(libc::EFAULT)
                        || e.kind() == io::ErrorKind::TimedOut);

                    if is_recoverable {
                        warn!("io_uring pipeline failed ({}), falling back to sendfile", uring_err);
                        // SAFETY: sfd and dfd are valid open file descriptors held
                        // alive by sf and _file_handle in the calling scope.
                        // Reset positions and truncate destination for clean retry.
                        unsafe {
                            libc::lseek(sfd, 0, libc::SEEK_SET);
                            libc::ftruncate(dfd, 0);
                            libc::lseek(dfd, 0, libc::SEEK_SET);
                        };
                        if src_file_size > 0 && !is_sparse {
                            // SAFETY: dfd is valid. Re-allocate space for non-sparse retry.
                            let _ = unsafe { libc::fallocate(dfd, 0, 0, src_file_size as i64) };
                        }
                        let sf_size = src_file_size;
                        let tl = target_label.clone();
                        let sendfile_res = spawn_blocking(move || {
                            sendfile_full(sfd, dfd, sf_size, "sendfile fallback")
                        }).await.map_err(FxcpError::Join)?;

                        match sendfile_res {
                            Ok(sf_stats) => {
                                if is_sparse {
                                    warn!("Sparse file copied via sendfile fallback  --  holes not preserved");
                                }
                                metrics::COPY_METHOD_SENDFILE_FALLBACK.with_label_values(&[&tl]).inc();
                                result_val = Ok(sf_stats);
                            }
                            Err(sf_err) => {
                                error!("sendfile fallback also failed: {}", sf_err);
                                result_val = Err(uring_err);
                            }
                        }
                    } else {
                        result_val = Err(uring_err);
                    }
                }
            }
        }

        if use_f2fs {
            let should_commit = result_val.is_ok();
            let commit_res = spawn_blocking(move || {
                let res = if should_commit {
                    // SAFETY: dfd is a valid fd with an active F2FS atomic write transaction.
                    if unsafe { libc::ioctl(dfd, F2FS_IOC_COMMIT_ATOMIC_WRITE) } < 0 {
                        error!("F2FS: Atomic Commit Failed!");
                        Err(FxcpError::Io(std::io::Error::last_os_error()))
                    } else {
                        Ok(())
                    }
                } else {
                    // SAFETY: dfd is a valid fd. Aborting the atomic transaction on error.
                    let _ = unsafe { libc::ioctl(dfd, F2FS_IOC_ABORT_ATOMIC_WRITE) };
                    Ok(())
                };
                Self::set_f2fs_pinning(dfd, false);
                res
            }).await.map_err(FxcpError::Join)?;
            
            if let Err(e) = commit_res
                && result_val.is_ok() { result_val = Err(e); }
        }

        if use_fixed {
            let _ = ring.submitter().unregister_files();
        }

        // Post-copy fsync: reuse the already-open dfd instead of re-opening the temp file.
        // Saves 1 OPEN + 1 CLOSE RPC on NFS, 2 syscalls on local filesystems.
        if !skip_fsync
            && let Ok(ref _stats) = result_val {
                let fsync_sqe = opcode::Fsync::new(types::Fd(dfd))
                    .build()
                    .user_data(POSTCOPY_FSYNC_OP);
                match Self::submit_and_wait_single(postcopy_ring, fsync_sqe, postcopy_async_fd).await {
                    Ok(_) => {},
                    Err(FxcpError::Io(ref e)) if e.raw_os_error() == Some(libc::EBADF)
                        || e.raw_os_error() == Some(libc::EFAULT) => {
                        debug!("Post-copy io_uring fsync failed ({}), falling back to blocking fsync", e);
                        // SAFETY: dfd is a valid open fd from prepare_destination_file.
                        unsafe { libc::fsync(dfd) };
                    }
                    Err(e) => return Ok(Err(e)),
                }
            }

        // Post-copy FADV_DONTNEED on source: release page cache after copy completes.
        // Best-effort  --  any error is logged and ignored.
        {
            let fadv_sqe = opcode::Fadvise::new(types::Fd(sfd), 0, libc::POSIX_FADV_DONTNEED)
                .build()
                .user_data(FADVISE_OP);
            // SAFETY: the SQE is fully built with a valid fd.
            if unsafe { ring.submission().push(&fadv_sqe) }.is_ok() {
                let _ = ring.submit();
            }
            // Fire-and-forget: we do NOT wait for the Fadvise completion.
        }

        drop(sf);
        debug!("execute_full_copy_logic: Finished");
        Ok(result_val)
    }

    async fn try_reflink(sfd: i32, dfd: i32, src_file_size: u64, src_path: PathBuf, target_label: String) -> Result<CopyStats> {
        spawn_blocking(move || {
            let start = Instant::now();
            // SAFETY: sfd and dfd are valid open file descriptors.
            let ret = unsafe { libc::ioctl(dfd, FICLONE, sfd) };
            if ret == 0 {
                 metrics::COPY_METHOD_REFLINK.with_label_values(&[&target_label]).inc();
                 return Ok(CopyStats { bytes_processed: src_file_size, bytes_zeros: 0, io_duration: start.elapsed(), ops_count: 1 });
            }
            let err = std::io::Error::last_os_error();
            if err.raw_os_error() != Some(libc::EOPNOTSUPP) && err.raw_os_error() != Some(libc::EXDEV) {
                 warn!("Reflink failed on supported target ({}). Fallback copy active. Path: {:?}", err, src_path);
            } else {
                 debug!("Reflink not supported or cross-device ({}), falling back.", err);
            }
            
            let mut total_reflinked = 0usize;
            let size_usize: usize = src_file_size.try_into().unwrap_or(usize::MAX);
            let mut ops = 0;
            
            if src_file_size > 0 {
                let mut off_in = 0i64;
                let mut off_out = 0i64;
                while total_reflinked < size_usize {
                    let remaining = size_usize - total_reflinked;
                    let chunk = std::cmp::min(remaining, crate::constants::COPY_FILE_RANGE_CHUNK_SIZE as usize);
                    // SAFETY: sfd/dfd are valid fds. off_in/off_out are valid
                    // mutable references to offsets updated by the kernel.
                    let ret = unsafe { libc::copy_file_range(sfd, &mut off_in, dfd, &mut off_out, chunk, 0) };
                    if ret < 0 {
                        let os_err = std::io::Error::last_os_error();
                        return Err(FxcpError::Io(io::Error::new(
                            os_err.kind(),
                            format!("copy_file_range failed for {}: {}", target_label, os_err),
                        )));
                    } else if ret == 0 {
                        break;
                    }
                    total_reflinked += ret as usize;
                    ops += 1;
                }
            }
            
            if total_reflinked == size_usize {
                 let is_network_fs = match statfs::statfs(&src_path) {
                    Ok(s) => { let magic = s.filesystem_type().0 as u64; magic == NFS_SUPER_MAGIC || magic == 0x517B || magic == 0xFF534D42 },
                    Err(_) => false,
                };
                if is_network_fs { metrics::COPY_METHOD_OFFLOAD.with_label_values(&[&target_label]).inc(); } else { metrics::COPY_METHOD_REFLINK.with_label_values(&[&target_label]).inc(); }
                Ok(CopyStats { bytes_processed: src_file_size, bytes_zeros: 0, io_duration: start.elapsed(), ops_count: ops })
            } else {
                Err(FxcpError::Io(io::Error::other("Reflink partial copy or EOF")))
            }
        }).await.map_err(FxcpError::Join)?
    }

    async fn try_reflink_range(sfd: i32, dfd: i32, src_offset: u64, len: u64, dst_offset: u64, target_label: String) -> Result<CopyStats> {
        spawn_blocking(move || {
            let start = Instant::now();
            let args = FileCloneRange { src_fd: sfd as i64, src_offset, src_length: len, dest_offset: dst_offset };
            // SAFETY: dfd is a valid fd. args is a repr(C) FileCloneRange
            // struct with valid source fd, offset, and length.
            let ret = unsafe { libc::ioctl(dfd, FICLONERANGE, &args) };
            if ret == 0 {
                 metrics::COPY_METHOD_REFLINK.with_label_values(&[&target_label]).inc();
                 Ok(CopyStats { bytes_processed: len, bytes_zeros: 0, io_duration: start.elapsed(), ops_count: 1 })
            } else {
                 let err = std::io::Error::last_os_error();
                 if err.raw_os_error() != Some(libc::EOPNOTSUPP) && err.raw_os_error() != Some(libc::EXDEV) {
                     warn!("Reflink range failed on supported target ({}). Fallback copy active.", err);
                 }
                 Err(FxcpError::Io(io::Error::new(
                     err.kind(),
                     format!("FICLONERANGE ioctl failed for {}: {}", target_label, err),
                 )))
            }
        }).await.map_err(FxcpError::Join)?
    }

    fn check_atomic_invariants(caps: &Arc<Capabilities>, offset: u64, length: u32, buffer_addr: usize) -> Result<()> {
        let min = caps.atomic_min_bytes.load(Ordering::Relaxed) as u64;
        let max = caps.atomic_max_bytes.load(Ordering::Relaxed) as u64;
        
        if min == 0 || max == 0 {
            return Err(FxcpError::Io(io::Error::new(io::ErrorKind::Unsupported, "Atomic writes not supported by hardware")));
        }
        
        let len_u64 = length as u64;
        if !len_u64.is_power_of_two() {
            return Err(FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, format!("Atomic write length {} is not power of 2", len_u64))));
        }
        
        if len_u64 < min || len_u64 > max {
            return Err(FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, format!("Atomic write length {} out of bounds ({}-{})", len_u64, min, max))));
        }
        
        if !offset.is_multiple_of(len_u64) {
            return Err(FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, format!("Atomic write offset {} not aligned to length {}", offset, len_u64))));
        }
        
        if !(buffer_addr as u64).is_multiple_of(len_u64) {
            return Err(FxcpError::Io(io::Error::new(io::ErrorKind::InvalidInput, "Buffer memory not aligned for atomic write")));
        }
        
        Ok(())
    }

    fn map_sparse_segments(
        sfd: RawFd,
        offset: u64,
        length: u64
    ) -> Result<Vec<FileSegment>> {
        trace!("map_sparse_segments: sfd={}, offset={}, len={}", sfd, offset, length);
        let mut segments = Vec::new();
        let mut current_offset = offset;
        let end_offset = offset + length;

        while current_offset < end_offset {
            // SAFETY: sfd is a valid open file descriptor.
            let data_offset_res = unsafe { libc::lseek(sfd, current_offset as i64, libc::SEEK_DATA) };
            if data_offset_res < 0 {
                let err = std::io::Error::last_os_error();
                match err.raw_os_error() {
                    Some(libc::ENXIO) => {
                        let hole_len = end_offset - current_offset;
                        if hole_len > 0 {
                            segments.push(FileSegment::Hole { offset: current_offset, len: hole_len });
                        }
                        break;
                    },
                    _ => return Err(FxcpError::Io(err)),
                }
            }
            let data_offset = data_offset_res as u64;
            
            if data_offset >= end_offset {
                let hole_len = end_offset - current_offset;
                if hole_len > 0 {
                    segments.push(FileSegment::Hole { offset: current_offset, len: hole_len });
                }
                break;
            }

            if data_offset > current_offset {
                let hole_len = data_offset - current_offset;
                segments.push(FileSegment::Hole { offset: current_offset, len: hole_len });
                current_offset = data_offset;
            }

            // SAFETY: sfd is a valid open file descriptor.
            let hole_offset_res = unsafe { libc::lseek(sfd, current_offset as i64, libc::SEEK_HOLE) };
            let hole_offset = if hole_offset_res < 0 {
                end_offset
            } else {
                hole_offset_res as u64
            };

            let segment_end = hole_offset.min(end_offset);
            let segment_len = segment_end - current_offset;
            
            if segment_len > 0 {
                segments.push(FileSegment::Data { offset: current_offset, len: segment_len });
                current_offset += segment_len;
            } else {
                warn!("map_sparse_segments: Infinite loop guard triggered. off={} len={}", current_offset, segment_len);
                break;
            }
        }
        trace!("map_sparse_segments: found {} segments", segments.len());
        Ok(segments)
    }

    async fn perform_delta_uring_pipelined(
        ring: &mut IoUring,
        sfd: i32,
        dfd: i32,
        offset: u64,
        length: u64,
        vdo_opt: bool,
        buffer_pool: &mut BufferPool,
        mut atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        _debug_path: PathBuf,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        use_atomic: bool,
        _vdo_stall_threshold: u32,
        source_uncached: bool,
        target_uncached: bool,
        governor: Option<Arc<Governor>>,
        _target_label: String,
        fsync_tracker: &mut FsyncLatencyTracker,
        buffer_limit: Option<usize>,
        skip_fsync: bool,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        use_fixed: bool,
        cached_segments: Option<Vec<FileSegment>>,
    ) -> Result<CopyStats> {
        debug!("perform_delta_uring_pipelined: Start offset={} len={}", offset, length);
        let mut stats = CopyStats::default();
        let start_time = Instant::now();
        
        let use_seek_hole = source_caps.seek_hole.load(Ordering::Relaxed);

        if use_seek_hole {
            let segments = if let Some(segs) = cached_segments {
                debug!("perform_delta_uring_pipelined: Using pre-computed {} segments from probe", segs.len());
                segs
            } else {
                debug!("perform_delta_uring_pipelined: Mapping sparse segments...");
                let mapped = spawn_blocking(move || {
                    Self::map_sparse_segments(sfd, offset, length)
                }).await.map_err(FxcpError::Join)??;
                debug!("perform_delta_uring_pipelined: Mapped {} segments", mapped.len());
                mapped
            };

            for segment in segments {
                match segment {
                    FileSegment::Hole { offset, len } => {
                        stats.bytes_zeros += len;
                        if len >= crate::constants::MIN_HOLE_PUNCH_SIZE {
                            let falloc_op = if use_fixed {
                                opcode::Fallocate::new(types::Fixed(DST_FIXED_SLOT), len)
                                    .offset(offset)
                                    .mode(libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE)
                                    .build().user_data(HOLE_OP)
                            } else {
                                opcode::Fallocate::new(types::Fd(dfd), len)
                                    .offset(offset)
                                    .mode(libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE)
                                    .build().user_data(HOLE_OP)
                            };
                            
                            let mut pushed = false;
                            for _ in 0..10 {
                                // SAFETY: the SQE is fully built with valid fd and buffer references.
                                if unsafe { ring.submission().push(&falloc_op) }.is_ok() {
                                    pushed = true;
                                    break;
                                }
                                let _ = ring.submit();
                                if ring.submission().is_full() {
                                    match tokio::time::timeout(Duration::from_millis(crate::constants::IO_URING_READABLE_TIMEOUT_MS), async_fd.readable()).await {
                                        Ok(Ok(mut guard)) => {
                                            let mut buf = [0u8; 8];
                                            // SAFETY: reading 8 bytes from the eventfd to drain readiness notification.
                                            let _ = unsafe { libc::read(async_fd.get_ref().as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, 8) };
                                            guard.clear_ready();
                                        },
                                        Ok(Err(e)) => return Err(FxcpError::Io(e)),
                                        Err(_) => {
                                            if let Some(_) = ring.completion().next() {
                                                // drain
                                            }
                                        }
                                    }
                                }
                            }
                            if !pushed {
                                 return Err(FxcpError::Io(io::Error::other("Submission queue full (fallocate)")));
                            }
                        }
                    },
                    FileSegment::Data { offset, len } => {
                        let atomic_pool_ref = atomic_pool.as_deref_mut();
                        let seg_stats = Self::process_data_segment(
                            ring, sfd, dfd, offset, len, vdo_opt, buffer_pool, atomic_pool_ref, async_fd.clone(),
                            source_caps, target_caps, use_atomic,
                            source_uncached, target_uncached, &governor,
                            &_target_label, buffer_limit,
                            segment_stall_timeout_secs, segment_overall_timeout_secs,
                            use_fixed,
                        ).await?;
                        stats.bytes_processed += seg_stats.bytes_processed;
                        stats.bytes_zeros += seg_stats.bytes_zeros;
                        stats.ops_count += seg_stats.ops_count;
                    }
                }
            }
        } else {
            let segment_stats = Self::process_data_segment(
                ring, sfd, dfd, offset, length, vdo_opt, buffer_pool, atomic_pool, async_fd.clone(),
                source_caps, target_caps, use_atomic,
                source_uncached, target_uncached, &governor,
                &_target_label, buffer_limit,
                segment_stall_timeout_secs, segment_overall_timeout_secs,
                use_fixed,
            ).await?;
            stats.bytes_processed += segment_stats.bytes_processed;
            stats.bytes_zeros += segment_stats.bytes_zeros;
            stats.ops_count += segment_stats.ops_count;
        }

        uring_fsync_with_timeout(ring, dfd, use_fixed, &async_fd, fsync_tracker, &_debug_path, skip_fsync).await?;

        metrics::COPY_METHOD_STANDARD.with_label_values(&[&_target_label]).inc();
        stats.io_duration = start_time.elapsed();
        debug!("perform_delta_uring_pipelined: Finished");
        Ok(stats)
    }

    async fn process_data_segment(
        ring: &mut IoUring,
        sfd: i32,
        dfd: i32,
        start_offset: u64,
        total_len: u64,
        vdo_opt: bool,
        buffer_pool: &mut BufferPool,
        atomic_pool: Option<&mut BufferPool>,
        async_fd: Arc<AsyncFd<RawFd>>,
        source_caps: &Arc<Capabilities>,
        target_caps: &Arc<Capabilities>,
        use_atomic: bool,
        source_uncached: bool,
        target_uncached: bool,
        governor: &Option<Arc<Governor>>,
        _target_label: &String,
        buffer_limit: Option<usize>,
        segment_stall_timeout_secs: u64,
        segment_overall_timeout_secs: u64,
        use_fixed: bool,
    ) -> Result<CopyStats> {
        debug!("process_data_segment ENTRY: offset={}, len={}", start_offset, total_len);
        
        let mut bytes_processed = 0u64;
        let mut bytes_zeros = 0u64;
        let mut ops_count = 0u64;
        
        let (active_pool, is_atomic_buffer) = if use_atomic {
            if let Some(pool) = atomic_pool {
                (pool, true)
            } else {
                (buffer_pool, true)
            }
        } else {
            (buffer_pool, false)
        };

        let chunk_size = active_pool.chunk_size() as u64;
        let capacity = active_pool.capacity();
        let buffers_count_u64 = buffer_limit.unwrap_or(capacity).min(capacity) as u64;
        
        let mut inflight_ops = 0;
        let mut offset = start_offset;
        let mut length = total_len;
        let mut atomic_intent = vec![false; capacity];
        let mut context_map = vec![0u64; capacity];
        let mut encountered_error: Option<FxcpError> = None;
        let atomic_max = target_caps.atomic_max_bytes.load(Ordering::Relaxed) as u64;
        let mut pending_submissions = 0;
        
        let mut batched_fallback_counter = metrics::BatchedCounter::new(
            metrics::ATOMIC_WRITE_FALLBACKS.clone(),
            64
        );

        let loop_start = std::time::Instant::now();
        let mut last_progress = std::time::Instant::now();
        let mut last_bytes_processed = 0u64;

        while length > 0 || inflight_ops > 0 {
            // Early exit: error already encountered and nothing in flight to drain.
            // Without this, the loop spins for segment_overall_timeout_secs (60s)
            // waiting for progress that can never happen (e.g. EBADF from io_uring).
            if encountered_error.is_some() && inflight_ops == 0 {
                break;
            }

            if loop_start.elapsed() > std::time::Duration::from_secs(segment_overall_timeout_secs) {
                error!("process_data_segment: Segment timeout after {}s (offset={}, len={}, inflight={})",
                       segment_overall_timeout_secs, offset, length, inflight_ops);
                return Err(FxcpError::Io(io::Error::new(
                    io::ErrorKind::TimedOut,
                    format!("Segment timeout: {}s with {} ops in flight", segment_overall_timeout_secs, inflight_ops)
                )));
            }

            if bytes_processed > last_bytes_processed {
                last_progress = std::time::Instant::now();
                last_bytes_processed = bytes_processed;
            } else if inflight_ops > 0 && last_progress.elapsed() > std::time::Duration::from_secs(segment_stall_timeout_secs) {
                error!("process_data_segment: STALL detected - no progress for {}s (inflight={}, free_buffers={}, sq_capacity={})",
                       segment_stall_timeout_secs, inflight_ops, active_pool.free_count(), ring.submission().capacity());
                return Err(FxcpError::Io(io::Error::other(
                    format!("Stall: no progress for {}s (inflight={}, free={})",
                            segment_stall_timeout_secs, inflight_ops, active_pool.free_count())
                )));
            }

            if let Some(gov) = &governor
                && gov.current_memory_usage_pct() > 0.90 { tokio::task::yield_now().await; }

            while inflight_ops < buffers_count_u64 && length > 0 && encountered_error.is_none() {
                 if ring.submission().is_full() {
                     if pending_submissions > 0 {
                         if let Err(e) = ring.submit() {
                             encountered_error = Some(FxcpError::Io(io::Error::other(e.to_string())));
                             break;
                         }
                         pending_submissions = 0;
                     }
                     if ring.submission().is_full() {
                         break;
                     }
                 }

                 let index = if let Some(idx) = active_pool.acquire() { idx } else {
                     break;
                 };
                 
                 if index as usize >= active_pool.capacity() {
                     error!("BufferPool Index Out of Bounds: {} (Capacity: {})", index, active_pool.capacity());
                     return Err(FxcpError::MemoryExhausted("BufferPool index corrupted".into()));
                 }

                 let mut current_len = u32::try_from((length as usize).min(chunk_size as usize))
                     .map_err(|_| FxcpError::Config(format!(
                         "io_uring chunk length {} exceeds u32::MAX", (length as usize).min(chunk_size as usize)
                     )))?;
                 
                 if use_atomic && is_atomic_buffer && atomic_max > 0 {
                     current_len = current_len.min(u32::try_from(atomic_max)
                         .map_err(|_| FxcpError::Config(format!(
                             "atomic_max {} exceeds u32::MAX", atomic_max
                         )))?);
                 }
                 let len = current_len;

                 if use_atomic && is_atomic_buffer {
                    let ptr_opt = active_pool.get_ptr(index);
                    if let Some(ptr) = ptr_opt {
                        let ptr_addr = ptr as usize;
                        if let Err(e) = Self::check_atomic_invariants(target_caps, offset, len, ptr_addr) {
                            active_pool.release(index);
                            return Err(e);
                        }
                    } else {
                        active_pool.release(index);
                        error!("BufferPool returned null pointer for index {}", index);
                        return Err(FxcpError::MemoryExhausted("BufferPool null pointer".into()));
                    }
                    atomic_intent[index as usize] = true;
                 } else {
                    atomic_intent[index as usize] = false;
                 }

                 let mut read_flags: i32 = 0;
                 if source_uncached && source_caps.uncached_io.load(Ordering::Relaxed) { read_flags |= RWF_UNCACHED; }
                 
                 context_map[index as usize] = offset;
                 let buf_ptr = match active_pool.get_ptr(index) {
                     Some(p) => p,
                     None => {
                         active_pool.release(index);
                         error!("BufferPool returned null pointer for index {}", index);
                         return Err(FxcpError::MemoryExhausted("BufferPool null pointer".into()));
                     }
                 };

                 let user_data = READ_OP | (index as u64);
                 let read_op = if use_fixed {
                    opcode::ReadFixed::new(types::Fixed(SRC_FIXED_SLOT), buf_ptr, len, index)
                        .offset(offset).rw_flags(read_flags).build().user_data(user_data)
                 } else {
                    opcode::Read::new(types::Fd(sfd), buf_ptr, len)
                        .offset(offset).rw_flags(read_flags).build().user_data(user_data)
                 };

                 // SAFETY: the SQE is fully built with valid fd and buffer references.
                 if unsafe { ring.submission().push(&read_op) }.is_err() {
                     active_pool.release(index);
                     if pending_submissions > 0 {
                         let _ = ring.submit();
                         pending_submissions = 0;
                     }
                     break;
                 }
                 inflight_ops += 1;
                 pending_submissions += 1;

                 offset += len as u64;
                 length -= len as u64;
                 ops_count += 1;
            }

            if pending_submissions > 0 {
                if let Err(e) = ring.submit()
                    && encountered_error.is_none() {
                        encountered_error = Some(FxcpError::Io(io::Error::other(e.to_string())));
                    }
                pending_submissions = 0;
            }

            loop {
                let peek_cqe = ring.completion().next();
                
                let cqe = if let Some(c) = peek_cqe {
                    c
                } else {
                    if length > 0 && active_pool.free_count() > 0 && !ring.submission().is_full() {
                        break;
                    }
                    
                    if inflight_ops > 0 {
                        static TIMEOUT_STREAK: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

                        match tokio::time::timeout(Duration::from_millis(crate::constants::IO_URING_READABLE_TIMEOUT_MS), async_fd.readable()).await {
                            Ok(Ok(mut guard)) => {
                                TIMEOUT_STREAK.store(0, std::sync::atomic::Ordering::Relaxed);
                                let mut buf = [0u8; 8];
                                // SAFETY: reading 8 bytes from the eventfd to drain readiness notification.
                                let _ = unsafe { libc::read(async_fd.get_ref().as_raw_fd(), buf.as_mut_ptr() as *mut libc::c_void, 8) };
                                guard.clear_ready();
                                if let Some(c) = ring.completion().next() { c } else { continue; }
                            },
                            Ok(Err(e)) => return Err(FxcpError::Io(e)),
                            Err(_timeout) => {
                                let streak = TIMEOUT_STREAK.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
                                if streak >= crate::constants::IOURING_COMPLETION_TIMEOUT_STREAK {
                                    let sq_len = ring.submission().len();
                                    let sq_cap = ring.submission().capacity();
                                    let cq_len = ring.completion().len();
                                    error!("process_data_segment: io_uring hung - no completions for 60s (inflight={}, sq_len={}, sq_cap={}, cq_len={})",
                                           inflight_ops, sq_len, sq_cap, cq_len);
                                    return Err(FxcpError::Io(io::Error::new(
                                        io::ErrorKind::TimedOut,
                                        "io_uring hung - no completions for 60s"
                                    )));
                                }
                                ring.submit()?;
                                if let Some(c) = ring.completion().next() { c } else { continue; }
                            }
                        }
                    } else {
                        break;
                    }
                };

                let user_data = cqe.user_data();
                if user_data == 0 {
                    continue;
                }

                let op_type = user_data & OP_TYPE_MASK;
                if op_type == HOLE_OP {
                    continue;
                }

                let index = (user_data & INDEX_MASK) as u16;
                let original_offset = if (index as usize) < context_map.len() {
                    context_map[index as usize]
                } else {
                    0
                };

                if cqe.result() < 0 {
                    let raw_err = -cqe.result();

                    if op_type == WRITE_OP && atomic_intent[index as usize]
                        && (raw_err == libc::EOPNOTSUPP || raw_err == libc::EINVAL) {
                            if raw_err == libc::EOPNOTSUPP {
                                warn!("Atomic Write Failed (EOPNOTSUPP). Disabling Atomic Writes for this session.");
                                target_caps.atomic_writes.store(false, Ordering::Relaxed);
                            } else {
                                warn!("Atomic Write Failed for offset {}: {} (Errno {}). Downgrading to Buffered I/O.", context_map[index as usize], io::Error::from_raw_os_error(raw_err), raw_err);
                            }
                            
                            batched_fallback_counter.inc();
                            atomic_intent[index as usize] = false;
                            
                            let retry_len = active_pool.get_len(index) as u32;
                            let retry_offset = context_map[index as usize];
                            
                            let mut write_flags: i32 = 0;
                            if target_uncached && target_caps.uncached_io.load(Ordering::Relaxed) { write_flags |= RWF_UNCACHED; }
                            
                            let user_data_write = WRITE_OP | (index as u64);
                            if let Some(buf_ptr) = active_pool.get_ptr(index) {
                                // SAFETY: buf_ptr is valid for retry_len bytes from the pool; fd/slot registered with ring.
                                let write_op = unsafe { build_write_sqe(use_fixed, dfd, buf_ptr, retry_len, index as u16, retry_offset, write_flags, user_data_write) };
                                
                                // SAFETY: the SQE is fully built with valid fd and buffer references.
                                if unsafe { ring.submission().push(&write_op) }.is_ok() {
                                    continue;
                                } else {
                                    encountered_error = Some(FxcpError::Io(io::Error::other("SQ full during atomic retry")));
                                }
                            } else {
                                error!("io_uring buffer pool index out of range (atomic retry): index={}", index);
                                encountered_error = Some(FxcpError::Io(io::Error::other("io_uring buffer pool index out of range (atomic retry)")));
                            }
                        }

                    if op_type != FALLOC_OP {
                        active_pool.release(index);
                    }
                    
                    if encountered_error.is_none() {
                        let err = io::Error::from_raw_os_error(-cqe.result());
                        if err.kind() != io::ErrorKind::Interrupted {
                             encountered_error = Some(FxcpError::Io(err));
                        }
                    }
                    // Only decrement for ops that were tracked in inflight_ops.
                    // FADVISE_OP is fire-and-forget (never incremented inflight_ops).
                    if op_type != FADVISE_OP {
                        inflight_ops -= 1;
                    }
                    continue;
                }

                if encountered_error.is_some() {
                    if op_type != FALLOC_OP {
                        active_pool.release(index);
                    }
                    // Only decrement for ops that were tracked in inflight_ops.
                    if op_type != FADVISE_OP {
                        inflight_ops -= 1;
                    }
                    continue;
                }

                let bytes_transferred = cqe.result() as u64;

                match op_type {
                    READ_OP => {
                        inflight_ops -= 1;
                        let mut wf: i32 = 0;
                        if target_uncached && target_caps.uncached_io.load(Ordering::Relaxed) { wf |= RWF_UNCACHED; }
                        if target_caps.atomic_writes.load(Ordering::Relaxed) && atomic_intent[index as usize] { wf |= RWF_ATOMIC; }

                        match handle_read_completion(ring, active_pool, index, bytes_transferred, original_offset, dfd, use_fixed, vdo_opt, wf) {
                            Ok((zeros_delta, processed_delta)) => {
                                pending_submissions += 1;
                                inflight_ops += 1;
                                bytes_zeros += zeros_delta;
                                bytes_processed += processed_delta;
                            }
                            Err(e) => {
                                encountered_error = Some(e);
                                continue;
                            }
                        }
                    },
                    WRITE_OP => {
                        inflight_ops -= 1;
                        bytes_processed += bytes_transferred;
                        active_pool.release(index);
                    },
                    FALLOC_OP => {
                        inflight_ops -= 1;
                        active_pool.release(index);
                    }
                    _ => {
                        warn!("process_data_segment: Unknown CQE op type {:x} (user_data={:x}), not tracked in inflight_ops — ignoring", op_type, user_data);
                        // Do NOT decrement inflight_ops — unknown ops were never
                        // incremented. Decrementing causes u64 underflow to MAX,
                        // which hangs the loop for 60s waiting for completions.
                    }
                }
            }
        }

        if let Some(err) = encountered_error {
            debug!("process_data_segment EXIT with ERROR: {:?}", err);
            return Err(err);
        }

        debug!("process_data_segment EXIT SUCCESS");
        Ok(CopyStats { bytes_processed, bytes_zeros, io_duration: Duration::ZERO, ops_count })
    }

    #[inline(always)]
    fn _is_block_zero(buf: &[u8]) -> bool {
        is_zero_block(buf)
    }
}

/// Apply or release a flock on a file, tracking the lock handle by inode in `lock_map`
pub fn apply_lock(path: &Path, lock_type: u32, lock_map: &DashMap<u64, std::fs::File>) -> Result<()> {
    if (lock_type & libc::LOCK_UN as u32) != 0 {
         if let Ok(meta) = std::fs::metadata(path) { lock_map.remove(&meta.ino()); }
         return Ok(());
    }
    
    let file = std::fs::File::open(path).map_err(FxcpError::Io)?;
    let fd = file.as_raw_fd();
    
    // SAFETY: fd is a valid open file descriptor. LOCK_NB makes the
    // call non-blocking so it cannot deadlock.
    let ret = unsafe { libc::flock(fd, (lock_type as i32) | libc::LOCK_NB) };
    
    if ret == 0 { if let Ok(meta) = file.metadata() { lock_map.insert(meta.ino(), file); } }
    else { debug!("Failed to apply lock on {:?}: {}", path, std::io::Error::last_os_error()); }

    Ok(())
}

// ---------------------------------------------------------------------------
// Delta copy  --  transfer only dirty ranges identified by Merkle tree diff
// ---------------------------------------------------------------------------
impl SmartCopier {
    /// Copy only the specified dirty ranges from src to dst.
    /// Each DirtyRange triggers an `optimized_copy_range()` call.
    pub async fn copy_delta(
        &mut self,
        src: &std::path::Path,
        dst: &std::path::Path,
        dirty_ranges: &[crate::hashing::DirtyRange],
        file_size: u64,
        label: &str,
    ) -> Result<CopyStats> {
        let mut total_stats = CopyStats::default();

        for range in dirty_ranges {
            if range.offset >= file_size {
                continue;
            }
            let stats = self.optimized_copy_range(
                src.to_path_buf(),
                dst.to_path_buf(),
                range.offset,
                range.length,
                file_size,
                label.to_string(),
                None,
                self.skip_fsync,
            ).await?;
            total_stats.bytes_processed += stats.bytes_processed;
            total_stats.ops_count += stats.ops_count;
        }

        if let Ok(dst_meta) = std::fs::metadata(dst)
            && dst_meta.len() > file_size {
                let stats = self.optimized_truncate(dst.to_path_buf(), file_size).await?;
                total_stats.bytes_processed += stats.bytes_processed;
                total_stats.ops_count += stats.ops_count;
            }

        Ok(total_stats)
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    use super::*;

    #[tokio::test]
    #[ignore = "flaky under parallel NFS load; run isolated with RUST_TEST_THREADS=1"]
    async fn test_iouring_short_read_no_corruption() {
        crate::metrics::initialize_metrics(512);

        // Use /var/tmp (real filesystem) instead of /tmp (tmpfs)  --  io_uring
        // intermittently returns ENOENT on tmpfs under parallel test load.
        let dir = tempfile::Builder::new()
            .prefix("fxcp_test_")
            .tempdir_in("/var/tmp")
            .unwrap();
        let src_path = dir.path().join("source.bin");
        let dst_path = dir.path().join("dest.bin");

        let data: Vec<u8> = (0..100_001u32).map(|i| (i % 251) as u8).collect();
        std::fs::write(&src_path, &data).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();
        copier
            .optimized_copy(
                src_path.clone(),
                dst_path.clone(),
                100_001,
                "test".to_string(),
                None,
                true,
                false,
            )
            .await
            .unwrap();

        let src_hash = blake3::hash(&std::fs::read(&src_path).unwrap());
        let dst_hash = blake3::hash(&std::fs::read(&dst_path).unwrap());
        assert_eq!(
            src_hash, dst_hash,
            "Copy must be byte-identical (short read corruption check)"
        );

        let src_len = std::fs::metadata(&src_path).unwrap().len();
        let dst_len = std::fs::metadata(&dst_path).unwrap().len();
        assert_eq!(src_len, dst_len, "File sizes must match");
    }

    #[tokio::test]
    async fn test_copy_delta_truncates_on_shrink() {
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src.bin");
        let dst = dir.path().join("dst.bin");
        std::fs::write(&src, vec![0xAAu8; 100_000]).unwrap();
        std::fs::write(&dst, vec![0xAAu8; 200_000]).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();
        copier.copy_delta(&src, &dst, &[], 100_000, "test").await.unwrap();

        let dst_len = std::fs::metadata(&dst).unwrap().len();
        assert_eq!(dst_len, 100_000, "dst should be truncated to source size");
    }

    #[tokio::test]
    async fn test_copy_delta_shrink_to_zero() {
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src.bin");
        let dst = dir.path().join("dst.bin");
        std::fs::write(&src, b"").unwrap();
        std::fs::write(&dst, vec![0xBBu8; 100_000]).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();
        copier.copy_delta(&src, &dst, &[], 0, "test").await.unwrap();

        let dst_len = std::fs::metadata(&dst).unwrap().len();
        assert_eq!(dst_len, 0, "dst should be truncated to 0");
    }

    #[tokio::test]
    async fn test_copy_delta_no_truncate_on_grow() {
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src.bin");
        let dst = dir.path().join("dst.bin");
        let src_data = vec![0xCCu8; 200_000];
        std::fs::write(&src, &src_data).unwrap();
        std::fs::write(&dst, vec![0xCCu8; 100_000]).unwrap();

        let dirty = vec![crate::hashing::DirtyRange { offset: 100_000, length: 100_000 }];
        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();
        copier.copy_delta(&src, &dst, &dirty, 200_000, "test").await.unwrap();

        let dst_len = std::fs::metadata(&dst).unwrap().len();
        assert_eq!(dst_len, 200_000, "dst should grow to 200KB, not be truncated");
    }

    #[test]
    fn test_prepare_destination_file_sparse_skip() {
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("sparse_target.dat");

        // is_sparse=true: fallocate should be skipped
        let (file, _fd, _) = SmartCopier::prepare_destination_file(
            path.clone(), false, false, 1_048_576, true, None, false
        ).unwrap();
        drop(file);

        use std::os::unix::fs::MetadataExt;
        let meta = std::fs::metadata(&path).unwrap();
        // With is_sparse=true, no fallocate -> minimal blocks allocated
        assert_eq!(meta.len(), 0, "truncate(true) sets size to 0, no fallocate for sparse");
        assert!(meta.blocks() < 16, "sparse file should have minimal block allocation");
    }

    #[test]
    fn test_prepare_destination_file_non_sparse_preallocates() {
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("dense_target.dat");

        // is_sparse=false: fallocate should be called
        let (file, _fd, _) = SmartCopier::prepare_destination_file(
            path.clone(), false, false, 65536, false, None, false
        ).unwrap();
        drop(file);

        use std::os::unix::fs::MetadataExt;
        let meta = std::fs::metadata(&path).unwrap();
        // On filesystems supporting fallocate, blocks should be pre-allocated.
        // 65536 bytes / 512 bytes per block = 128 blocks minimum.
        // Some FS may report more due to alignment. On tmpfs fallocate may not
        // allocate blocks, so we just verify the file was created successfully.
        assert!(meta.len() == 0 || meta.blocks() > 0,
            "non-sparse destination should have blocks after fallocate (or be on tmpfs)");
    }

    #[tokio::test]
    async fn test_inflight_counter_balanced_after_copy() {
        // Tests that inflight_ops reaches 0 after a copy completes.
        // If inflight_ops underflows or leaks, the copy loop hangs until timeout.
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src_path = dir.path().join("source_balanced.bin");
        let dst_path = dir.path().join("dest_balanced.bin");

        // Use a size that exercises multiple io_uring chunks (>64KB)
        let data: Vec<u8> = (0..200_003u32).map(|i| (i % 251) as u8).collect();
        std::fs::write(&src_path, &data).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();
        // If inflight_ops doesn't reach 0, this will hang until the stall timeout
        copier
            .optimized_copy(
                src_path.clone(),
                dst_path.clone(),
                200_003,
                "test_balanced".to_string(),
                None,
                true,
                false,
            )
            .await
            .unwrap();

        // Verify byte-identical copy (proves the loop completed correctly)
        let src_hash = blake3::hash(&std::fs::read(&src_path).unwrap());
        let dst_hash = blake3::hash(&std::fs::read(&dst_path).unwrap());
        assert_eq!(src_hash, dst_hash, "Copy must be byte-identical (inflight balance check)");
    }

    #[tokio::test]
    async fn test_inflight_unknown_op_type_no_hang() {
        // Regression test for the _ => arm bug in process_data_segment.
        // Before the fix: unknown CQE op types caused inflight_ops to never reach 0,
        // hanging the loop for 60 seconds (600 * 100ms timeouts).
        // After the fix: unknown op types decrement inflight_ops and release the buffer.
        //
        // This test verifies the copy pipeline completes within a tight timeout.
        // If the _ => arm bug were triggered, this would timeout at 10s instead of 60s.
        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src_path = dir.path().join("source_nohang.bin");
        let dst_path = dir.path().join("dest_nohang.bin");

        // 150KB — exercises io_uring Tier 3 path (>64KB)
        let data: Vec<u8> = (0..153_600u32).map(|i| (i % 127) as u8).collect();
        std::fs::write(&src_path, &data).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();

        // 10-second timeout: well above normal copy time (~100ms), well below the
        // 60-second hang that the _ => arm bug would cause.
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            copier.optimized_copy(
                src_path.clone(),
                dst_path.clone(),
                153_600,
                "test_nohang".to_string(),
                None,
                true,
                false,
            ),
        )
        .await;

        assert!(result.is_ok(), "Copy timed out — possible inflight_ops hang (unknown op type bug)");
        assert!(result.unwrap().is_ok(), "Copy failed with error");

        let src_hash = blake3::hash(&std::fs::read(&src_path).unwrap());
        let dst_hash = blake3::hash(&std::fs::read(&dst_path).unwrap());
        assert_eq!(src_hash, dst_hash, "Copy must be byte-identical");
    }

    #[tokio::test]
    async fn test_separate_postcopy_ring_no_cqe_leak() {
        use std::os::unix::io::AsRawFd;
        let ring = io_uring::IoUring::new(256).unwrap();
        let ring_fd = ring.as_raw_fd();
        let postcopy_ring: io_uring::IoUring = io_uring::IoUring::builder()
            .setup_attach_wq(ring_fd)
            .build(8)
            .unwrap();
        assert_ne!(ring.as_raw_fd(), postcopy_ring.as_raw_fd(),
            "postcopy ring must have a different fd from data ring");

        crate::metrics::initialize_metrics(512);
        let dir = tempfile::tempdir().unwrap();
        let src_path = dir.path().join("src_postcopy.bin");
        let dst_path = dir.path().join("dst_postcopy.bin");
        let data: Vec<u8> = (0..128 * 1024u32).map(|i| (i % 251) as u8).collect();
        std::fs::write(&src_path, &data).unwrap();

        let mut copier = crate::sync::create_copier(dir.path(), dir.path())
            .await
            .unwrap();

        let result = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            copier.optimized_copy(
                src_path.clone(), dst_path.clone(),
                128 * 1024, "test_postcopy".to_string(),
                None, false, false,
            ),
        ).await;

        assert!(result.is_ok(), "Copy timed out — postcopy ring may be broken");
        assert!(result.unwrap().is_ok(), "Copy failed");
        let src_hash = blake3::hash(&std::fs::read(&src_path).unwrap());
        let dst_hash = blake3::hash(&std::fs::read(&dst_path).unwrap());
        assert_eq!(src_hash, dst_hash, "Copy must be byte-identical");
    }

    #[test]
    fn test_iouring_read_write_fallback_no_ebadf() {
        // Given: a temp file with known content and an io_uring ring WITHOUT register_files
        use std::os::unix::io::AsRawFd;
        let dir = tempfile::Builder::new()
            .prefix("fxcp_iouring_fallback_")
            .tempdir_in("/var/tmp")
            .unwrap();
        let src_path = dir.path().join("read_test.bin");
        let dst_path = dir.path().join("write_test.bin");
        let payload = b"EBADF fallback verification payload";
        std::fs::write(&src_path, payload).unwrap();
        std::fs::write(&dst_path, vec![0u8; payload.len()]).unwrap();

        let src_file = std::fs::File::open(&src_path).unwrap();
        let dst_file = std::fs::OpenOptions::new()
            .write(true)
            .open(&dst_path)
            .unwrap();
        let sfd = src_file.as_raw_fd();
        let dfd = dst_file.as_raw_fd();

        let mut ring = io_uring::IoUring::new(8).unwrap();
        // Deliberately NOT calling register_files — simulates the use_fixed=false path

        // When: submitting Read(Fd) then Write(Fd) opcodes with raw fds
        let mut read_buf = vec![0u8; payload.len()];
        let read_sqe = opcode::Read::new(types::Fd(sfd), read_buf.as_mut_ptr(), read_buf.len() as u32)
            .offset(0)
            .build()
            .user_data(1);

        unsafe { ring.submission().push(&read_sqe).unwrap(); }
        ring.submit_and_wait(1).unwrap();

        let read_cqe = ring.completion().next().unwrap();
        // Then: Read succeeds (no EBADF)
        assert!(
            read_cqe.result() >= 0,
            "Read(Fd) without register_files returned error {}: expected success (EBADF root cause would give -9)",
            read_cqe.result()
        );
        assert_eq!(read_cqe.result() as usize, payload.len());

        let write_sqe = opcode::Write::new(types::Fd(dfd), read_buf.as_ptr(), read_buf.len() as u32)
            .offset(0)
            .build()
            .user_data(2);

        unsafe { ring.submission().push(&write_sqe).unwrap(); }
        ring.submit_and_wait(1).unwrap();

        let write_cqe = ring.completion().next().unwrap();
        // Then: Write succeeds (no EBADF)
        assert!(
            write_cqe.result() >= 0,
            "Write(Fd) without register_files returned error {}: expected success (EBADF root cause would give -9)",
            write_cqe.result()
        );
        assert_eq!(write_cqe.result() as usize, payload.len());

        // Then: written content matches source
        let written = std::fs::read(&dst_path).unwrap();
        assert_eq!(&written, payload, "Written content must match source payload");
    }

    #[test]
    fn test_reflink_or_copy_preserves_content() {
        // Verify byte-for-byte content match after reflink_or_copy
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src.bin");
        let dst = dir.path().join("dst.bin");
        let data: Vec<u8> = (0..100_000u32).map(|i| (i % 251) as u8).collect();
        std::fs::write(&src, &data).unwrap();
        let bytes = reflink_or_copy(&src, &dst).unwrap();
        assert_eq!(bytes, data.len() as u64);
        let result = std::fs::read(&dst).unwrap();
        assert_eq!(result, data, "reflink_or_copy must preserve content exactly");
    }

    #[test]
    fn test_reflink_or_copy_zero_byte_file() {
        // Zero-byte files must be handled without error
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("empty.bin");
        let dst = dir.path().join("empty_dst.bin");
        std::fs::write(&src, b"").unwrap();
        let bytes = reflink_or_copy(&src, &dst).unwrap();
        assert_eq!(bytes, 0);
        assert_eq!(std::fs::metadata(&dst).unwrap().len(), 0);
    }

    #[test]
    fn test_reflink_or_copy_missing_src_returns_error() {
        // Missing source must return an error, not panic
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("nonexistent.bin");
        let dst = dir.path().join("dst.bin");
        let result = reflink_or_copy(&src, &dst);
        assert!(result.is_err(), "missing src must return Err");
    }

    #[test]
    fn test_reflink_or_copy_same_device_xfs_shared_extents() {
        // On XFS with reflink=1, verify FICLONE fires (shared extents).
        // Skip if /var/tmp is not XFS or doesn't support reflink.
        use std::process::Command;
        let dir = tempfile::Builder::new()
            .prefix("fxcp_reflink_test_")
            .tempdir_in("/var/tmp")
            .unwrap();
        let src = dir.path().join("src.bin");
        let dst = dir.path().join("dst.bin");
        // 64KB — above the 4096-byte threshold in determine_copy_strategy
        let data = vec![0xABu8; 65536];
        std::fs::write(&src, &data).unwrap();
        let bytes = reflink_or_copy(&src, &dst).unwrap();
        assert_eq!(bytes, 65536);
        // Verify content
        assert_eq!(std::fs::read(&dst).unwrap(), data);
        // Check if filefrag shows shared extents (XFS reflink=1 only)
        let out = Command::new("filefrag").arg("-v").arg(&dst).output();
        if let Ok(out) = out {
            let s = String::from_utf8_lossy(&out.stdout);
            if s.contains("shared") {
                // FICLONE fired — shared extents confirmed
                println!("FICLONE confirmed: shared extents detected");
            }
            // If no "shared" — FICLONE fell back to sendfile (cross-device or non-reflink FS)
            // This is acceptable — the test verifies content correctness regardless
        }
    }
}