libdictenstein 0.1.0

//! Concurrency controls for Persistent ART.
//!
//! This module provides advanced concurrency primitives for the Persistent ART:
//!
//! - **Optimistic Lock Coupling**: Lock-free reads with version validation
//! - **Read-Write Locks**: Fine-grained locking for tree traversal
//! - **Epoch-Based Reclamation**: Safe memory reclamation for concurrent access
//!
//! # Architecture
//!
//! The concurrency model follows these principles:
//!
//! 1. **Readers don't block readers**: Multiple concurrent reads are always allowed
//! 2. **Writers acquire exclusive access**: Write operations use fine-grained locks
//! 3. **Optimistic reads**: Readers proceed without locks, validate versions after
//! 4. **Lock coupling**: Writers hold parent lock while acquiring child lock
//!
//! # Optimistic Lock Coupling
//!
//! ```text
//! ┌───────────────────────────────────────────────────────────────────┐
//! │                     Reader (Optimistic)                           │
//! │   ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐                      │
//! │   │Read v│ → │ Read │ → │Read v│ → │ Read │ → Validate versions  │
//! │   │ (A)  │   │ data │   │ (B)  │   │ data │                      │
//! │   └──────┘   └──────┘   └──────┘   └──────┘                      │
//! └───────────────────────────────────────────────────────────────────┘
//!
//! ┌───────────────────────────────────────────────────────────────────┐
//! │                     Writer (Lock Coupling)                         │
//! │   ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐                      │
//! │   │Lock A│ → │Lock B│ → │Unlock│ → │Modify│ → Unlock B           │
//! │   │      │   │      │   │  A   │   │  B   │                      │
//! │   └──────┘   └──────┘   └──────┘   └──────┘                      │
//! └───────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Usage
//!
//! ```text
//! use libdictenstein::persistent_artrie::concurrency::*;
//!
//! let node = OptimisticNode::new(data);
//!
//! // Optimistic read
//! loop {
//!     let guard = node.read_optimistic();
//!     let value = guard.read(|data| data.clone());
//!     if guard.validate() {
//!         // Value is consistent
//!         break;
//!     }
//!     // Retry - writer modified during read
//! }
//!
//! // Write with lock coupling
//! let parent_guard = parent.write();
//! let child_guard = child.write();
//! drop(parent_guard); // Release parent before modifying child
//! child_guard.modify(|data| { /* ... */ });
//! ```

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};

/// Version number for optimistic concurrency control.
///
/// The version is incremented on every write. Odd versions indicate
/// an in-progress write, even versions indicate a stable state.
#[derive(Debug)]
pub struct OptimisticVersion {
    version: AtomicU64,
}

impl OptimisticVersion {
    /// Create a new version counter.
    pub fn new() -> Self {
        OptimisticVersion {
            version: AtomicU64::new(0),
        }
    }

    /// Get the current version.
    pub fn get(&self) -> u64 {
        self.version.load(Ordering::Acquire)
    }

    /// Check if the version is stable (not being modified).
    pub fn is_stable(&self) -> bool {
        self.version.load(Ordering::Acquire) % 2 == 0
    }

    /// Begin a write operation (increment to odd).
    pub fn begin_write(&self) -> u64 {
        loop {
            let observed = self.version.load(Ordering::Acquire);
            if observed % 2 != 0 {
                std::hint::spin_loop();
                continue;
            }

            if self
                .version
                .compare_exchange(observed, observed + 1, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
            {
                return observed;
            }

            std::hint::spin_loop();
        }
    }

    /// End a write operation (increment to even).
    pub fn end_write(&self) {
        self.version.fetch_add(1, Ordering::Release);
    }

    /// Check if version changed since observed value.
    pub fn validate(&self, observed: u64) -> bool {
        self.version.load(Ordering::Acquire) == observed
    }
}

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

/// Optimistic read guard for lock-free reads.
///
/// This guard captures a version before reading and validates after.
/// If validation fails, the read must be retried.
pub struct OptimisticReadGuard<'a> {
    version: &'a OptimisticVersion,
    observed: u64,
}

impl<'a> OptimisticReadGuard<'a> {
    /// Create a new guard, capturing the current version.
    pub fn new(version: &'a OptimisticVersion) -> Self {
        // Wait for stable version
        let mut observed = version.get();
        while observed % 2 != 0 {
            std::hint::spin_loop();
            observed = version.get();
        }

        OptimisticReadGuard { version, observed }
    }

    /// Get the observed version.
    pub fn observed(&self) -> u64 {
        self.observed
    }

    /// Validate that no write occurred since the guard was created.
    pub fn validate(&self) -> bool {
        self.version.validate(self.observed)
    }
}

/// Write guard that manages version updates.
pub struct WriteGuard<'a> {
    version: &'a OptimisticVersion,
}

impl<'a> WriteGuard<'a> {
    /// Acquire a write guard.
    pub fn new(version: &'a OptimisticVersion) -> Self {
        version.begin_write();
        WriteGuard { version }
    }
}

impl<'a> Drop for WriteGuard<'a> {
    fn drop(&mut self) {
        self.version.end_write();
    }
}

/// Lock coupling coordinator for safe tree traversal.
///
/// During tree modifications, we use "lock coupling" (also called "crabbing"):
/// - Acquire lock on parent
/// - Acquire lock on child
/// - Release lock on parent (if child is "safe")
/// - A node is "safe" if it won't be modified by the current operation
pub struct LockCoupling {
    /// Maximum depth of held locks
    max_depth: usize,
    /// Current lock depth
    current_depth: AtomicUsize,
}

