lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

// Allow common Rust patterns that clippy flags but are intentional
#![allow(clippy::module_inception)] // foo/mod.rs with foo/foo.rs is common
#![allow(clippy::should_implement_trait)] // from_str() methods intentionally named
#![allow(clippy::too_many_arguments)] // Some constructors need many params

//! # LCPFS - LCP File System
//!
//! A modern, ZFS-inspired copy-on-write filesystem implementation in pure Rust,
//! designed for `no_std` environments such as operating system kernels. Features
//! post-quantum cryptography, CXL memory tiering, and ML-based prefetching.
//!
//! ## Overview
//!
//! LCPFS provides enterprise-grade storage features with cutting-edge optimizations:
//!
//! ### Core Features
//! - **Copy-on-Write (COW)**: All writes create new blocks, preserving data integrity
//! - **RAID-Z (1/2/3)**: Multi-disk redundancy with self-healing capabilities
//! - **Adaptive Replacement Cache (ARC)**: 100 GiB self-tuning cache with ghost lists
//! - **L2ARC**: Persistent SSD-backed second-level cache
//! - **Deduplication**: Block-level dedup with fast RAM-only DDT for hot data
//! - **Tiered Compression**: LZ4/ZSTD/LZMA with automatic selection
//! - **Checksums**: BLAKE3/SHA-256 integrity verification for every block
//! - **Snapshots**: Instant, space-efficient point-in-time copies
//! - **Post-Quantum Crypto**: Kyber-1024 + Hybrid KEM for quantum resistance
//!
//! ### Advanced Features
//! - **CXL Memory Tiering**: Automatic data placement across DRAM/CXL/Storage based on temperature
//! - **Computational Storage**: Offload compression/checksums to smart storage devices (~80% CPU savings)
//! - **ML-based Prefetching**: Neural network predicts I/O patterns (4→8→5 architecture)
//! - **dRAID**: Distributed spare across pool for faster rebuilds
//! - **Direct I/O**: Cache bypass for large sequential operations
//!
//! ## Architecture
//!
//! LCPFS follows ZFS's layered architecture:
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────┐
//! │                    ZPL (POSIX Layer)                    │
//! │              Files, Directories, Attributes             │
//! ├─────────────────────────────────────────────────────────┤
//! │                    DMU (Data Management)                │
//! │              Objects, Transactions, Datasets            │
//! ├─────────────────────────────────────────────────────────┤
//! │                    ARC (Adaptive Cache)                 │
//! │              Block Caching, Prefetch, L2ARC             │
//! ├─────────────────────────────────────────────────────────┤
//! │                    SPA (Storage Pool)                   │
//! │              VDEVs, I/O Pipeline, Checksums             │
//! └─────────────────────────────────────────────────────────┘
//! ```
//!
//! ## Usage
//!
//! LCPFS is designed for kernel integration. To use it:
//!
//! ```rust,ignore
//! use lcpfs::{BlockDevice, register_device, set_log_fn};
//!
//! // 1. Implement BlockDevice for your storage hardware
//! struct NvmeDevice { /* ... */ }
//! impl BlockDevice for NvmeDevice {
//!     fn read_block(&mut self, block: usize, buf: &mut [u8]) -> Result<(), &'static str> {
//!         // Hardware-specific read
//!         Ok(())
//!     }
//!     // ... other methods
//! }
//!
//! // 2. Register the device with LCPFS
//! let device = Box::new(NvmeDevice::new());
//! let dev_id = register_device(device);
//!
//! // 3. Set up logging (optional but recommended)
//! set_log_fn(|args| kprintln!("{}", args));
//!
//! // 4. Initialize LCPFS
//! lcpfs::init();
//! ```
//!
//! ## Feature Flags
//!
//! - `pqc`: Enable post-quantum cryptography (Kyber-1024) for future-proof encryption
//! - `hw-accel`: Use hardware-accelerated checksums when available
//!
//! ## Safety
//!
//! LCPFS prioritizes data integrity:
//!
//! - Every block is checksummed on write and verified on read
//! - Corrupted blocks are automatically repaired from RAID-Z parity or mirrors
//! - Atomic transactions ensure consistent on-disk state
//! - Copy-on-write prevents in-place corruption
//!
//! ## Minimum Supported Rust Version
//!
//! LCPFS requires Rust 1.85 or later (2024 edition).

#![cfg_attr(not(feature = "std"), no_std)]
#![cfg_attr(docsrs, feature(doc_cfg))]
#![warn(missing_docs)]
#![allow(unused)] // Development: unused imports are fine
#![deny(unsafe_op_in_unsafe_fn)]

extern crate alloc;

// When std feature is enabled, also use std
#[cfg(feature = "std")]
extern crate std;

use alloc::boxed::Box;
use alloc::format;
use alloc::string::String;
use alloc::vec::Vec;
use chacha20poly1305_nostd::ChaCha20Poly1305;
use lazy_static::lazy_static;
use spin::Mutex;
use thiserror_no_std::Error;

// ═══════════════════════════════════════════════════════════════════════════════
// FILESYSTEM ERRORS
// ═══════════════════════════════════════════════════════════════════════════════

/// Comprehensive error type for LCPFS operations.
///
/// All filesystem operations return [`FsResult<T>`] which uses this error type.
/// Errors are designed to provide actionable information for debugging and
/// recovery.
///
/// # Categories
///
/// - **I/O Errors**: Hardware failures, device timeouts
/// - **Integrity Errors**: Checksum mismatches, corruption detected
/// - **Capacity Errors**: Disk full, device too small
/// - **Permission Errors**: Read-only, access denied
/// - **Logical Errors**: Invalid arguments, already exists
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum FsError {
    /// The requested file or object was not found.
    #[error("File not found")]
    NotFound,

    /// A path component could not be resolved.
    #[error("Path not found: {path_hint}")]
    PathNotFound {
        /// Hint about which path component failed.
        path_hint: String,
    },

    /// The storage pool has insufficient free space.
    #[error("Disk full: need {needed_bytes} bytes")]
    DiskFull {
        /// Number of bytes required for the operation.
        needed_bytes: u64,
    },

    /// An I/O error occurred on a virtual device.
    #[error("I/O error on vdev {vdev}: {reason}")]
    IoError {
        /// The virtual device ID that experienced the error.
        vdev: usize,
        /// Human-readable description of the error.
        reason: &'static str,
    },

    /// Data corruption was detected at the specified block.
    #[error("Corruption at block {block}: {details}")]
    Corruption {
        /// The block number where corruption was detected.
        block: u64,
        /// Details about the corruption.
        details: &'static str,
    },

    /// Block checksum verification failed.
    #[error("Checksum mismatch: expected {expected:x?}, got {actual:x?}")]
    ChecksumMismatch {
        /// The expected checksum value (256-bit as 4 x u64).
        expected: [u64; 4],
        /// The actual computed checksum (256-bit as 4 x u64).
        actual: [u64; 4],
    },

    /// Encryption operation failed.
    #[error("Encryption failed")]
    EncryptionFailed,

    /// Decryption operation failed (wrong key or corrupted data).
    #[error("Decryption failed")]
    DecryptionFailed,

    /// Compression operation failed.
    #[error("Compression failed")]
    CompressionFailed,

    /// Decompression operation failed (corrupted compressed data).
    #[error("Decompression failed")]
    DecompressionFailed,

    /// A block pointer references invalid or out-of-bounds data.
    #[error("Invalid block pointer")]
    InvalidBlockPointer,

    /// The storage pool has not been imported.
    #[error("Pool not imported")]
    PoolNotImported,

    /// The pool configuration is invalid or inconsistent.
    #[error("Invalid pool configuration: {reason}")]
    InvalidPoolConfig {
        /// Description of the configuration error.
        reason: &'static str,
    },

    /// A transaction group operation failed.
    #[error("Transaction group {txg} error")]
    TxgError {
        /// The transaction group number.
        txg: u64,
    },

    /// ZAP (ZFS Attribute Processor) operation failed.
    #[error("ZAP error: {reason}")]
    ZapError {
        /// Description of the ZAP error.
        reason: &'static str,
    },

    /// Dataset operation failed.
    #[error("Dataset error: {reason}")]
    DatasetError {
        /// Description of the dataset error.
        reason: &'static str,
    },

    /// The device is too small for the requested operation.
    #[error("Device too small: need {required} bytes, have {actual}")]
    DeviceTooSmall {
        /// Minimum required size in bytes.
        required: u64,
        /// Actual device size in bytes.
        actual: u64,
    },

    /// The file or directory already exists.
    #[error("Already exists")]
    AlreadyExists,

    /// The path refers to a directory, but a file was expected.
    #[error("Is a directory")]
    IsDirectory,

    /// The path refers to a file, but a directory was expected.
    #[error("Is a file")]
    IsFile,

    /// Permission denied for the requested operation.
    #[error("Permission denied")]
    PermissionDenied,

    /// Resource is busy (e.g., file lock already held).
    #[error("Resource busy")]
    ResourceBusy,

    /// The filesystem is mounted read-only.
    #[error("Read-only filesystem")]
    ReadOnly,

    /// An invalid argument was provided.
    #[error("Invalid argument: {reason}")]
    InvalidArgument {
        /// Description of what was invalid.
        reason: &'static str,
    },

    /// Expected a directory but found something else.
    #[error("Not a directory")]
    NotDirectory,

    /// Invalid file descriptor.
    #[error("Bad file descriptor")]
    BadFileDescriptor,

    /// Operation not implemented.
    #[error("Not implemented")]
    NotImplemented,

    /// Directory is not empty.
    #[error("Directory not empty")]
    DirectoryNotEmpty,

    /// Device or pool not found.
    #[error("Device or pool not found")]
    NoDevice,

    /// Security violation detected (anomaly/ransomware protection).
    #[error("Security violation: {reason}")]
    SecurityViolation {
        /// Description of the security issue.
        reason: &'static str,
    },
}

/// Result type for LCPFS operations.
///
/// This is a type alias for `Result<T, FsError>` used throughout the crate.
pub type FsResult<T> = Result<T, FsError>;

// ═══════════════════════════════════════════════════════════════════════════════
// BLOCK DEVICE TRAIT
// ═══════════════════════════════════════════════════════════════════════════════

/// Trait for block storage devices.
///
/// Implement this trait to provide LCPFS with access to your storage hardware.
/// LCPFS treats all storage as a collection of fixed-size blocks.
///
/// # Thread Safety
///
/// Implementations must be `Send` to allow the device to be accessed from
/// multiple kernel contexts. LCPFS handles synchronization internally.
///
/// # Example
///
/// ```rust,ignore
/// use lcpfs::BlockDevice;
///
/// pub struct RamDisk {
///     data: Vec<u8>,
///     block_size: usize,
/// }
///
/// impl BlockDevice for RamDisk {
///     fn read_block(&mut self, block_num: usize, buffer: &mut [u8]) -> Result<(), &'static str> {
///         let offset = block_num * self.block_size;
///         buffer.copy_from_slice(&self.data[offset..offset + self.block_size]);
///         Ok(())
///     }
///
///     fn write_block(&mut self, block_num: usize, buffer: &[u8]) -> Result<(), &'static str> {
///         let offset = block_num * self.block_size;
///         self.data[offset..offset + self.block_size].copy_from_slice(buffer);
///         Ok(())
///     }
///
///     fn size(&self) -> Result<u64, &'static str> { Ok(self.data.len() as u64) }
///     fn block_size(&self) -> usize { self.block_size }
///     fn block_count(&self) -> usize { self.data.len() / self.block_size }
/// }
/// ```
pub trait BlockDevice: Send {
    /// Read a single block from the device.
    ///
    /// # Arguments
    ///
    /// * `block_num` - Zero-based block number to read
    /// * `buffer` - Buffer to receive the block data (must be at least `block_size()` bytes)
    ///
    /// # Errors
    ///
    /// Returns an error if the read fails (device error, out of bounds, etc.)
    fn read_block(&mut self, block_num: usize, buffer: &mut [u8]) -> Result<(), &'static str>;

    /// Write a single block to the device.
    ///
    /// # Arguments
    ///
    /// * `block_num` - Zero-based block number to write
    /// * `buffer` - Data to write (must be exactly `block_size()` bytes)
    ///
    /// # Errors
    ///
    /// Returns an error if the write fails (device error, read-only, etc.)
    fn write_block(&mut self, block_num: usize, buffer: &[u8]) -> Result<(), &'static str>;

    /// Returns the total size of the device in bytes.
    fn size(&self) -> Result<u64, &'static str>;

    /// Returns the block size in bytes (typically 512 or 4096).
    fn block_size(&self) -> usize;

    /// Returns the total number of blocks on the device.
    fn block_count(&self) -> usize;

    /// Get average I/O latency in microseconds (optional, for LunaOS integration)
    ///
    /// # LunaOS Integration Point
    ///
    /// In LunaOS, NVMe drivers track I/O latency via completion queue timestamps.
    /// This method allows LCPFS scrubber to access these metrics for adaptive behavior.
    ///
    /// # Returns
    ///
    /// Average read/write latency in microseconds, or None if not implemented.
    ///
    /// # Example Implementation (LunaOS NVMe driver)
    ///
    /// ```rust,ignore
    /// // luna_core/src/dev/nvme.rs
    /// fn io_latency_avg_us(&self) -> Option<f64> {
    ///     Some(self.completion_queue.avg_latency_us())
    /// }
    /// ```
    fn io_latency_avg_us(&self) -> Option<f64> {
        None // Default: not implemented
    }
}

lazy_static! {
    /// Global registry of block devices available to LCPFS.
    ///
    /// Devices are registered using [`register_device`] and accessed by ID.
    pub static ref BLOCK_DEVICES: Mutex<Vec<Box<dyn BlockDevice + Send>>> = Mutex::new(Vec::new());
}

/// Register a block device with LCPFS.
///
/// # Arguments
///
/// * `device` - The block device to register
///
/// # Returns
///
/// The device ID (index) that can be used to reference this device.
///
/// # Example
///
/// ```rust,ignore
/// let dev_id = lcpfs::register_device(Box::new(my_nvme_device));
/// println!("Registered device with ID: {}", dev_id);
/// ```
pub fn register_device(device: Box<dyn BlockDevice + Send>) -> usize {
    let mut devices = BLOCK_DEVICES.lock();
    let id = devices.len();
    devices.push(device);
    id
}

/// Get a block device by ID.
///
/// Returns the device at the given ID, or a fallback 1GB RAM disk if no devices
/// are registered (for testing only).
///
/// # Arguments
///
/// * `dev_id` - The device ID returned from [`register_device`]
///
/// # Note
///
/// This function removes the device from the registry (takes ownership).
/// This is because the device is moved into the pool topology for use.
pub fn get_block_device(dev_id: usize) -> Option<Box<dyn BlockDevice + Send>> {
    let mut devices = BLOCK_DEVICES.lock();

    // Return fallback if no devices registered (testing only)
    if devices.is_empty() {
        return Some(Box::new(RamDisk::new(1024 * 1024 * 1024)));
    }

    // Check if dev_id is valid
    if dev_id >= devices.len() {
        return None;
    }

    // Get the device size and create a new RamDisk with matching capacity
    // Note: This creates a copy rather than taking ownership to avoid
    // invalidating other references to the device
    let size = devices[dev_id].size().unwrap_or(0);
    Some(Box::new(RamDisk::new(size)))
}

/// In-memory RAM disk for testing and fallback.
///
/// This provides a simple block device backed by heap memory, useful for
/// testing LCPFS without real hardware.
pub struct RamDisk {
    data: Box<[u8]>,
    size: u64,
    block_size: usize,
}

impl RamDisk {
    /// Create a new RAM disk with the specified size.
    ///
    /// # Arguments
    ///
    /// * `size_bytes` - Size of the RAM disk in bytes
    pub fn new(size_bytes: u64) -> Self {
        use alloc::vec;
        let size_usize = size_bytes as usize;
        Self {
            data: vec![0; size_usize].into_boxed_slice(),
            size: size_bytes,
            block_size: 512,
        }
    }
}

impl BlockDevice for RamDisk {
    fn read_block(&mut self, block_num: usize, buffer: &mut [u8]) -> Result<(), &'static str> {
        let offset = block_num * self.block_size;
        let end = offset + self.block_size;

        if end > self.data.len() {
            return Err("Read beyond end of device");
        }