impl LockCoupling {
    /// Create a new lock coupling coordinator.
    pub fn new(max_depth: usize) -> Self {
        LockCoupling {
            max_depth,
            current_depth: AtomicUsize::new(0),
        }
    }

    /// Attempt to acquire a lock at the next level.
    ///
    /// Returns true if successful, false if max depth reached.
    pub fn try_descend(&self) -> bool {
        let current = self.current_depth.load(Ordering::Relaxed);
        if current >= self.max_depth {
            return false;
        }
        self.current_depth.fetch_add(1, Ordering::Relaxed);
        true
    }

    /// Release a level (when ascending or completing).
    pub fn ascend(&self) {
        let current = self.current_depth.load(Ordering::Relaxed);
        if current > 0 {
            self.current_depth.fetch_sub(1, Ordering::Relaxed);
        }
    }

    /// Get current depth.
    pub fn depth(&self) -> usize {
        self.current_depth.load(Ordering::Relaxed)
    }

    /// Reset to root.
    pub fn reset(&self) {
        self.current_depth.store(0, Ordering::Relaxed);
    }
}

impl Default for LockCoupling {
    fn default() -> Self {
        Self::new(64) // Default max depth
    }
}

/// Epoch-based reclamation for safe memory management.
///
/// This provides a way to safely deallocate memory that might be
/// accessed by concurrent readers. Memory is not freed until all
/// readers that might see it have finished.
#[derive(Debug)]
pub struct EpochManager {
    /// Global epoch counter
    global_epoch: AtomicU64,
    /// Number of active readers
    active_readers: AtomicUsize,
    /// Condvar notified when active_readers reaches zero.
    /// Used by `wait_for_quiescence` to avoid polling.
    quiescence_condvar: std::sync::Condvar,
    /// Mutex paired with `quiescence_condvar`.
    quiescence_mutex: std::sync::Mutex<()>,
}

impl EpochManager {
    /// Create a new epoch manager.
    pub fn new() -> Self {
        EpochManager {
            global_epoch: AtomicU64::new(0),
            active_readers: AtomicUsize::new(0),
            quiescence_condvar: std::sync::Condvar::new(),
            quiescence_mutex: std::sync::Mutex::new(()),
        }
    }

    /// Enter a read epoch.
    ///
    /// Returns the current epoch for validation.
    pub fn enter_read(&self) -> u64 {
        // SeqCst: the pin-publish (this increment) must be globally ordered BEFORE
        // any pointer the reader subsequently loads. This is the StoreLoad barrier
        // epoch-based reclamation requires — it pairs with the `fence(SeqCst)` a
        // reclaimer issues after unlinking a node and before reading
        // `active_reader_count`. Together they guarantee that if the reclaimer's
        // scan fails to observe this reader, the reader is guaranteed to observe
        // the reclaimer's unlink (and re-fault a fresh node) rather than deref a
        // freed one. `AcqRel`/`Acquire` alone permit StoreLoad reordering and are
        // NOT sufficient.
        self.active_readers.fetch_add(1, Ordering::SeqCst);
        self.global_epoch.load(Ordering::Acquire)
    }

    /// Exit a read epoch.
    ///
    /// When the last reader exits, notifies any threads waiting for quiescence.
    pub fn exit_read(&self) {
        // SeqCst: a reader's pointer accesses must be globally ordered BEFORE this
        // un-pin, so a reclaimer that observes the count drop to zero is guaranteed
        // the reader is done touching any node it held.
        let prev = self.active_readers.fetch_sub(1, Ordering::SeqCst);
        // If we were the last reader (prev was 1, now 0), notify waiters
        if prev == 1 {
            // Lock briefly to synchronize with condvar wait
            let _guard = self
                .quiescence_mutex
                .lock()
                .expect("quiescence_mutex poisoned");
            self.quiescence_condvar.notify_all();
        }
    }

    /// Advance the global epoch.
    ///
    /// This should be called periodically by a background thread
    /// to enable garbage collection.
    pub fn advance(&self) -> u64 {
        self.global_epoch.fetch_add(1, Ordering::AcqRel)
    }

    /// Get the current epoch.
    pub fn current_epoch(&self) -> u64 {
        self.global_epoch.load(Ordering::Acquire)
    }

    /// Check if there are any active readers.
    ///
    /// `SeqCst`: a reclaimer reads this AFTER a `fence(SeqCst)` that follows its
    /// unlink; the SeqCst load participates in the same total order as readers'
    /// SeqCst `enter_read`, closing the StoreLoad gap (see [`Self::enter_read`]).
    pub fn has_active_readers(&self) -> bool {
        self.active_readers.load(Ordering::SeqCst) > 0
    }

    /// Get the number of active readers.
    ///
    /// `SeqCst` for the same EBR StoreLoad reason as [`Self::has_active_readers`].
    pub fn active_reader_count(&self) -> usize {
        self.active_readers.load(Ordering::SeqCst)
    }

    /// Wait for epoch quiescence (no active readers).
    ///
    /// This method advances the epoch and waits until all readers from the
    /// old epoch have completed. It is used before performing operations
    /// that require exclusive access to the data (e.g., node eviction).
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum time to wait for quiescence
    /// * `poll_interval` - How often to check for quiescence
    ///
    /// # Returns
    ///
    /// * `Ok(old_epoch)` - Quiescence achieved, returns the old epoch
    /// * `Err(timeout_duration)` - Timed out waiting for readers to drain
    ///
    /// # Example
    ///
    /// ```text
    /// use std::time::Duration;
    /// use libdictenstein::persistent_artrie::concurrency::EpochManager;
    ///
    /// let manager = EpochManager::new();
    ///
    /// // Wait for quiescence with 100ms timeout
    /// match manager.wait_for_quiescence(
    ///     Duration::from_millis(100),
    ///     Duration::from_micros(100),
    /// ) {
    ///     Ok(old_epoch) => {
    ///         // Safe to modify data - no old-epoch readers
    ///         println!("Quiescence achieved at epoch {}", old_epoch);
    ///     }
    ///     Err(elapsed) => {
    ///         // Timed out - readers still active
    ///         println!("Timed out after {:?}", elapsed);
    ///     }
    /// }
    /// ```
    pub fn wait_for_quiescence(
        &self,
        timeout: std::time::Duration,
        _poll_interval: std::time::Duration,
    ) -> Result<u64, std::time::Duration> {
        use std::time::Instant;

        let start = Instant::now();

        // Advance epoch to create a boundary
        let old_epoch = self.advance();

        // Fast path: no readers active
        if !self.has_active_readers() {
            return Ok(old_epoch);
        }

        // Slow path: wait on condvar for readers to drain
        let mut guard = self
            .quiescence_mutex
            .lock()
            .expect("quiescence_mutex poisoned");
        loop {
            if !self.has_active_readers() {
                return Ok(old_epoch);
            }

            let remaining = timeout.saturating_sub(start.elapsed());
            if remaining.is_zero() {
                return Err(start.elapsed());
            }

            let (new_guard, wait_result) = self
                .quiescence_condvar
                .wait_timeout(guard, remaining)
                .expect("quiescence_mutex poisoned");
            guard = new_guard;

            if !self.has_active_readers() {
                return Ok(old_epoch);
            }

            if wait_result.timed_out() {
                return Err(start.elapsed());
            }
        }
    }

    /// Try to achieve quiescence without blocking.
    ///
    /// Advances the epoch and immediately checks if there are active readers.
    /// Returns `Some(old_epoch)` if quiescence is achieved, `None` otherwise.
    ///
    /// This is useful for non-blocking eviction attempts where the caller
    /// can retry later if quiescence is not immediately available.
    pub fn try_quiescence(&self) -> Option<u64> {
        let old_epoch = self.advance();
        if self.has_active_readers() {
            None
        } else {
            Some(old_epoch)
        }
    }
}

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

/// RAII guard for epoch-protected reads.
pub struct EpochGuard<'a> {
    manager: &'a EpochManager,
    #[allow(dead_code)]
    epoch: u64,
}

impl<'a> EpochGuard<'a> {
    /// Create a new epoch guard.
    pub fn new(manager: &'a EpochManager) -> Self {
        let epoch = manager.enter_read();
        EpochGuard { manager, epoch }
    }
}

impl<'a> Drop for EpochGuard<'a> {
    fn drop(&mut self) {
        self.manager.exit_read();
    }
}

/// MVCC-Lite read context for snapshot isolation.
///
/// Provides a lightweight form of Multi-Version Concurrency Control (MVCC)
/// using epoch-based snapshots. When created, it captures the current epoch
/// and tracks observed node versions. At validation time, it verifies that
/// all observed nodes still have the same versions.
///
/// # Isolation Guarantees
///
/// | Property         | Guarantee                                      |
/// |------------------|------------------------------------------------|
/// | Dirty Read       | Prevented (epoch snapshot)                     |
/// | Non-Repeatable   | Prevented (version validation)                 |
/// | Phantom Read     | Possible (new insertions not tracked)          |
///
/// # Example
///
/// ```text
/// use libdictenstein::persistent_artrie::concurrency::{EpochManager, MvccReadContext};
///
/// let epoch_manager = EpochManager::new();
///
/// // Create a read context for snapshot isolation
/// let mut ctx = MvccReadContext::new(&epoch_manager);
///
/// // Perform reads, observing node versions
/// ctx.observe_node(node_id, node_version);
///
/// // Validate that no observed nodes were modified
/// if ctx.validate(|id| get_node_version(id)) {
///     // Read was consistent
/// } else {
///     // A concurrent write occurred - retry
/// }
/// ```
pub struct MvccReadContext<'a> {
    /// Reference to the epoch manager
    epoch_manager: &'a EpochManager,
    /// Snapshot epoch captured at context creation
    snapshot_epoch: u64,
    /// Observed node versions: (node_id, version) pairs
    node_versions: Vec<(usize, u64)>,
}

impl<'a> MvccReadContext<'a> {
    /// Create a new MVCC read context.
    ///
    /// Captures the current epoch as a snapshot point and registers
    /// with the epoch manager as an active reader.
    pub fn new(epoch_manager: &'a EpochManager) -> Self {
        let snapshot_epoch = epoch_manager.enter_read();
        MvccReadContext {
            epoch_manager,
            snapshot_epoch,
            node_versions: Vec::with_capacity(16), // Pre-allocate for typical path length
        }
    }

    /// Record an observed node and its version.
    ///
    /// This should be called for each node accessed during the read operation.
    /// At validation time, we verify these versions haven't changed.
    #[inline]
    pub fn observe_node(&mut self, node_id: usize, version: u64) {
        self.node_versions.push((node_id, version));
    }