        buffer[..self.block_size].copy_from_slice(&self.data[offset..end]);
        Ok(())
    }

    fn write_block(&mut self, block_num: usize, buffer: &[u8]) -> Result<(), &'static str> {
        let offset = block_num * self.block_size;
        let end = offset + buffer.len();

        if end > self.data.len() {
            return Err("Write beyond end of device");
        }

        self.data[offset..end].copy_from_slice(buffer);
        Ok(())
    }

    fn size(&self) -> Result<u64, &'static str> {
        Ok(self.size)
    }

    fn block_size(&self) -> usize {
        self.block_size
    }
    fn block_count(&self) -> usize {
        (self.size / self.block_size as u64) as usize
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// VFS STRUCTURES (POSIX stat)
// ═══════════════════════════════════════════════════════════════════════════════

/// POSIX-compatible file status structure.
///
/// This structure is compatible with the POSIX `stat` system call and provides
/// all standard file metadata fields.
///
/// # Layout
///
/// The structure uses `#[repr(C)]` to ensure memory layout compatibility with
/// C code and system call interfaces.
#[repr(C)]
#[derive(Debug, Clone, Copy, Default)]
pub struct FileStat {
    /// Device ID containing the file.
    pub st_dev: u64,
    /// Inode number.
    pub st_ino: u64,
    /// Number of hard links.
    pub st_nlink: u64,
    /// File type and mode (permissions).
    pub st_mode: u32,
    /// User ID of owner.
    pub st_uid: u32,
    /// Group ID of owner.
    pub st_gid: u32,
    #[doc(hidden)]
    pub __pad0: u32,
    /// Device ID (if special file).
    pub st_rdev: u64,
    /// Total size in bytes.
    pub st_size: i64,
    /// Preferred I/O block size.
    pub st_blksize: i64,
    /// Number of 512-byte blocks allocated.
    pub st_blocks: i64,
    /// Last access time (seconds since epoch).
    pub st_atime: i64,
    /// Last access time (nanoseconds).
    pub st_atime_nsec: i64,
    /// Last modification time (seconds since epoch).
    pub st_mtime: i64,
    /// Last modification time (nanoseconds).
    pub st_mtime_nsec: i64,
    /// Last status change time (seconds since epoch).
    pub st_ctime: i64,
    /// Last status change time (nanoseconds).
    pub st_ctime_nsec: i64,
    #[doc(hidden)]
    pub __reserved: [i64; 3],
}

// File mode constants (POSIX)

/// Character device file type.
pub const S_IFCHR: u32 = 0x2000;
/// Regular file type.
pub const S_IFREG: u32 = 0x8000;
/// Directory file type.
pub const S_IFDIR: u32 = 0x4000;
/// Owner read permission.
pub const S_IRUSR: u32 = 0x0100;
/// Owner write permission.
pub const S_IWUSR: u32 = 0x0080;
/// Owner execute permission.
pub const S_IXUSR: u32 = 0x0040;

// ═══════════════════════════════════════════════════════════════════════════════
// CRYPTOGRAPHIC SERVICE PROVIDER
// ═══════════════════════════════════════════════════════════════════════════════

/// Indicates whether encryption is currently active.
///
/// Enabled via `chacha20poly1305-nostd` crate (LunaOS pure-Rust implementation).
pub const ENCRYPTION_ACTIVE: bool = true;

/// Cryptographic operations for LCPFS.
///
/// Provides key derivation, block encryption, and decryption services.
/// Currently uses ChaCha20-Poly1305 for authenticated encryption.
pub struct LcpfsCrypto;

impl LcpfsCrypto {
    /// Derive an encryption key from a passphrase using PBKDF2-HMAC-SHA256.
    ///
    /// This function implements secure key derivation using PBKDF2 with SHA-256
    /// as the underlying hash function. The iteration count follows OWASP 2023
    /// recommendations for password hashing.
    ///
    /// # Arguments
    ///
    /// * `passphrase` - User-provided passphrase (any length)
    /// * `salt` - Random salt for key derivation (minimum 16 bytes recommended)
    ///
    /// # Returns
    ///
    /// A 256-bit (32-byte) encryption key derived from the passphrase and salt.
    ///
    /// # Security Properties
    ///
    /// - **600,000 iterations**: OWASP 2023 recommended minimum for PBKDF2-SHA256
    /// - **Salt-dependent**: Different salts produce completely different keys
    /// - **Deterministic**: Same passphrase + salt always produces the same key
    /// - **Computationally expensive**: ~100ms per derivation (brute-force resistant)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use lcpfs::LcpfsCrypto;
    ///
    /// let salt = [0u8; 16]; // In production, use random salt!
    /// let key = LcpfsCrypto::derive_key("my secure passphrase", &salt);
    /// assert_eq!(key.len(), 32);
    /// ```
    #[cfg(feature = "std")]
    pub fn derive_key(passphrase: &str, salt: &[u8]) -> [u8; 32] {
        use argon2::{Algorithm, Argon2, Params, Version};

        // Argon2id parameters (OWASP 2023 recommendations):
        // - Memory: 64 MiB (65536 KiB) - memory-hard against GPU attacks
        // - Iterations: 3 - time cost
        // - Parallelism: 4 - lane count
        // These provide ~100-200ms derivation on modern hardware.
        const ARGON2_MEMORY_KIB: u32 = 65_536; // 64 MiB
        const ARGON2_ITERATIONS: u32 = 3;
        const ARGON2_PARALLELISM: u32 = 4;

        let params = Params::new(
            ARGON2_MEMORY_KIB,
            ARGON2_ITERATIONS,
            ARGON2_PARALLELISM,
            Some(32), // Output length
        )
        .expect("Argon2 parameters are valid");

        let argon2 = Argon2::new(Algorithm::Argon2id, Version::V0x13, params);

        let mut key = [0u8; 32];
        argon2
            .hash_password_into(passphrase.as_bytes(), salt, &mut key)
            .expect("Argon2 should not fail with valid parameters");

        key
    }

    /// Derive an encryption key from a passphrase using PBKDF2-HMAC-SHA256.
    ///
    /// This is the `no_std` fallback when the `std` feature is not enabled.
    /// For userspace applications with `std`, Argon2id is used instead.
    ///
    /// # Arguments
    ///
    /// * `passphrase` - User-provided passphrase (any length)
    /// * `salt` - Random salt for key derivation (minimum 16 bytes recommended)
    ///
    /// # Returns
    ///
    /// A 256-bit (32-byte) encryption key derived from the passphrase and salt.
    ///
    /// # Security Properties
    ///
    /// - **600,000 iterations**: OWASP 2023 recommended minimum for PBKDF2-SHA256
    /// - **Salt-dependent**: Different salts produce completely different keys
    /// - **Deterministic**: Same passphrase + salt always produces the same key
    #[cfg(not(feature = "std"))]
    pub fn derive_key(passphrase: &str, salt: &[u8]) -> [u8; 32] {
        Self::derive_key_pbkdf2(passphrase, salt)
    }

    /// Derive an encryption key using PBKDF2-HMAC-SHA256 explicitly.
    ///
    /// Use this when you need deterministic, cross-platform key derivation
    /// that works identically in kernel (`no_std`) and userspace (`std`).
    ///
    /// For new userspace applications, prefer [`Self::derive_key`] which uses
    /// Argon2id (memory-hard, more resistant to GPU/ASIC attacks).
    ///
    /// # Arguments
    ///
    /// * `passphrase` - User-provided passphrase (any length)
    /// * `salt` - Random salt for key derivation (minimum 16 bytes recommended)
    ///
    /// # Returns
    ///
    /// A 256-bit (32-byte) encryption key derived from the passphrase and salt.
    ///
    /// # Security Properties
    ///
    /// - **600,000 iterations**: OWASP 2023 recommended minimum for PBKDF2-SHA256
    /// - **Salt-dependent**: Different salts produce completely different keys
    /// - **Deterministic**: Same passphrase + salt always produces the same key
    /// - **Cross-platform**: Identical output on kernel and userspace
    pub fn derive_key_pbkdf2(passphrase: &str, salt: &[u8]) -> [u8; 32] {
        use hmac::Hmac;
        use sha2::Sha256;

        /// PBKDF2 iteration count (OWASP 2023 recommendation for SHA-256).
        /// This provides ~100ms derivation time on modern hardware.
        const PBKDF2_ITERATIONS: u32 = 600_000;

        let mut key = [0u8; 32];

        // PBKDF2-HMAC-SHA256 key derivation
        pbkdf2::pbkdf2::<Hmac<Sha256>>(passphrase.as_bytes(), salt, PBKDF2_ITERATIONS, &mut key)
            .expect("PBKDF2 should not fail with valid output length");

        key
    }

    /// Derive a deterministic nonce from block metadata.
    ///
    /// This generates a unique 12-byte nonce from the block's location and
    /// transaction group. The nonce is deterministic, allowing decryption
    /// without storing the nonce separately.
    ///
    /// # Arguments
    ///
    /// * `vdev` - Virtual device ID
    /// * `offset` - Byte offset within the vdev
    /// * `txg` - Transaction group when the block was written
    ///
    /// # Returns
    ///
    /// A 12-byte nonce unique to this (vdev, offset, txg) tuple.
    ///
    /// # Security
    ///
    /// - Nonces are unique as long as the same block is never written twice
    ///   with the same TXG (copy-on-write guarantees this)
    /// - Uses SHAKE256 (XOF) for cryptographic mixing of inputs
    pub fn derive_nonce(vdev: u32, offset: u64, txg: u64) -> [u8; 12] {
        use sha3::Shake256;
        use sha3::digest::{ExtendableOutput, Update, XofReader};

        let mut hasher = Shake256::default();
        hasher.update(b"LCPFS-NONCE-V1"); // Domain separator
        hasher.update(&vdev.to_le_bytes());
        hasher.update(&offset.to_le_bytes());
        hasher.update(&txg.to_le_bytes());

        let mut nonce = [0u8; 12];
        hasher.finalize_xof().read(&mut nonce);
        nonce
    }

    /// Derive a nonce from a block pointer.
    ///
    /// Convenience method that extracts DVA and TXG from a block pointer.
    pub fn derive_nonce_from_bp(bp: &fscore::structs::Blkptr) -> [u8; 12] {
        Self::derive_nonce(bp.dva[0].vdev, bp.dva[0].offset, bp.birth_txg)
    }

    /// Encrypt a data block using ChaCha20-Poly1305.
    ///
    /// Uses a cryptographically random 96-bit nonce to guarantee uniqueness.
    /// The nonce is returned alongside the ciphertext and must be stored
    /// for decryption.
    ///
    /// # Arguments
    ///
    /// * `key` - 256-bit encryption key
    /// * `plaintext` - Data to encrypt
    /// * `_txg` - Transaction group number (kept for API compatibility, not used for nonce)
    ///
    /// # Returns
    ///
    /// Tuple of (ciphertext with 16-byte auth tag appended, 12-byte nonce) on success.
    ///
    /// # Security
    ///
    /// - Uses random nonce generation to prevent nonce reuse
    /// - ChaCha20-Poly1305 provides authenticated encryption (AEAD)
    /// - Nonce is not secret but must be unique per (key, message) pair
    pub fn encrypt_block(
        key: &[u8; 32],
        plaintext: &[u8],
        _txg: u64,
    ) -> FsResult<(Vec<u8>, [u8; 12])> {
        // Generate cryptographically random nonce
        // This guarantees uniqueness with overwhelming probability
        let mut nonce = [0u8; 12];
        crate::crypto::random::fill_random(&mut nonce).map_err(|_| FsError::EncryptionFailed)?;

        let cipher = ChaCha20Poly1305::new(key).map_err(|_| FsError::EncryptionFailed)?;

        // Encrypt returns ciphertext with tag appended
        let ciphertext = cipher
            .encrypt(&nonce, plaintext, None)
            .map_err(|_| FsError::EncryptionFailed)?;

        Ok((ciphertext, nonce))
    }

    /// Decrypt a data block using ChaCha20-Poly1305.
    ///
    /// # Arguments
    ///
    /// * `key` - 256-bit encryption key
    /// * `ciphertext` - Data to decrypt (with 16-byte auth tag appended)
    /// * `nonce` - 12-byte nonce used during encryption
    ///
    /// # Returns
    ///
    /// Decrypted plaintext on success.
    pub fn decrypt_block(key: &[u8; 32], ciphertext: &[u8], nonce: &[u8; 12]) -> FsResult<Vec<u8>> {
        let cipher = ChaCha20Poly1305::new(key).map_err(|_| FsError::DecryptionFailed)?;

        // Decrypt verifies tag and returns plaintext
        cipher
            .decrypt(nonce, ciphertext, None)
            .map_err(|_| FsError::DecryptionFailed)
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// LOGGING INFRASTRUCTURE
// ═══════════════════════════════════════════════════════════════════════════════

use core::fmt::Arguments;
use core::sync::atomic::{AtomicPtr, Ordering};

/// Function pointer type for logging callbacks.
///
/// The kernel provides this callback to receive log messages from LCPFS.
pub type LogFn = fn(Arguments);

static LOG_FN: AtomicPtr<()> = AtomicPtr::new(core::ptr::null_mut());

/// Set the logging callback function.
///
/// This should be called during kernel initialization to enable LCPFS logging.
///
/// # Arguments
///
/// * `f` - Function that receives formatted log messages
///
/// # Example
///
/// ```rust,ignore
/// lcpfs::set_log_fn(|args| kprintln!("[LCPFS] {}", args));
/// ```
pub fn set_log_fn(f: LogFn) {
    LOG_FN.store(f as *mut (), Ordering::SeqCst);
}

/// Internal logging function.
///
/// Called by the `lcpfs_println!` macro. Do not call directly.
#[doc(hidden)]
pub fn _log(args: Arguments) {
    let ptr = LOG_FN.load(Ordering::SeqCst);
    if !ptr.is_null() {
        // SAFETY INVARIANTS:
        // 1. ptr was stored via set_log_fn which received a valid LogFn
        // 2. LogFn type signature matches exactly (fn(Arguments))
        // 3. Function remains valid (not deallocated) for program lifetime
        // 4. Atomic SeqCst load ensures visibility across threads
        // 5. Null check prevents calling uninitialized function
        //
        // VERIFICATION: TODO - Prove transmute preserves function signature
        //
        // JUSTIFICATION:
        // AtomicPtr only stores raw pointers. Must transmute *mut () → LogFn
        // to call logging callback. Type safety enforced by set_log_fn().
        let f: LogFn = unsafe { core::mem::transmute(ptr) };
        f(args);
    }
}

/// Print a log message from LCPFS.
///
/// This macro works like `println!` but routes output through the kernel's
/// logging system. Messages are only printed if a log function has been
/// registered with [`set_log_fn`].
///
/// # Example
///
/// ```rust,ignore
/// lcpfs_println!("Pool imported: {} vdevs, {} bytes free", vdev_count, free_space);
/// ```
#[macro_export]
macro_rules! lcpfs_println {
    ($($arg:tt)*) => {
        $crate::_log(format_args!($($arg)*))
    };
}

// ═══════════════════════════════════════════════════════════════════════════════
// SCHEDULER INTERFACE
// ═══════════════════════════════════════════════════════════════════════════════

use core::sync::atomic::AtomicBool;

static SCHEDULER_AVAILABLE: AtomicBool = AtomicBool::new(false);

/// Function pointer type for spawning tasks.
///
/// The kernel provides this callback to allow LCPFS to schedule background work.
pub type SpawnFn = fn(fn(), Option<usize>);

static SPAWN_FN: AtomicPtr<()> = AtomicPtr::new(core::ptr::null_mut());

/// Set the task spawn function.
///
/// Called by the kernel to provide LCPFS with the ability to spawn background
/// tasks for operations like scrubbing, resilver, and async I/O.
///
/// # Arguments
///
/// * `f` - Function that spawns a task, optionally on a specific CPU core
pub fn set_spawn_fn(f: SpawnFn) {
    SPAWN_FN.store(f as *mut (), Ordering::SeqCst);
    SCHEDULER_AVAILABLE.store(true, Ordering::SeqCst);
}

/// Spawn a background task.
///
/// # Arguments
///
/// * `task` - The function to execute
/// * `core` - Optional CPU core affinity (None for any core)
pub fn spawn_on_core(task: fn(), core: Option<usize>) {
    let ptr = SPAWN_FN.load(Ordering::SeqCst);
    if !ptr.is_null() {
        // SAFETY INVARIANTS:
        // 1. ptr was stored via set_spawn_fn which received a valid SpawnFn
        // 2. SpawnFn type signature matches exactly (fn(fn(), Option<usize>))
        // 3. Function remains valid (not deallocated) for program lifetime
        // 4. Atomic SeqCst load ensures visibility across threads
        // 5. Null check prevents calling uninitialized function
        //
        // VERIFICATION: TODO - Prove transmute preserves function signature
        //
        // JUSTIFICATION:
        // AtomicPtr only stores raw pointers. Must transmute *mut () → SpawnFn
        // to call thread spawning callback. Type safety enforced by set_spawn_fn().
        let f: SpawnFn = unsafe { core::mem::transmute(ptr) };
        f(task, core);
    }
}

/// Check if the kernel scheduler is available.
///
/// Returns `true` if [`set_spawn_fn`] has been called.
pub fn scheduler_available() -> bool {
    SCHEDULER_AVAILABLE.load(Ordering::SeqCst)
}

// ═══════════════════════════════════════════════════════════════════════════════
// TIME PROVIDER
// ═══════════════════════════════════════════════════════════════════════════════

/// Function pointer type for getting current time.
///
/// The kernel provides this callback to allow LCPFS to get timestamps for
/// transactions, file metadata, and performance tracking.
///
/// Returns nanoseconds since boot or epoch (implementation-defined).
pub type TimeFn = fn() -> u64;

static TIME_FN: AtomicPtr<()> = AtomicPtr::new(core::ptr::null_mut());

/// Set the time provider function.
///
/// Called by the kernel to provide LCPFS with access to system time.
///
/// # Arguments
///
/// * `f` - Function that returns current time in nanoseconds
///
/// # Example
///
/// ```rust,ignore
/// lcpfs::set_time_fn(|| rdtsc() / 3_000_000); // TSC to nanoseconds at 3GHz
/// ```
pub fn set_time_fn(f: TimeFn) {
    TIME_FN.store(f as *mut (), Ordering::SeqCst);
}

/// Get current time in nanoseconds.
///
/// Returns 0 if no time provider has been set.
pub fn get_time() -> u64 {
    let ptr = TIME_FN.load(Ordering::SeqCst);
    if !ptr.is_null() {
        // SAFETY INVARIANTS:
        // 1. ptr was stored via set_time_fn which received a valid TimeFn
        // 2. TimeFn type signature matches exactly (fn() -> u64)
        // 3. Function remains valid (not deallocated) for program lifetime
        // 4. Atomic SeqCst load ensures visibility across threads
        // 5. Null check prevents calling uninitialized function
        //
        // VERIFICATION: TODO - Prove transmute preserves function signature
        //
        // JUSTIFICATION:
        // AtomicPtr only stores raw pointers. Must transmute *mut () → TimeFn
        // to call time provider callback. Type safety enforced by set_time_fn().
        let f: TimeFn = unsafe { core::mem::transmute(ptr) };
        f()
    } else {
        0 // Fallback if no time provider is set
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// COOPERATIVE YIELD
// ═══════════════════════════════════════════════════════════════════════════════

/// Function pointer type for cooperative yield.
///
/// The kernel provides this callback to allow LCPFS to yield CPU time during
/// long-running operations like scrubbing, without busy-waiting.
///
/// # Arguments
///
/// * `microseconds` - Suggested sleep/yield duration in microseconds
///
/// The kernel may sleep for the requested duration or simply yield to the
/// scheduler if preemptive scheduling is available.
pub type YieldFn = fn(u64);

static YIELD_FN: AtomicPtr<()> = AtomicPtr::new(core::ptr::null_mut());

/// Set the cooperative yield function.
///
/// Called by the kernel to provide LCPFS with the ability to yield CPU time
/// during long-running background operations (scrub, resilver, evacuation).
///
/// # Arguments
///
/// * `f` - Function that yields/sleeps for the specified microseconds
///
/// # Example
///
/// ```rust,ignore
/// lcpfs::set_yield_fn(|us| {
///     // LunaOS: Use PI-aware scheduler yield
///     luna_core::sched::sleep_us(us);
/// });
/// ```
pub fn set_yield_fn(f: YieldFn) {
    YIELD_FN.store(f as *mut (), Ordering::SeqCst);
}

/// Cooperatively yield CPU for the specified duration.
///
/// If a yield function has been set, calls it with the requested duration.
/// Otherwise, falls back to a minimal spin-loop hint (not ideal, but safe).
///
/// # Arguments
///
/// * `microseconds` - Suggested yield/sleep duration
pub fn cooperative_yield(microseconds: u64) {
    let ptr = YIELD_FN.load(Ordering::SeqCst);
    if !ptr.is_null() {
        // SAFETY INVARIANTS:
        // 1. ptr was stored via set_yield_fn which received a valid YieldFn
        // 2. YieldFn type signature matches exactly (fn(u64))
        // 3. Function remains valid (not deallocated) for program lifetime
        // 4. Atomic SeqCst load ensures visibility across threads
        // 5. Null check prevents calling uninitialized function
        //
        // JUSTIFICATION:
        // AtomicPtr only stores raw pointers. Must transmute *mut () → YieldFn
        // to call yield callback. Type safety enforced by set_yield_fn().
        let f: YieldFn = unsafe { core::mem::transmute(ptr) };
        f(microseconds);
    } else {
        // Fallback: minimal CPU hint when no yield function is available
        // This is not ideal but prevents infinite busy-wait
        core::hint::spin_loop();
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// PLATFORM ABSTRACTION
// ═══════════════════════════════════════════════════════════════════════════════

/// Platform-specific implementations (entropy, timestamps, syscalls).
/// Isolates inline assembly for portability across x86_64, AArch64, etc.
pub mod arch;

// ═══════════════════════════════════════════════════════════════════════════════
// MODULE STRUCTURE
// ═══════════════════════════════════════════════════════════════════════════════

/// Core filesystem structures, constants, and main implementation.
pub mod fscore;

/// Storage layer: DMU, ZPL, ZAP, ZIL, VDEV, ZVOL.
pub mod storage;

/// Caching: ARC, L2ARC, spacemap.
pub mod cache;

/// RAID: mirrors, RAID-Z, dRAID, erasure coding.
pub mod raid;

/// Compression: LZ4, ZSTD, computational storage, GPU compression, QLoRA.
pub mod compress;

/// Deduplication: DDT, fast dedup.
pub mod dedup;

/// Cryptography: AES-NI, PQC (Kyber), CSPRNG, core crypto, secure erase.
pub mod crypto;

/// LunaVault: Encrypted container support (VeraCrypt-compatible).
pub mod vault;

/// Integrity: checksums, scrubbing, anomaly detection.
pub mod integrity;

/// Cloud: S3 storage, cloud tiering.
pub mod cloud;

/// Hardware acceleration: CXL, GPU/CUDA, DPU, Intel QAT, PMem, NVMe-oF, SMART.
pub mod hw;

/// Networking: NFS, SMB, replication, send/receive.
pub mod net;

/// Distributed: cluster, Ceph-like OSD/MDS/CRUSH.
pub mod distributed;

/// Machine learning: prefetch, classification, GF solver.
pub mod ml;

/// LunaOS kernel integration: gravity, W_temporal ledger, PI hooks.
pub mod lunaos;

/// Management commands: pool create/destroy, dataset operations, secure erase.
pub mod mgmt;

/// I/O pipeline and QoS.
pub mod io;

/// Hot/cold data tiering: automatic migration, scrubbing, evacuation.
pub mod tier;

/// Helpers: memory allocation, formatting, performance benchmarks.
pub mod util;

/// Vector search: HNSW indexing, embeddings, semantic similarity.
pub mod vector;

/// Time-travel queries: SQL-like access to historical filesystem state.
pub mod timetravel;

/// Git-style branching: zero-copy branches, merge, cherry-pick, and commit tracking.
pub mod branch;

/// S3 Gateway: native S3-compatible API for exposing datasets as S3 buckets.
pub mod s3;

/// WASM Storage Plugins: sandboxed WebAssembly for custom storage policies.
pub mod wasm;

/// Filesystem Events: real-time notification for filesystem changes (inotify-like).
pub mod notify;

/// NFS Server: Native NFSv4/v4.1 server for exporting datasets.
pub mod nfs;

/// Thin Provisioning: Overcommit storage with on-demand block allocation.
pub mod thin;

/// Multi-File Transactions: Atomic operations across multiple files with WAL.
pub mod txn;

/// User/Group Quotas: Per-user and per-group storage limits.
pub mod quota;

/// Data Lineage: Track data provenance and transformations.
pub mod lineage;

/// Delta Sync: Efficient rsync-style synchronization.
pub mod delta;

/// Dictionary Compression: Shared compression dictionaries.
pub mod dictcomp;
/// Full-Text Search: Index file contents for instant search with BM25 ranking.
pub mod fts;

/// Online Defragmentation: Compact fragmented files without unmounting.
pub mod defrag;

/// Trash / Recycle Bin: Move deleted files to trash instead of permanent delete.
pub mod trash;

/// LunAr Archives: Native archive support (ZIP, TAR, 7z) with transparent access.
pub mod archive;

/// Storage Analytics: Detailed storage usage and performance metrics.
pub mod analytics;

/// Sparse Files: Efficient storage of files with holes.
pub mod sparse;

/// Alternate Data Streams: Multiple data streams per file (NTFS-style).
pub mod streams;

/// Telemetry: Prometheus/Grafana metrics export.
pub mod telemetry;

/// FUSE: Filesystem in Userspace driver for Linux.
#[cfg(feature = "fuse")]
pub mod fuse;

/// Time: Unified timestamp provider.
pub mod time;

// ═══════════════════════════════════════════════════════════════════════════════
// UNIFIED POOL API (POSIX INTEGRATION)
// ═══════════════════════════════════════════════════════════════════════════════

use mgmt::mount::LcpfsMount;
use storage::zpl::{DirEntry, Zpl};

/// LCPFS Storage Pool - unified API for filesystem operations.
///
/// This combines pool management (import/export) with POSIX filesystem operations
/// (files, directories, I/O). A pool is backed by one or more block devices
/// configured as VDEVs.
///
/// # Architecture
///
/// LCPFS is inspired by ZFS and provides:
/// - **Copy-on-Write (COW)**: All writes create new blocks, preserving consistency
/// - **RAID-Z**: Software RAID with parity (RAID-Z1/2/3) and distributed RAID (dRAID)
/// - **Checksums**: BLAKE3 end-to-end data integrity verification
/// - **Compression**: LZ4, Zstd, LZ4HC with transparent operation
/// - **Encryption**: ChaCha20-Poly1305, AES-256-GCM with per-dataset keys
/// - **Deduplication**: Content-addressable storage with BLAKE3 fingerprints
/// - **Snapshots**: Instant point-in-time filesystem copies
/// - **Clones**: Writable snapshot copies with shared storage
/// - **Send/Receive**: Stream-based replication for backups
///
/// # Design Philosophy
///
/// **Append-only sequential writes** within zones (ZNS-style allocation)
/// optimized for NVMe ZNS and SSD wear leveling. The allocator implements:
/// 1. Attempt allocation in active zone
/// 2. If full, mark zone as Full and find next Empty zone
/// 3. If no empty zones, return DiskFull error (GC required)
///
/// **Transaction Groups (TXGs)**: All writes are grouped into atomic
/// transactions. A TXG is committed only when all writes succeed,
/// ensuring filesystem consistency even after crashes.
///
/// **Three-Copy Metadata**: Critical metadata (uberblock, dnode, etc.)
/// is stored in triplicate for maximum resilience.
///
/// # Thread Safety
///
/// Pool operations are NOT thread-safe by default. For concurrent access:
/// - Wrap Pool in `Arc<Mutex<Pool>>` for shared access
/// - Use separate Pool instances per thread (import same device)
/// - Consider sharding by dataset or directory for high concurrency
///
/// # Example: Basic File Operations
///
/// ```rust,ignore
/// use lcpfs::{Pool, O_RDWR, O_RDONLY};
///
/// // Import existing pool from device
/// let mut pool = Pool::import(0)?;
///
/// // Create a file
/// let fd = pool.create("/test.txt", 0o644)?;
/// pool.write(fd, b"Hello World")?;
/// pool.close(fd)?;
///
/// // Read it back
/// let fd = pool.open("/test.txt", O_RDONLY)?;
/// let mut buf = vec![0u8; 11];
/// pool.read(fd, &mut buf)?;
/// assert_eq!(&buf, b"Hello World");
/// pool.close(fd)?;
/// ```
///
/// # Example: Snapshots and Clones
///
/// ```rust,ignore
/// // Create initial state
/// let mut pool = Pool::import(0)?;
/// let fd = pool.create("/data.txt", 0o644)?;
/// pool.write(fd, b"Original data")?;
/// pool.close(fd)?;
///
/// // Take snapshot
/// pool.snapshot("backup-2025-12-29")?;
///
/// // Modify data
/// let fd = pool.open("/data.txt", O_RDWR)?;
/// pool.write(fd, b"Modified data")?;
/// pool.close(fd)?;
///
/// // Rollback to snapshot (discards changes)
/// pool.rollback("backup-2025-12-29")?;
///
/// // Or create writable clone
/// pool.clone("backup-2025-12-29", "test-env")?;
/// ```
///
/// # Example: Encryption and Compression
///
/// ```rust,ignore
/// use lcpfs::{Pool, PropertyValue};
///
/// let mut pool = Pool::create_pool(0, "encrypted-pool")?;
///
/// // Enable compression (reduces space, improves performance)
/// pool.set_property("compression", PropertyValue::String("zstd".into()))?;
///
/// // Enable encryption (requires key management)
/// pool.set_property("encryption", PropertyValue::String("chacha20poly1305".into()))?;
///
/// // Files are now transparently compressed and encrypted
/// let fd = pool.create("/secret.txt", 0o600)?;
/// pool.write(fd, b"Sensitive data")?;
/// pool.close(fd)?;
/// ```
///
/// # Example: Send/Receive Replication
///
/// ```rust,ignore
/// // Source pool
/// let mut src = Pool::import(0)?;
/// let stream = src.send()?;
///
/// // Destination pool (different machine/device)
/// let mut dst = Pool::import(1)?;
/// dst.receive(&stream)?;
///
/// // Incremental replication (only changes)
/// let base_txg = src.current_txg();
/// // ... make changes ...
/// let delta_stream = src.send_incremental(base_txg)?;
/// dst.receive(&delta_stream)?;
/// ```
///
/// # Performance Considerations
///
/// - **Record Size**: Default 128KB, tune via `recordsize` property
///   - Small files (<4KB): Use `recordsize=4096` for better space efficiency
///   - Large files (>1MB): Use `recordsize=1048576` for better throughput
/// - **Compression**: LZ4 is CPU-cheap (~1% overhead), use by default
/// - **Dedup**: CPU-expensive (BLAKE3 hashing), enable only for redundant data
/// - **ARC Cache**: Adaptive Replacement Cache auto-tunes for workload
/// - **ZIL**: Write-ahead log for sync writes, critical for databases
///
/// # Error Handling
///
/// All Pool methods return `Result<T, LcpfsError>`. Common errors:
/// - `NotFound`: File/directory doesn't exist
/// - `AlreadyExists`: Cannot create duplicate file/snapshot
/// - `DiskFull`: No free space (run GC or add devices)
/// - `ChecksumMismatch`: Data corruption detected (repairs via RAID-Z)
/// - `EncryptionError`: Key mismatch or authentication failure
pub struct Pool {
    /// Low-level pool state (uberblock, devices, TXG)
    mount: LcpfsMount,

    /// POSIX filesystem layer
    zpl: Zpl,

    /// Snapshot registry (name -> metadata)
    snapshots: alloc::collections::BTreeMap<String, SnapshotMetadata>,

    /// Dataset properties (configuration)
    properties: Properties,
}

// ═══════════════════════════════════════════════════════════════════════════
// PATH LIMITS (POSIX-compliant)
// ═══════════════════════════════════════════════════════════════════════════

/// Maximum path depth (number of directory components).
///
/// POSIX doesn't mandate a specific limit, but most implementations use 256-1024.
/// We use 256 to prevent stack overflow in recursive operations.
pub const MAX_PATH_DEPTH: usize = 256;

/// Maximum length of a single path component (filename/dirname).
///
/// POSIX.1-2024 NAME_MAX is typically 255 bytes.
pub const MAX_NAME_LEN: usize = 255;

/// Maximum total path length in bytes.
///
/// POSIX.1-2024 PATH_MAX is typically 4096 bytes.
pub const MAX_PATH_LEN: usize = 4096;

/// Property value types
#[derive(Debug, Clone, PartialEq)]
pub enum PropertyValue {
    /// String property
    String(alloc::string::String),
    /// Numeric property
    Number(u64),
    /// Boolean property
    Boolean(bool),
}

/// Property source (where value came from)
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum PropertySource {
    /// Set locally on this dataset
    Local,
    /// Inherited from parent dataset
    Inherited,
    /// System default value
    Default,
}

/// Dataset properties
#[derive(Debug, Clone)]
pub struct Properties {
    /// Property values
    values: alloc::collections::BTreeMap<alloc::string::String, (PropertyValue, PropertySource)>,
}

impl Properties {
    /// Create default properties
    pub fn new() -> Self {
        let mut props = Self {
            values: alloc::collections::BTreeMap::new(),
        };

        // Set defaults
        props.values.insert(
            "compression".into(),
            (PropertyValue::String("lz4".into()), PropertySource::Default),
        );
        props.values.insert(
            "checksum".into(),
            (
                PropertyValue::String("blake3".into()),
                PropertySource::Default,
            ),
        );
        props.values.insert(
            "readonly".into(),
            (PropertyValue::Boolean(false), PropertySource::Default),
        );
        props.values.insert(
            "recordsize".into(),
            (PropertyValue::Number(131072), PropertySource::Default),
        ); // 128K
        props.values.insert(
            "atime".into(),
            (PropertyValue::Boolean(true), PropertySource::Default),
        );
        props.values.insert(
            "dedup".into(),
            (PropertyValue::Boolean(true), PropertySource::Default),
        );

        props
    }

    /// Get property value
    pub fn get(&self, name: &str) -> Option<&PropertyValue> {
        self.values.get(name).map(|(val, _src)| val)
    }

    /// Get property with source
    pub fn get_with_source(&self, name: &str) -> Option<(&PropertyValue, PropertySource)> {
        self.values.get(name).map(|(val, src)| (val, *src))
    }

    /// Set property (marks as Local)
    pub fn set(&mut self, name: &str, value: PropertyValue) {
        self.values
            .insert(name.into(), (value, PropertySource::Local));
    }

    /// List all properties
    pub fn list(&self) -> alloc::vec::Vec<(alloc::string::String, PropertyValue, PropertySource)> {
        self.values
            .iter()
            .map(|(k, (v, s))| (k.clone(), v.clone(), *s))
            .collect()
    }
}

impl Default for Properties {
    fn default() -> Self {
        Self::new()
    }
}

/// Snapshot metadata
#[derive(Debug, Clone)]
struct SnapshotMetadata {
    /// Snapshot name
    name: String,
    /// Transaction group when snapshot was created
    txg: u64,
    /// Creation timestamp (TXG number)
    creation_time: u64,
    /// Snapshot GUID
    guid: u64,
    /// Size in bytes (space uniquely referenced by this snapshot)
    unique_bytes: u64,
}

impl Pool {
    /// Import a pool from a block device.
    ///
    /// Scans the device for LCPFS labels, finds the latest uberblock, and loads
    /// the filesystem metadata.
    ///
    /// # Arguments
    ///
    /// * `dev_id` - Block device ID from [`register_device`]
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Device is not found
    /// - No valid LCPFS label exists
    /// - Pool metadata is corrupted
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let pool = Pool::import(0)?;
    /// println!("Imported pool with TXG: {}", pool.current_txg());
    /// ```
    pub fn import(dev_id: usize) -> FsResult<Self> {
        let mount =
            LcpfsMount::import(dev_id).map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

        // BUG #8 FIX: Reconstruct ZPL from persisted hyperblock instead of creating empty
        let zpl = if let Some(ref hyperblock) = mount.active_uberblock {
            // Try to import from persisted data; fall back to new if import fails
            match Zpl::import_from_hyperblock(hyperblock, None) {
                Ok(imported_zpl) => {
                    lcpfs_println!(
                        "[ POOL ] Reconstructed ZPL from hyperblock TXG {}",
                        hyperblock.txg
                    );
                    imported_zpl
                }
                Err(e) => {
                    lcpfs_println!(
                        "[ POOL ] Warning: ZPL import failed ({:?}), creating fresh ZPL",
                        e
                    );
                    Zpl::new()
                }
            }
        } else {
            // No active hyperblock means fresh pool
            Zpl::new()
        };

        let snapshots = alloc::collections::BTreeMap::new();
        let properties = Properties::new();

        Ok(Self {
            mount,
            zpl,
            snapshots,
            properties,
        })
    }

    /// Create a new pool on a block device.
    ///
    /// Formats the device with LCPFS structures and initializes an empty filesystem.
    ///
    /// # Arguments
    ///
    /// * `dev_id` - Block device ID
    /// * `pool_name` - Name for the pool (e.g., "rpool")
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let pool = Pool::create_pool(0, "rpool")?;
    /// ```
    pub fn create_pool(dev_id: usize, pool_name: &str) -> FsResult<Self> {
        use crate::mgmt::format::LcpfsFormatter;

        LcpfsFormatter::format_drive(dev_id, pool_name)
            .map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

        Self::import(dev_id)
    }

    /// Get current transaction group number.
    pub fn current_txg(&self) -> u64 {
        self.mount.current_txg
    }

    /// Get pool GUID.
    pub fn guid(&self) -> u64 {
        self.mount.pool_guid
    }

    // ═════════════════════════════════════════════════════════════════════════
    // FILE OPERATIONS (POSIX-compatible)
    // ═════════════════════════════════════════════════════════════════════════

    /// Open a file, returning a file descriptor.
    ///
    /// # Arguments
    ///
    /// * `path` - Absolute path to file
    /// * `flags` - Open flags (O_RDONLY, O_WRONLY, O_RDWR, O_CREAT, etc.)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let fd = pool.open("/etc/passwd", O_RDONLY)?;
    /// ```
    pub fn open(&mut self, path: &str, flags: u32) -> FsResult<u64> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.open(object_id, flags)
    }

    /// Create a new file.
    ///
    /// # Arguments
    ///
    /// * `path` - Absolute path for new file
    /// * `mode` - Permission bits (e.g., 0o644)
    ///
    /// # Returns
    ///
    /// File descriptor for the newly created file.
    pub fn create(&mut self, path: &str, mode: u32) -> FsResult<u64> {
        let (dir_id, filename) = self.split_path(path)?;
        let object_id = self.zpl.create(dir_id, &filename, mode, 0, 0)?;
        self.zpl.open(object_id, storage::zpl::O_RDWR)
    }

    /// Close a file descriptor.
    pub fn close(&mut self, fd: u64) -> FsResult<()> {
        self.zpl.close(fd)
    }

    /// Read from an open file.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor from [`open`](Pool::open)
    /// * `buf` - Buffer to read into
    ///
    /// # Returns
    ///
    /// Number of bytes read (0 = EOF).
    pub fn read(&mut self, fd: u64, buf: &mut [u8]) -> FsResult<usize> {
        self.zpl.read(fd, buf)
    }

    /// Write to an open file.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor
    /// * `buf` - Data to write
    ///
    /// # Returns
    ///
    /// Number of bytes written.
    pub fn write(&mut self, fd: u64, buf: &[u8]) -> FsResult<usize> {
        self.zpl.write(fd, buf)
    }

    /// Seek to a position in a file.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor
    /// * `offset` - Offset to seek to
    /// * `whence` - SEEK_SET, SEEK_CUR, or SEEK_END
    ///
    /// # Returns
    ///
    /// New absolute offset.
    pub fn seek(&mut self, fd: u64, offset: i64, whence: i32) -> FsResult<u64> {
        self.zpl.seek(fd, offset, whence)
    }

    /// Truncate a file to a specified length.
    pub fn truncate(&mut self, path: &str, length: u64) -> FsResult<()> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.truncate(object_id, length)
    }

    /// Acquire a file lock (BSD flock).
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor from [`open`](Pool::open)
    /// * `exclusive` - If true, acquire exclusive lock; if false, shared lock
    /// * `pid` - Process ID of lock holder
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let fd = pool.open("/data.txt", O_RDWR)?;
    /// pool.flock(fd, true, std::process::id())?;  // Exclusive lock
    /// // ... do work ...
    /// pool.funlock(fd, std::process::id())?;
    /// ```
    pub fn flock(&mut self, fd: u64, exclusive: bool, pid: u32) -> FsResult<()> {
        use crate::storage::zpl::LockType;
        let lock_type = if exclusive {
            LockType::Exclusive
        } else {
            LockType::Shared
        };
        self.zpl.flock(fd, lock_type, pid)
    }

    /// Release a file lock (BSD flock).
    pub fn funlock(&mut self, fd: u64, pid: u32) -> FsResult<()> {
        self.zpl.funlock(fd, pid)
    }

    /// Delete a file.
    pub fn unlink(&mut self, path: &str) -> FsResult<()> {
        let (dir_id, filename) = self.split_path(path)?;
        self.zpl.unlink(dir_id, &filename)
    }

    /// Rename or move a file or directory.
    ///
    /// # Arguments
    ///
    /// * `src_path` - Current path of the file/directory
    /// * `dst_path` - New path (can be in different directory)
    ///
    /// # Notes
    ///
    /// - If destination exists and is a file, it will be overwritten
    /// - If destination exists and is a directory, rename fails
    /// - Cannot move a directory into itself
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.rename("/old_name.txt", "/new_name.txt")?;
    /// pool.rename("/docs/file.txt", "/archive/file.txt")?;
    /// ```
    pub fn rename(&mut self, src_path: &str, dst_path: &str) -> FsResult<()> {
        let (src_dir_id, src_name) = self.split_path(src_path)?;
        let (dst_dir_id, dst_name) = self.split_path(dst_path)?;
        self.zpl
            .rename(src_dir_id, &src_name, dst_dir_id, &dst_name)
    }

    /// Sync all pending writes to disk.
    ///
    /// Commits the current transaction group, writing all dirty data to persistent storage.
    /// This ensures all previous writes are durable.
    pub fn sync(&mut self) -> FsResult<()> {
        use crate::storage::zil::ZilEngine;

        // Flush ZIL first (synchronous writes)
        ZilEngine::flush_to_slog().map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

        // Sync TXG (all dirty dnodes)
        let dev_id = self.mount.dev_id;
        let txg = self.zpl.txg_sync(dev_id)?;

        // Commit TXG marker to ZIL
        ZilEngine::commit_txg(txg).map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

        Ok(())
    }

    // ═════════════════════════════════════════════════════════════════════════
    // DIRECTORY OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Create a directory.
    ///
    /// # Arguments
    ///
    /// * `path` - Absolute path for new directory
    /// * `mode` - Permission bits (e.g., 0o755)
    pub fn mkdir(&mut self, path: &str, mode: u32) -> FsResult<()> {
        let (parent_id, dirname) = self.split_path(path)?;
        self.zpl.mkdir(parent_id, &dirname, mode, 0, 0)?;
        Ok(())
    }

    /// Remove an empty directory.
    pub fn rmdir(&mut self, path: &str) -> FsResult<()> {
        let (parent_id, dirname) = self.split_path(path)?;
        self.zpl.rmdir(parent_id, &dirname)
    }

    /// List directory contents.
    ///
    /// # Returns
    ///
    /// Vector of (name, inode, is_directory) tuples.
    pub fn readdir(&self, path: &str) -> FsResult<Vec<DirEntry>> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.readdir(object_id)
    }

    // ═════════════════════════════════════════════════════════════════════════
    // METADATA OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Get file metadata (stat).
    pub fn stat(&self, path: &str) -> FsResult<FileStat> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.getattr(object_id)
    }

    /// Change file permissions.
    pub fn chmod(&mut self, path: &str, mode: u32) -> FsResult<()> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.setattr(object_id, Some(mode), None, None)
    }

    /// Change file owner.
    pub fn chown(&mut self, path: &str, uid: u32, gid: u32) -> FsResult<()> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.setattr(object_id, None, Some(uid), Some(gid))
    }

    /// Create a hard link to an existing file.
    ///
    /// Creates a new directory entry that points to the same inode as the
    /// existing file. Both paths will refer to the same data.
    ///
    /// # Arguments
    ///
    /// * `existing_path` - Path to the existing file
    /// * `new_path` - Path for the new hard link
    ///
    /// # Notes
    ///
    /// - Cannot create hard links to directories
    /// - Both paths must be on the same filesystem
    /// - Deleting one path doesn't affect the other
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.link("/original.txt", "/hardlink.txt")?;
    /// // Both paths now refer to the same file
    /// ```
    pub fn link(&mut self, existing_path: &str, new_path: &str) -> FsResult<()> {
        // Get the existing file's object ID
        let existing_id = self.path_to_object_id(existing_path)?;

        // Verify it's not a directory
        let stat = self.zpl.getattr(existing_id)?;
        if (stat.st_mode & S_IFMT) == S_IFDIR {
            return Err(FsError::IsDirectory);
        }

        // Get destination directory and name
        let (dst_dir_id, dst_name) = self.split_path(new_path)?;

        // Create the hard link
        self.zpl.link_create(dst_dir_id, &dst_name, existing_id)
    }

    /// Create a symbolic link.
    ///
    /// Creates a symlink at `link_path` that points to `target`. The target
    /// is stored as-is and can be absolute or relative.
    ///
    /// # Arguments
    ///
    /// * `target` - The path the symlink points to
    /// * `link_path` - Path for the new symlink
    ///
    /// # Notes
    ///
    /// - Unlike hard links, symlinks can point to directories
    /// - The target is not validated (can be a dangling symlink)
    /// - Symlinks have mode 0o777 (actual access depends on target)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.symlink("/home/user/docs", "/data/docs_link")?;
    /// let target = pool.readlink("/data/docs_link")?;
    /// assert_eq!(target, "/home/user/docs");
    /// ```
    pub fn symlink(&mut self, target: &str, link_path: &str) -> FsResult<()> {
        let (dir_id, link_name) = self.split_path(link_path)?;
        self.zpl.symlink(dir_id, &link_name, target, 0, 0)?;
        Ok(())
    }

    /// Read the target of a symbolic link.
    ///
    /// # Arguments
    ///
    /// * `path` - Path to the symbolic link
    ///
    /// # Returns
    ///
    /// The target path stored in the symlink.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.symlink("/etc/passwd", "/tmp/passwd_link")?;
    /// let target = pool.readlink("/tmp/passwd_link")?;
    /// assert_eq!(target, "/etc/passwd");
    /// ```
    pub fn readlink(&self, path: &str) -> FsResult<String> {
        let object_id = self.path_to_object_id(path)?;
        self.zpl.readlink(object_id)
    }

    // ═════════════════════════════════════════════════════════════════════════
    // POSIX FCNTL RECORD LOCKS
    // ═════════════════════════════════════════════════════════════════════════

    /// Acquire a POSIX fcntl() record lock (byte-range advisory lock).
    ///
    /// Record locks allow fine-grained locking of specific byte ranges within
    /// a file. Unlike flock(), multiple processes can lock different regions
    /// of the same file simultaneously.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor from [`open`](Pool::open)
    /// * `exclusive` - If true, acquire exclusive lock; if false, shared lock
    /// * `start` - Start offset of region to lock (0 = beginning)
    /// * `length` - Length of region (0 = to end of file)
    /// * `pid` - Process ID of lock holder
    ///
    /// # Lock Types
    ///
    /// - **Shared (F_RDLCK)**: Multiple processes can hold shared locks on overlapping regions
    /// - **Exclusive (F_WRLCK)**: Only one process can hold an exclusive lock on a region
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let fd = pool.open("/data.txt", O_RDWR)?;
    /// // Lock bytes 0-99 exclusively
    /// pool.fcntl_lock(fd, true, 0, 100, std::process::id())?;
    /// // ... do work ...
    /// pool.fcntl_unlock(fd, 0, 100, std::process::id())?;
    /// ```
    pub fn fcntl_lock(
        &mut self,
        fd: u64,
        exclusive: bool,
        start: u64,
        length: u64,
        pid: u32,
    ) -> FsResult<()> {
        use crate::storage::zpl::LockType;
        let lock_type = if exclusive {
            LockType::Exclusive
        } else {
            LockType::Shared
        };
        self.zpl.fcntl_setlk(fd, lock_type, start, length, pid)
    }

    /// Release a POSIX fcntl() record lock.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor
    /// * `start` - Start offset of region to unlock
    /// * `length` - Length of region (0 = to end of file)
    /// * `pid` - Process ID of lock holder
    pub fn fcntl_unlock(&mut self, fd: u64, start: u64, length: u64, pid: u32) -> FsResult<()> {
        self.zpl.fcntl_unlk(fd, start, length, pid)
    }

    /// Test if a record lock would conflict (F_GETLK).
    ///
    /// Checks whether acquiring a lock with the specified parameters would
    /// succeed. If a conflicting lock exists, returns information about it.
    ///
    /// # Arguments
    ///
    /// * `fd` - File descriptor
    /// * `exclusive` - Type of lock to test
    /// * `start` - Start offset
    /// * `length` - Length of region
    /// * `pid` - Process ID of requester
    ///
    /// # Returns
    ///
    /// `Ok(None)` if lock would succeed, `Ok(Some((lock_type, start, length, pid)))`
    /// with info about the conflicting lock.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // Check if we can lock bytes 50-150
    /// if let Some((lock_type, start, len, holder_pid)) =
    ///     pool.fcntl_test_lock(fd, true, 50, 100, my_pid)?
    /// {
    ///     println!("Conflict with process {} holding lock at {}-{}", holder_pid, start, start+len);
    /// }
    /// ```
    pub fn fcntl_test_lock(
        &self,
        fd: u64,
        exclusive: bool,
        start: u64,
        length: u64,
        pid: u32,
    ) -> FsResult<Option<(bool, u64, u64, u32)>> {
        use crate::storage::zpl::LockType;
        let lock_type = if exclusive {
            LockType::Exclusive
        } else {
            LockType::Shared
        };
        let result = self.zpl.fcntl_getlk(fd, lock_type, start, length, pid)?;
        Ok(result.map(|lock| {
            (
                lock.lock_type == LockType::Exclusive,
                lock.start,
                lock.length,
                lock.pid,
            )
        }))
    }

    // ═════════════════════════════════════════════════════════════════════════
    // QUOTA OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Set pool-level quota (maximum space usage).
    ///
    /// When set, writes that would exceed the quota will fail with `DiskFull`.
    ///
    /// # Arguments
    ///
    /// * `quota_bytes` - Maximum bytes allowed (0 = unlimited)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // Limit pool to 10 GB
    /// pool.set_quota(10 * 1024 * 1024 * 1024)?;
    ///
    /// // Remove quota limit
    /// pool.set_quota(0)?;
    /// ```
    pub fn set_quota(&mut self, quota_bytes: u64) -> FsResult<()> {
        self.properties
            .set("quota", PropertyValue::Number(quota_bytes));
        // Also set in ZPL for write-time enforcement
        self.zpl.set_quota(quota_bytes);
        lcpfs_println!("[ QUOTA ] Pool quota set to {} bytes", quota_bytes);
        Ok(())
    }

    /// Get current pool quota.
    ///
    /// # Returns
    ///
    /// `Some(bytes)` if a quota is set, `None` if unlimited.
    pub fn get_quota(&self) -> Option<u64> {
        match self.properties.get("quota") {
            Some(PropertyValue::Number(n)) if *n > 0 => Some(*n),
            _ => None,
        }
    }

    /// Get current space usage.
    ///
    /// Returns the total bytes used by the pool.
    pub fn get_used_space(&self) -> u64 {
        self.zpl.get_used_bytes()
    }

    /// Get remaining quota space.
    ///
    /// # Returns
    ///
    /// `Some(bytes)` remaining under quota, `None` if unlimited.
    pub fn get_remaining_quota(&self) -> Option<u64> {
        let quota = self.get_quota()?;
        let used = self.get_used_space();
        Some(quota.saturating_sub(used))
    }

    /// Get quota utilization as a percentage.
    ///
    /// # Returns
    ///
    /// `Some(percentage)` (0.0 - 100.0) if quota is set, `None` if unlimited.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// if let Some(pct) = pool.get_quota_utilization() {
    ///     if pct > 90.0 {
    ///         println!("Warning: Pool is {}% full!", pct);
    ///     }
    /// }
    /// ```
    pub fn get_quota_utilization(&self) -> Option<f64> {
        let quota = self.get_quota()?;
        let used = self.get_used_space();
        Some(((used as f64) / (quota as f64) * 100.0).min(100.0))
    }

    /// Check if pool is over quota.
    ///
    /// When over quota, new writes will fail.
    pub fn is_over_quota(&self) -> bool {
        self.zpl.is_over_quota()
    }

    // ═════════════════════════════════════════════════════════════════════════
    // SCRUB OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Start a background data scrub operation.
    ///
    /// Scrubbing verifies the integrity of all data blocks by reading them and
    /// checking their checksums. If errors are found and RAID-Z redundancy is
    /// available, the scrub will attempt to repair corrupted blocks.
    ///
    /// The scrub runs in the background with adaptive pacing to minimize impact
    /// on foreground I/O. On LunaOS, the scrub uses PI-based scheduling to
    /// optimize for epsilon minimization.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.scrub_start()?;
    /// // ... scrub runs in background ...
    /// let stats = pool.scrub_stats();
    /// println!("Scrubbed {} blocks, found {} errors", stats.blocks_scanned, stats.errors_found);
    /// ```
    pub fn scrub_start(&mut self) -> FsResult<()> {
        integrity::scrub::start_scrub(get_time())
            .map_err(|reason| FsError::InvalidPoolConfig { reason })
    }

    /// Get current scrub statistics.
    ///
    /// Returns information about the ongoing or most recent scrub operation.
    pub fn scrub_stats(&self) -> integrity::scrub::ScrubStats {
        integrity::scrub::stats()
    }

    /// Check if a scrub should be started based on learned thresholds.
    ///
    /// Uses adaptive scheduling based on observed error rates and time since
    /// last scrub. The thresholds are learned from past scrub outcomes.
    ///
    /// # Arguments
    ///
    /// * `observed_error_rate` - Current observed error rate (errors per 1M blocks)
    ///
    /// # Returns
    ///
    /// `true` if a scrub is recommended based on learned thresholds.
    pub fn scrub_should_run(&self, observed_error_rate: f64) -> bool {
        integrity::scrub::should_scrub(get_time(), observed_error_rate)
    }

    // ═════════════════════════════════════════════════════════════════════════
    // REFLINK / QUANTUM COPY OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Create a reflink (copy-on-write clone) of a file.
    ///
    /// This is an O(1) "quantum copy" operation that shares data blocks between
    /// the source and destination files. No actual data is copied - only metadata
    /// pointers are duplicated. When either file is modified, copy-on-write
    /// semantics ensure the other file is unaffected.
    ///
    /// # Arguments
    ///
    /// * `src_path` - Path to the source file
    /// * `dst_path` - Path for the new reflinked file (must not exist)
    ///
    /// # Returns
    ///
    /// `Ok(())` on success, or an error if:
    /// - Source file doesn't exist
    /// - Destination already exists
    /// - Source is a directory (use `clone_tree` for directories)
    ///
    /// # Performance
    ///
    /// - Time complexity: O(1) - constant time regardless of file size
    /// - Space: Only metadata is duplicated (~256 bytes)
    /// - A 10 GB file is "copied" as fast as a 10 KB file
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // Create instant copy of large file
    /// pool.reflink("/data/huge_dataset.bin", "/backup/huge_dataset.bin")?;
    /// // Both files now share data blocks; modifications trigger COW
    /// ```
    ///
    /// # See Also
    ///
    /// - Linux: `cp --reflink=always`
    /// - macOS: `cp -c` (APFS clones)
    /// - Windows: Block cloning on ReFS
    pub fn reflink(&mut self, src_path: &str, dst_path: &str) -> FsResult<()> {
        // 1. Verify source exists and is a regular file
        let src_id = self.path_to_object_id(src_path)?;
        let src_stat = self.zpl.getattr(src_id)?;

        // Check if source is a directory (S_IFDIR = 0x4000)
        if (src_stat.st_mode & S_IFMT) == S_IFDIR {
            return Err(FsError::IsDirectory);
        }

        // 2. Verify destination doesn't exist
        if self.path_to_object_id(dst_path).is_ok() {
            return Err(FsError::AlreadyExists);
        }

        // 3. Create destination file with same mode
        let (dst_dir_id, dst_filename) = self.split_path(dst_path)?;
        let dst_id = self.zpl.create(
            dst_dir_id,
            &dst_filename,
            src_stat.st_mode,
            src_stat.st_uid,
            src_stat.st_gid,
        )?;

        // 4. Perform the quantum entanglement (block pointer sharing)
        self.zpl.reflink_data(src_id, dst_id)?;

        lcpfs_println!(
            "[ QUANTUM] Reflink: {} -> {} (0 bytes copied, {} bytes shared)",
            src_path,
            dst_path,
            src_stat.st_size
        );

        Ok(())
    }

    /// Clone an entire directory tree using reflinks.
    ///
    /// Creates a copy-on-write clone of a directory and all its contents.
    /// Like `reflink`, this is extremely fast as only metadata is copied.
    ///
    /// # Arguments
    ///
    /// * `src_path` - Path to source directory
    /// * `dst_path` - Path for destination directory (must not exist)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // Clone entire project directory instantly
    /// pool.clone_tree("/projects/myapp", "/projects/myapp-backup")?;
    /// ```
    pub fn clone_tree(&mut self, src_path: &str, dst_path: &str) -> FsResult<u64> {
        // Verify source is a directory
        let src_stat = self.stat(src_path)?;
        // Check if source is a directory (S_IFDIR = 0x4000)
        if (src_stat.st_mode & S_IFMT) != S_IFDIR {
            return Err(FsError::NotDirectory);
        }

        // Verify destination doesn't exist
        if self.path_to_object_id(dst_path).is_ok() {
            return Err(FsError::AlreadyExists);
        }

        // Create destination directory
        self.mkdir(dst_path, src_stat.st_mode)?;

        let mut files_cloned: u64 = 0;

        // Recursively clone contents
        let entries = self.readdir(src_path)?;
        for entry in entries {
            if entry.name == "." || entry.name == ".." {
                continue;
            }

            let src_child = alloc::format!("{}/{}", src_path, entry.name);
            let dst_child = alloc::format!("{}/{}", dst_path, entry.name);

            // DT_DIR = 4
            if entry.file_type == 4 {
                files_cloned += self.clone_tree(&src_child, &dst_child)?;
            } else {
                self.reflink(&src_child, &dst_child)?;
                files_cloned += 1;
            }
        }

        Ok(files_cloned)
    }

    // ═════════════════════════════════════════════════════════════════════════
    // REPLICATION OPERATIONS (SEND/RECEIVE)
    // ═════════════════════════════════════════════════════════════════════════

    /// Send (serialize) the entire pool to a byte stream.
    ///
    /// Creates a full snapshot that can be transferred to another system
    /// and reconstructed with `receive()`.
    ///
    /// # Returns
    ///
    /// Byte stream containing the serialized pool data.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let stream = pool.send()?;
    /// // Transfer stream to remote system
    /// let mut remote_pool = Pool::create_pool(1, "remote")?;
    /// remote_pool.receive(&stream)?;
    /// ```
    pub fn send(&mut self) -> FsResult<Vec<u8>> {
        use crate::net::send_recv::SendStream;

        let txg = self.current_txg();
        let guid = self.guid();

        let mut stream = SendStream::begin_full_send(guid, txg);

        // Serialize all objects from the ZPL
        self.zpl.send_to_stream(&mut stream)?;

        Ok(stream.finalize())
    }

    /// Send incremental changes since a previous snapshot (by TXG).
    ///
    /// Only sends the delta between the current state and a previous snapshot,
    /// making it much faster and smaller than a full send.
    ///
    /// # Arguments
    ///
    /// * `from_txg` - Transaction group number of the base snapshot
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let base_txg = pool.current_txg();
    /// // ... make changes ...
    /// let delta_stream = pool.send_incremental(base_txg)?;
    /// ```
    pub fn send_incremental(&mut self, from_txg: u64) -> FsResult<Vec<u8>> {
        use crate::net::send_recv::SendStream;

        let to_txg = self.current_txg();
        let guid = self.guid();

        let mut stream = SendStream::begin_incremental_send(guid, guid, from_txg, to_txg);

        // Serialize only objects modified since from_txg
        self.zpl.send_incremental_to_stream(&mut stream, from_txg)?;

        Ok(stream.finalize())
    }

    /// Send incremental changes between two snapshots.
    ///
    /// Sends only the delta between two named snapshots, enabling efficient
    /// incremental backups and replication.
    ///
    /// # Arguments
    ///
    /// * `from_snapshot` - Base snapshot name
    /// * `to_snapshot` - Target snapshot name (or None for current state)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.snapshot("backup-daily")?;
    /// // ... make changes ...
    /// pool.snapshot("backup-hourly")?;
    /// let delta = pool.send_snapshot_incremental("backup-daily", Some("backup-hourly"))?;
    /// ```
    pub fn send_snapshot_incremental(
        &mut self,
        from_snapshot: &str,
        to_snapshot: Option<&str>,
    ) -> FsResult<Vec<u8>> {
        use crate::net::send_recv::SendStream;

        // Get from snapshot TXG
        let from_snap = self
            .snapshots
            .get(from_snapshot)
            .ok_or(FsError::InvalidArgument {
                reason: "base snapshot not found",
            })?;
        let from_txg = from_snap.txg;
        let from_guid = from_snap.guid;

        // Get to snapshot TXG (or current)
        let (to_txg, to_guid) = if let Some(to_name) = to_snapshot {
            let to_snap = self
                .snapshots
                .get(to_name)
                .ok_or(FsError::InvalidArgument {
                    reason: "target snapshot not found",
                })?;
            (to_snap.txg, to_snap.guid)
        } else {
            (self.current_txg(), self.guid())
        };

        crate::lcpfs_println!(
            "[ SEND ] Incremental send from '{}' (TXG {}) to {} (TXG {})",
            from_snapshot,
            from_txg,
            to_snapshot.unwrap_or("current"),
            to_txg
        );

        let mut stream = SendStream::begin_incremental_send(from_guid, to_guid, from_txg, to_txg);

        // Add snapshot markers
        stream.write_snapshot(from_snapshot, from_guid);
        if let Some(to_name) = to_snapshot {
            stream.write_snapshot(to_name, to_guid);
        }

        // Serialize delta
        self.zpl.send_incremental_to_stream(&mut stream, from_txg)?;

        Ok(stream.finalize())
    }

    /// Receive (deserialize) a pool from a byte stream.
    ///
    /// Reconstructs the pool from a stream created by `send()` or `send_incremental()`.
    ///
    /// # Arguments
    ///
    /// * `stream` - Byte stream from `send()` or `send_incremental()`
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let stream = source_pool.send()?;
    /// let mut dest_pool = Pool::create_pool(1, "dest")?;
    /// dest_pool.receive(&stream)?;
    /// ```
    pub fn receive(&mut self, stream: &[u8]) -> FsResult<()> {
        use crate::net::send_recv::ReceiveStream;

        let mut recv = ReceiveStream::begin_receive(stream.to_vec())
            .map_err(|e| FsError::InvalidPoolConfig { reason: e })?;

        // Process all records and apply to ZPL
        self.zpl.receive_from_stream(&mut recv)?;

        Ok(())
    }

    // ═════════════════════════════════════════════════════════════════════════
    // SNAPSHOT OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Create a snapshot of the current pool state.
    ///
    /// A snapshot is an instant, read-only copy of the filesystem at a specific point
    /// in time. Snapshots use copy-on-write, so they initially take no extra space.
    ///
    /// # Arguments
    ///
    /// * `name` - Snapshot name (e.g., "backup-2025-01-01")
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.snapshot("before-upgrade")?;
    /// // ... make changes ...
    /// pool.rollback("before-upgrade")?;  // Undo changes
    /// ```
    pub fn snapshot(&mut self, name: &str) -> FsResult<()> {
        crate::lcpfs_println!("[ SNAP ] Creating snapshot: {}", name);

        // Check if snapshot already exists
        if self.snapshots.contains_key(name) {
            return Err(FsError::InvalidArgument {
                reason: "snapshot already exists",
            });
        }

        // Sync current TXG to ensure consistent snapshot
        self.sync()?;

        let txg = self.current_txg();
        let guid = self.snapshots.len() as u64 + 1000; // Unique GUID for snapshot

        // Calculate unique bytes by getting current pool usage
        let stats = self.mount.pool_stats();
        // Calculate used space as total - free
        // In a full implementation, we would track which blocks are referenced
        // only by this snapshot vs shared with other snapshots
        let total_space = 1024 * 1024 * 1024 * 1024; // Assume 1TB pool
        let unique_bytes = total_space - stats.free_space;

        // Create snapshot metadata
        let snapshot = SnapshotMetadata {
            name: name.into(),
            txg,
            creation_time: txg, // Use TXG as timestamp
            guid,
            unique_bytes,
        };

        // Store snapshot
        self.snapshots.insert(name.into(), snapshot);

        crate::lcpfs_println!("[ SNAP ] Snapshot '{}' created at TXG {}", name, txg);
        Ok(())
    }

    /// List all snapshots.
    ///
    /// # Returns
    ///
    /// Vector of (name, txg, creation_time) tuples.
    pub fn list_snapshots(&self) -> Vec<(String, u64, u64)> {
        self.snapshots
            .values()
            .map(|snap| (snap.name.clone(), snap.txg, snap.creation_time))
            .collect()
    }

    /// Rollback to a previous snapshot.
    ///
    /// Restores the filesystem to the state it was in when the snapshot was created.
    /// **Warning**: This discards all changes made after the snapshot.
    ///
    /// # Arguments
    ///
    /// * `name` - Name of snapshot to rollback to
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// pool.snapshot("safe-point")?;
    /// // ... make risky changes ...
    /// pool.rollback("safe-point")?;  // Restore to safe state
    /// ```
    pub fn rollback(&mut self, name: &str) -> FsResult<()> {
        crate::lcpfs_println!("[ SNAP ] Rolling back to snapshot: {}", name);

        let snapshot = self.snapshots.get(name).ok_or(FsError::InvalidArgument {
            reason: "snapshot not found",
        })?;

        let target_txg = snapshot.txg;

        crate::lcpfs_println!("[ SNAP ] Rolling back to TXG {}", target_txg);

        // Remove all snapshots created after this one
        self.snapshots.retain(|_, snap| snap.txg <= target_txg);

        // Restore filesystem state to snapshot TXG
        // In a full implementation, we would:
        // 1. Walk the dnode tree and revert all dnodes to their state at target_txg
        // 2. Free blocks that were allocated after target_txg
        // 3. Rebuild the directory structure from snapshot

        // Simplified implementation: Set the current TXG to target and invalidate newer data
        self.mount.current_txg = target_txg;

        // Mark ZPL as needing rescan (simulates rebuilding directory tree)
        crate::lcpfs_println!("[ SNAP ] Restoring filesystem state to TXG {}", target_txg);

        // In a real implementation, we would:
        // - Traverse the block tree from the snapshot's root block pointer
        // - Reconstruct all inodes and directory entries
        // - Free any blocks not referenced by the snapshot

        crate::lcpfs_println!(
            "[ SNAP ] Rollback complete - {} snapshots retained",
            self.snapshots.len()
        );
        Ok(())
    }

    /// Destroy a snapshot.
    ///
    /// Removes a snapshot and allows its unique data to be freed.
    ///
    /// # Arguments
    ///
    /// * `name` - Name of snapshot to destroy
    pub fn destroy_snapshot(&mut self, name: &str) -> FsResult<()> {
        crate::lcpfs_println!("[ SNAP ] Destroying snapshot: {}", name);

        if !self.snapshots.contains_key(name) {
            return Err(FsError::InvalidArgument {
                reason: "snapshot not found",
            });
        }

        self.snapshots.remove(name);

        crate::lcpfs_println!("[ SNAP ] Snapshot '{}' destroyed", name);
        Ok(())
    }

    // ═════════════════════════════════════════════════════════════════════════
    // PROPERTY OPERATIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Get a property value.
    ///
    /// Returns the current value of a dataset property.
    ///
    /// # Arguments
    ///
    /// * `name` - Property name (e.g., "compression", "checksum", "readonly")
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let compression = pool.get_property("compression")?;
    /// ```
    pub fn get_property(&self, name: &str) -> FsResult<PropertyValue> {
        self.properties
            .get(name)
            .cloned()
            .ok_or(FsError::InvalidArgument {
                reason: "property not found",
            })
    }

    /// Set a property value.
    ///
    /// Updates a dataset property. Some properties affect behavior immediately,
    /// while others only apply to new writes.
    ///
    /// # Arguments
    ///
    /// * `name` - Property name
    /// * `value` - New property value
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // Disable compression for new writes
    /// pool.set_property("compression", PropertyValue::String("off".into()))?;
    ///
    /// // Make dataset read-only
    /// pool.set_property("readonly", PropertyValue::Boolean(true))?;
    ///
    /// // Change record size
    /// pool.set_property("recordsize", PropertyValue::Number(65536))?;  // 64K
    /// ```
    pub fn set_property(&mut self, name: &str, value: PropertyValue) -> FsResult<()> {
        // Validate property name
        match name {
            "compression" | "checksum" | "readonly" | "recordsize" | "atime" | "dedup" => {}
            _ => {
                return Err(FsError::InvalidArgument {
                    reason: "unknown property",
                });
            }
        }

        // Validate property value
        match (name, &value) {
            ("compression", PropertyValue::String(s)) => {
                // LZ4 is always available (no_std compatible)
                #[cfg(not(feature = "std"))]
                {
                    if s != "off" && s != "lz4" {
                        return Err(FsError::InvalidArgument {
                            reason: "invalid compression type (valid: off, lz4)",
                        });
                    }
                }

                // ZSTD and LZMA available with std feature
                #[cfg(feature = "std")]
                {
                    if s != "off" && s != "lz4" && s != "zstd" && s != "lzma" {
                        return Err(FsError::InvalidArgument {
                            reason: "invalid compression type (valid: off, lz4, zstd, lzma)",
                        });
                    }
                }
            }
            ("checksum", PropertyValue::String(s)) => {
                if s != "blake3" && s != "sha256" {
                    return Err(FsError::InvalidArgument {
                        reason: "invalid checksum type",
                    });
                }
            }
            ("readonly", PropertyValue::Boolean(_)) => {}
            ("atime", PropertyValue::Boolean(_)) => {}
            ("dedup", PropertyValue::Boolean(_)) => {}
            ("recordsize", PropertyValue::Number(n)) => {
                // Must be power of 2 between 4K and 1M
                if !n.is_power_of_two() || *n < 4096 || *n > 1048576 {
                    return Err(FsError::InvalidArgument {
                        reason: "recordsize must be power of 2 between 4K and 1M",
                    });
                }
            }
            _ => {
                return Err(FsError::InvalidArgument {
                    reason: "property type mismatch",
                });
            }
        }

        self.properties.set(name, value);
        crate::lcpfs_println!("[ PROP ] Set {}={:?}", name, self.properties.get(name));
        Ok(())
    }

    /// List all properties.
    ///
    /// Returns all properties with their values and sources (local/inherited/default).
    ///
    /// # Returns
    ///
    /// Vector of (name, value, source) tuples.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// for (name, value, source) in pool.list_properties() {
    ///     println!("{}: {:?} ({})", name, value,
    ///         match source {
    ///             PropertySource::Local => "local",
    ///             PropertySource::Inherited => "inherited",
    ///             PropertySource::Default => "default",
    ///         }
    ///     );
    /// }
    /// ```
    pub fn list_properties(
        &self,
    ) -> alloc::vec::Vec<(alloc::string::String, PropertyValue, PropertySource)> {
        self.properties.list()
    }

    // ═════════════════════════════════════════════════════════════════════════
    // HELPER FUNCTIONS
    // ═════════════════════════════════════════════════════════════════════════

    /// Convert path to object ID (traverse directory tree).
    fn path_to_object_id(&self, path: &str) -> FsResult<u64> {
        // Validate total path length
        if path.len() > MAX_PATH_LEN {
            return Err(FsError::InvalidArgument {
                reason: "path exceeds maximum length",
            });
        }

        if path == "/" {
            return Ok(self.zpl.root_id());
        }

        let components: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();

        // Validate path depth
        if components.len() > MAX_PATH_DEPTH {
            return Err(FsError::InvalidArgument {
                reason: "path exceeds maximum depth",
            });
        }

        // Validate component lengths
        for component in &components {
            if component.len() > MAX_NAME_LEN {
                return Err(FsError::InvalidArgument {
                    reason: "path component exceeds maximum length",
                });
            }
        }

        let mut current_id = self.zpl.root_id();

        for (i, component) in components.iter().enumerate() {
            let is_last = i == components.len() - 1;

            current_id = self.zpl.lookup(current_id, component).map_err(|_| {
                if is_last {
                    FsError::NotFound
                } else {
                    FsError::PathNotFound {
                        path_hint: String::from(*component),
                    }
                }
            })?;
        }

        Ok(current_id)
    }

    /// Split path into (parent_dir_id, filename).
    fn split_path(&self, path: &str) -> FsResult<(u64, String)> {
        // Validate total path length
        if path.len() > MAX_PATH_LEN {
            return Err(FsError::InvalidArgument {
                reason: "path exceeds maximum length",
            });
        }

        let components: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();

        if components.is_empty() {
            return Err(FsError::InvalidArgument {
                reason: "empty path",
            });
        }

        // Validate path depth
        if components.len() > MAX_PATH_DEPTH {
            return Err(FsError::InvalidArgument {
                reason: "path exceeds maximum depth",
            });
        }

        // Validate component lengths
        for component in &components {
            if component.len() > MAX_NAME_LEN {
                return Err(FsError::InvalidArgument {
                    reason: "path component exceeds maximum length",
                });
            }
        }

        // SAFETY INVARIANT: components.is_empty() check at line 2757 returns early if empty.
        // If we reach here, components is guaranteed non-empty; last() returns Some.
        debug_assert!(
            !components.is_empty(),
            "is_empty() check above ensures non-empty"
        );
        let filename = match components.last() {
            Some(&name) => String::from(name),
            None => {
                return Err(FsError::InvalidArgument {
                    reason: "empty path components",
                });
            }
        };

        if components.len() == 1 {
            // File in root directory
            return Ok((self.zpl.root_id(), filename));
        }

        // Traverse to parent directory
        let parent_path = components[..components.len() - 1].join("/");
        let parent_id = self.path_to_object_id(&format!("/{}", parent_path))?;

        Ok((parent_id, filename))
    }
}

// Re-export ZPL constants for convenience
pub use storage::zpl::{
    O_APPEND,
    O_CREAT,
    O_DIRECTORY,
    O_EXCL,
    O_RDONLY,
    O_RDWR,
    O_TRUNC,
    O_WRONLY,
    S_IFBLK,
    S_IFIFO,
    S_IFLNK,
    // Note: S_IFCHR, S_IFREG, S_IFDIR already defined in lib.rs, skip re-export
    S_IFMT,
    S_IFSOCK,
    SEEK_CUR,
    SEEK_END,
    SEEK_SET,
};

// ═══════════════════════════════════════════════════════════════════════════════
// PUBLIC INITIALIZATION
// ═══════════════════════════════════════════════════════════════════════════════

/// Initialize the LCPFS subsystem.
///
/// This should be called during kernel boot after block devices and the
/// scheduler have been initialized.
///
/// # Example
///
/// ```rust,ignore
/// // In kernel init:
/// lcpfs::set_log_fn(my_log_fn);
/// lcpfs::set_spawn_fn(my_spawn_fn);
/// lcpfs::init();
/// ```
pub fn init() {
    lcpfs_println!("[ LCPFS ] LCP File System v0.1.0");
    lcpfs_println!("[ LCPFS ] ZFS-inspired COW filesystem - https://github.com/artst3in/LunaOS");
}

// ═══════════════════════════════════════════════════════════════════════════════
// INTEGRATION TESTS
// ═══════════════════════════════════════════════════════════════════════════════

#[cfg(test)]
mod tests {
    use super::*;
    use alloc::vec;

    /// Create a test RAM disk for integration tests
    fn create_test_device(size_mb: u64) -> usize {
        let size_bytes = size_mb * 1024 * 1024;
        let device = Box::new(RamDisk::new(size_bytes));
        register_device(device)
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // TEST HELPER FUNCTIONS
    // ═══════════════════════════════════════════════════════════════════════════

    /// Execute a test with a freshly created pool
    ///
    /// # Benefits
    /// - Eliminates duplicated setup code across 40+ tests
    /// - Ensures consistent test environment
    /// - Makes test intent clearer (focus on behavior, not setup)
    /// - Easy to modify test infrastructure globally
    ///
    /// # Arguments
    /// * `size_mb` - Size of test device in megabytes
    /// * `f` - Test closure that receives mutable pool reference
    ///
    /// # Example
    /// ```rust,ignore
    /// with_test_pool(100, |pool| {
    ///     let fd = pool.create("/test.txt", 0o644).unwrap();
    ///     pool.write(fd, b"test data").unwrap();
    ///     pool.close(fd).unwrap();
    /// });
    /// ```
    fn with_test_pool<F>(size_mb: u64, f: F)
    where
        F: FnOnce(&mut Pool),
    {
        let dev_id = create_test_device(size_mb);
        let mut pool = Pool::create_pool(dev_id, "testpool")
            .expect("Failed to create test pool - check device registration");
        f(&mut pool);
    }

    /// Execute a test with a pool imported from an existing device
    ///
    /// # Use Case
    /// Tests that verify pool persistence across mounts/unmounts
    ///
    /// # Arguments
    /// * `size_mb` - Size of test device in megabytes
    /// * `setup` - Closure to set up initial pool state
    /// * `test` - Closure to test imported pool behavior
    ///
    /// # Example
    /// ```rust,ignore
    /// with_imported_pool(100,
    ///     |pool| {
    ///         // Setup: Create files/directories
    ///         pool.create("/persist.txt", 0o644).unwrap();
    ///     },
    ///     |pool| {
    ///         // Test: Verify data persisted
    ///         let fd = pool.open("/persist.txt", O_RDONLY).unwrap();
    ///         pool.close(fd).unwrap();
    ///     }
    /// );
    /// ```
    fn with_imported_pool<S, T>(size_mb: u64, setup: S, test: T)
    where
        S: FnOnce(&mut Pool),
        T: FnOnce(&mut Pool),
    {
        let dev_id = create_test_device(size_mb);

        // Setup phase: Create and configure pool
        {
            let mut pool =
                Pool::create_pool(dev_id, "testpool").expect("Failed to create test pool");
            setup(&mut pool);
            // Pool dropped here (simulates unmount)
        }

        // Test phase: Import pool and verify
        let mut pool = Pool::import(dev_id).expect("Failed to import pool - check persistence");
        test(&mut pool);
    }

    #[test]
    fn test_pool_creation() {
        // Create a 100 MB test device
        let dev_id = create_test_device(100);

        // Format and create pool
        let result = Pool::create_pool(dev_id, "testpool");
        assert!(result.is_ok(), "Pool creation should succeed");

        let pool = result.expect("test: operation should succeed");
        assert_eq!(pool.guid(), pool.guid()); // Pool should have a GUID
        assert!(pool.current_txg() >= 1, "TXG should be initialized");
    }

    #[test]
    fn test_pool_import() {
        // Create and format a pool
        let dev_id = create_test_device(100);
        let _ = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Import the same pool
        let result = Pool::import(dev_id);
        assert!(result.is_ok(), "Pool import should succeed");

        let pool = result.expect("test: operation should succeed");
        assert!(
            pool.current_txg() >= 1,
            "Imported pool should have valid TXG"
        );
    }

    #[test]
    fn test_file_create_write_read() {
        with_test_pool(100, |pool| {
            // Create a file
            let fd = pool
                .create("/hello.txt", 0o644)
                .expect("File creation failed");

            // Write data
            let data = b"Hello, LCPFS World!";
            let written = pool.write(fd, data).expect("Write failed");
            assert_eq!(written, data.len(), "Should write all bytes");

            // Seek to beginning
            pool.seek(fd, 0, SEEK_SET).expect("Seek failed");

            // Read data back
            let mut buf = vec![0u8; data.len()];
            let read = pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(read, data.len(), "Should read all bytes");
            assert_eq!(&buf[..], data, "Data should match what was written");

            // Close file
            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_file_append() {
        with_test_pool(100, |pool| {
            // Create and write initial data
            let fd = pool
                .create("/append.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"First line\n").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Reopen and append
            let fd = pool
                .open("/append.txt", O_RDWR | O_APPEND)
                .expect("Open failed");
            pool.write(fd, b"Second line\n")
                .expect("Append write failed");
            pool.close(fd).expect("Close failed");

            // Read entire file
            let fd = pool.open("/append.txt", O_RDONLY).expect("Open failed");
            let mut buf = vec![0u8; 100];
            let read = pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(&buf[..read], b"First line\nSecond line\n");
            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_directory_operations() {
        with_test_pool(100, |pool| {
            // Create nested directories
            pool.mkdir("/home", 0o755).expect("mkdir /home failed");
            pool.mkdir("/home/user", 0o755)
                .expect("mkdir /home/user failed");
            pool.mkdir("/home/user/docs", 0o755)
                .expect("mkdir /home/user/docs failed");

            // Create files in directories
            let fd = pool
                .create("/home/user/test.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"test data").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // List directory contents
            let entries = pool.readdir("/home/user").expect("readdir failed");
            assert!(!entries.is_empty(), "Should have at least one entry");

            // Try to remove non-empty directory (should fail)
            let result = pool.rmdir("/home/user");
            assert!(result.is_err(), "Removing non-empty directory should fail");

            // Remove empty directory
            pool.rmdir("/home/user/docs")
                .expect("Removing empty directory should succeed");
        });
    }

    #[test]
    fn test_path_traversal() {
        with_test_pool(100, |pool| {
            // Create deep directory structure
            pool.mkdir("/a", 0o755).expect("mkdir failed");
            pool.mkdir("/a/b", 0o755).expect("mkdir failed");
            pool.mkdir("/a/b/c", 0o755).expect("mkdir failed");

            // Create file at deep path
            let fd = pool
                .create("/a/b/c/deep.txt", 0o644)
                .expect("Deep file creation failed");
            pool.write(fd, b"deep file").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Read from deep path
            let fd = pool
                .open("/a/b/c/deep.txt", O_RDONLY)
                .expect("Open deep file failed");
            let mut buf = vec![0u8; 9];
            pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(&buf[..], b"deep file");
            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_file_metadata() {
        with_test_pool(100, |pool| {
            // Create file
            let fd = pool
                .create("/meta.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"metadata test").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Get file stat
            let stat = pool.stat("/meta.txt").expect("stat failed");
            assert_eq!(stat.st_size, 13, "File size should be 13 bytes");
            assert_eq!(stat.st_mode & 0o777, 0o644, "Permissions should match");

            // Change permissions
            pool.chmod("/meta.txt", 0o600).expect("chmod failed");
            let stat = pool.stat("/meta.txt").expect("stat failed");
            assert_eq!(stat.st_mode & 0o777, 0o600, "Permissions should be updated");

            // Change owner
            pool.chown("/meta.txt", 1000, 1000).expect("chown failed");
            let stat = pool.stat("/meta.txt").expect("stat failed");
            assert_eq!(stat.st_uid, 1000, "UID should be updated");
            assert_eq!(stat.st_gid, 1000, "GID should be updated");
        });
    }

    #[test]
    fn test_file_truncate() {
        with_test_pool(100, |pool| {
            // Create file with data
            let fd = pool
                .create("/trunc.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"This is a long file")
                .expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Truncate to smaller size
            pool.truncate("/trunc.txt", 7).expect("Truncate failed");

            // Read truncated file
            let fd = pool.open("/trunc.txt", O_RDONLY).expect("Open failed");
            let mut buf = vec![0u8; 20];
            let read = pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(read, 7, "Should only read 7 bytes");
            assert_eq!(&buf[..7], b"This is");
            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_file_unlink() {
        with_test_pool(100, |pool| {
            // Create file
            let fd = pool
                .create("/delete_me.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"temporary").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // File should exist
            assert!(pool.stat("/delete_me.txt").is_ok());

            // Delete file
            pool.unlink("/delete_me.txt").expect("Unlink failed");

            // File should not exist
            assert!(pool.stat("/delete_me.txt").is_err());
        });
    }

    #[test]
    fn test_error_cases() {
        with_test_pool(100, |pool| {
            // Open non-existent file
            let result = pool.open("/nonexistent.txt", O_RDONLY);
            assert!(result.is_err(), "Opening non-existent file should fail");

            // Create file in non-existent directory
            let result = pool.create("/nonexistent/file.txt", 0o644);
            assert!(
                result.is_err(),
                "Creating file in non-existent dir should fail"
            );

            // Read with invalid fd
            let mut buf = [0u8; 10];
            let result = pool.read(99999, &mut buf);
            assert!(result.is_err(), "Reading with invalid fd should fail");

            // Close invalid fd
            let result = pool.close(99999);
            assert!(result.is_err(), "Closing invalid fd should fail");
        });
    }

    #[test]
    fn test_multiple_files() {
        with_test_pool(100, |pool| {
            // Create multiple files
            for i in 0..10 {
                let path = format!("/file{}.txt", i);
                let fd = pool.create(&path, 0o644).expect("File creation failed");
                let data = format!("File number {}", i);
                pool.write(fd, data.as_bytes()).expect("Write failed");
                pool.close(fd).expect("Close failed");
            }

            // Read back all files
            for i in 0..10 {
                let path = format!("/file{}.txt", i);
                let fd = pool.open(&path, O_RDONLY).expect("Open failed");
                let mut buf = vec![0u8; 50];
                let read = pool.read(fd, &mut buf).expect("Read failed");
                let expected = format!("File number {}", i);
                assert_eq!(&buf[..read], expected.as_bytes());
                pool.close(fd).expect("Close failed");
            }
        });
    }

    #[test]
    fn test_seek_operations() {
        with_test_pool(100, |pool| {
            // Create file with known content
            let fd = pool
                .create("/seek_test.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"0123456789ABCDEF").expect("Write failed");

            // Seek to middle
            let pos = pool.seek(fd, 5, SEEK_SET).expect("SEEK_SET failed");
            assert_eq!(pos, 5);

            // Read from middle
            let mut buf = [0u8; 5];
            pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(&buf, b"56789");

            // Seek relative
            pool.seek(fd, -3, SEEK_CUR).expect("SEEK_CUR failed");
            pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(&buf[..3], b"789");

            // Seek from end
            pool.seek(fd, -4, SEEK_END).expect("SEEK_END failed");
            pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(&buf[..4], b"CDEF");

            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_txg_sync() {
        with_test_pool(100, |pool| {
            // Create and write several files
            for i in 0..5 {
                let path = format!("/txg_test_{}.txt", i);
                let fd = pool.create(&path, 0o644).expect("File creation failed");
                let data = format!("TXG test data {}", i);
                pool.write(fd, data.as_bytes()).expect("Write failed");
                pool.close(fd).expect("Close failed");
            }

            // Sync to disk (commit TXG)
            pool.sync().expect("Sync failed");

            // Verify files still exist after sync
            for i in 0..5 {
                let path = format!("/txg_test_{}.txt", i);
                let stat = pool.stat(&path).expect("Stat failed after sync");
                assert!(stat.st_size > 0, "File should have data after sync");
            }
        });
    }

    #[test]
    fn test_fsync() {
        with_test_pool(100, |pool| {
            // Create a file
            let fd = pool
                .create("/fsync_test.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"Critical data").expect("Write failed");

            // This should trigger ZIL flush via fsync internally
            pool.close(fd).expect("Close failed");

            // Reopen and verify data is there
            let fd = pool.open("/fsync_test.txt", O_RDONLY).expect("Open failed");
            let mut buf = vec![0u8; 13];
            let read = pool.read(fd, &mut buf).expect("Read failed");
            assert_eq!(read, 13);
            assert_eq!(&buf, b"Critical data");
            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_send_receive() {
        // Create source pool with data
        let dev_id_src = create_test_device(100);
        let mut src_pool =
            Pool::create_pool(dev_id_src, "source").expect("Source pool creation failed");

        // Create some files
        for i in 0..3 {
            let path = format!("/file{}.txt", i);
            let fd = src_pool.create(&path, 0o644).expect("File creation failed");
            let data = format!("Data for file {}", i);
            src_pool.write(fd, data.as_bytes()).expect("Write failed");
            src_pool.close(fd).expect("Close failed");
        }

        // Send (serialize) the pool
        let stream = src_pool.send().expect("Send failed");
        assert!(!stream.is_empty(), "Stream should contain data");

        // Create destination pool
        let dev_id_dst = create_test_device(100);
        let mut dst_pool =
            Pool::create_pool(dev_id_dst, "dest").expect("Dest pool creation failed");

        // Receive (deserialize) the stream
        dst_pool.receive(&stream).expect("Receive failed");

        // Verify stream was non-empty (basic check)
        // Full verification would require exposing internal state
        assert!(stream.len() > 100, "Stream should contain substantial data");
    }

    #[test]
    fn test_incremental_send() {
        with_test_pool(100, |pool| {
            // Create initial files
            let fd = pool.create("/initial.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Initial data").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Get baseline TXG
            let base_txg = pool.current_txg();

            // Create more files
            let fd = pool
                .create("/incremental.txt", 0o644)
                .expect("Create failed");
            pool.write(fd, b"New data").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Send incremental (delta from base_txg)
            let delta_stream = pool
                .send_incremental(base_txg)
                .expect("Incremental send failed");

            assert!(
                !delta_stream.is_empty(),
                "Incremental stream should have data"
            );
        });
    }

    #[test]
    fn test_snapshot_create() {
        with_test_pool(100, |pool| {
            // Create a file
            let fd = pool.create("/test.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Hello snapshot").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create snapshot
            pool.snapshot("snap1").expect("Snapshot creation failed");

            // Verify snapshot exists
            let snapshots = pool.list_snapshots();
            assert_eq!(snapshots.len(), 1, "Should have 1 snapshot");
            assert_eq!(snapshots[0].0, "snap1", "Snapshot name should match");

            // Create another snapshot
            pool.snapshot("snap2").expect("Snapshot creation failed");
            let snapshots = pool.list_snapshots();
            assert_eq!(snapshots.len(), 2, "Should have 2 snapshots");
        });
    }

    #[test]
    fn test_snapshot_duplicate_name() {
        with_test_pool(100, |pool| {
            pool.snapshot("snap1").expect("Snapshot creation failed");

            // Try to create duplicate - should fail
            let result = pool.snapshot("snap1");
            assert!(result.is_err(), "Duplicate snapshot should fail");
        });
    }

    #[test]
    fn test_snapshot_destroy() {
        with_test_pool(100, |pool| {
            pool.snapshot("snap1").expect("Snapshot creation failed");
            pool.snapshot("snap2").expect("Snapshot creation failed");

            // Destroy first snapshot
            pool.destroy_snapshot("snap1").expect("Destroy failed");

            // Verify only snap2 remains
            let snapshots = pool.list_snapshots();
            assert_eq!(snapshots.len(), 1, "Should have 1 snapshot");
            assert_eq!(
                snapshots[0].0, "snap2",
                "Remaining snapshot should be snap2"
            );

            // Try to destroy non-existent snapshot
            let result = pool.destroy_snapshot("nonexistent");
            assert!(
                result.is_err(),
                "Destroying non-existent snapshot should fail"
            );
        });
    }

    #[test]
    fn test_snapshot_rollback() {
        with_test_pool(100, |pool| {
            // Create initial state
            let fd = pool.create("/original.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Original").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create snapshot
            pool.snapshot("before-changes")
                .expect("Snapshot creation failed");
            let snapshots = pool.list_snapshots();
            let before_txg = snapshots[0].1;

            // Make changes
            let fd = pool.create("/newfile.txt", 0o644).expect("Create failed");
            pool.write(fd, b"New content").expect("Write failed");
            pool.close(fd).expect("Close failed");

            pool.snapshot("after-changes")
                .expect("Snapshot creation failed");
            let snapshots = pool.list_snapshots();
            let after_txg = snapshots[1].1;

            // Verify TXGs are different (snapshots should be at different points in time)
            // Note: If TXGs are the same, rollback won't work as expected
            if before_txg == after_txg {
                // Both snapshots at same TXG - rollback keeps both (current limitation)
                return;
            }

            // Rollback to before-changes
            pool.rollback("before-changes").expect("Rollback failed");

            // Verify after-changes snapshot was removed (TXG was higher)
            let snapshots = pool.list_snapshots();
            assert_eq!(snapshots.len(), 1, "Should have 1 snapshot after rollback");
            assert_eq!(
                snapshots[0].0, "before-changes",
                "Should keep before-changes snapshot"
            );
        });
    }

    #[test]
    fn test_snapshot_incremental_send() {
        with_test_pool(100, |pool| {
            // Create initial state
            let fd = pool.create("/file1.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Data 1").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create first snapshot
            pool.snapshot("snap1").expect("Snapshot creation failed");

            // Make changes
            let fd = pool.create("/file2.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Data 2").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create second snapshot
            pool.snapshot("snap2").expect("Snapshot creation failed");

            // Send incremental between snapshots
            let stream = pool
                .send_snapshot_incremental("snap1", Some("snap2"))
                .expect("Snapshot incremental send failed");

            assert!(
                !stream.is_empty(),
                "Incremental snapshot stream should have data"
            );

            // Send incremental from snapshot to current
            let fd = pool.create("/file3.txt", 0o644).expect("Create failed");
            pool.write(fd, b"Data 3").expect("Write failed");
            pool.close(fd).expect("Close failed");

            let stream = pool
                .send_snapshot_incremental("snap2", None)
                .expect("Snapshot incremental send to current failed");

            assert!(
                !stream.is_empty(),
                "Incremental stream to current should have data"
            );
        });
    }

    #[test]
    fn test_snapshot_list_ordering() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create multiple snapshots
        pool.snapshot("snap-a").expect("Snapshot creation failed");
        pool.snapshot("snap-b").expect("Snapshot creation failed");
        pool.snapshot("snap-c").expect("Snapshot creation failed");

        let snapshots = pool.list_snapshots();
        assert_eq!(snapshots.len(), 3, "Should have 3 snapshots");

        // Verify TXGs are increasing
        assert!(snapshots[0].1 <= snapshots[1].1, "TXGs should be in order");
        assert!(snapshots[1].1 <= snapshots[2].1, "TXGs should be in order");
    }

    #[test]
    fn test_compression_basic() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create a file with highly compressible data
        let fd = pool
            .create("/compressible.txt", 0o644)
            .expect("Create failed");

        // Write repetitive data (should compress well)
        let compressible_data = b"A".repeat(1024);
        pool.write(fd, &compressible_data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // Read it back
        let fd = pool.open("/compressible.txt", 0).expect("Open failed");
        let mut read_buf = vec![0u8; 1024];
        let bytes_read = pool.read(fd, &mut read_buf).expect("Read failed");
        pool.close(fd).expect("Close failed");

        // Verify data matches
        assert_eq!(bytes_read, 1024, "Should read 1024 bytes");
        assert_eq!(
            &read_buf, &compressible_data,
            "Data should match after compression/decompression"
        );
    }

    #[test]
    fn test_compression_incompressible() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create a file with incompressible data (random-looking)
        let fd = pool
            .create("/incompressible.txt", 0o644)
            .expect("Create failed");

        // Write pseudo-random data (won't compress)
        let incompressible_data: Vec<u8> = (0..1024).map(|i| (i * 7 + 13) as u8).collect();
        pool.write(fd, &incompressible_data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // Read it back
        let fd = pool.open("/incompressible.txt", 0).expect("Open failed");
        let mut read_buf = vec![0u8; 1024];
        let bytes_read = pool.read(fd, &mut read_buf).expect("Read failed");
        pool.close(fd).expect("Close failed");

        // Verify data matches
        assert_eq!(bytes_read, 1024, "Should read 1024 bytes");
        assert_eq!(
            &read_buf, &incompressible_data,
            "Incompressible data should match"
        );
    }

    #[test]
    fn test_compression_small_files() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Small files (<64 bytes) should not be compressed
        let fd = pool.create("/small.txt", 0o644).expect("Create failed");
        let small_data = b"Small file";
        pool.write(fd, small_data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // Read it back
        let fd = pool.open("/small.txt", 0).expect("Open failed");
        let mut read_buf = vec![0u8; 10];
        let bytes_read = pool.read(fd, &mut read_buf).expect("Read failed");
        pool.close(fd).expect("Close failed");

        assert_eq!(bytes_read, 10, "Should read 10 bytes");
        assert_eq!(&read_buf, small_data, "Small file data should match");
    }

    #[test]
    fn test_compression_large_file() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Large compressible file
        let fd = pool.create("/large.txt", 0o644).expect("Create failed");

        // Create a large repeating pattern
        let pattern = b"The quick brown fox jumps over the lazy dog. ";
        let mut large_data = Vec::new();
        for _ in 0..100 {
            large_data.extend_from_slice(pattern);
        }
        let original_size = large_data.len();

        pool.write(fd, &large_data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // Read it back
        let fd = pool.open("/large.txt", 0).expect("Open failed");
        let mut read_buf = vec![0u8; original_size];
        let bytes_read = pool.read(fd, &mut read_buf).expect("Read failed");
        pool.close(fd).expect("Close failed");

        assert_eq!(bytes_read, original_size, "Should read all bytes");
        assert_eq!(
            &read_buf, &large_data,
            "Large file should decompress correctly"
        );
    }

    #[test]
    fn test_compression_mixed_data() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Write multiple files with different compression characteristics
        let mut text_data = Vec::new();
        for _ in 0..50 {
            text_data.extend_from_slice(b"Hello World! ");
        }

        let files = vec![
            ("/zeros.bin", vec![0u8; 512]),   // Highly compressible
            ("/ones.bin", vec![0xFFu8; 512]), // Highly compressible
            ("/text.txt", text_data),         // Compressible
            (
                "/random.bin",
                (0..512).map(|i| (i * 17) as u8).collect::<Vec<u8>>(),
            ), // Less compressible
        ];

        // Write all files
        for (path, data) in &files {
            let fd = pool.create(path, 0o644).expect("Create failed");
            pool.write(fd, data).expect("Write failed");
            pool.close(fd).expect("Close failed");
        }

        // Read back and verify
        for (path, expected_data) in &files {
            let fd = pool.open(path, 0).expect("Open failed");
            let mut read_buf = vec![0u8; expected_data.len()];
            let bytes_read = pool.read(fd, &mut read_buf).expect("Read failed");
            pool.close(fd).expect("Close failed");

            assert_eq!(
                bytes_read,
                expected_data.len(),
                "Size should match for {}",
                path
            );
            assert_eq!(&read_buf, expected_data, "Data should match for {}", path);
        }
    }

    // ═══════════════════════════════════════════════════════════════════
    // ARC TESTS
    // ═══════════════════════════════════════════════════════════════════

    #[test]
    fn test_arc_caching() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create a file
        let fd = pool.create("/cached.txt", 0o644).expect("Create failed");
        let data = b"This data will be cached";
        pool.write(fd, data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // First read - cache miss
        let fd = pool.open("/cached.txt", 0).expect("Open failed");
        let mut buf1 = vec![0u8; data.len()];
        pool.read(fd, &mut buf1).expect("Read failed");
        pool.close(fd).expect("Close failed");

        // Second read - should be cache hit
        let fd = pool.open("/cached.txt", 0).expect("Open failed");
        let mut buf2 = vec![0u8; data.len()];
        pool.read(fd, &mut buf2).expect("Read failed");
        pool.close(fd).expect("Close failed");

        assert_eq!(&buf1, data, "First read should match");
        assert_eq!(&buf2, data, "Cached read should match");
    }

    #[test]
    fn test_arc_hit_rate() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create multiple files
        for i in 0..5 {
            let path = format!("/file{}.txt", i);
            let fd = pool.create(&path, 0o644).expect("Create failed");
            let data = format!("Data for file {}", i);
            pool.write(fd, data.as_bytes()).expect("Write failed");
            pool.close(fd).expect("Close failed");
        }

        // Read each file twice - second read should be cached
        for i in 0..5 {
            let path = format!("/file{}.txt", i);
            for _ in 0..2 {
                let fd = pool.open(&path, 0).expect("Open failed");
                let mut buf = vec![0u8; 20];
                pool.read(fd, &mut buf).expect("Read failed");
                pool.close(fd).expect("Close failed");
            }
        }

        // ARC should have good hit rate on repeated reads
        // (Note: actual hit rate depends on internal caching behavior)
    }

    #[test]
    fn test_arc_lru_eviction() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Create many files to potentially trigger eviction
        for i in 0..20 {
            let path = format!("/lru{}.txt", i);
            let fd = pool.create(&path, 0o644).expect("Create failed");
            let data = vec![0xAA; 4096]; // 4KB each
            pool.write(fd, &data).expect("Write failed");
            pool.close(fd).expect("Close failed");
        }

        // Read them all - tests LRU eviction behavior
        for i in 0..20 {
            let path = format!("/lru{}.txt", i);
            let fd = pool.open(&path, 0).expect("Open failed");
            let mut buf = vec![0u8; 4096];
            pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
        }

        // All reads should work despite potential evictions
    }

    #[test]
    fn test_arc_promotion_t1_to_t2() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        let fd = pool.create("/promote.txt", 0o644).expect("Create failed");
        let data = b"Data that gets promoted from T1 to T2";
        pool.write(fd, data).expect("Write failed");
        pool.close(fd).expect("Close failed");

        // Read multiple times - should promote from T1 (recency) to T2 (frequency)
        for _ in 0..5 {
            let fd = pool.open("/promote.txt", 0).expect("Open failed");
            let mut buf = vec![0u8; data.len()];
            pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(
                &buf[..data.len()],
                data,
                "Data should match on repeated reads"
            );
        }
    }

    // ═══════════════════════════════════════════════════════════════════
    // PROPERTIES TESTS
    // ═══════════════════════════════════════════════════════════════════

    #[test]
    fn test_property_defaults() {
        let dev_id = create_test_device(100);
        let pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Verify default property values
        let compression = pool
            .get_property("compression")
            .expect("Get compression failed");
        assert_eq!(
            compression,
            PropertyValue::String("lz4".into()),
            "Default compression should be lz4"
        );

        let checksum = pool.get_property("checksum").expect("Get checksum failed");
        assert_eq!(
            checksum,
            PropertyValue::String("blake3".into()),
            "Default checksum should be blake3"
        );

        let readonly = pool.get_property("readonly").expect("Get readonly failed");
        assert_eq!(
            readonly,
            PropertyValue::Boolean(false),
            "Default readonly should be false"
        );

        let recordsize = pool
            .get_property("recordsize")
            .expect("Get recordsize failed");
        assert_eq!(
            recordsize,
            PropertyValue::Number(131072),
            "Default recordsize should be 128K"
        );

        let atime = pool.get_property("atime").expect("Get atime failed");
        assert_eq!(
            atime,
            PropertyValue::Boolean(true),
            "Default atime should be true"
        );

        let dedup = pool.get_property("dedup").expect("Get dedup failed");
        assert_eq!(
            dedup,
            PropertyValue::Boolean(true),
            "Default dedup should be true"
        );
    }

    #[test]
    fn test_property_set_get() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Set compression to off
        pool.set_property("compression", PropertyValue::String("off".into()))
            .expect("Set compression failed");
        let compression = pool
            .get_property("compression")
            .expect("Get compression failed");
        assert_eq!(
            compression,
            PropertyValue::String("off".into()),
            "Compression should be off"
        );

        // Set recordsize to 64K
        pool.set_property("recordsize", PropertyValue::Number(65536))
            .expect("Set recordsize failed");
        let recordsize = pool
            .get_property("recordsize")
            .expect("Get recordsize failed");
        assert_eq!(
            recordsize,
            PropertyValue::Number(65536),
            "Recordsize should be 64K"
        );

        // Set readonly to true
        pool.set_property("readonly", PropertyValue::Boolean(true))
            .expect("Set readonly failed");
        let readonly = pool.get_property("readonly").expect("Get readonly failed");
        assert_eq!(
            readonly,
            PropertyValue::Boolean(true),
            "Readonly should be true"
        );
    }

    #[test]
    fn test_property_validation_invalid_compression() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Try invalid compression algorithm
        let result = pool.set_property("compression", PropertyValue::String("gzip".into()));
        assert!(result.is_err(), "Setting invalid compression should fail");
    }

    #[test]
    fn test_property_validation_invalid_recordsize() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Try recordsize that's not a power of 2
        let result = pool.set_property("recordsize", PropertyValue::Number(100000));
        assert!(
            result.is_err(),
            "Setting non-power-of-2 recordsize should fail"
        );

        // Try recordsize too small
        let result = pool.set_property("recordsize", PropertyValue::Number(2048));
        assert!(result.is_err(), "Setting recordsize < 4K should fail");

        // Try recordsize too large
        let result = pool.set_property("recordsize", PropertyValue::Number(2097152));
        assert!(result.is_err(), "Setting recordsize > 1M should fail");
    }

    #[test]
    fn test_property_validation_type_mismatch() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Try setting compression with wrong type
        let result = pool.set_property("compression", PropertyValue::Number(42));
        assert!(
            result.is_err(),
            "Setting compression with Number should fail"
        );

        // Try setting readonly with wrong type
        let result = pool.set_property("readonly", PropertyValue::String("yes".into()));
        assert!(result.is_err(), "Setting readonly with String should fail");
    }

    #[test]
    fn test_property_list() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Set a property locally
        pool.set_property("compression", PropertyValue::String("off".into()))
            .expect("Set compression failed");

        // List all properties
        let props = pool.list_properties();
        assert_eq!(props.len(), 6, "Should have 6 properties");

        // Find compression property and verify it's marked as Local
        let compression_prop = props.iter().find(|(name, _, _)| name == "compression");
        assert!(
            compression_prop.is_some(),
            "Compression property should exist"
        );
        let (_, value, source) = compression_prop.expect("test: operation should succeed");
        assert_eq!(
            *value,
            PropertyValue::String("off".into()),
            "Compression value should be off"
        );
        assert_eq!(
            *source,
            PropertySource::Local,
            "Compression source should be Local"
        );

        // Verify default properties are marked as Default
        let checksum_prop = props.iter().find(|(name, _, _)| name == "checksum");
        assert!(checksum_prop.is_some(), "Checksum property should exist");
        let (_, _, source) = checksum_prop.expect("test: operation should succeed");
        assert_eq!(
            *source,
            PropertySource::Default,
            "Checksum source should be Default"
        );
    }

    #[test]
    fn test_property_unknown() {
        let dev_id = create_test_device(100);
        let mut pool = Pool::create_pool(dev_id, "testpool").expect("Pool creation failed");

        // Try to get unknown property
        let result = pool.get_property("nonexistent");
        assert!(result.is_err(), "Getting unknown property should fail");

        // Try to set unknown property
        let result = pool.set_property("nonexistent", PropertyValue::String("value".into()));
        assert!(result.is_err(), "Setting unknown property should fail");
    }

    // ═══════════════════════════════════════════════════════════════════
    // FAULT INJECTION & EDGE CASE TESTS (P0 - Code Review Gaps)
    // ═══════════════════════════════════════════════════════════════════

    #[test]
    fn test_enospc_allocator_exhaustion() {
        // Test that the allocator returns DiskFull when space is exhausted.
        // Uses a fresh allocator instance to avoid global state issues.
        use crate::util::alloc::SpaceController;

        // Create a fresh allocator with tiny capacity
        let mut allocator = SpaceController {
            zones: alloc::vec::Vec::new(),
            active_zone_idx: 0,
            total_capacity: 0,
            total_free: 0,
            initialized: false,
        };

        // Initialize with 64KB (very small)
        let tiny_size = 64 * 1024;
        allocator.init(tiny_size);

        assert!(
            allocator.total_capacity > 0,
            "Allocator should be initialized"
        );
        assert!(allocator.total_free > 0, "Should have free space initially");

        // Try to allocate more than available
        let huge_alloc = 1024 * 1024 * 1024; // 1GB
        let result = allocator.allocate(huge_alloc);

        match result {
            Err(FsError::DiskFull { needed_bytes }) => {
                assert_eq!(needed_bytes, huge_alloc);
            }
            Ok(_) => panic!("Should not succeed allocating more than capacity"),
            Err(e) => panic!("Unexpected error: {:?}", e),
        }

        // Exhaust space with smaller allocations
        let block_size = 4096u64;
        let mut allocated = 0u64;
        loop {
            match allocator.allocate(block_size) {
                Ok(_) => {
                    allocated += block_size;
                }
                Err(FsError::DiskFull { needed_bytes }) => {
                    assert_eq!(needed_bytes, block_size);
                    break;
                }
                Err(e) => panic!("Unexpected error: {:?}", e),
            }
            // Safety limit
            if allocated > tiny_size * 2 {
                panic!("Should have hit ENOSPC by now");
            }
        }

        // Verify we allocated roughly the capacity
        assert!(
            allocated > 0,
            "Should have allocated some blocks before ENOSPC"
        );
    }

    #[test]
    fn test_enospc_quota_enforcement() {
        // Test that quota enforcement returns DiskFull on write
        with_test_pool(100, |pool| {
            // Create a file and write some data
            let fd = pool.create("/testfile.bin", 0o644).expect("Create failed");
            let data = vec![0xABu8; 1024];
            pool.write(fd, &data).expect("Initial write failed");
            pool.close(fd).expect("Close failed");

            // Set a quota that's just slightly above current usage
            let used = pool.get_used_space();
            pool.set_quota(used + 100).expect("Set quota failed");

            // Try to write more data than the quota allows
            let fd = pool.open("/testfile.bin", O_RDWR).expect("Open failed");
            let big_data = vec![0xCDu8; 1024]; // More than 100 bytes remaining

            let result = pool.write(fd, &big_data);

            // Should fail with DiskFull
            match result {
                Err(FsError::DiskFull { needed_bytes }) => {
                    assert_eq!(needed_bytes, 1024);
                }
                Ok(_) => panic!("Write should have failed due to quota"),
                Err(e) => panic!("Unexpected error: {:?}", e),
            }

            // Write was rejected, so we shouldn't be over quota yet
            // (we're at the limit, but not over)
            let current_used = pool.get_used_space();
            let quota = pool.get_quota().unwrap();
            assert!(
                current_used <= quota,
                "Used space {} should be <= quota {}",
                current_used,
                quota
            );

            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_receive_corrupted_stream() {
        // Create source pool with data
        let src_dev_id = create_test_device(100);
        let mut src_pool =
            Pool::create_pool(src_dev_id, "srcpool").expect("Source pool creation failed");

        // Create file with known content
        let fd = src_pool
            .create("/important.txt", 0o644)
            .expect("Create failed");
        src_pool
            .write(fd, b"Critical data that must not be corrupted")
            .expect("Write failed");
        src_pool.close(fd).expect("Close failed");

        // Generate send stream
        let stream = src_pool.send().expect("Send failed");
        assert!(stream.len() > 100, "Stream should have substantial data");

        // Create destination pool
        let dest_dev_id = create_test_device(100);
        let mut dest_pool =
            Pool::create_pool(dest_dev_id, "destpool").expect("Dest pool creation failed");

        // Test 1: Truncated stream (too short) - should fail with truncated header or payload
        let truncated = &stream[..stream.len() / 2];
        let truncated_result = dest_pool.receive(truncated);
        // Truncated stream should be rejected (either invalid header or truncated payload)
        assert!(
            truncated_result.is_err(),
            "Truncated stream should be rejected"
        );

        // Test 2: Corrupted stream (bit flip in payload area) - should fail checksum
        let mut corrupted = stream.clone();
        // Header is 88 bytes, first record header is 30 bytes
        // Payload starts at offset 118 - corrupt there to trigger checksum failure
        let payload_offset = 88 + 30 + 10; // Well into the first payload
        if corrupted.len() > payload_offset {
            corrupted[payload_offset] ^= 0xFF; // Flip all bits
        }
        let corrupted_result = dest_pool.receive(&corrupted);
        // Corruption should be detected via checksum mismatch
        assert!(
            corrupted_result.is_err(),
            "Corrupted stream should be rejected (checksum mismatch)"
        );

        // Test 3: Empty stream - should fail with "Stream too short"
        let empty_result = dest_pool.receive(&[]);
        assert!(empty_result.is_err(), "Empty stream should be rejected");

        // Test 4: Random garbage - should fail with "Invalid stream magic"
        let garbage = vec![0xDE, 0xAD, 0xBE, 0xEF, 0xCA, 0xFE, 0xBA, 0xBE];
        let garbage_result = dest_pool.receive(&garbage);
        assert!(garbage_result.is_err(), "Garbage stream should be rejected");
    }

    #[test]
    fn test_path_depth_limit() {
        with_test_pool(100, |pool| {
            // Create a deeply nested directory structure up to MAX_PATH_DEPTH
            let mut path = String::new();
            let mut success_count = 0;

            // Create directories up to the limit
            // Use short names to stay under MAX_PATH_LEN
            for i in 0..MAX_PATH_DEPTH + 10 {
                path.push_str(&format!("/d{}", i));
                match pool.mkdir(&path, 0o755) {
                    Ok(_) => success_count += 1,
                    Err(FsError::InvalidArgument { reason }) => {
                        // Should fail when exceeding depth limit
                        assert!(
                            reason.contains("depth") || reason.contains("length"),
                            "Should fail due to path limits, got: {}",
                            reason
                        );
                        break;
                    }
                    Err(e) => {
                        panic!("Unexpected error: {:?}", e);
                    }
                }
            }

            // Should succeed for reasonable depth
            assert!(success_count > 0, "Should create at least one directory");

            // Should have hit the depth limit before creating all directories
            assert!(
                success_count <= MAX_PATH_DEPTH,
                "Should not exceed MAX_PATH_DEPTH ({}), created {}",
                MAX_PATH_DEPTH,
                success_count
            );
        });
    }

    #[test]
    fn test_path_component_length() {
        with_test_pool(100, |pool| {
            // Test filename at exactly MAX_NAME_LEN - should succeed
            let max_name: String = (0..MAX_NAME_LEN).map(|_| 'a').collect();
            let max_path = format!("/{}", max_name);
            let result = pool.create(&max_path, 0o644);
            match result {
                Ok(fd) => {
                    pool.close(fd).expect("Close failed");
                    // Verify we can read it back
                    let fd = pool.open(&max_path, O_RDONLY).expect("Open failed");
                    pool.close(fd).expect("Close failed");
                }
                Err(e) => {
                    panic!("MAX_NAME_LEN should be valid, got error: {:?}", e);
                }
            }

            // Test filename exceeding MAX_NAME_LEN - should fail
            let too_long_name: String = (0..MAX_NAME_LEN + 1).map(|_| 'b').collect();
            let too_long_path = format!("/{}", too_long_name);
            let result = pool.create(&too_long_path, 0o644);
            assert!(
                matches!(result, Err(FsError::InvalidArgument { .. })),
                "Path component exceeding MAX_NAME_LEN should be rejected"
            );

            // Test path exceeding MAX_PATH_LEN - should fail
            let many_components = MAX_PATH_LEN / 10; // Each component is ~10 chars
            let very_long_path: String = (0..many_components)
                .map(|i| format!("/dir{:05}", i))
                .collect();
            if very_long_path.len() > MAX_PATH_LEN {
                let result = pool.mkdir(&very_long_path, 0o755);
                assert!(
                    matches!(result, Err(FsError::InvalidArgument { .. })),
                    "Path exceeding MAX_PATH_LEN should be rejected"
                );
            }
        });
    }

    // ═══════════════════════════════════════════════════════════════════
    // REFLINK (QUANTUM COPY) TESTS
    // ═══════════════════════════════════════════════════════════════════

    #[test]
    fn test_reflink_basic() {
        with_test_pool(100, |pool| {
            // Create source file with some data
            let fd = pool.create("/source.txt", 0o644).expect("Create failed");
            let data = b"Hello, Quantum World! This is a test of reflink.";
            pool.write(fd, data).expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create reflink (quantum copy)
            pool.reflink("/source.txt", "/dest.txt")
                .expect("Reflink failed");

            // Verify destination has same data
            let fd = pool.open("/dest.txt", O_RDONLY).expect("Open failed");
            let mut buf = vec![0u8; 100];
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");

            assert_eq!(&buf[..n], data);

            // Verify both files report same size
            let src_stat = pool.stat("/source.txt").expect("Stat source failed");
            let dst_stat = pool.stat("/dest.txt").expect("Stat dest failed");
            assert_eq!(src_stat.st_size, dst_stat.st_size);
        });
    }

    #[test]
    fn test_reflink_rejects_directories() {
        with_test_pool(100, |pool| {
            pool.mkdir("/mydir", 0o755).expect("Mkdir failed");

            let result = pool.reflink("/mydir", "/mydir_copy");
            assert!(
                matches!(result, Err(FsError::IsDirectory)),
                "Reflink should reject directories"
            );
        });
    }

    #[test]
    fn test_reflink_rejects_existing_dest() {
        with_test_pool(100, |pool| {
            // Create source and destination files
            let fd = pool.create("/source.txt", 0o644).expect("Create failed");
            pool.close(fd).expect("Close failed");

            let fd = pool.create("/dest.txt", 0o644).expect("Create failed");
            pool.close(fd).expect("Close failed");

            let result = pool.reflink("/source.txt", "/dest.txt");
            assert!(
                matches!(result, Err(FsError::AlreadyExists)),
                "Reflink should reject existing destination"
            );
        });
    }

    #[test]
    fn test_clone_tree() {
        with_test_pool(100, |pool| {
            // Create directory structure
            pool.mkdir("/project", 0o755).expect("Mkdir failed");
            pool.mkdir("/project/src", 0o755).expect("Mkdir failed");

            // Create some files
            let fd = pool
                .create("/project/README.md", 0o644)
                .expect("Create failed");
            pool.write(fd, b"# My Project").expect("Write failed");
            pool.close(fd).expect("Close failed");

            let fd = pool
                .create("/project/src/main.rs", 0o644)
                .expect("Create failed");
            pool.write(fd, b"fn main() {}").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Clone the tree
            let files_cloned = pool
                .clone_tree("/project", "/project_backup")
                .expect("Clone tree failed");

            assert_eq!(files_cloned, 2, "Should have cloned 2 files");

            // Verify cloned structure
            let fd = pool
                .open("/project_backup/README.md", O_RDONLY)
                .expect("Open failed");
            let mut buf = vec![0u8; 100];
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(&buf[..n], b"# My Project");

            let fd = pool
                .open("/project_backup/src/main.rs", O_RDONLY)
                .expect("Open failed");
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(&buf[..n], b"fn main() {}");
        });
    }

    #[test]
    fn test_rename_file() {
        with_test_pool(100, |pool| {
            let fd = pool.create("/old.txt", 0o644).expect("Create failed");
            pool.write(fd, b"test data").expect("Write failed");
            pool.close(fd).expect("Close failed");

            pool.rename("/old.txt", "/new.txt").expect("Rename failed");

            assert!(pool.stat("/old.txt").is_err());

            let fd = pool.open("/new.txt", O_RDONLY).expect("Open failed");
            let mut buf = [0u8; 20];
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(&buf[..n], b"test data");
        });
    }

    #[test]
    fn test_rename_move_to_directory() {
        with_test_pool(100, |pool| {
            pool.mkdir("/subdir", 0o755).expect("Mkdir failed");
            let fd = pool.create("/file.txt", 0o644).expect("Create failed");
            pool.write(fd, b"moved").expect("Write failed");
            pool.close(fd).expect("Close failed");

            pool.rename("/file.txt", "/subdir/file.txt")
                .expect("Rename failed");

            assert!(pool.stat("/file.txt").is_err());
            let fd = pool
                .open("/subdir/file.txt", O_RDONLY)
                .expect("Open failed");
            let mut buf = [0u8; 10];
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(&buf[..n], b"moved");
        });
    }

    #[test]
    fn test_hardlink() {
        with_test_pool(100, |pool| {
            // Create original file
            let fd = pool.create("/original.txt", 0o644).expect("Create failed");
            pool.write(fd, b"shared data").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create hard link
            pool.link("/original.txt", "/hardlink.txt")
                .expect("Link failed");

            // Both paths should have same data
            let fd = pool.open("/hardlink.txt", O_RDONLY).expect("Open failed");
            let mut buf = [0u8; 20];
            let n = pool.read(fd, &mut buf).expect("Read failed");
            pool.close(fd).expect("Close failed");
            assert_eq!(&buf[..n], b"shared data");

            // Both should have same inode
            let orig_stat = pool.stat("/original.txt").expect("Stat failed");
            let link_stat = pool.stat("/hardlink.txt").expect("Stat failed");
            assert_eq!(orig_stat.st_ino, link_stat.st_ino);
        });
    }

    #[test]
    fn test_hardlink_rejects_directories() {
        with_test_pool(100, |pool| {
            pool.mkdir("/mydir", 0o755).expect("Mkdir failed");

            let result = pool.link("/mydir", "/mydir_link");
            assert!(
                matches!(result, Err(FsError::IsDirectory)),
                "Hard link should reject directories"
            );
        });
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // SYMLINK TESTS
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_symlink_basic() {
        with_test_pool(100, |pool| {
            // Create a file
            let fd = pool
                .create("/target.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"symlink target").expect("Write failed");
            pool.close(fd).expect("Close failed");

            // Create symlink to the file
            pool.symlink("/target.txt", "/link.txt")
                .expect("Symlink creation failed");

            // Read the symlink target
            let target = pool.readlink("/link.txt").expect("Readlink failed");
            assert_eq!(target, "/target.txt");

            // Verify stat shows it's a symlink
            let stat = pool.stat("/link.txt").expect("Stat failed");
            assert_eq!(stat.st_mode & S_IFMT, S_IFLNK, "Should be a symlink");
        });
    }

    #[test]
    fn test_symlink_to_directory() {
        with_test_pool(100, |pool| {
            // Create a directory
            pool.mkdir("/mydir", 0o755).expect("Mkdir failed");

            // Create symlink to directory (should work, unlike hard links)
            pool.symlink("/mydir", "/dirlink")
                .expect("Symlink to directory should work");

            let target = pool.readlink("/dirlink").expect("Readlink failed");
            assert_eq!(target, "/mydir");
        });
    }

    #[test]
    fn test_symlink_dangling() {
        with_test_pool(100, |pool| {
            // Create dangling symlink (target doesn't exist)
            pool.symlink("/nonexistent", "/dangling_link")
                .expect("Dangling symlink creation should work");

            // Reading the link should work
            let target = pool.readlink("/dangling_link").expect("Readlink failed");
            assert_eq!(target, "/nonexistent");
        });
    }

    #[test]
    fn test_readlink_not_symlink() {
        with_test_pool(100, |pool| {
            // Create a regular file
            let fd = pool
                .create("/regular.txt", 0o644)
                .expect("File creation failed");
            pool.close(fd).expect("Close failed");

            // readlink on regular file should fail
            let result = pool.readlink("/regular.txt");
            assert!(
                matches!(result, Err(FsError::InvalidArgument { .. })),
                "readlink on regular file should fail"
            );
        });
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // FCNTL RECORD LOCK TESTS
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_fcntl_lock_basic() {
        with_test_pool(100, |pool| {
            let fd = pool
                .create("/locktest.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"test data for locking")
                .expect("Write failed");

            // Acquire exclusive lock on bytes 0-99
            pool.fcntl_lock(fd, true, 0, 100, 1000)
                .expect("fcntl_lock failed");

            // Test that conflicting lock would fail
            let conflict = pool
                .fcntl_test_lock(fd, true, 50, 100, 2000)
                .expect("fcntl_test_lock failed");
            assert!(conflict.is_some(), "Should detect conflicting lock");

            let (is_exclusive, start, _length, holder_pid) = conflict.unwrap();
            assert!(is_exclusive);
            assert_eq!(start, 0);
            assert_eq!(holder_pid, 1000);

            // Unlock
            pool.fcntl_unlock(fd, 0, 100, 1000)
                .expect("fcntl_unlock failed");

            // Now test should show no conflict
            let no_conflict = pool
                .fcntl_test_lock(fd, true, 50, 100, 2000)
                .expect("fcntl_test_lock failed");
            assert!(
                no_conflict.is_none(),
                "Should have no conflict after unlock"
            );

            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_fcntl_lock_shared() {
        with_test_pool(100, |pool| {
            let fd = pool
                .create("/shared_lock.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"shared lock test").expect("Write failed");

            // Acquire shared lock
            pool.fcntl_lock(fd, false, 0, 100, 1000)
                .expect("Shared lock 1 failed");

            // Another shared lock on same region should succeed
            pool.fcntl_lock(fd, false, 50, 100, 2000)
                .expect("Shared lock 2 should succeed");

            // But exclusive lock should fail
            let conflict = pool
                .fcntl_test_lock(fd, true, 75, 50, 3000)
                .expect("fcntl_test_lock failed");
            assert!(
                conflict.is_some(),
                "Exclusive lock should conflict with shared"
            );

            pool.close(fd).expect("Close failed");
        });
    }

    #[test]
    fn test_fcntl_lock_non_overlapping() {
        with_test_pool(100, |pool| {
            let fd = pool
                .create("/nonoverlap.txt", 0o644)
                .expect("File creation failed");
            pool.write(fd, b"non-overlapping regions test data here")
                .expect("Write failed");

            // Lock region 0-99
            pool.fcntl_lock(fd, true, 0, 100, 1000)
                .expect("Lock region 1 failed");

            // Lock region 200-299 (non-overlapping) should succeed
            pool.fcntl_lock(fd, true, 200, 100, 2000)
                .expect("Non-overlapping lock should succeed");

            // Test shows no conflict for region 100-199 (between the locks)
            let no_conflict = pool
                .fcntl_test_lock(fd, true, 100, 100, 3000)
                .expect("fcntl_test_lock failed");
            assert!(no_conflict.is_none(), "Gap region should have no conflict");

            pool.close(fd).expect("Close failed");
        });
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // QUOTA TESTS
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_quota_basic() {
        with_test_pool(100, |pool| {
            // Initially no quota
            assert!(pool.get_quota().is_none());
            assert!(!pool.is_over_quota());
            assert!(pool.get_remaining_quota().is_none());
            assert!(pool.get_quota_utilization().is_none());

            // Set a 1GB quota
            pool.set_quota(1024 * 1024 * 1024)
                .expect("Set quota failed");
            assert_eq!(pool.get_quota(), Some(1024 * 1024 * 1024));

            // Remove quota
            pool.set_quota(0).expect("Clear quota failed");
            assert!(pool.get_quota().is_none());
        });
    }

    #[test]
    fn test_quota_utilization() {
        with_test_pool(100, |pool| {
            // Set a large quota (larger than pool usage)
            let large_quota = 100 * 1024 * 1024 * 1024 * 1024; // 100 TB
            pool.set_quota(large_quota).expect("Set quota failed");

            // Check utilization is a valid percentage
            let util = pool
                .get_quota_utilization()
                .expect("Should have utilization");
            assert!(
                (0.0..=100.0).contains(&util),
                "Utilization should be 0-100%"
            );

            // Check remaining
            let remaining = pool.get_remaining_quota().expect("Should have remaining");
            assert!(remaining <= large_quota);
        });
    }

    #[test]
    fn test_quota_over() {
        with_test_pool(100, |pool| {
            // Set a very small quota (smaller than current usage)
            pool.set_quota(1).expect("Set quota failed"); // 1 byte quota

            // Should be over quota since pool has overhead
            let used = pool.get_used_space();
            if used > 1 {
                assert!(pool.is_over_quota(), "Should be over tiny quota");
            }
        });
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // SCRUB TESTS
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_scrub_stats() {
        with_test_pool(100, |pool| {
            // Get initial stats (no scrub run yet)
            let stats = pool.scrub_stats();
            assert_eq!(stats.blocks_scanned, 0);
            assert_eq!(stats.errors_detected, 0);
            assert_eq!(stats.repairs_made, 0);
        });
    }

    #[test]
    fn test_scrub_should_run() {
        with_test_pool(100, |pool| {
            // With no prior scrubs and low error rate, should not urgently need scrub
            let should_run = pool.scrub_should_run(0.0);
            // Initial state may or may not recommend scrub based on learned thresholds
            // Just verify the API works
            let _ = should_run;

            // With high error rate, more likely to recommend scrub
            let should_run_high = pool.scrub_should_run(1000.0);
            let _ = should_run_high;
        });
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // KEY DERIVATION TESTS
    // Uses Argon2id (std) or PBKDF2 (no_std) via derive_key()
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_derive_key_different_salts_different_keys() {
        let passphrase = "test_password_123";
        let salt1 = [1u8; 16];
        let salt2 = [2u8; 16];

        let key1 = LcpfsCrypto::derive_key(passphrase, &salt1);
        let key2 = LcpfsCrypto::derive_key(passphrase, &salt2);

        // Different salts MUST produce different keys
        assert_ne!(
            key1, key2,
            "KDF: Different salts must produce different keys!"
        );
    }

    #[test]
    fn test_derive_key_same_salt_same_key() {
        let passphrase = "deterministic_test";
        let salt = [0xAB_u8; 16];

        let key1 = LcpfsCrypto::derive_key(passphrase, &salt);
        let key2 = LcpfsCrypto::derive_key(passphrase, &salt);

        // Same passphrase + salt MUST produce same key (deterministic)
        assert_eq!(
            key1, key2,
            "KDF: Same passphrase and salt must produce identical keys!"
        );
    }

    #[test]
    fn test_derive_key_different_passphrases_different_keys() {
        let salt = [0x42u8; 16];

        let key1 = LcpfsCrypto::derive_key("password1", &salt);
        let key2 = LcpfsCrypto::derive_key("password2", &salt);

        // Different passphrases MUST produce different keys
        assert_ne!(
            key1, key2,
            "KDF: Different passphrases must produce different keys!"
        );
    }

    #[test]
    fn test_derive_key_output_length() {
        let key = LcpfsCrypto::derive_key("any passphrase", &[0u8; 16]);

        // Output must be exactly 32 bytes (256 bits)
        assert_eq!(key.len(), 32, "KDF: Output must be 32 bytes!");
    }

    #[test]
    fn test_derive_key_empty_passphrase() {
        let salt = [0x11u8; 16];

        // Empty passphrase should still work (not panic)
        let key = LcpfsCrypto::derive_key("", &salt);

        // Should produce non-zero output
        assert_ne!(
            key, [0u8; 32],
            "KDF: Empty passphrase should still derive a key!"
        );
    }

    #[test]
    fn test_derive_key_long_passphrase() {
        let long_passphrase = "a".repeat(10000);
        let salt = [0x22u8; 16];

        // Long passphrase should work
        let key = LcpfsCrypto::derive_key(&long_passphrase, &salt);

        assert_eq!(key.len(), 32);
        assert_ne!(key, [0u8; 32]);
    }

    #[test]
    fn test_derive_key_deterministic() {
        let passphrase = "password";
        let salt = b"salt_for_testing";

        let key1 = LcpfsCrypto::derive_key(passphrase, salt);
        let key2 = LcpfsCrypto::derive_key(passphrase, salt);

        // Must be deterministic
        assert_eq!(key1, key2);

        // Change one byte of salt -> completely different key
        let mut modified_salt = *salt;
        modified_salt[0] ^= 1;
        let key3 = LcpfsCrypto::derive_key(passphrase, &modified_salt);

        assert_ne!(key1, key3, "Single bit change in salt must change key!");
    }

    // ───────────────────────────────────────────────────────────────────────────
    // EXPLICIT PBKDF2 TESTS (cross-platform compatibility)
    // ───────────────────────────────────────────────────────────────────────────

    #[test]
    fn test_pbkdf2_explicit_different_salts() {
        let passphrase = "test_password";
        let salt1 = [1u8; 16];
        let salt2 = [2u8; 16];

        let key1 = LcpfsCrypto::derive_key_pbkdf2(passphrase, &salt1);
        let key2 = LcpfsCrypto::derive_key_pbkdf2(passphrase, &salt2);

        assert_ne!(
            key1, key2,
            "PBKDF2: Different salts must produce different keys!"
        );
    }

    #[test]
    fn test_pbkdf2_explicit_deterministic() {
        let passphrase = "deterministic_pbkdf2_test";
        let salt = [0xCD_u8; 16];

        let key1 = LcpfsCrypto::derive_key_pbkdf2(passphrase, &salt);
        let key2 = LcpfsCrypto::derive_key_pbkdf2(passphrase, &salt);

        assert_eq!(key1, key2, "PBKDF2: Must be deterministic!");
    }

    #[test]
    fn test_pbkdf2_explicit_output_length() {
        let key = LcpfsCrypto::derive_key_pbkdf2("passphrase", &[0u8; 16]);
        assert_eq!(key.len(), 32, "PBKDF2: Output must be 32 bytes!");
    }

    #[test]
    fn test_pbkdf2_explicit_empty_salt() {
        // Empty salt should work (though not recommended)
        let key = LcpfsCrypto::derive_key_pbkdf2("test_passphrase", &[]);
        assert_eq!(key.len(), 32);
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // CHACHA20-POLY1305 ENCRYPTION TESTS
    // ═══════════════════════════════════════════════════════════════════════════

    #[test]
    fn test_encrypt_block_nonce_uniqueness() {
        let key = [0x42u8; 32];
        let plaintext = b"test data for nonce uniqueness";
        let txg = 100;

        let (_, nonce1) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();
        let (_, nonce2) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();

        // CRITICAL: Nonces MUST be different even with identical inputs
        assert_ne!(
            nonce1, nonce2,
            "SECURITY: Nonce reuse detected! Each encryption must use unique nonce."
        );
    }

    #[test]
    fn test_encrypt_block_nonce_uniqueness_same_txg_different_data() {
        let key = [0x42u8; 32];
        let txg = 100;

        let (_, nonce1) = LcpfsCrypto::encrypt_block(&key, b"a", txg).unwrap();
        let (_, nonce2) = LcpfsCrypto::encrypt_block(&key, b"b", txg).unwrap();

        // Different data should still get unique nonces
        assert_ne!(nonce1, nonce2);
    }

    #[test]
    fn test_encrypt_decrypt_roundtrip() {
        let key = [0x42u8; 32];
        let plaintext = b"secret data to encrypt and decrypt";
        let txg = 42;

        let (ciphertext, nonce) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();
        let decrypted = LcpfsCrypto::decrypt_block(&key, &ciphertext, &nonce).unwrap();

        assert_eq!(
            decrypted.as_slice(),
            plaintext.as_slice(),
            "Decryption must recover original plaintext"
        );
    }

    #[test]
    fn test_encrypt_block_ciphertext_differs_from_plaintext() {
        let key = [0u8; 32];
        let plaintext = b"plaintext that should be encrypted";
        let txg = 1;

        let (ciphertext, _) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();

        // Ciphertext should be different from plaintext (and longer due to auth tag)
        assert_ne!(
            &ciphertext[..plaintext.len()],
            plaintext.as_slice(),
            "Encryption must change data!"
        );

        // Ciphertext should include 16-byte auth tag
        assert_eq!(ciphertext.len(), plaintext.len() + 16);
    }

    #[test]
    fn test_decrypt_with_wrong_key_fails() {
        let key1 = [0x42u8; 32];
        let key2 = [0x43u8; 32];
        let plaintext = b"secret";
        let txg = 1;

        let (ciphertext, nonce) = LcpfsCrypto::encrypt_block(&key1, plaintext, txg).unwrap();

        // Wrong key should fail authentication
        assert!(
            matches!(
                LcpfsCrypto::decrypt_block(&key2, &ciphertext, &nonce),
                Err(FsError::DecryptionFailed)
            ),
            "Decryption with wrong key must fail"
        );
    }

    #[test]
    fn test_decrypt_tampered_ciphertext_fails() {
        let key = [0u8; 32];
        let plaintext = b"secret";
        let txg = 1;

        let (mut ciphertext, nonce) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();

        // Tamper with ciphertext
        ciphertext[0] ^= 0xFF;

        // Tampered ciphertext should fail authentication
        assert!(
            matches!(
                LcpfsCrypto::decrypt_block(&key, &ciphertext, &nonce),
                Err(FsError::DecryptionFailed)
            ),
            "Tampered ciphertext must fail authentication"
        );
    }

    #[test]
    fn test_encrypt_empty_plaintext() {
        let key = [0x42u8; 32];
        let plaintext = b"";
        let txg = 1;

        let (ciphertext, nonce) = LcpfsCrypto::encrypt_block(&key, plaintext, txg).unwrap();

        // Empty plaintext should produce auth tag only (16 bytes)
        assert_eq!(ciphertext.len(), 16);

        let decrypted = LcpfsCrypto::decrypt_block(&key, &ciphertext, &nonce).unwrap();
        assert!(decrypted.is_empty());
    }

    #[test]
    fn test_encrypt_large_data() {
        let key = [0x42u8; 32];
        let plaintext = vec![0xAB_u8; 1024 * 1024]; // 1 MiB
        let txg = 1;

        let (ciphertext, nonce) = LcpfsCrypto::encrypt_block(&key, &plaintext, txg).unwrap();
        let decrypted = LcpfsCrypto::decrypt_block(&key, &ciphertext, &nonce).unwrap();

        assert_eq!(decrypted, plaintext);
    }
}