    /// Get the snapshot epoch.
    pub fn epoch(&self) -> u64 {
        self.snapshot_epoch
    }

    /// Get the number of observed nodes.
    pub fn observed_count(&self) -> usize {
        self.node_versions.len()
    }

    /// Validate that no observed nodes were modified.
    ///
    /// Takes a function that retrieves the current version for a node ID.
    /// Returns true if all observed versions match current versions.
    ///
    /// # Arguments
    ///
    /// * `get_version` - Function that returns the current version for a node ID
    pub fn validate<F>(&self, get_version: F) -> bool
    where
        F: Fn(usize) -> Option<u64>,
    {
        for &(node_id, expected_version) in &self.node_versions {
            match get_version(node_id) {
                Some(current_version) if current_version == expected_version => continue,
                _ => return false, // Version changed or node deleted
            }
        }
        true
    }

    /// Validate with a mutable closure (for caching lookups).
    pub fn validate_mut<F>(&self, mut get_version: F) -> bool
    where
        F: FnMut(usize) -> Option<u64>,
    {
        for &(node_id, expected_version) in &self.node_versions {
            match get_version(node_id) {
                Some(current_version) if current_version == expected_version => continue,
                _ => return false,
            }
        }
        true
    }

    /// Clear observed nodes for reuse.
    pub fn reset(&mut self) {
        self.node_versions.clear();
        // Note: We keep the same epoch - this is for retrying within the same read
    }
}

impl<'a> Drop for MvccReadContext<'a> {
    fn drop(&mut self) {
        self.epoch_manager.exit_read();
    }
}

/// Retry statistics for optimistic operations.
#[derive(Debug, Default)]
pub struct RetryStats {
    /// Number of successful reads
    pub successful: AtomicU64,
    /// Number of retries due to concurrent modifications
    pub retries: AtomicU64,
    /// Maximum retries in a single operation
    pub max_retries: AtomicU64,
}

impl RetryStats {
    /// Create new retry stats.
    pub fn new() -> Self {
        Self::default()
    }

    /// Record a successful operation.
    pub fn record_success(&self, retry_count: u64) {
        self.successful.fetch_add(1, Ordering::Relaxed);
        if retry_count > 0 {
            self.retries.fetch_add(retry_count, Ordering::Relaxed);
            loop {
                let current = self.max_retries.load(Ordering::Relaxed);
                if retry_count <= current {
                    break;
                }
                if self
                    .max_retries
                    .compare_exchange_weak(
                        current,
                        retry_count,
                        Ordering::Relaxed,
                        Ordering::Relaxed,
                    )
                    .is_ok()
                {
                    break;
                }
            }
        }
    }

    /// Get the retry rate.
    pub fn retry_rate(&self) -> f64 {
        let successful = self.successful.load(Ordering::Relaxed);
        let retries = self.retries.load(Ordering::Relaxed);
        if successful + retries == 0 {
            0.0
        } else {
            retries as f64 / (successful + retries) as f64
        }
    }

    /// Get a snapshot of the stats.
    pub fn snapshot(&self) -> RetryStatsSnapshot {
        RetryStatsSnapshot {
            successful: self.successful.load(Ordering::Relaxed),
            retries: self.retries.load(Ordering::Relaxed),
            max_retries: self.max_retries.load(Ordering::Relaxed),
        }
    }
}

/// Immutable snapshot of retry stats.
#[derive(Debug, Clone, Copy)]
pub struct RetryStatsSnapshot {
    /// Successful operations
    pub successful: u64,
    /// Total retries
    pub retries: u64,
    /// Maximum retries in single operation
    pub max_retries: u64,
}

/// Atomic statistics for trie operations.
///
/// Provides thread-safe counters for monitoring trie performance and behavior.
/// All counters use relaxed ordering for minimal overhead, as exact counts
/// aren't required - approximate trends are sufficient for monitoring.
#[derive(Debug, Default)]
pub struct TrieStats {
    /// Number of read operations (get, contains, prefix iterations)
    pub reads: AtomicU64,
    /// Number of write operations (insert, remove)
    pub writes: AtomicU64,
    /// Number of read retries due to concurrent modifications
    pub read_retries: AtomicU64,
    /// Number of buffer cache hits
    pub cache_hits: AtomicU64,
    /// Number of buffer cache misses (disk reads)
    pub cache_misses: AtomicU64,
    /// Number of WAL writes
    pub wal_writes: AtomicU64,
    /// Number of WAL syncs (fsync calls)
    pub wal_syncs: AtomicU64,
}

impl TrieStats {
    /// Create new statistics counters (all zeroed).
    pub fn new() -> Self {
        Self::default()
    }

    /// Record a read operation.
    #[inline]
    pub fn record_read(&self) {
        self.reads.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a write operation.
    #[inline]
    pub fn record_write(&self) {
        self.writes.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a read retry.
    #[inline]
    pub fn record_read_retry(&self) {
        self.read_retries.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a cache hit.
    #[inline]
    pub fn record_cache_hit(&self) {
        self.cache_hits.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a cache miss.
    #[inline]
    pub fn record_cache_miss(&self) {
        self.cache_misses.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a WAL write.
    #[inline]
    pub fn record_wal_write(&self) {
        self.wal_writes.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a WAL sync.
    #[inline]
    pub fn record_wal_sync(&self) {
        self.wal_syncs.fetch_add(1, Ordering::Relaxed);
    }

    /// Get the cache hit ratio (0.0 to 1.0).
    pub fn cache_hit_ratio(&self) -> f64 {
        let hits = self.cache_hits.load(Ordering::Relaxed);
        let misses = self.cache_misses.load(Ordering::Relaxed);
        let total = hits + misses;
        if total == 0 {
            0.0
        } else {
            hits as f64 / total as f64
        }
    }

    /// Get the read retry ratio (0.0 to 1.0).
    pub fn read_retry_ratio(&self) -> f64 {
        let reads = self.reads.load(Ordering::Relaxed);
        let retries = self.read_retries.load(Ordering::Relaxed);
        if reads == 0 {
            0.0
        } else {
            retries as f64 / reads as f64
        }
    }

    /// Get a snapshot of the stats.
    pub fn snapshot(&self) -> TrieStatsSnapshot {
        TrieStatsSnapshot {
            reads: self.reads.load(Ordering::Relaxed),
            writes: self.writes.load(Ordering::Relaxed),
            read_retries: self.read_retries.load(Ordering::Relaxed),
            cache_hits: self.cache_hits.load(Ordering::Relaxed),
            cache_misses: self.cache_misses.load(Ordering::Relaxed),
            wal_writes: self.wal_writes.load(Ordering::Relaxed),
            wal_syncs: self.wal_syncs.load(Ordering::Relaxed),
        }
    }

    /// Reset all counters to zero.
    pub fn reset(&self) {
        self.reads.store(0, Ordering::Relaxed);
        self.writes.store(0, Ordering::Relaxed);
        self.read_retries.store(0, Ordering::Relaxed);
        self.cache_hits.store(0, Ordering::Relaxed);
        self.cache_misses.store(0, Ordering::Relaxed);
        self.wal_writes.store(0, Ordering::Relaxed);
        self.wal_syncs.store(0, Ordering::Relaxed);
    }
}

/// Immutable snapshot of trie stats.
#[derive(Debug, Clone, Copy)]
pub struct TrieStatsSnapshot {
    /// Number of read operations
    pub reads: u64,
    /// Number of write operations
    pub writes: u64,
    /// Number of read retries
    pub read_retries: u64,
    /// Number of cache hits
    pub cache_hits: u64,
    /// Number of cache misses
    pub cache_misses: u64,
    /// Number of WAL writes
    pub wal_writes: u64,
    /// Number of WAL syncs
    pub wal_syncs: u64,
}

impl TrieStatsSnapshot {
    /// Get the cache hit ratio (0.0 to 1.0).
    pub fn cache_hit_ratio(&self) -> f64 {
        let total = self.cache_hits + self.cache_misses;
        if total == 0 {
            0.0
        } else {
            self.cache_hits as f64 / total as f64
        }
    }

    /// Get the read retry ratio (0.0 to 1.0).
    pub fn read_retry_ratio(&self) -> f64 {
        if self.reads == 0 {
            0.0
        } else {
            self.read_retries as f64 / self.reads as f64
        }
    }
}

/// Thread-safe optimistic data wrapper.
///
/// Provides optimistic reads and exclusive writes for any data type.
pub struct OptimisticCell<T> {
    /// Version for optimistic concurrency
    version: OptimisticVersion,
    /// The data guarded against Rust-level read/write races.
    data: RwLock<T>,
}

// SAFETY: RwLock serializes mutable access and allows shared read access only
// when T is safe to share.
unsafe impl<T: Send + Sync> Sync for OptimisticCell<T> {}
unsafe impl<T: Send> Send for OptimisticCell<T> {}

impl<T> OptimisticCell<T> {
    /// Create a new optimistic cell.
    pub fn new(data: T) -> Self {
        OptimisticCell {
            version: OptimisticVersion::new(),
            data: RwLock::new(data),
        }
    }

    /// Attempt an optimistic read.
    ///
    /// Returns the result and whether it was successful.
    /// If not successful, the caller should retry.
    pub fn try_read<R, F>(&self, f: F) -> Option<R>
    where
        F: FnOnce(&T) -> R,
    {
        let guard = OptimisticReadGuard::new(&self.version);

        let data = self.data.read().expect("OptimisticCell read lock poisoned");
        let result = f(&data);
        drop(data);

        if guard.validate() {
            Some(result)
        } else {
            None
        }
    }

    /// Perform an optimistic read with retries.
    ///
    /// Retries until successful or max_retries reached.
    pub fn read_with_retry<R, F>(&self, f: F, max_retries: usize) -> Option<R>
    where
        F: Fn(&T) -> R,
    {
        for _ in 0..max_retries {
            if let Some(result) = self.try_read(&f) {
                return Some(result);
            }
            std::hint::spin_loop();
        }
        None
    }

    /// Perform an exclusive write.
    pub fn write<R, F>(&self, f: F) -> R
    where
        F: FnOnce(&mut T) -> R,
    {
        let _guard = WriteGuard::new(&self.version);

        let mut data = self
            .data
            .write()
            .expect("OptimisticCell write lock poisoned");
        f(&mut data)
    }

    /// Get the current version.
    pub fn version(&self) -> u64 {
        self.version.get()
    }

    /// Check if currently being written.
    pub fn is_locked(&self) -> bool {
        !self.version.is_stable()
    }
}

/// Shared ownership optimistic cell.
pub type SharedOptimisticCell<T> = Arc<OptimisticCell<T>>;

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;

    #[test]
    fn test_optimistic_version() {
        let version = OptimisticVersion::new();

        assert_eq!(version.get(), 0);
        assert!(version.is_stable());

        // Start write
        version.begin_write();
        assert!(!version.is_stable());
        assert_eq!(version.get(), 1);

        // End write
        version.end_write();
        assert!(version.is_stable());
        assert_eq!(version.get(), 2);
    }

    #[test]
    fn test_optimistic_read_guard() {
        let version = OptimisticVersion::new();

        let guard = OptimisticReadGuard::new(&version);
        assert_eq!(guard.observed(), 0);
        assert!(guard.validate());

        // Concurrent write invalidates guard
        version.begin_write();
        assert!(!guard.validate());

        version.end_write();
        assert!(!guard.validate()); // Still invalid - version changed
    }

    #[test]
    fn test_lock_coupling() {
        let lc = LockCoupling::new(3);

        assert_eq!(lc.depth(), 0);

        assert!(lc.try_descend());
        assert_eq!(lc.depth(), 1);

        assert!(lc.try_descend());
        assert_eq!(lc.depth(), 2);

        assert!(lc.try_descend());
        assert_eq!(lc.depth(), 3);

        // Max depth reached
        assert!(!lc.try_descend());
        assert_eq!(lc.depth(), 3);

        lc.ascend();
        assert_eq!(lc.depth(), 2);

        lc.reset();
        assert_eq!(lc.depth(), 0);
    }

    #[test]
    fn test_epoch_manager() {
        let epoch = EpochManager::new();

        assert_eq!(epoch.current_epoch(), 0);
        assert!(!epoch.has_active_readers());

        // Enter read
        let e1 = epoch.enter_read();
        assert_eq!(e1, 0);
        assert!(epoch.has_active_readers());
        assert_eq!(epoch.active_reader_count(), 1);

        // Advance epoch
        let e2 = epoch.advance();
        assert_eq!(e2, 0);
        assert_eq!(epoch.current_epoch(), 1);

        // Exit read
        epoch.exit_read();
        assert!(!epoch.has_active_readers());
    }

    #[test]
    fn test_epoch_guard() {
        let epoch = EpochManager::new();

        {
            let _guard = EpochGuard::new(&epoch);
            assert!(epoch.has_active_readers());
        }

        assert!(!epoch.has_active_readers());
    }

    #[test]
    fn test_retry_stats() {
        let stats = RetryStats::new();

        stats.record_success(0);
        stats.record_success(2);
        stats.record_success(5);

        let snapshot = stats.snapshot();
        assert_eq!(snapshot.successful, 3);
        assert_eq!(snapshot.retries, 7);
        assert_eq!(snapshot.max_retries, 5);
    }

    #[test]
    fn test_optimistic_cell_read() {
        let cell = OptimisticCell::new(42);

        let result = cell.try_read(|v| *v).expect("read should succeed");
        assert_eq!(result, 42);
    }

    #[test]
    fn test_optimistic_cell_write() {
        let cell = OptimisticCell::new(42);

        cell.write(|v| *v = 100);

        let result = cell.try_read(|v| *v).expect("read should succeed");
        assert_eq!(result, 100);
    }

    #[test]
    fn test_optimistic_cell_concurrent() {
        let cell = Arc::new(OptimisticCell::new(0));
        let cell_clone = cell.clone();

        // Writer thread
        let writer = thread::spawn(move || {
            for i in 1..=100 {
                cell_clone.write(|v| *v = i);
            }
        });

        // Reader thread
        let reader = thread::spawn(move || {
            let mut max_seen = 0;
            for _ in 0..1000 {
                if let Some(v) = cell.read_with_retry(|v| *v, 10) {
                    if v > max_seen {
                        max_seen = v;
                    }
                }
            }
            max_seen
        });

        writer.join().expect("writer panicked");
        let max_seen = reader.join().expect("reader panicked");

        // Should have seen some writes (exact value depends on timing)
        assert!(max_seen >= 0);
    }

    #[test]
    fn test_optimistic_cell_is_locked() {
        let cell = OptimisticCell::new(42);

        assert!(!cell.is_locked());

        // During write, it should be locked
        // (Hard to test reliably due to timing)
        cell.write(|_| {
            // Inside write, the cell is locked
            // But we can't easily check from inside
        });

        assert!(!cell.is_locked());
    }

    // =========================================================================
    // Edge case tests for branch coverage
    // =========================================================================

    /// Test OptimisticReadGuard spin loop when version is odd (write in progress).
    /// This covers lines 130-132 in the spin loop.
    #[test]
    fn test_optimistic_read_guard_waits_for_stable_version() {
        use std::time::Duration;

        let version = Arc::new(OptimisticVersion::new());
        let version_clone = Arc::clone(&version);

        // Start a write (makes version odd)
        version.begin_write();
        assert!(
            !version.is_stable(),
            "Version should be unstable (odd) during write"
        );

        // Spawn a reader thread that will spin
        let reader_handle = thread::spawn(move || {
            // This should spin until version becomes even
            let guard = OptimisticReadGuard::new(&version_clone);
            guard.observed()
        });

        // Give the reader thread time to start spinning
        thread::sleep(Duration::from_millis(10));

        // End the write (makes version even)
        version.end_write();

        // Reader should now complete
        let observed = reader_handle.join().expect("reader thread should complete");
        assert_eq!(observed % 2, 0, "Observed version should be even (stable)");
        assert_eq!(
            observed, 2,
            "Version should be 2 after begin_write + end_write"
        );
    }

    /// Test OptimisticReadGuard with already-stable version (no spin).
    #[test]
    fn test_optimistic_read_guard_immediate_stable() {
        let version = OptimisticVersion::new();
        assert!(version.is_stable());

        // Should not need to spin
        let guard = OptimisticReadGuard::new(&version);
        assert_eq!(guard.observed(), 0);
        assert!(guard.validate());
    }

    /// Test LockCoupling::ascend when depth is already 0.
    #[test]
    fn test_lock_coupling_ascend_at_zero() {
        let lc = LockCoupling::new(5);
        assert_eq!(lc.depth(), 0);

        // Ascend when already at 0 should be a no-op
        lc.ascend();
        assert_eq!(lc.depth(), 0, "Depth should remain 0 after ascend from 0");

        // Ascend again should still be safe
        lc.ascend();
        assert_eq!(lc.depth(), 0);
    }

    /// Test EpochManager wait_for_quiescence with immediate success.
    #[test]
    fn test_epoch_wait_for_quiescence_immediate() {
        use std::time::Duration;

        let epoch = EpochManager::new();
        assert!(!epoch.has_active_readers());

        // No readers, should succeed immediately
        let result =
            epoch.wait_for_quiescence(Duration::from_millis(100), Duration::from_micros(100));

        assert!(result.is_ok(), "Should achieve quiescence immediately");
        let old_epoch = result.unwrap();
        assert_eq!(old_epoch, 0, "Old epoch should be 0");
        assert_eq!(
            epoch.current_epoch(),
            1,
            "Current epoch should be 1 after advance"
        );
    }

    /// Test EpochManager wait_for_quiescence with active readers (timeout).
    #[test]
    fn test_epoch_wait_for_quiescence_timeout() {
        use std::sync::atomic::AtomicBool;
        use std::time::Duration;

        let epoch = Arc::new(EpochManager::new());
        let epoch_clone = Arc::clone(&epoch);

        // Barrier: signals that the reader has entered the epoch
        let reader_entered = Arc::new(AtomicBool::new(false));
        let reader_entered_clone = Arc::clone(&reader_entered);

        // Start a reader that will hold for a while
        let reader_handle = thread::spawn(move || {
            let _e = epoch_clone.enter_read();
            // Signal that we've entered the epoch
            reader_entered_clone.store(true, std::sync::atomic::Ordering::Release);
            // Hold for longer than the timeout
            thread::sleep(Duration::from_millis(500));
            epoch_clone.exit_read();
        });

        // Spin until reader is definitely inside enter_read()
        while !reader_entered.load(std::sync::atomic::Ordering::Acquire) {
            thread::yield_now();
        }

        // Try to wait for quiescence with short timeout
        let result =
            epoch.wait_for_quiescence(Duration::from_millis(50), Duration::from_micros(100));

        assert!(result.is_err(), "Should timeout with active reader");
        let elapsed = result.unwrap_err();
        assert!(
            elapsed >= Duration::from_millis(50),
            "Should have waited at least 50ms"
        );

        // Let the reader finish
        reader_handle.join().expect("reader should complete");
    }

    /// Test EpochManager try_quiescence with no readers.
    #[test]
    fn test_epoch_try_quiescence_success() {
        let epoch = EpochManager::new();

        let result = epoch.try_quiescence();
        assert!(result.is_some(), "Should succeed with no readers");
        assert_eq!(result.unwrap(), 0, "Old epoch should be 0");
    }

    /// Test EpochManager try_quiescence with active readers.
    #[test]
    fn test_epoch_try_quiescence_with_readers() {
        let epoch = EpochManager::new();

        // Enter a read
        let _e = epoch.enter_read();
        assert!(epoch.has_active_readers());

        let result = epoch.try_quiescence();
        assert!(result.is_none(), "Should fail with active reader");

        // Exit the read
        epoch.exit_read();

        // Now should succeed
        let result = epoch.try_quiescence();
        assert!(result.is_some(), "Should succeed after reader exits");
    }

    /// Test RetryStats with no retries.
    #[test]
    fn test_retry_stats_no_retries() {
        let stats = RetryStats::new();

        stats.record_success(0);
        stats.record_success(0);
        stats.record_success(0);

        let snapshot = stats.snapshot();
        assert_eq!(snapshot.successful, 3);
        assert_eq!(snapshot.retries, 0);
        assert_eq!(snapshot.max_retries, 0);
        assert!((stats.retry_rate() - 0.0).abs() < 0.001);
    }

    /// Test RetryStats retry rate calculation edge case.
    #[test]
    fn test_retry_stats_empty() {
        let stats = RetryStats::new();

        // No operations recorded yet
        assert!(
            (stats.retry_rate() - 0.0).abs() < 0.001,
            "Retry rate should be 0 with no ops"
        );
    }

    /// Test TrieStats cache_hit_ratio with zero accesses.
    #[test]
    fn test_trie_stats_cache_hit_ratio_zero() {
        let stats = TrieStats::new();

        // No cache accesses
        assert!((stats.cache_hit_ratio() - 0.0).abs() < 0.001);

        let snapshot = stats.snapshot();
        assert!((snapshot.cache_hit_ratio() - 0.0).abs() < 0.001);
    }

    /// Test TrieStats read_retry_ratio with zero reads.
    #[test]
    fn test_trie_stats_read_retry_ratio_zero() {
        let stats = TrieStats::new();

        // No reads
        assert!((stats.read_retry_ratio() - 0.0).abs() < 0.001);

        let snapshot = stats.snapshot();
        assert!((snapshot.read_retry_ratio() - 0.0).abs() < 0.001);
    }

    /// Test TrieStats reset functionality.
    #[test]
    fn test_trie_stats_reset() {
        let stats = TrieStats::new();

        stats.record_read();
        stats.record_write();
        stats.record_cache_hit();
        stats.record_cache_miss();
        stats.record_read_retry();
        stats.record_wal_write();
        stats.record_wal_sync();

        let snapshot = stats.snapshot();
        assert_eq!(snapshot.reads, 1);
        assert_eq!(snapshot.writes, 1);
        assert_eq!(snapshot.cache_hits, 1);

        stats.reset();

        let snapshot = stats.snapshot();
        assert_eq!(snapshot.reads, 0);
        assert_eq!(snapshot.writes, 0);
        assert_eq!(snapshot.cache_hits, 0);
        assert_eq!(snapshot.cache_misses, 0);
        assert_eq!(snapshot.read_retries, 0);
        assert_eq!(snapshot.wal_writes, 0);
        assert_eq!(snapshot.wal_syncs, 0);
    }

    /// Test MvccReadContext validation with changed version.
    #[test]
    fn test_mvcc_read_context_validation_changed() {
        let epoch = EpochManager::new();
        let mut ctx = MvccReadContext::new(&epoch);

        // Observe some nodes
        ctx.observe_node(1, 10);
        ctx.observe_node(2, 20);
        ctx.observe_node(3, 30);

        assert_eq!(ctx.observed_count(), 3);

        // Validate with matching versions
        let valid = ctx.validate(|id| match id {
            1 => Some(10),
            2 => Some(20),
            3 => Some(30),
            _ => None,
        });
        assert!(valid, "Should be valid when versions match");

        // Validate with one changed version
        let invalid = ctx.validate(|id| match id {
            1 => Some(10),
            2 => Some(21), // Changed!
            3 => Some(30),
            _ => None,
        });
        assert!(!invalid, "Should be invalid when a version changed");

        // Validate with missing node
        let missing = ctx.validate(|id| match id {
            1 => Some(10),
            2 => None, // Deleted!
            3 => Some(30),
            _ => None,
        });
        assert!(!missing, "Should be invalid when a node is missing");
    }

    /// Test MvccReadContext reset.
    #[test]
    fn test_mvcc_read_context_reset() {
        let epoch = EpochManager::new();
        let mut ctx = MvccReadContext::new(&epoch);

        ctx.observe_node(1, 10);
        ctx.observe_node(2, 20);
        assert_eq!(ctx.observed_count(), 2);

        ctx.reset();
        assert_eq!(ctx.observed_count(), 0);

        // Epoch should be preserved
        assert_eq!(ctx.epoch(), 0);
    }

    /// Test MvccReadContext validate_mut.
    #[test]
    fn test_mvcc_read_context_validate_mut() {
        let epoch = EpochManager::new();
        let mut ctx = MvccReadContext::new(&epoch);

        ctx.observe_node(1, 10);
        ctx.observe_node(2, 20);

        // Use mutable closure that tracks calls
        let mut call_count = 0;
        let valid = ctx.validate_mut(|id| {
            call_count += 1;
            match id {
                1 => Some(10),
                2 => Some(20),
                _ => None,
            }
        });

        assert!(valid);
        assert_eq!(call_count, 2, "Should have called get_version twice");
    }

    /// Test OptimisticCell read_with_retry reaches max retries.
    #[test]
    fn test_optimistic_cell_read_with_retry_fails() {
        let cell = Arc::new(OptimisticCell::new(42));
        let cell_clone = Arc::clone(&cell);

        // Continuously write to cause read failures
        let writer_running = Arc::new(std::sync::atomic::AtomicBool::new(true));
        let writer_flag = Arc::clone(&writer_running);

        let writer = thread::spawn(move || {
            while writer_flag.load(std::sync::atomic::Ordering::Relaxed) {
                cell_clone.write(|v| *v += 1);
                std::hint::spin_loop();
            }
        });

        // Try to read with very few retries - might fail due to constant writes
        // This is probabilistic, but with enough writes we should see some failures
        let mut failures = 0;
        for _ in 0..100 {
            if cell.read_with_retry(|v| *v, 1).is_none() {
                failures += 1;
            }
        }

        writer_running.store(false, std::sync::atomic::Ordering::Relaxed);
        writer.join().expect("writer should complete");

        // We expect at least some failures when max_retries is only 1
        // This test verifies the retry logic works, not a specific failure count
        // If all 100 reads succeeded with just 1 retry each while writes are happening,
        // that's also valid behavior
    }

    /// Test WriteGuard drop behavior.
    #[test]
    fn test_write_guard_drop_increments_version() {
        let version = OptimisticVersion::new();
        assert_eq!(version.get(), 0);

        {
            let _guard = WriteGuard::new(&version);
            assert_eq!(version.get(), 1); // Odd during write
            assert!(!version.is_stable());
        }

        // After guard drops, version should be even
        assert_eq!(version.get(), 2);
        assert!(version.is_stable());
    }

    /// Test OptimisticVersion validation fails when version changes.
    #[test]
    fn test_optimistic_version_validate_after_change() {
        let version = OptimisticVersion::new();

        let observed = version.get();
        assert!(version.validate(observed));

        // Change the version
        version.begin_write();
        version.end_write();

        // Old observation should no longer be valid
        assert!(!version.validate(observed));
        assert!(version.validate(version.get()));
    }
}