clock_ring.rs

//! Clock-sweep ring for second-chance eviction.
//!
//! Uses a fixed-size slot array and a hand pointer to evict the first
//! unreferenced entry encountered. Accesses set a referenced bit that
//! grants a second chance before eviction.
//!
//! ## Architecture
//!
//! ```text
//! ┌───────────────────────────────────────────────────────────────────────────┐
//! │                           ClockRing<K, V>                                 │
//! │                                                                           │
//! │   ┌─────────────────────────────┐   ┌─────────────────────────────────┐   │
//! │   │  index: HashMap<K, usize>   │   │  slots: Vec<Option<Entry>>      │   │
//! │   │                             │   │                                 │   │
//! │   │  ┌───────────┬──────────┐   │   │   ┌─────┬─────┬─────┬─────┐     │   │
//! │   │  │    Key    │  Index   │   │   │   │  0  │  1  │  2  │  3  │     │   │
//! │   │  ├───────────┼──────────┤   │   │   ├─────┼─────┼─────┼─────┤     │   │
//! │   │  │  "key_a"  │    0     │ ──┼───┼──►│ A,1 │ B,0 │ C,1 │None │     │   │
//! │   │  │  "key_b"  │    1     │ ──┼───┼──►│     │     │     │     │     │   │
//! │   │  │  "key_c"  │    2     │ ──┼───┼──►│     │     │     │     │     │   │
//! │   │  └───────────┴──────────┘   │   │   └─────┴─────┴─────┴─────┘     │   │
//! │   └─────────────────────────────┘   │                ▲                │   │
//! │                                     │                │ hand = 1       │   │
//! │                                     └────────────────┼────────────────┘   │
//! │                                                      │                    │
//! │   Entry: { key, value }  +  referenced: Vec<bool>    │                    │
//! │   "A,1" = slot 0: Entry { key: "key_a" }, ref[0]=1   │                    │
//! │   "B,0" = slot 1: Entry { key: "key_b" }, ref[1]=0  ◄┘ (next victim)      │
//! │                                                                           │
//! └───────────────────────────────────────────────────────────────────────────┘
//!
//! Clock Sweep Algorithm
//! ─────────────────────
//!
//!   Insert "key_d" when full (hand at slot 1):
//!
//!     Step 1: Check slot[1] ("B", ref=0)
//!             ref=0 → EVICT, insert here
//!
//!     Result: slot[1] = Entry { key: "key_d", ref=false }
//!             hand advances to 2
//!             Return: Some(("key_b", value_b))
//!
//!   Insert "key_e" when full (hand at slot 2):
//!
//!     Step 1: Check slot[2] ("C", ref=1)
//!             ref=1 → clear ref, advance hand
//!     Step 2: Check slot[3] (None)
//!             Empty → insert here
//!
//!     Result: slot[3] = Entry { key: "key_e", ref=false }
//!             hand advances to 0
//!             Return: None (used empty slot)
//!
//! Second-Chance Behavior
//! ──────────────────────
//!
//!   ┌────────────────────────────────────────────────────────────────────────┐
//!   │  Access via get()/touch() → Sets referenced = true                     │
//!   │                                                                        │
//!   │  Eviction scan:                                                        │
//!   │    ref=1 → Clear to ref=0, skip (second chance granted)                │
//!   │    ref=0 → Evict this entry                                            │
//!   │                                                                        │
//!   │  Effect: Recently accessed entries survive longer                      │
//!   └────────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! ## Key Components
//!
//! - [`ClockRing`]: Single-threaded CLOCK cache
//! - [`ConcurrentClockRing`]: Thread-safe wrapper with `RwLock`
//! - [`Iter`], [`IterMut`], [`IntoIter`]: Iterators over entries
//! - [`Keys`], [`Values`], [`ValuesMut`]: Iterators over keys or values
//!
//! ## Operations
//!
//! | Operation       | Time        | Notes                                  |
//! |-----------------|-------------|----------------------------------------|
//! | [`insert`]      | O(1) amort. | Bounded scan with reference clearing   |
//! | [`get`]         | O(1)        | Returns value, sets reference bit      |
//! | [`peek`]        | O(1)        | Returns value, does NOT set ref bit    |
//! | [`touch`]       | O(1)        | Sets reference bit only                |
//! | [`remove`]      | O(1)        | Clears slot + index entry              |
//! | [`pop_victim`]  | O(1) amort. | Evicts next unreferenced entry         |
//! | [`peek_victim`] | O(n) worst  | Finds next victim without modifying    |
//! | [`iter`] / [`keys`] / [`values`] | O(n) | Borrowed iteration over entries |
//! | [`into_iter`]   | O(n)        | Consuming iteration over entries       |
//!
//! [`insert`]: ClockRing::insert
//! [`get`]: ClockRing::get
//! [`peek`]: ClockRing::peek
//! [`touch`]: ClockRing::touch
//! [`remove`]: ClockRing::remove
//! [`pop_victim`]: ClockRing::pop_victim
//! [`peek_victim`]: ClockRing::peek_victim
//! [`iter`]: ClockRing::iter
//! [`keys`]: ClockRing::keys
//! [`values`]: ClockRing::values
//! [`into_iter`]: ClockRing#impl-IntoIterator
//!
//! ## Use Cases
//!
//! - **Page replacement**: Classic use case for CLOCK algorithm
//! - **Buffer pool**: Database buffer management
//! - **Web cache**: Approximate LRU with lower overhead
//!
//! ## Example Usage
//!
//! ```
//! use cachekit::ds::ClockRing;
//!
//! let mut cache = ClockRing::new(3);
//!
//! // Insert entries
//! cache.insert("page_1", "content_1");
//! cache.insert("page_2", "content_2");
//! cache.insert("page_3", "content_3");
//!
//! // Access sets the reference bit (second chance)
//! cache.get(&"page_1");  // page_1 now has ref=1
//!
//! // Insert when full - evicts unreferenced entry
//! let evicted = cache.insert("page_4", "content_4");
//! // page_2 or page_3 evicted (page_1 protected by ref bit)
//!
//! assert!(cache.contains(&"page_1"));  // Survived due to reference bit
//! assert!(cache.contains(&"page_4"));  // Newly inserted
//! ```
//!
//! ## Use Case: Page Buffer
//!
//! ```
//! use cachekit::ds::ClockRing;
//!
//! struct PageBuffer {
//!     cache: ClockRing<u64, Vec<u8>>,  // page_id -> page_data
//! }
//!
//! impl PageBuffer {
//!     fn new(capacity: usize) -> Self {
//!         Self { cache: ClockRing::new(capacity) }
//!     }
//!
//!     fn read_page(&mut self, page_id: u64) -> Option<&Vec<u8>> {
//!         // get() sets reference bit, giving this page a second chance
//!         self.cache.get(&page_id)
//!     }
//!
//!     fn load_page(&mut self, page_id: u64, data: Vec<u8>) {
//!         if let Some((evicted_id, _evicted_data)) = self.cache.insert(page_id, data) {
//!             // Could write back dirty page here
//!             println!("Evicted page {}", evicted_id);
//!         }
//!     }
//! }
//!
//! let mut buffer = PageBuffer::new(100);
//! buffer.load_page(1, vec![0; 4096]);
//! buffer.load_page(2, vec![0; 4096]);
//! assert!(buffer.read_page(1).is_some());
//! ```
//!
//! ## Thread Safety
//!
//! - [`ClockRing`]: Not thread-safe, use in single-threaded contexts
//! - [`ConcurrentClockRing`]: Thread-safe via `parking_lot::RwLock`
//!
//! ## Implementation Notes
//!
//! - Fixed-size slot array; no reallocation during operation
//! - Reference bits stored in a separate `Vec<bool>` for cache-friendly sweeps
//! - Keys mapped to slot indices via [`std::collections::HashMap`]; the default
//!   DoS-resistant [`RandomState`] hasher is used unless the caller opts into
//!   another via [`ClockRing::with_hasher`]. Key is cloned once per new insertion.
//! - Hand pointer advances after each insert/eviction
//! - [`insert`] is O(1) amortized: each access sets at most one ref bit, so the
//!   total clearing work across N inserts is bounded by N
//! - `debug_validate_invariants()` available in debug/test builds
//!
//! ## Memory Budgeting
//!
//! `ClockRing` bounds the *number* of entries (`capacity`), not the
//! aggregate byte footprint of stored values. [`ClockRing::approx_bytes`]
//! reports only the in-line cost of `Option<Entry<K, V>>` slots plus
//! the index map; it counts `size_of::<V>()` and **does not follow
//! heap pointers owned by `V`**. A ring of `Vec<u8>` or `String` values
//! can therefore consume orders of magnitude more memory than
//! `approx_bytes` suggests.
//!
//! For workloads with variable-sized values, enforce a byte budget
//! at the call site (e.g. track `value.len()` on insert and evict via
//! [`ClockRing::pop_victim`] until under budget) rather than relying
//! on slot count alone.
//!
//! ## Timing Side Channels
//!
//! Lookup methods ([`contains`](ClockRing::contains),
//! [`peek`](ClockRing::peek), [`get`](ClockRing::get),
//! [`touch`](ClockRing::touch), …) take time that depends on whether
//! the key is present and on the hasher's probe chain length. This is
//! standard for any [`HashMap`]-backed cache and is **not** mitigated
//! here — the default [`RandomState`] randomizes the probe chain
//! across processes (defeating cross-process collision attacks) but
//! does not produce constant-time lookups.
//!
//! Do not use `ClockRing` as the backing store for data structures
//! whose key set must remain confidential against a co-located
//! attacker that can measure operation latency (e.g. session-token
//! caches in a multi-tenant process where the tenant boundary is the
//! key space). For such use cases, gate access behind an explicit
//! constant-time membership check or use a constant-time data
//! structure.
//!
//! [`RandomState`]: std::collections::hash_map::RandomState
//! [`HashMap`]: std::collections::HashMap

use std::borrow::Borrow;
use std::collections::HashMap;
use std::collections::hash_map::RandomState;
use std::hash::{BuildHasher, Hash};

#[cfg(feature = "concurrency")]
use parking_lot::RwLock;

/// Coarse upper bound on `capacity` accepted by [`ClockRing::new`] /
/// [`ClockRing::try_new`].
///
/// This is a *first* guard that rejects obviously pathological values
/// cheaply. It is intentionally permissive: for any key/value size
/// larger than a few bytes, allocating `MAX_CAPACITY` slots would still
/// exhaust memory. The constructors defend against that separately by
/// using [`Vec::try_reserve_exact`] / [`HashMap::try_reserve`], so
/// out-of-memory conditions surface as
/// [`ClockRingError::AllocationFailed`] rather than aborting the
/// process.
pub const MAX_CAPACITY: usize = isize::MAX as usize / 64;

/// Error returned by [`ClockRing::try_new`] / [`ClockRing::try_with_hasher`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ClockRingError {
    /// The requested capacity exceeds [`MAX_CAPACITY`].
    CapacityTooLarge {
        /// The capacity that was requested.
        requested: usize,
        /// The configured upper bound.
        max: usize,
    },
    /// The allocator could not satisfy the per-slot reservation for the
    /// requested capacity.
    ///
    /// Returned instead of aborting the process when
    /// `capacity * size_of::<Option<Entry<K, V>>>()` exceeds what the
    /// allocator can provide (including the case where the byte count
    /// overflows `isize::MAX`).
    AllocationFailed {
        /// The capacity whose allocation failed.
        requested: usize,
    },
}

impl std::fmt::Display for ClockRingError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ClockRingError::CapacityTooLarge { requested, max } => {
                write!(f, "ClockRing capacity {requested} exceeds maximum {max}")
            },
            ClockRingError::AllocationFailed { requested } => {
                write!(
                    f,
                    "ClockRing failed to allocate backing storage for capacity {requested}"
                )
            },
        }
    }
}

impl std::error::Error for ClockRingError {}

/// Acknowledges that the caller is supplying a hasher to
/// [`ClockRing::with_hasher`] / [`ClockRing::try_with_hasher`] (and
/// the [`ConcurrentClockRing`] equivalents) under the assumption that
/// keys are **not attacker-controlled**.
///
/// Required at every call site as a deliberate, code-reviewable
/// signal: a non-randomized hasher (e.g. `rustc_hash::FxBuildHasher`,
/// unseeded `ahash::AHasher`, FNV) reintroduces hash-collision DoS —
/// adversarial keys can degrade [`HashMap`]-backed lookups to O(n)
/// regardless of the CLOCK sweep cost.
///
/// Acceptable inputs to a `KeysAreTrusted`-tagged ring include:
/// authenticated user/row IDs, opaque handles minted internally,
/// values already passed through a keyed hash (HMAC), or any source
/// the application's threat model treats as integrity-protected.
/// For attacker-controlled keys, use [`ClockRing::new`] /
/// [`ClockRing::try_new`], which default to the DoS-resistant
/// [`RandomState`] hasher.
///
/// # Example
///
/// ```
/// use cachekit::ds::{ClockRing, KeysAreTrusted};
/// use std::collections::hash_map::RandomState;
///
/// let ring: ClockRing<u64, &str> =
///     ClockRing::with_hasher(64, RandomState::new(), KeysAreTrusted::new());
/// assert_eq!(ring.capacity(), 64);
/// ```
///
/// [`HashMap`]: std::collections::HashMap
/// [`RandomState`]: std::collections::hash_map::RandomState
#[derive(Debug, Clone, Copy, Default)]
pub struct KeysAreTrusted(());

impl KeysAreTrusted {
    /// Constructs the acknowledgement.
    ///
    /// See the [type-level docs](KeysAreTrusted) for the security
    /// implications of passing this to a hasher-configurable
    /// constructor.
    #[inline]
    pub const fn new() -> Self {
        Self(())
    }
}

/// Advance an index by one position modulo `cap`, overflow-safe.
///
/// Using `checked_add` + fallback to `wrapping_add % cap` avoids the
/// `(idx + 1) % cap` overflow that is theoretically reachable on
/// pathologically large `cap` values.
///
/// Returns `0` when `cap == 0` so that the function is total: current
/// callers all guard on non-zero capacity, but making `step` itself
/// safe to call with any input prevents a release-mode division-by-zero
/// panic if a future call site forgets the guard.
#[inline]
fn step(idx: usize, cap: usize) -> usize {
    if cap == 0 {
        return 0;
    }
    match idx.checked_add(1) {
        Some(next) => next % cap,
        None => 0,
    }
}

/// Clock entry holding key and value.
///
/// Reference bits are stored separately in [`ClockRing::referenced`] for
/// cache-friendly sweep access.
#[derive(Debug, Clone)]
struct Entry<K, V> {
    key: K,
    value: V,
}

/// Fixed-size ring implementing the CLOCK (second-chance) eviction algorithm.
///
/// Provides O(1) amortized insertion with automatic eviction of unreferenced
/// entries. Accessed entries receive a "second chance" via a reference bit.
///
/// Lookup methods ([`get`](Self::get), [`peek`](Self::peek),
/// [`contains`](Self::contains), etc.) accept any borrowed form of the key
/// via [`Borrow`], so a `ClockRing<String, V>` can be
/// queried with `&str`.
///
/// Implements [`Clone`], [`Debug`], [`Extend`]`<(K, V)>`, and
/// [`IntoIterator`]. See [`iter`](Self::iter), [`keys`](Self::keys),
/// [`values`](Self::values) for borrowed iteration.
///
/// # Type Parameters
///
/// - `K`: Key type, must be `Eq + Hash + Clone`
/// - `V`: Value type
/// - `S`: Hash builder. Defaults to [`RandomState`], a DoS-resistant
///   randomized hasher. Swap via [`ClockRing::with_hasher`] for a faster
///   non-randomized hasher when keys are fully trusted.
///
/// # Hash-collision resistance
///
/// `ClockRing` defaults to the standard library's randomized
/// [`RandomState`]. This makes it safe to use as a cache keyed by
/// attacker-controlled data (URL paths, user IDs, header values, etc.)
/// without worrying about adversarial hash collisions degrading lookups
/// to O(n). Callers that fully trust their key source can opt into a
/// faster non-randomized hasher (e.g. `rustc_hash::FxBuildHasher`) via
/// [`ClockRing::with_hasher`] / [`ClockRing::try_with_hasher`], which
/// require a [`KeysAreTrusted`] acknowledgement to make the trade-off
/// visible at every call site.
///
/// # Example
///
/// ```
/// use cachekit::ds::ClockRing;
///
/// let mut ring = ClockRing::new(2);
///
/// // Insert two entries
/// ring.insert("a", 1);
/// ring.insert("b", 2);
///
/// // Access "a" - sets reference bit
/// ring.get(&"a");
///
/// // Insert "c" - "b" is evicted (no reference bit)
/// let evicted = ring.insert("c", 3);
/// assert_eq!(evicted, Some(("b", 2)));
///
/// assert!(ring.contains(&"a"));
/// assert!(ring.contains(&"c"));
/// ```
///
/// # Use Case: Simple Cache with Eviction Callback
///
/// ```
/// use cachekit::ds::ClockRing;
///
/// let mut cache = ClockRing::new(3);
/// let mut eviction_count = 0;
///
/// for i in 0..10 {
///     if let Some((_key, _value)) = cache.insert(i, format!("value_{}", i)) {
///         eviction_count += 1;
///     }
/// }
///
/// assert_eq!(cache.len(), 3);
/// assert_eq!(eviction_count, 7);  // 10 inserts - 3 capacity = 7 evictions
/// ```
#[must_use]
pub struct ClockRing<K, V, S = RandomState> {
    slots: Vec<Option<Entry<K, V>>>,
    referenced: Vec<bool>,
    index: HashMap<K, usize, S>,
    hand: usize,
    len: usize,
    #[cfg(feature = "metrics")]
    sweep_hand_advances: u64,
    #[cfg(feature = "metrics")]
    sweep_ref_bit_resets: u64,
}

/// Hand-written so stored keys and values never appear in debug
/// output. A derived `Debug` would recurse through every `Entry<K,
/// V>` and `HashMap<K, usize, S>` entry, making `tracing::debug!`,
/// `dbg!`, and panic-unwind backtraces into a secret-exposure channel
/// for caches that hold session tokens, API keys, or any other
/// sensitive material. Callers that need full contents can iterate
/// via [`ClockRing::iter`] and print entries they've vetted.
impl<K, V, S> std::fmt::Debug for ClockRing<K, V, S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut dbg = f.debug_struct("ClockRing");
        dbg.field("len", &self.len)
            .field("capacity", &self.slots.len())
            .field("hand", &self.hand);
        #[cfg(feature = "metrics")]
        {
            dbg.field("sweep_hand_advances", &self.sweep_hand_advances)
                .field("sweep_ref_bit_resets", &self.sweep_ref_bit_resets);
        }
        dbg.finish_non_exhaustive()
    }
}

/// Thread-safe wrapper around [`ClockRing`] using `parking_lot::RwLock`.
///
/// Provides the same functionality as [`ClockRing`] but safe for concurrent
/// access. Uses closure-based value access since references cannot outlive
/// lock guards. Lookup methods accept any borrowed form of the key via
/// [`Borrow`].
///
/// # Example
///
/// ```
/// use std::sync::Arc;
/// use std::thread;
/// use cachekit::ds::ConcurrentClockRing;
///
/// let cache = Arc::new(ConcurrentClockRing::new(100));
///
/// let handles: Vec<_> = (0..4).map(|t| {
///     let cache = Arc::clone(&cache);
///     thread::spawn(move || {
///         for i in 0..25 {
///             let key = t * 25 + i;
///             cache.insert(key, key * 10);
///         }
///     })
/// }).collect();
///
/// for h in handles {
///     h.join().unwrap();
/// }
///
/// assert!(cache.len() <= 100);
/// ```
///
/// # Iteration
///
/// `ConcurrentClockRing` does not directly expose iterators (holding
/// a lock for the duration of iteration would hurt concurrency). Call
/// [`into_inner`](Self::into_inner) to unwrap the inner [`ClockRing`]
/// and iterate it.
///
/// # Non-blocking Operations
///
/// All operations have `try_*` variants that return `None` if the lock
/// cannot be acquired immediately:
///
/// ```
/// use cachekit::ds::ConcurrentClockRing;
///
/// let cache = ConcurrentClockRing::new(10);
/// cache.insert("key", 42);
///
/// // Non-blocking read
/// if let Some(Some(val)) = cache.try_get_with(&"key", |v| *v) {
///     assert_eq!(val, 42);
/// }
/// ```
///
/// # Closures run under the lock
///
/// Every `*_with` / `try_*_with` method invokes the caller-supplied
/// closure **while holding the internal `RwLock` guard**. This has
/// important consequences:
///
/// - The closure **must not** call back into the same
///   `ConcurrentClockRing` (or any structure that might). The lock is
///   *not* reentrant, so re-entry will either deadlock (`*_with`) or
///   spuriously fail by returning `None` (`try_*_with`).
/// - A slow closure stalls every other reader/writer of this cache.
///   Keep closures short; clone data out if you need to perform I/O
///   or further work.
/// - `get_with` / `get_mut_with` / `touch` / `update` /
///   `remove` / `pop_victim` and their `try_*` variants hold the
///   **write** lock across the closure because they mutate the
///   reference bits; `peek_with` / `peek_victim_with` hold the read
///   lock.
/// - If the closure panics the lock is released cleanly
///   (`parking_lot::RwLock` does not poison), but any observable state
///   changes performed *before* the closure (e.g. setting the
///   reference bit in `get_with`) are retained.
///
/// # `Drop for V` runs outside the lock
///
/// [`insert`](Self::insert) / [`try_insert`](Self::try_insert) release
/// the internal write lock before dropping the value replaced by an
/// update (and the caller is responsible for dropping the evicted
/// entry after the method returns). This keeps an expensive or
/// reentrant `Drop for V` from stalling other threads or deadlocking
/// by calling back into the same cache. The same is true for
/// [`remove`](Self::remove): the returned value is dropped on the
/// caller's side of the lock.
#[cfg(feature = "concurrency")]
#[must_use]
pub struct ConcurrentClockRing<K, V, S = RandomState> {
    inner: RwLock<ClockRing<K, V, S>>,
}

/// See the [`ClockRing`] `Debug` impl for the rationale: stored keys
/// and values are redacted. Acquires the inner read lock, so calling
/// `{:?}` under a held write guard will deadlock on the same thread
/// that holds the guard — matching the behavior of
/// `RwLock::read()` elsewhere in this module.
#[cfg(feature = "concurrency")]
impl<K, V, S> std::fmt::Debug for ConcurrentClockRing<K, V, S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let inner = self.inner.read();
        let mut dbg = f.debug_struct("ConcurrentClockRing");
        dbg.field("len", &inner.len)
            .field("capacity", &inner.slots.len())
            .field("hand", &inner.hand);
        #[cfg(feature = "metrics")]
        {
            dbg.field("sweep_hand_advances", &inner.sweep_hand_advances)
                .field("sweep_ref_bit_resets", &inner.sweep_ref_bit_resets);
        }
        dbg.finish_non_exhaustive()
    }
}

#[cfg(feature = "concurrency")]
impl<K, V> ConcurrentClockRing<K, V, RandomState>
where
    K: Eq + Hash + Clone,
{
    /// Creates a new ring with `capacity` slots and the default
    /// DoS-resistant hasher ([`RandomState`]).
    ///
    /// # Panics
    ///
    /// Panics if `capacity > MAX_CAPACITY`. Use
    /// [`try_new`](Self::try_new) for a fallible version.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<String, i32> = ConcurrentClockRing::new(100);
    /// assert_eq!(cache.capacity(), 100);
    /// assert!(cache.is_empty());
    /// ```
    pub fn new(capacity: usize) -> Self {
        Self {
            inner: RwLock::new(ClockRing::new(capacity)),
        }
    }

    /// Fallible version of [`new`](Self::new).
    ///
    /// Returns [`ClockRingError::CapacityTooLarge`] if
    /// `capacity > MAX_CAPACITY`.
    pub fn try_new(capacity: usize) -> Result<Self, ClockRingError> {
        Ok(Self {
            inner: RwLock::new(ClockRing::try_new(capacity)?),
        })
    }
}

#[cfg(feature = "concurrency")]
impl<K, V, S> ConcurrentClockRing<K, V, S>
where
    K: Eq + Hash + Clone,
    S: BuildHasher,
{
    /// Creates a new ring with `capacity` slots using the provided
    /// `hash_builder`.
    ///
    /// The [`KeysAreTrusted`] argument is a deliberate acknowledgement
    /// that a non-randomized hasher is safe for this workload — see
    /// its type-level docs for the DoS trade-off. Use
    /// [`new`](Self::new) / [`try_new`](Self::try_new) for the default
    /// DoS-resistant [`RandomState`] hasher.
    ///
    /// # Panics
    ///
    /// Panics if `capacity > MAX_CAPACITY`. Use
    /// [`try_with_hasher`](Self::try_with_hasher) for a fallible version.
    pub fn with_hasher(capacity: usize, hash_builder: S, ack: KeysAreTrusted) -> Self {
        Self {
            inner: RwLock::new(ClockRing::with_hasher(capacity, hash_builder, ack)),
        }
    }

    /// Fallible version of [`with_hasher`](Self::with_hasher).
    ///
    /// The [`KeysAreTrusted`] argument is a deliberate acknowledgement
    /// that a non-randomized hasher is safe for this workload — see
    /// its type-level docs for the DoS trade-off.
    pub fn try_with_hasher(
        capacity: usize,
        hash_builder: S,
        ack: KeysAreTrusted,
    ) -> Result<Self, ClockRingError> {
        Ok(Self {
            inner: RwLock::new(ClockRing::try_with_hasher(capacity, hash_builder, ack)?),
        })
    }

    /// Returns the configured capacity (number of slots).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<&str, i32> = ConcurrentClockRing::new(50);
    /// assert_eq!(cache.capacity(), 50);
    /// ```
    pub fn capacity(&self) -> usize {
        let ring = self.inner.read();
        ring.capacity()
    }

    /// Returns the number of occupied slots.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// assert_eq!(cache.len(), 0);
    ///
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    /// assert_eq!(cache.len(), 2);
    /// ```
    pub fn len(&self) -> usize {
        let ring = self.inner.read();
        ring.len()
    }

    /// Returns `true` if there are no entries.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<&str, i32> = ConcurrentClockRing::new(10);
    /// assert!(cache.is_empty());
    ///
    /// cache.insert("key", 42);
    /// assert!(!cache.is_empty());
    /// ```
    pub fn is_empty(&self) -> bool {
        let ring = self.inner.read();
        ring.is_empty()
    }

    /// Returns `true` if `key` is present.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("key", 42);
    ///
    /// assert!(cache.contains(&"key"));
    /// assert!(!cache.contains(&"missing"));
    /// ```
    pub fn contains<Q>(&self, key: &Q) -> bool
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let ring = self.inner.read();
        ring.contains(key)
    }

    /// Accesses value without setting the reference bit.
    ///
    /// Unlike [`get_with`](Self::get_with), this does not grant a second chance.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("key", vec![1, 2, 3]);
    ///
    /// let sum = cache.peek_with(&"key", |v| v.iter().sum::<i32>());
    /// assert_eq!(sum, Some(6));
    /// ```
    pub fn peek_with<Q, R>(&self, key: &Q, f: impl FnOnce(&V) -> R) -> Option<R>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let ring = self.inner.read();
        ring.peek(key).map(f)
    }

    /// Accesses value and sets the reference bit (grants second chance).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    ///
    /// // Access "a" - sets reference bit
    /// cache.get_with(&"a", |v| *v);
    ///
    /// // Insert "c" - "b" evicted (no ref bit), "a" survives
    /// cache.insert("c", 3);
    /// assert!(cache.contains(&"a"));
    /// assert!(!cache.contains(&"b"));
    /// ```
    pub fn get_with<Q, R>(&self, key: &Q, f: impl FnOnce(&V) -> R) -> Option<R>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.write();
        ring.get(key).map(f)
    }

    /// Accesses value mutably and sets the reference bit (grants second chance).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("key", vec![1, 2, 3]);
    ///
    /// cache.get_mut_with(&"key", |v| v.push(4));
    /// let sum = cache.peek_with(&"key", |v| v.iter().sum::<i32>());
    /// assert_eq!(sum, Some(10));
    /// ```
    pub fn get_mut_with<Q, R>(&self, key: &Q, f: impl FnOnce(&mut V) -> R) -> Option<R>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.write();
        ring.get_mut(key).map(f)
    }

    /// Sets the reference bit for `key`; returns `false` if missing.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("key", 42);
    ///
    /// assert!(cache.touch(&"key"));       // Sets reference bit
    /// assert!(!cache.touch(&"missing"));  // Key not found
    /// ```
    pub fn touch<Q>(&self, key: &Q) -> bool
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.write();
        ring.touch(key)
    }

    /// Updates the value for an existing key, returning the old value.
    ///
    /// Sets the reference bit on update. Returns `None` if the key doesn't exist.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    /// cache.insert("a", 1);
    ///
    /// assert_eq!(cache.update(&"a", 10), Some(1));
    /// assert_eq!(cache.update(&"missing", 99), None);
    /// ```
    pub fn update<Q>(&self, key: &Q, value: V) -> Option<V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.write();
        ring.update(key, value)
    }

    /// Inserts or updates `key`, evicting if necessary.
    ///
    /// Returns `Some((evicted_key, evicted_value))` if an entry was evicted.
    ///
    /// Any value replaced by an **update** (same key already present)
    /// is dropped *after* the internal write lock has been released, so
    /// a slow or reentrant `Drop for V` cannot stall other
    /// readers/writers or deadlock by calling back into this cache.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    /// assert_eq!(cache.insert("a", 1), None);  // No eviction
    /// assert_eq!(cache.insert("b", 2), None);  // No eviction
    ///
    /// // Full - must evict
    /// let evicted = cache.insert("c", 3);
    /// assert!(evicted.is_some());
    /// ```
    pub fn insert(&self, key: K, value: V) -> Option<(K, V)> {
        let (replaced, evicted) = {
            let mut ring = self.inner.write();
            ring.insert_swap(key, value)
        };
        drop(replaced);
        evicted
    }

    /// Removes `key` and returns its value, if present.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("key", 42);
    ///
    /// assert_eq!(cache.remove(&"key"), Some(42));
    /// assert_eq!(cache.remove(&"key"), None);  // Already removed
    /// ```
    pub fn remove<Q>(&self, key: &Q) -> Option<V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.write();
        ring.remove(key)
    }

    /// Peeks the next eviction candidate without modifying state.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    /// cache.touch(&"a");  // "a" now has ref bit
    ///
    /// // "b" is next victim (no ref bit)
    /// let victim_key = cache.peek_victim_with(|k, _v| *k);
    /// assert_eq!(victim_key, Some("b"));
    /// ```
    pub fn peek_victim_with<R>(&self, f: impl FnOnce(&K, &V) -> R) -> Option<R> {
        let ring = self.inner.read();
        ring.peek_victim().map(|(key, value)| f(key, value))
    }

    /// Evicts the next candidate and returns it.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(3);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    /// cache.insert("c", 3);
    ///
    /// let evicted = cache.pop_victim();
    /// assert!(evicted.is_some());
    /// assert_eq!(cache.len(), 2);
    /// ```
    pub fn pop_victim(&self) -> Option<(K, V)> {
        let mut ring = self.inner.write();
        ring.pop_victim()
    }

    /// Clears all entries without releasing capacity.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    ///
    /// cache.clear();
    /// assert!(cache.is_empty());
    /// ```
    pub fn clear(&self) {
        let mut ring = self.inner.write();
        ring.clear();
    }

    /// Clears all entries and shrinks internal storage.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(100);
    /// cache.insert("a", 1);
    ///
    /// cache.clear_shrink();
    /// assert!(cache.is_empty());
    /// ```
    pub fn clear_shrink(&self) {
        let mut ring = self.inner.write();
        ring.clear_shrink();
    }

    /// Returns an approximate memory footprint in bytes.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<u64, u64> = ConcurrentClockRing::new(100);
    /// assert!(cache.approx_bytes() > 0);
    /// ```
    #[must_use]
    pub fn approx_bytes(&self) -> usize {
        let ring = self.inner.read();
        ring.approx_bytes()
    }

    /// Reserves capacity for at least `additional` more index entries.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<String, i32> = ConcurrentClockRing::new(10);
    /// cache.reserve_index(100);
    /// ```
    pub fn reserve_index(&self, additional: usize) {
        let mut ring = self.inner.write();
        ring.reserve_index(additional);
    }

    /// Shrinks internal storage to fit current contents.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache: ConcurrentClockRing<&str, i32> = ConcurrentClockRing::new(100);
    /// cache.insert("a", 1);
    /// cache.shrink_to_fit();
    /// ```
    pub fn shrink_to_fit(&self) {
        let mut ring = self.inner.write();
        ring.shrink_to_fit();
    }

    /// Non-blocking version of [`update`](Self::update).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    ///
    /// assert_eq!(cache.try_update(&"a", 10), Some(Some(1)));
    /// assert_eq!(cache.try_update(&"missing", 99), Some(None));
    /// ```
    pub fn try_update<Q>(&self, key: &Q, value: V) -> Option<Option<V>>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.try_write()?;
        Some(ring.update(key, value))
    }

    /// Non-blocking version of [`insert`](Self::insert).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    ///
    /// assert_eq!(cache.try_insert("a", 1), Some(None));
    /// assert_eq!(cache.try_insert("b", 2), Some(None));
    /// ```
    pub fn try_insert(&self, key: K, value: V) -> Option<Option<(K, V)>> {
        let (replaced, evicted) = {
            let mut ring = self.inner.try_write()?;
            ring.insert_swap(key, value)
        };
        drop(replaced);
        Some(evicted)
    }

    /// Non-blocking version of [`remove`](Self::remove).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    ///
    /// assert_eq!(cache.try_remove(&"a"), Some(Some(1)));
    /// assert_eq!(cache.try_remove(&"a"), Some(None));
    /// ```
    pub fn try_remove<Q>(&self, key: &Q) -> Option<Option<V>>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.try_write()?;
        Some(ring.remove(key))
    }

    /// Non-blocking version of [`touch`](Self::touch).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    ///
    /// assert_eq!(cache.try_touch(&"a"), Some(true));
    /// assert_eq!(cache.try_touch(&"missing"), Some(false));
    /// ```
    pub fn try_touch<Q>(&self, key: &Q) -> Option<bool>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.try_write()?;
        Some(ring.touch(key))
    }

    /// Non-blocking version of [`peek_with`](Self::peek_with).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 42);
    ///
    /// assert_eq!(cache.try_peek_with(&"a", |v| *v), Some(Some(42)));
    /// assert_eq!(cache.try_peek_with(&"missing", |v| *v), Some(None));
    /// ```
    pub fn try_peek_with<Q, R>(&self, key: &Q, f: impl FnOnce(&V) -> R) -> Option<Option<R>>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let ring = self.inner.try_read()?;
        Some(ring.peek(key).map(f))
    }

    /// Non-blocking version of [`get_with`](Self::get_with).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 42);
    ///
    /// assert_eq!(cache.try_get_with(&"a", |v| *v), Some(Some(42)));
    /// assert_eq!(cache.try_get_with(&"missing", |v| *v), Some(None));
    /// ```
    pub fn try_get_with<Q, R>(&self, key: &Q, f: impl FnOnce(&V) -> R) -> Option<Option<R>>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.try_write()?;
        Some(ring.get(key).map(f))
    }

    /// Non-blocking version of [`get_mut_with`](Self::get_mut_with).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", vec![1, 2]);
    ///
    /// cache.try_get_mut_with(&"a", |v| v.push(3));
    /// let sum = cache.peek_with(&"a", |v| v.iter().sum::<i32>());
    /// assert_eq!(sum, Some(6));
    /// ```
    pub fn try_get_mut_with<Q, R>(&self, key: &Q, f: impl FnOnce(&mut V) -> R) -> Option<Option<R>>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let mut ring = self.inner.try_write()?;
        Some(ring.get_mut(key).map(f))
    }

    /// Non-blocking version of [`peek_victim_with`](Self::peek_victim_with).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(2);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    ///
    /// if let Some(Some(key)) = cache.try_peek_victim_with(|k, _v| *k) {
    ///     assert!(key == "a" || key == "b");
    /// }
    /// ```
    pub fn try_peek_victim_with<R>(&self, f: impl FnOnce(&K, &V) -> R) -> Option<Option<R>> {
        let ring = self.inner.try_read()?;
        Some(ring.peek_victim().map(|(key, value)| f(key, value)))
    }

    /// Non-blocking version of [`pop_victim`](Self::pop_victim).
    ///
    /// Returns `None` if the lock could not be acquired.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(3);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    ///
    /// if let Some(evicted) = cache.try_pop_victim() {
    ///     assert!(evicted.is_some());
    /// }
    /// ```
    pub fn try_pop_victim(&self) -> Option<Option<(K, V)>> {
        let mut ring = self.inner.try_write()?;
        Some(ring.pop_victim())
    }

    /// Non-blocking clear. Returns `true` if successful.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    ///
    /// assert!(cache.try_clear());
    /// assert!(cache.is_empty());
    /// ```
    pub fn try_clear(&self) -> bool {
        if let Some(mut ring) = self.inner.try_write() {
            ring.clear();
            true
        } else {
            false
        }
    }

    /// Non-blocking clear and shrink. Returns `true` if successful.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(100);
    /// cache.insert("a", 1);
    ///
    /// assert!(cache.try_clear_shrink());
    /// assert!(cache.is_empty());
    /// ```
    pub fn try_clear_shrink(&self) -> bool {
        if let Some(mut ring) = self.inner.try_write() {
            ring.clear_shrink();
            true
        } else {
            false
        }
    }
}

// ---------------------------------------------------------------------------
// Iteration methods — no trait bounds needed on K
// ---------------------------------------------------------------------------

impl<K, V, S> ClockRing<K, V, S> {
    /// Returns an iterator over `(&K, &V)` pairs in slot order.
    ///
    /// Does **not** set reference bits (like [`peek`](Self::peek)).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// let pairs: Vec<_> = ring.iter().collect();
    /// assert_eq!(pairs.len(), 2);
    /// ```
    pub fn iter(&self) -> Iter<'_, K, V> {
        Iter {
            inner: self.slots.iter(),
        }
    }

    /// Returns an iterator over `(&K, &mut V)` pairs in slot order.
    ///
    /// Does **not** set reference bits.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// for (_key, value) in ring.iter_mut() {
    ///     *value += 10;
    /// }
    /// assert_eq!(ring.peek(&"a"), Some(&11));
    /// ```
    pub fn iter_mut(&mut self) -> IterMut<'_, K, V> {
        IterMut {
            inner: self.slots.iter_mut(),
        }
    }

    /// Returns an iterator over keys in slot order.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// let keys: Vec<_> = ring.keys().collect();
    /// assert_eq!(keys.len(), 2);
    /// ```
    pub fn keys(&self) -> Keys<'_, K, V> {
        Keys { inner: self.iter() }
    }

    /// Returns an iterator over values in slot order.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// let sum: i32 = ring.values().sum();
    /// assert_eq!(sum, 3);
    /// ```
    pub fn values(&self) -> Values<'_, K, V> {
        Values { inner: self.iter() }
    }

    /// Returns an iterator over mutable values in slot order.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// for value in ring.values_mut() {
    ///     *value *= 2;
    /// }
    /// assert_eq!(ring.peek(&"a"), Some(&2));
    /// assert_eq!(ring.peek(&"b"), Some(&4));
    /// ```
    pub fn values_mut(&mut self) -> ValuesMut<'_, K, V> {
        ValuesMut {
            inner: self.iter_mut(),
        }
    }
}

impl<K, V> ClockRing<K, V, RandomState>
where
    K: Eq + Hash + Clone,
{
    /// Creates a new ring with `capacity` slots and the default
    /// DoS-resistant hasher ([`RandomState`]).
    ///
    /// # Panics
    ///
    /// Panics if `capacity > MAX_CAPACITY`. Use
    /// [`try_new`](Self::try_new) for a fallible version.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let ring: ClockRing<String, i32> = ClockRing::new(100);
    /// assert_eq!(ring.capacity(), 100);
    /// assert!(ring.is_empty());
    /// ```
    pub fn new(capacity: usize) -> Self {
        Self::try_new(capacity).expect("ClockRing::new: capacity exceeds MAX_CAPACITY")
    }

    /// Fallible version of [`new`](Self::new).
    ///
    /// Returns [`ClockRingError::CapacityTooLarge`] when
    /// `capacity > MAX_CAPACITY`.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::{ClockRing, MAX_CAPACITY};
    ///
    /// let ring = ClockRing::<u32, u32>::try_new(16).unwrap();
    /// assert_eq!(ring.capacity(), 16);
    ///
    /// assert!(ClockRing::<u32, u32>::try_new(MAX_CAPACITY + 1).is_err());
    /// ```
    pub fn try_new(capacity: usize) -> Result<Self, ClockRingError> {
        // `RandomState` *is* DoS-resistant, so the acknowledgement
        // constructed here is truthful — it's the marker that the
        // caller has chosen a hasher appropriate for untrusted keys.
        Self::try_with_hasher(capacity, RandomState::default(), KeysAreTrusted::new())
    }
}

impl<K, V, S> ClockRing<K, V, S>
where
    K: Eq + Hash + Clone,
    S: BuildHasher,
{
    /// Creates a new ring with `capacity` slots using the supplied
    /// `hash_builder`.
    ///
    /// Use this to opt into a faster non-randomized hasher (e.g.
    /// `rustc_hash::FxBuildHasher`) **only when keys are fully
    /// trusted**. With attacker-controlled keys, a non-randomized hasher
    /// enables hash-collision DoS.
    ///
    /// The [`KeysAreTrusted`] argument is a deliberate, greppable
    /// acknowledgement of that trade-off; see its type-level docs for
    /// guidance on when it's appropriate.
    ///
    /// # Panics
    ///
    /// Panics if `capacity > MAX_CAPACITY`. Use
    /// [`try_with_hasher`](Self::try_with_hasher) for a fallible version.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::{ClockRing, KeysAreTrusted};
    /// use std::collections::hash_map::RandomState;
    ///
    /// let ring: ClockRing<u64, &str> =
    ///     ClockRing::with_hasher(64, RandomState::new(), KeysAreTrusted::new());
    /// assert_eq!(ring.capacity(), 64);
    /// ```
    #[track_caller]
    pub fn with_hasher(capacity: usize, hash_builder: S, ack: KeysAreTrusted) -> Self {
        Self::try_with_hasher(capacity, hash_builder, ack)
            .expect("ClockRing::with_hasher: capacity exceeds MAX_CAPACITY")
    }

    /// Fallible version of [`with_hasher`](Self::with_hasher).
    ///
    /// The [`KeysAreTrusted`] argument is a deliberate, greppable
    /// acknowledgement that a non-randomized hasher is safe for this
    /// workload; see its type-level docs for guidance.
    ///
    /// Returns [`ClockRingError::CapacityTooLarge`] when `capacity`
    /// exceeds [`MAX_CAPACITY`], or [`ClockRingError::AllocationFailed`]
    /// when the allocator cannot satisfy the backing-storage reservation
    /// for `capacity`. This never aborts the process on hostile
    /// capacity values, even when `capacity * size_of::<Option<Entry<K,
    /// V>>>()` would overflow `isize::MAX`.
    pub fn try_with_hasher(
        capacity: usize,
        hash_builder: S,
        _ack: KeysAreTrusted,
    ) -> Result<Self, ClockRingError> {
        if capacity > MAX_CAPACITY {
            return Err(ClockRingError::CapacityTooLarge {
                requested: capacity,
                max: MAX_CAPACITY,
            });
        }

        let alloc_err = || ClockRingError::AllocationFailed {
            requested: capacity,
        };

        let mut slots: Vec<Option<Entry<K, V>>> = Vec::new();
        slots.try_reserve_exact(capacity).map_err(|_| alloc_err())?;
        slots.resize_with(capacity, || None);

        let mut referenced: Vec<bool> = Vec::new();
        referenced
            .try_reserve_exact(capacity)
            .map_err(|_| alloc_err())?;
        referenced.resize(capacity, false);

        let mut index = HashMap::with_hasher(hash_builder);
        index.try_reserve(capacity).map_err(|_| alloc_err())?;

        Ok(Self {
            slots,
            referenced,
            index,
            hand: 0,
            len: 0,
            #[cfg(feature = "metrics")]
            sweep_hand_advances: 0,
            #[cfg(feature = "metrics")]
            sweep_ref_bit_resets: 0,
        })
    }

    /// Returns the configured capacity (number of slots).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let ring: ClockRing<&str, i32> = ClockRing::new(50);
    /// assert_eq!(ring.capacity(), 50);
    /// ```
    pub fn capacity(&self) -> usize {
        self.slots.len()
    }

    /// Reserves capacity for at least `additional` more index entries.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring: ClockRing<String, i32> = ClockRing::new(10);
    /// ring.reserve_index(100);  // Pre-allocate index capacity
    /// ```
    pub fn reserve_index(&mut self, additional: usize) {
        self.index.reserve(additional);
    }

    /// Shrinks internal storage to fit current contents.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring: ClockRing<&str, i32> = ClockRing::new(100);
    /// ring.insert("a", 1);
    /// ring.shrink_to_fit();
    /// ```
    pub fn shrink_to_fit(&mut self) {
        self.index.shrink_to_fit();
        self.slots.shrink_to_fit();
        self.referenced.shrink_to_fit();
    }

    /// Clears all entries without releasing capacity.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// ring.clear();
    /// assert!(ring.is_empty());
    /// assert!(!ring.contains(&"a"));
    /// ```
    pub fn clear(&mut self) {
        self.index.clear();
        for slot in &mut self.slots {
            *slot = None;
        }
        self.referenced.fill(false);
        self.len = 0;
        self.hand = 0;
        #[cfg(feature = "metrics")]
        {
            self.sweep_hand_advances = 0;
            self.sweep_ref_bit_resets = 0;
        }
    }

    /// Clears all entries and shrinks internal storage.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(100);
    /// ring.insert("a", 1);
    ///
    /// ring.clear_shrink();
    /// assert!(ring.is_empty());
    /// ```
    pub fn clear_shrink(&mut self) {
        self.clear();
        self.index.shrink_to_fit();
        self.slots.shrink_to_fit();
        self.referenced.shrink_to_fit();
    }

    /// Cumulative hand advances during sweep operations.
    #[cfg(feature = "metrics")]
    #[inline]
    pub fn sweep_hand_advances(&self) -> u64 {
        self.sweep_hand_advances
    }

    /// Cumulative reference-bit resets during sweep operations.
    #[cfg(feature = "metrics")]
    #[inline]
    pub fn sweep_ref_bit_resets(&self) -> u64 {
        self.sweep_ref_bit_resets
    }

    /// Returns an approximate memory footprint in bytes.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let ring: ClockRing<u64, u64> = ClockRing::new(100);
    /// let bytes = ring.approx_bytes();
    /// assert!(bytes > 0);
    /// ```
    #[must_use]
    pub fn approx_bytes(&self) -> usize {
        let index_bytes = self
            .index
            .capacity()
            .saturating_mul(std::mem::size_of::<(K, usize)>());
        let slots_bytes = self
            .slots
            .capacity()
            .saturating_mul(std::mem::size_of::<Option<Entry<K, V>>>());
        let ref_bytes = self
            .referenced
            .capacity()
            .saturating_mul(std::mem::size_of::<bool>());
        std::mem::size_of::<Self>()
            .saturating_add(index_bytes)
            .saturating_add(slots_bytes)
            .saturating_add(ref_bytes)
    }

    /// Returns the number of occupied slots.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// assert_eq!(ring.len(), 0);
    ///
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    /// assert_eq!(ring.len(), 2);
    /// ```
    pub fn len(&self) -> usize {
        self.len
    }

    /// Returns `true` if there are no entries.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring: ClockRing<&str, i32> = ClockRing::new(10);
    /// assert!(ring.is_empty());
    ///
    /// ring.insert("key", 42);
    /// assert!(!ring.is_empty());
    /// ```
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Returns `true` if `key` is present.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// ring.insert("key", 42);
    ///
    /// assert!(ring.contains(&"key"));
    /// assert!(!ring.contains(&"missing"));
    /// ```
    pub fn contains<Q>(&self, key: &Q) -> bool
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        self.index.contains_key(key)
    }

    /// Returns the value without setting the reference bit.
    ///
    /// Unlike [`get`](Self::get), this does not grant a second chance.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// ring.insert("key", 42);
    ///
    /// // peek doesn't set reference bit
    /// assert_eq!(ring.peek(&"key"), Some(&42));
    /// assert_eq!(ring.peek(&"missing"), None);
    /// ```
    pub fn peek<Q>(&self, key: &Q) -> Option<&V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = *self.index.get(key)?;
        self.slots.get(idx)?.as_ref().map(|entry| &entry.value)
    }

    /// Returns the value and sets the reference bit (grants second chance).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(2);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// // Access "a" - sets reference bit
    /// assert_eq!(ring.get(&"a"), Some(&1));
    ///
    /// // Insert "c" - "b" evicted (no ref bit), "a" survives
    /// ring.insert("c", 3);
    /// assert!(ring.contains(&"a"));
    /// assert!(!ring.contains(&"b"));
    /// ```
    #[must_use]
    pub fn get<Q>(&mut self, key: &Q) -> Option<&V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = *self.index.get(key)?;
        let entry = self.slots.get(idx)?.as_ref()?;
        self.referenced[idx] = true;
        Some(&entry.value)
    }

    /// Returns a mutable reference to the value and sets the reference bit.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// ring.insert("key", vec![1, 2, 3]);
    ///
    /// if let Some(v) = ring.get_mut(&"key") {
    ///     v.push(4);
    /// }
    /// assert_eq!(ring.peek(&"key"), Some(&vec![1, 2, 3, 4]));
    /// ```
    #[must_use]
    pub fn get_mut<Q>(&mut self, key: &Q) -> Option<&mut V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = *self.index.get(key)?;
        let slot = self.slots.get_mut(idx)?;
        if slot.is_none() {
            return None;
        }
        self.referenced[idx] = true;
        slot.as_mut().map(|entry| &mut entry.value)
    }

    /// Sets the reference bit for `key`; returns `false` if missing.
    ///
    /// Use this to mark an entry as recently accessed without retrieving its value.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(2);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// // Touch "a" without reading value
    /// assert!(ring.touch(&"a"));
    /// assert!(!ring.touch(&"missing"));
    ///
    /// // "a" survives eviction due to reference bit
    /// let evicted = ring.insert("c", 3);
    /// assert_eq!(evicted, Some(("b", 2)));
    /// ```
    pub fn touch<Q>(&mut self, key: &Q) -> bool
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = match self.index.get(key) {
            Some(idx) => *idx,
            None => return false,
        };
        match self.slots.get(idx) {
            Some(Some(_)) => {
                self.referenced[idx] = true;
                true
            },
            _ => false,
        }
    }

    /// Updates the value for an existing key, returning the old value.
    ///
    /// Sets the reference bit on update. Returns `None` if the key doesn't exist.
    /// This method never evicts - use [`insert`](Self::insert) for that.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(2);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// // Update existing key
    /// assert_eq!(ring.update(&"a", 10), Some(1));
    /// assert_eq!(ring.peek(&"a"), Some(&10));
    ///
    /// // Key doesn't exist - returns None
    /// assert_eq!(ring.update(&"missing", 99), None);
    /// ```
    pub fn update<Q>(&mut self, key: &Q, value: V) -> Option<V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = *self.index.get(key)?;
        let entry = self.slots.get_mut(idx)?.as_mut()?;
        let old = std::mem::replace(&mut entry.value, value);
        self.referenced[idx] = true;
        Some(old)
    }

    /// Inserts or updates `key`, evicting if necessary.
    ///
    /// Returns `Some((evicted_key, evicted_value))` when a full ring evicts an entry.
    /// Returns `None` in all other cases:
    /// - Inserted into an empty slot (no eviction needed)
    /// - Updated an existing key (old value is **dropped**, use [`update`](Self::update) to retrieve it)
    /// - Zero-capacity ring (entry is silently discarded)
    ///
    /// Key is cloned once per new insertion (stored in both the slot and the index).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(2);
    ///
    /// // Insert into empty slots - no eviction
    /// assert_eq!(ring.insert("a", 1), None);
    /// assert_eq!(ring.insert("b", 2), None);
    ///
    /// // Update existing - no eviction, old value dropped
    /// assert_eq!(ring.insert("a", 10), None);
    /// assert_eq!(ring.peek(&"a"), Some(&10));
    ///
    /// // Insert new when full - evicts unreferenced entry
    /// let evicted = ring.insert("c", 3);
    /// assert!(evicted.is_some());
    /// ```
    pub fn insert(&mut self, key: K, value: V) -> Option<(K, V)> {
        let (replaced, evicted) = self.insert_swap(key, value);
        drop(replaced);
        evicted
    }

    /// Lower-level sibling of [`insert`](Self::insert) that *returns*
    /// any value displaced by an update instead of dropping it inline.
    ///
    /// Semantics by state:
    /// - **Zero capacity**: returns `(None, None)` (value is discarded).
    /// - **Update of an existing key**: returns `(Some(old_value), None)`.
    /// - **Insert into an empty slot**: returns `(None, None)`.
    /// - **Insert into a full ring**: returns
    ///   `(None, Some((evicted_key, evicted_value)))`.
    /// - **Invariant violation** (the index maps `key` to a slot index
    ///   that is out of bounds for `slots`; reachable only via state
    ///   corruption, never from safe use of this module): returns
    ///   `(Some(value), None)` handing the caller's value back
    ///   untouched rather than silently dropping it. A
    ///   `debug_assert!` in the same branch surfaces the root-cause
    ///   corruption in debug/test/fuzz builds.
    ///
    /// This method is useful when the caller wants to control *when*
    /// the dropped value is destroyed — notably
    /// [`ConcurrentClockRing::insert`] uses it to run the replaced
    /// value's `Drop` **after** releasing the internal write lock, so a
    /// slow or reentrant `Drop for V` cannot deadlock or stall other
    /// threads.
    pub fn insert_swap(&mut self, key: K, value: V) -> (Option<V>, Option<(K, V)>) {
        if self.capacity() == 0 {
            return (None, None);
        }

        if let Some(&idx) = self.index.get(&key) {
            if let Some(entry) = self.slots.get_mut(idx).and_then(|slot| slot.as_mut()) {
                let old = std::mem::replace(&mut entry.value, value);
                self.referenced[idx] = true;
                return (Some(old), None);
            }
            // Broken invariant: the index claims `key` lives at `idx`
            // but the slot is empty. Rather than silently dropping the
            // caller's `value`, repair state in place: place the value
            // in the empty slot, refresh the index entry, and
            // recompute `len` from ground truth (slot occupancy). The
            // `debug_assert!` surfaces the root-cause corruption in
            // tests and fuzzers without crashing release builds.
            debug_assert!(false, "index entry points to empty slot in insert_swap");
            self.index.remove(&key);
            if idx < self.slots.len() {
                let entry_key = key.clone();
                self.slots[idx] = Some(Entry {
                    key: entry_key,
                    value,
                });
                if let Some(referenced) = self.referenced.get_mut(idx) {
                    *referenced = false;
                }
                self.index.insert(key, idx);
                self.len = self.slots.iter().filter(|slot| slot.is_some()).count();
                let cap = self.slots.len();
                self.hand = step(idx, cap);
                return (None, None);
            }
            // Deeper corruption: the index maps `key` to an out-of-bounds
            // slot index. No valid slot exists to install the value in,
            // so we hand it back via the `replaced` position rather than
            // silently dropping it. For `V` owning a file descriptor,
            // lock, `Arc<_>`, or any resource with observable Drop, a
            // silent drop would leak a resource across a single
            // corruption event without surfacing any error. Surfacing it
            // through the return keeps `insert_swap`'s
            // no-silent-value-loss contract intact; `ClockRing::insert`
            // will still drop it at the caller boundary (matching the
            // update path), and `ConcurrentClockRing::insert` will drop
            // it after releasing the write lock.
            return (Some(value), None);
        }

        if self.len < self.capacity() {
            let cap = self.capacity();
            let mut idx = self.hand % cap;
            for _ in 0..cap {
                if self.slots[idx].is_none() {
                    let entry_key = key.clone();
                    self.slots[idx] = Some(Entry {
                        key: entry_key,
                        value,
                    });
                    self.referenced[idx] = false;
                    self.index.insert(key, idx);
                    self.len = self.len.saturating_add(1);
                    self.hand = step(idx, cap);
                    return (None, None);
                }
                idx = step(idx, cap);
            }
            unreachable!("len < capacity but no empty slot found");
        }

        // Full — CLOCK sweep to find and evict a victim.
        // One pass clears all ref bits; the second finds an unreferenced slot.
        let cap = self.capacity();
        for _ in 0..cap.saturating_mul(2) {
            let idx = self.hand;
            if self.referenced[idx] {
                self.referenced[idx] = false;
                #[cfg(feature = "metrics")]
                {
                    self.sweep_ref_bit_resets = self.sweep_ref_bit_resets.saturating_add(1);
                }
                self.advance_hand();
                #[cfg(feature = "metrics")]
                {
                    self.sweep_hand_advances = self.sweep_hand_advances.saturating_add(1);
                }
                continue;
            }

            // Clone the new key *before* any destructive state change.
            // A panicking `K::Clone` (including OOM in a heap-allocating
            // Clone impl) reached from here must leave the ring
            // invariants intact — otherwise the next `insert` can land
            // on an empty slot with `len == capacity` and hit the
            // `unreachable!()` below, turning a transient allocation
            // failure into a persistent-panic DoS for every caller
            // sharing this ring (notably `ConcurrentClockRing`).
            let entry_key = key.clone();

            // Remove the evictee's index entry **before** emptying the
            // slot. `HashMap::remove` calls the user-supplied `Hash` /
            // `Eq` on the evictee key; if either panics we must not
            // have mutated the slot yet. Pre-fix, the slot was taken
            // first and a panic here left `len == capacity` with one
            // empty slot — the next insert then fired the
            // `unreachable!("occupied slot missing under full ring")`
            // below, turning a transient hasher panic into a
            // persistent-panic DoS shared across every thread holding
            // a `ConcurrentClockRing` handle.
            let victim_key_ref = match &self.slots[idx] {
                Some(entry) => &entry.key,
                None => unreachable!("occupied slot missing under full ring"),
            };
            let removed = self.index.remove(victim_key_ref);
            debug_assert_eq!(
                removed,
                Some(idx),
                "index/slot disagreement in insert_swap eviction"
            );

            // From here on every step is infallible until the final
            // `index.insert`: take + install + clear-ref bit.
            let evicted = match self.slots[idx].take() {
                Some(entry) => entry,
                // Invariant: we just confirmed occupancy above, so
                // this cannot happen.
                None => unreachable!("slot vanished between borrow and take"),
            };

            self.slots[idx] = Some(Entry {
                key: entry_key,
                value,
            });
            self.referenced[idx] = false;
            // The index was pre-sized at construction via
            // `try_reserve(capacity)` and `len` never exceeds
            // `capacity`, so this `insert` does not rehash and cannot
            // fail an allocation here. If a user-supplied hasher
            // nevertheless panics, the slot is already occupied with
            // the new entry and the index is missing only the new
            // key's pointer; subsequent lookups via the absent key
            // return `None` and `insert_swap`'s repair path restores
            // the mapping on the next write of the same key.
            self.index.insert(key, idx);
            self.advance_hand();
            #[cfg(feature = "metrics")]
            {
                self.sweep_hand_advances = self.sweep_hand_advances.saturating_add(1);
            }
            return (None, Some((evicted.key, evicted.value)));
        }
        unreachable!("insert sweep exceeded 2*capacity without finding victim");
    }

    /// Peeks the next eviction candidate without modifying state.
    ///
    /// Scans from the hand position to find the first unreferenced entry.
    /// Returns `None` if the ring is empty **or** every occupied entry has its
    /// reference bit set (unlike [`pop_victim`](Self::pop_victim), this method
    /// does not clear reference bits).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(2);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    /// ring.touch(&"a");  // "a" now has reference bit
    ///
    /// // "b" is the next victim (no reference bit)
    /// let victim = ring.peek_victim();
    /// assert_eq!(victim, Some((&"b", &2)));
    /// ```
    #[must_use]
    pub fn peek_victim(&self) -> Option<(&K, &V)> {
        if self.capacity() == 0 || self.len == 0 {
            return None;
        }
        let cap = self.capacity();
        let mut idx = self.hand % cap;
        for _ in 0..cap {
            if let Some(entry) = self.slots.get(idx).and_then(|slot| slot.as_ref()) {
                if !self.referenced[idx] {
                    return Some((&entry.key, &entry.value));
                }
            }
            idx = step(idx, cap);
        }
        None
    }

    /// Evicts the next candidate (first unreferenced slot) and returns it.
    ///
    /// Clears reference bits as it scans, giving referenced entries a second
    /// chance. Uses a `2 × capacity` sweep budget so that even when every entry
    /// is referenced, all bits are cleared in the first pass and a victim is
    /// found in the second.
    ///
    /// Returns `None` only when the ring is empty.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    /// ring.insert("c", 3);
    ///
    /// // Evict one entry
    /// let evicted = ring.pop_victim();
    /// assert!(evicted.is_some());
    /// assert_eq!(ring.len(), 2);
    /// ```
    pub fn pop_victim(&mut self) -> Option<(K, V)> {
        if self.capacity() == 0 || self.len == 0 {
            return None;
        }
        let cap = self.capacity();
        for _ in 0..cap.saturating_mul(2) {
            let idx = self.hand;
            if self.slots[idx].is_some() {
                if self.referenced[idx] {
                    self.referenced[idx] = false;
                    #[cfg(feature = "metrics")]
                    {
                        self.sweep_ref_bit_resets = self.sweep_ref_bit_resets.saturating_add(1);
                    }
                    self.advance_hand();
                    #[cfg(feature = "metrics")]
                    {
                        self.sweep_hand_advances = self.sweep_hand_advances.saturating_add(1);
                    }
                    continue;
                }

                // Remove the index entry **before** emptying the slot,
                // so a panicking `K::Hash` / `K::Eq` in
                // `HashMap::remove` leaves slot/len/referenced
                // untouched. Pre-fix, `slots[idx].take()` ran first
                // and a panic here left `len == capacity` with an
                // empty slot, which weaponized the next insert's
                // `unreachable!("occupied slot missing under full
                // ring")` into a persistent-panic DoS.
                let victim_key_ref = match &self.slots[idx] {
                    Some(entry) => &entry.key,
                    None => unreachable!("slot reported occupied but is empty"),
                };
                let removed = self.index.remove(victim_key_ref);
                debug_assert_eq!(removed, Some(idx), "index/slot disagreement in pop_victim");

                // From here on, every step is infallible: take + set
                // + decrement + advance.
                let evicted = match self.slots[idx].take() {
                    Some(entry) => entry,
                    None => unreachable!("slot reported occupied but take() returned None"),
                };
                self.referenced[idx] = false;
                self.len = self
                    .len
                    .checked_sub(1)
                    .unwrap_or_else(|| unreachable!("len underflow in pop_victim"));
                self.advance_hand();
                #[cfg(feature = "metrics")]
                {
                    self.sweep_hand_advances = self.sweep_hand_advances.saturating_add(1);
                }
                return Some((evicted.key, evicted.value));
            }
            self.advance_hand();
            #[cfg(feature = "metrics")]
            {
                self.sweep_hand_advances = self.sweep_hand_advances.saturating_add(1);
            }
        }
        None
    }

    #[cfg(any(test, debug_assertions))]
    /// Returns a debug snapshot of slot occupancy in ring order.
    pub fn debug_snapshot_slots(&self) -> Vec<Option<(&K, bool)>> {
        self.slots
            .iter()
            .enumerate()
            .map(|(idx, slot)| {
                slot.as_ref()
                    .map(|entry| (&entry.key, self.referenced[idx]))
            })
            .collect()
    }

    /// Removes `key` and returns its value, if present.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(10);
    /// ring.insert("key", 42);
    ///
    /// assert_eq!(ring.remove(&"key"), Some(42));
    /// assert_eq!(ring.remove(&"key"), None);  // Already removed
    /// assert!(!ring.contains(&"key"));
    /// ```
    pub fn remove<Q>(&mut self, key: &Q) -> Option<V>
    where
        K: Borrow<Q>,
        Q: Hash + Eq + ?Sized,
    {
        let idx = self.index.remove(key)?;
        let entry = self.slots.get_mut(idx)?.take()?;
        self.referenced[idx] = false;
        self.len = self
            .len
            .checked_sub(1)
            .unwrap_or_else(|| unreachable!("len underflow in remove"));
        Some(entry.value)
    }

    /// Callers must ensure capacity > 0 before calling.
    fn advance_hand(&mut self) {
        self.hand = step(self.hand, self.slots.len());
    }

    #[cfg(any(test, debug_assertions))]
    pub fn debug_validate_invariants(&self) {
        let slot_count = self.slots.iter().filter(|slot| slot.is_some()).count();
        assert_eq!(self.len, slot_count);
        assert_eq!(self.len, self.index.len());
        assert_eq!(self.referenced.len(), self.slots.len());

        if self.capacity() == 0 {
            assert_eq!(self.hand, 0);
        } else {
            assert!(self.hand < self.capacity());
        }

        for (key, &idx) in &self.index {
            assert!(idx < self.slots.len());
            let entry = self.slots[idx]
                .as_ref()
                .expect("index points to empty slot");
            assert!(&entry.key == key);
        }

        for idx in 0..self.slots.len() {
            if self.slots[idx].is_none() {
                assert!(!self.referenced[idx], "empty slot has referenced bit set");
            }
        }
    }
}

/// Bulk-inserts `(K, V)` pairs via repeated [`ClockRing::insert`].
///
/// # Memory-budget caveat
///
/// `ClockRing` bounds the **number** of entries, not the aggregate
/// byte footprint of stored values — see the `Memory Budgeting`
/// section of the module docs. When `V` owns heap allocations (e.g.
/// `Vec<u8>`, `String`) and the iterator is attacker-influenced, this
/// impl will happily replace already-stored values with arbitrarily
/// large new ones up to `capacity` times before any single eviction
/// returns, so aggregate memory use is bounded only by `capacity *
/// max(size_of(V_i))` across the iterator. Evicted entries are
/// **silently dropped** — the caller never sees them.
///
/// For attacker-influenced values, enforce a byte budget at the call
/// site (track `value.len()` on insert and evict via
/// [`ClockRing::pop_victim`] until under budget) instead of using
/// [`Extend`].
impl<K, V, S> Extend<(K, V)> for ClockRing<K, V, S>
where
    K: Eq + Hash + Clone,
    S: BuildHasher,
{
    fn extend<I: IntoIterator<Item = (K, V)>>(&mut self, iter: I) {
        for (key, value) in iter {
            let _ = self.insert(key, value);
        }
    }
}

// ---------------------------------------------------------------------------
// Clone — manual impl so that behavior (and its failure mode) is explicit.
// ---------------------------------------------------------------------------

impl<K, V, S> Clone for ClockRing<K, V, S>
where
    K: Clone,
    V: Clone,
    S: Clone,
{
    /// Clones the ring using the global allocator's **infallible** paths.
    ///
    /// # Allocator behavior
    ///
    /// Unlike [`try_new`](Self::try_new) /
    /// [`try_with_hasher`](Self::try_with_hasher), which surface
    /// allocator failure as [`ClockRingError::AllocationFailed`], this
    /// impl delegates to [`Vec::clone`] and [`HashMap::clone`] which
    /// route through the infallible allocation path. On OOM the
    /// global alloc-error handler fires — aborting the process by
    /// default. Callers that reach `clone` with attacker-influenced
    /// capacity should prefer [`try_clone`](Self::try_clone), which
    /// uses `try_reserve_exact` / `try_reserve` and returns an error
    /// instead of aborting.
    fn clone(&self) -> Self {
        Self {
            slots: self.slots.clone(),
            referenced: self.referenced.clone(),
            index: self.index.clone(),
            hand: self.hand,
            len: self.len,
            #[cfg(feature = "metrics")]
            sweep_hand_advances: self.sweep_hand_advances,
            #[cfg(feature = "metrics")]
            sweep_ref_bit_resets: self.sweep_ref_bit_resets,
        }
    }
}

impl<K, V, S> ClockRing<K, V, S>
where
    K: Eq + Hash + Clone,
    V: Clone,
    S: Clone + BuildHasher,
{
    /// Fallible clone that surfaces allocator failure as
    /// [`ClockRingError::AllocationFailed`] rather than aborting the
    /// process.
    ///
    /// Matches the [`try_new`](Self::try_new) /
    /// [`try_with_hasher`](Self::try_with_hasher) contract: every
    /// backing allocation is performed via
    /// [`Vec::try_reserve_exact`] / [`HashMap::try_reserve`], so no
    /// step falls back to the infallible alloc-error handler.
    ///
    /// Prefer this over [`Clone::clone`] when the ring's capacity is
    /// attacker-influenced (see the `Clone` impl's docs for the
    /// abort-on-OOM trade-off it inherits from `Vec`/`HashMap`).
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(4);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// let copy = ring.try_clone().unwrap();
    /// assert_eq!(copy.len(), ring.len());
    /// assert_eq!(copy.peek(&"a"), Some(&1));
    /// ```
    pub fn try_clone(&self) -> Result<Self, ClockRingError> {
        let cap = self.capacity();
        let alloc_err = || ClockRingError::AllocationFailed { requested: cap };

        let mut slots: Vec<Option<Entry<K, V>>> = Vec::new();
        slots.try_reserve_exact(cap).map_err(|_| alloc_err())?;
        for slot in &self.slots {
            slots.push(slot.clone());
        }

        let mut referenced: Vec<bool> = Vec::new();
        referenced.try_reserve_exact(cap).map_err(|_| alloc_err())?;
        referenced.extend_from_slice(&self.referenced);

        let mut index = HashMap::with_hasher(self.index.hasher().clone());
        index.try_reserve(cap).map_err(|_| alloc_err())?;
        for (k, &v) in &self.index {
            index.insert(k.clone(), v);
        }

        Ok(Self {
            slots,
            referenced,
            index,
            hand: self.hand,
            len: self.len,
            #[cfg(feature = "metrics")]
            sweep_hand_advances: self.sweep_hand_advances,
            #[cfg(feature = "metrics")]
            sweep_ref_bit_resets: self.sweep_ref_bit_resets,
        })
    }
}

// ---------------------------------------------------------------------------
// Iterator types (C-ITER-TY: names match the methods that produce them)
// ---------------------------------------------------------------------------

/// Iterator over `(&K, &V)` pairs of a [`ClockRing`].
///
/// Created by [`ClockRing::iter`].
#[derive(Debug)]
pub struct Iter<'a, K, V> {
    inner: std::slice::Iter<'a, Option<Entry<K, V>>>,
}

impl<'a, K, V> Iterator for Iter<'a, K, V> {
    type Item = (&'a K, &'a V);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next() {
                Some(Some(entry)) => return Some((&entry.key, &entry.value)),
                Some(None) => continue,
                None => return None,
            }
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, self.inner.size_hint().1)
    }
}

/// Mutable iterator over `(&K, &mut V)` pairs of a [`ClockRing`].
///
/// Created by [`ClockRing::iter_mut`].
#[derive(Debug)]
pub struct IterMut<'a, K, V> {
    inner: std::slice::IterMut<'a, Option<Entry<K, V>>>,
}

impl<'a, K, V> Iterator for IterMut<'a, K, V> {
    type Item = (&'a K, &'a mut V);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next() {
                Some(Some(entry)) => return Some((&entry.key, &mut entry.value)),
                Some(None) => continue,
                None => return None,
            }
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, self.inner.size_hint().1)
    }
}

/// Owning iterator over `(K, V)` pairs of a [`ClockRing`].
///
/// Created by calling [`IntoIterator::into_iter`] on a `ClockRing`
/// (or equivalently, `for (k, v) in ring { ... }`).
#[derive(Debug)]
pub struct IntoIter<K, V> {
    inner: std::vec::IntoIter<Option<Entry<K, V>>>,
}

impl<K, V> Iterator for IntoIter<K, V> {
    type Item = (K, V);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next() {
                Some(Some(entry)) => return Some((entry.key, entry.value)),
                Some(None) => continue,
                None => return None,
            }
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, self.inner.size_hint().1)
    }
}

/// Iterator over keys of a [`ClockRing`].
///
/// Created by [`ClockRing::keys`].
#[derive(Debug)]
pub struct Keys<'a, K, V> {
    inner: Iter<'a, K, V>,
}

impl<'a, K, V> Iterator for Keys<'a, K, V> {
    type Item = &'a K;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|(k, _)| k)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.inner.size_hint()
    }
}

/// Iterator over values of a [`ClockRing`].
///
/// Created by [`ClockRing::values`].
#[derive(Debug)]
pub struct Values<'a, K, V> {
    inner: Iter<'a, K, V>,
}

impl<'a, K, V> Iterator for Values<'a, K, V> {
    type Item = &'a V;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|(_, v)| v)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.inner.size_hint()
    }
}

/// Mutable iterator over values of a [`ClockRing`].
///
/// Created by [`ClockRing::values_mut`].
#[derive(Debug)]
pub struct ValuesMut<'a, K, V> {
    inner: IterMut<'a, K, V>,
}

impl<'a, K, V> Iterator for ValuesMut<'a, K, V> {
    type Item = &'a mut V;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|(_, v)| v)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.inner.size_hint()
    }
}

// ---------------------------------------------------------------------------
// IntoIterator impls (C-ITER: iter, iter_mut, into_iter)
// ---------------------------------------------------------------------------

impl<K, V, S> IntoIterator for ClockRing<K, V, S> {
    type Item = (K, V);
    type IntoIter = IntoIter<K, V>;

    /// Consumes the ring, returning an iterator over all `(K, V)` pairs.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ClockRing;
    ///
    /// let mut ring = ClockRing::new(3);
    /// ring.insert("a", 1);
    /// ring.insert("b", 2);
    ///
    /// let pairs: Vec<_> = ring.into_iter().collect();
    /// assert_eq!(pairs.len(), 2);
    /// ```
    fn into_iter(self) -> Self::IntoIter {
        IntoIter {
            inner: self.slots.into_iter(),
        }
    }
}

impl<'a, K, V, S> IntoIterator for &'a ClockRing<K, V, S> {
    type Item = (&'a K, &'a V);
    type IntoIter = Iter<'a, K, V>;

    fn into_iter(self) -> Self::IntoIter {
        self.iter()
    }
}

impl<'a, K, V, S> IntoIterator for &'a mut ClockRing<K, V, S> {
    type Item = (&'a K, &'a mut V);
    type IntoIter = IterMut<'a, K, V>;

    fn into_iter(self) -> Self::IntoIter {
        self.iter_mut()
    }
}

// ---------------------------------------------------------------------------
// ConcurrentClockRing::into_inner (C-CONV: into_ for owned → owned)
// ---------------------------------------------------------------------------

#[cfg(feature = "concurrency")]
impl<K, V, S> ConcurrentClockRing<K, V, S> {
    /// Consumes the concurrent wrapper, returning the inner [`ClockRing`].
    ///
    /// This is useful when you need to iterate or inspect a concurrent ring
    /// after all shared references have been dropped.
    ///
    /// # Example
    ///
    /// ```
    /// use cachekit::ds::ConcurrentClockRing;
    ///
    /// let cache = ConcurrentClockRing::new(10);
    /// cache.insert("a", 1);
    /// cache.insert("b", 2);
    ///
    /// let ring = cache.into_inner();
    /// let pairs: Vec<_> = ring.iter().collect();
    /// assert_eq!(pairs.len(), 2);
    /// ```
    pub fn into_inner(self) -> ClockRing<K, V, S> {
        self.inner.into_inner()
    }
}

#[cfg(test)]
#[allow(unused_must_use)]
mod tests {
    use super::*;

    #[test]
    fn step_helper_wraps_and_handles_usize_max() {
        assert_eq!(step(0, 4), 1);
        assert_eq!(step(3, 4), 0);
        // Overflow path: idx + 1 would overflow usize::MAX. step must
        // fall back to 0 rather than panic or wrap silently.
        assert_eq!(step(usize::MAX, 4), 0);
    }

    #[test]
    fn step_helper_is_total_for_zero_capacity() {
        // step must be safe to call with cap == 0 (returns 0) rather
        // than panicking on `next % 0`. Current callers always guard
        // on non-zero capacity, but step itself should be total so
        // future callers cannot trip a release-mode panic.
        assert_eq!(step(0, 0), 0);
        assert_eq!(step(42, 0), 0);
        assert_eq!(step(usize::MAX, 0), 0);
    }

    #[test]
    fn try_new_rejects_capacity_above_max() {
        let err = ClockRing::<u32, u32>::try_new(MAX_CAPACITY + 1).unwrap_err();
        match err {
            ClockRingError::CapacityTooLarge { requested, max } => {
                assert_eq!(requested, MAX_CAPACITY + 1);
                assert_eq!(max, MAX_CAPACITY);
            },
            other => panic!("expected CapacityTooLarge, got {other:?}"),
        }

        // Happy path: small capacity allocates cleanly.
        let ring = ClockRing::<u32, u32>::try_new(8).unwrap();
        assert_eq!(ring.capacity(), 8);
    }

    #[test]
    fn try_new_returns_allocation_failed_on_hostile_capacity() {
        // `[u8; 1024]` keeps `Option<Entry<K, V>>` comfortably above 64 B,
        // so `MAX_CAPACITY * size_of::<Option<Entry<K, V>>>()` exceeds
        // `isize::MAX`. Pre-`try_reserve_exact` this would abort the
        // process via `capacity_overflow`; we now surface it as a
        // regular `ClockRingError::AllocationFailed`.
        let err = ClockRing::<u32, [u8; 1024]>::try_new(MAX_CAPACITY).unwrap_err();
        match err {
            ClockRingError::AllocationFailed { requested } => {
                assert_eq!(requested, MAX_CAPACITY);
            },
            other => panic!("expected AllocationFailed, got {other:?}"),
        }
    }

    #[test]
    #[cfg(not(debug_assertions))]
    fn insert_swap_repairs_stale_index_instead_of_dropping_value() {
        // Security hardening regression: if the index maps a key to
        // an idx but the slot at idx is empty (invariant broken by a
        // malformed `Hash`/`Eq`/`Clone` impl on `K`, or a future
        // refactor), the previous implementation silently dropped the
        // caller's value. The hardened path repairs state in place.
        //
        // Gated on release builds because the repair path is preceded
        // by a `debug_assert!(false, ...)` that fires in debug to
        // surface the root-cause corruption during testing.
        let mut ring: ClockRing<&str, i32> = ClockRing::new(4);
        ring.insert("a", 1);
        let idx = *ring.index.get("a").unwrap();

        // Manually corrupt state: clear the slot but leave the index
        // entry, mimicking a broken-invariant scenario.
        ring.slots[idx] = None;

        let (replaced, evicted) = ring.insert_swap("a", 99);
        assert_eq!(replaced, None);
        assert_eq!(evicted, None);
        // The value we passed in must not have been silently dropped.
        assert_eq!(ring.peek(&"a"), Some(&99));
        ring.debug_validate_invariants();
    }

    #[test]
    #[cfg(not(debug_assertions))]
    fn insert_swap_returns_value_when_index_points_out_of_bounds() {
        // Security hardening regression: if the index maps a key to an
        // out-of-bounds slot idx (state corruption reachable only via a
        // malformed `Hash`/`Eq`/`Clone` impl on `K` or unsafe code —
        // never from safe use of this module), the previous
        // implementation silently dropped the caller's value. With `V`
        // owning a file descriptor, lock, or `Arc<_>`, that would leak
        // the resource without surfacing any error. The hardened path
        // now returns `(Some(value), None)` so the caller receives
        // their value back via the replaced-value position.
        //
        // Gated on release builds because the fallthrough path is
        // preceded by a `debug_assert!(false, ...)` that fires in debug
        // to surface the root-cause corruption during testing.
        let mut ring: ClockRing<&str, i32> = ClockRing::new(4);
        ring.insert("a", 1);

        // Manually corrupt state: rewrite the index entry to point past
        // the end of `slots`, simulating an invariant violation. The
        // slot-side entry is cleared so slot occupancy still matches
        // `len == 0` post-remove, keeping the rest of the ring
        // internally consistent.
        ring.index.insert("a", ring.slots.len() + 5);
        ring.slots[0] = None;
        ring.len = 0;

        let (replaced, evicted) = ring.insert_swap("a", 99);
        assert_eq!(
            replaced,
            Some(99),
            "value must be handed back via the replaced position \
             rather than silently dropped on invariant violation",
        );
        assert_eq!(evicted, None);
        // The corruption itself isn't fully repaired (no valid slot to
        // install into), but `insert_swap` removed the stale index
        // entry so subsequent operations don't observe the OOB mapping.
        assert!(!ring.contains(&"a"));
    }

    #[test]
    fn debug_impl_does_not_leak_keys_or_values() {
        // Security hardening: the derived `Debug` for `ClockRing`
        // recursed through every stored `Entry<K, V>` and
        // `HashMap<K, usize, S>` entry, turning `tracing::debug!`,
        // `dbg!`, and panic backtraces into a secret-exposure channel
        // for caches keyed on session tokens / API keys or holding
        // sensitive values. The hand-written `Debug` redacts contents.
        let mut ring: ClockRing<&str, &str> = ClockRing::new(4);
        ring.insert("session-token-key-do-not-leak", "bearer-secret-do-not-leak");
        ring.insert("api-key-do-not-leak", "plaintext-password-do-not-leak");

        let rendered = format!("{ring:?}");
        assert!(
            !rendered.contains("do-not-leak"),
            "Debug impl leaked key or value into output: {rendered}",
        );

        // Safety: still expose aggregate shape for ops dashboards /
        // snapshot tests.
        assert!(rendered.contains("ClockRing"));
        assert!(rendered.contains("len"));
        assert!(rendered.contains("capacity"));
        assert!(rendered.contains("hand"));
    }

    #[cfg(feature = "concurrency")]
    #[test]
    fn concurrent_debug_impl_does_not_leak_keys_or_values() {
        let cache: ConcurrentClockRing<&str, &str> = ConcurrentClockRing::new(4);
        cache.insert("session-token-key-do-not-leak", "bearer-secret-do-not-leak");
        cache.insert("api-key-do-not-leak", "plaintext-password-do-not-leak");

        let rendered = format!("{cache:?}");
        assert!(
            !rendered.contains("do-not-leak"),
            "Debug impl leaked key or value into output: {rendered}",
        );
        assert!(rendered.contains("ConcurrentClockRing"));
        assert!(rendered.contains("len"));
        assert!(rendered.contains("capacity"));
    }

    #[test]
    fn get_does_not_set_ref_bit_on_missing_key() {
        // Regression: get/touch/get_mut must not touch the reference
        // vector when the key is absent or its slot is empty.
        let mut ring = ClockRing::<&str, i32>::new(4);
        ring.insert("a", 1);
        // `a` was just inserted, so its ref bit is clear. A failed
        // lookup must not silently promote any slot.
        assert!(ring.get(&"missing").is_none());
        assert!(!ring.touch(&"missing"));
        assert!(ring.get_mut(&"missing").is_none());
        // `a` still has ref bit cleared → it's a valid victim candidate.
        assert_eq!(ring.peek_victim(), Some((&"a", &1)));
    }

    #[test]
    #[should_panic(expected = "capacity exceeds MAX_CAPACITY")]
    fn new_panics_on_oversized_capacity() {
        let _ = ClockRing::<u32, u32>::new(MAX_CAPACITY + 1);
    }

    #[test]
    fn approx_bytes_saturates_without_panicking() {
        // Just exercise the saturating path on a normal ring; the main
        // guarantee is that we don't panic in debug builds.
        let ring: ClockRing<u64, u64> = ClockRing::new(128);
        let bytes = ring.approx_bytes();
        assert!(bytes > 0);
    }

    #[test]
    fn with_hasher_uses_custom_hasher() {
        use std::collections::hash_map::RandomState;
        // Build with an explicit RandomState instance. Exercises the
        // generic `with_hasher` path end-to-end, including the
        // `KeysAreTrusted` acknowledgement.
        let mut ring: ClockRing<&str, i32, RandomState> =
            ClockRing::with_hasher(4, RandomState::new(), KeysAreTrusted::new());
        assert_eq!(ring.insert("a", 1), None);
        assert_eq!(ring.insert("b", 2), None);
        assert_eq!(ring.peek(&"a"), Some(&1));
        assert_eq!(ring.peek(&"b"), Some(&2));
    }

    #[test]
    fn try_with_hasher_requires_ack_and_builds() {
        use std::collections::hash_map::RandomState;
        // The `KeysAreTrusted` marker is required at the call site
        // — this test is the ergonomics regression: `::new()` is all
        // that's needed, no unsafe blocks, no separate builder step.
        let ring: ClockRing<u32, u32, RandomState> =
            ClockRing::try_with_hasher(8, RandomState::new(), KeysAreTrusted::new()).unwrap();
        assert_eq!(ring.capacity(), 8);

        // Equivalent via `Default`, for callers who prefer that form.
        let ring2: ClockRing<u32, u32, RandomState> =
            ClockRing::try_with_hasher(8, RandomState::new(), KeysAreTrusted::default()).unwrap();
        assert_eq!(ring2.capacity(), 8);
    }

    #[test]
    fn clock_ring_eviction_prefers_unreferenced() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.touch(&"a");
        let evicted = ring.insert("c", 3);

        assert_eq!(evicted, Some(("b", 2)));
        assert!(ring.contains(&"a"));
        assert!(ring.contains(&"c"));
    }

    #[test]
    fn clock_ring_zero_capacity_is_noop() {
        let mut ring = ClockRing::<&str, i32>::new(0);
        assert!(ring.is_empty());
        assert_eq!(ring.capacity(), 0);
        assert_eq!(ring.insert("a", 1), None);
        assert!(ring.is_empty());
        assert!(ring.peek(&"a").is_none());
        assert!(ring.get(&"a").is_none());
        assert!(!ring.contains(&"a"));
    }

    #[test]
    fn clock_ring_insert_and_peek_no_eviction() {
        let mut ring = ClockRing::new(3);
        assert_eq!(ring.insert("a", 1), None);
        assert_eq!(ring.insert("b", 2), None);
        assert_eq!(ring.insert("c", 3), None);
        assert_eq!(ring.len(), 3);
        assert!(ring.contains(&"a"));
        assert!(ring.contains(&"b"));
        assert!(ring.contains(&"c"));
        assert_eq!(ring.peek(&"a"), Some(&1));
        assert_eq!(ring.peek(&"b"), Some(&2));
        assert_eq!(ring.peek(&"c"), Some(&3));
    }

    #[test]
    fn clock_ring_update_existing_key_does_not_grow() {
        let mut ring = ClockRing::new(2);
        assert_eq!(ring.insert("a", 1), None);
        assert_eq!(ring.insert("b", 2), None);
        assert_eq!(ring.len(), 2);

        assert_eq!(ring.insert("a", 10), None);
        assert_eq!(ring.len(), 2);
        assert_eq!(ring.peek(&"a"), Some(&10));
    }

    #[test]
    fn clock_ring_update_returns_old_value() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        // Update existing key returns old value
        assert_eq!(ring.update(&"a", 10), Some(1));
        assert_eq!(ring.peek(&"a"), Some(&10));
        assert_eq!(ring.len(), 2);

        // Update non-existent key returns None
        assert_eq!(ring.update(&"missing", 99), None);
        assert_eq!(ring.len(), 2);

        // Update sets reference bit - verify by eviction
        let mut ring2 = ClockRing::new(2);
        ring2.insert("x", 1);
        ring2.insert("y", 2);
        ring2.update(&"x", 10); // Sets ref bit on x
        let evicted = ring2.insert("z", 3);
        assert_eq!(evicted, Some(("y", 2))); // y evicted, not x
    }

    #[test]
    fn clock_ring_get_sets_referenced_and_eviction_skips_it() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        assert_eq!(ring.get(&"a"), Some(&1));
        let evicted = ring.insert("c", 3);

        assert_eq!(evicted, Some(("b", 2)));
        assert!(ring.contains(&"a"));
        assert!(ring.contains(&"c"));
        assert!(!ring.contains(&"b"));
    }

    #[test]
    fn clock_ring_touch_marks_referenced() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        assert!(ring.touch(&"b"));

        let evicted = ring.insert("c", 3);
        assert_eq!(evicted, Some(("a", 1)));
        assert!(ring.contains(&"b"));
        assert!(ring.contains(&"c"));
    }

    #[test]
    fn clock_ring_remove_clears_slot_and_updates_len() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);
        assert_eq!(ring.len(), 3);

        assert_eq!(ring.remove(&"b"), Some(2));
        assert_eq!(ring.len(), 2);
        assert!(!ring.contains(&"b"));
        assert!(ring.peek(&"b").is_none());

        let evicted = ring.insert("d", 4);
        assert_eq!(evicted, None, "ring has free slot, must not evict");
        assert!(ring.contains(&"d"));
        assert!(!ring.contains(&"b"));
        assert_eq!(ring.len(), 3);
    }

    #[test]
    fn clock_ring_eviction_cycles_with_hand_wrap() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);

        let evicted1 = ring.insert("c", 3);
        assert!(matches!(evicted1, Some(("a", 1)) | Some(("b", 2))));
        assert_eq!(ring.len(), 2);

        let evicted2 = ring.insert("d", 4);
        assert!(matches!(
            evicted2,
            Some(("a", 1)) | Some(("b", 2)) | Some(("c", 3))
        ));
        assert_eq!(ring.len(), 2);
        assert!(ring.contains(&"d"));
    }

    #[test]
    fn clock_ring_debug_invariants_hold() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.get(&"a");
        ring.insert("c", 3);
        ring.remove(&"b");
        ring.debug_validate_invariants();
    }

    #[test]
    fn clock_ring_peek_and_pop_victim() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);

        // All entries are inserted unreferenced; any is a valid victim.
        let peeked = ring.peek_victim();
        assert!(matches!(
            peeked,
            Some((&"a", &1)) | Some((&"b", &2)) | Some((&"c", &3))
        ));

        let evicted = ring.pop_victim();
        assert!(matches!(
            evicted,
            Some(("a", 1)) | Some(("b", 2)) | Some(("c", 3))
        ));
        assert_eq!(ring.len(), 2);
        ring.debug_validate_invariants();
    }

    #[test]
    fn clock_ring_peek_skips_referenced_entries() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.touch(&"a");

        let peeked = ring.peek_victim();
        assert_eq!(peeked, Some((&"b", &2)));
    }

    #[test]
    fn clock_ring_pop_victim_clears_referenced_then_eviction() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.touch(&"a");
        ring.touch(&"b");

        // 2*cap sweep budget: clears all refs, then evicts
        let evicted = ring.pop_victim();
        assert!(matches!(evicted, Some(("a", 1)) | Some(("b", 2))));
        assert_eq!(ring.len(), 1);
    }

    #[test]
    fn clock_ring_debug_snapshot_slots() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.touch(&"a");
        let snapshot = ring.debug_snapshot_slots();
        assert_eq!(snapshot.len(), 2);
        assert!(
            snapshot
                .iter()
                .any(|slot| matches!(slot, &Some((&"a", true))))
        );
    }

    #[test]
    fn clock_ring_pop_victim_all_referenced_evicts_in_one_call() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);
        ring.touch(&"a");
        ring.touch(&"b");
        ring.touch(&"c");

        // 2*cap sweep budget: clears all refs then evicts in one call
        let evicted = ring.pop_victim();
        assert!(evicted.is_some());
        assert_eq!(ring.len(), 2);
        ring.debug_validate_invariants();
    }

    #[cfg(feature = "concurrency")]
    #[test]
    fn concurrent_clock_ring_try_ops() {
        let ring = ConcurrentClockRing::new(2);
        assert_eq!(ring.try_insert("a", 1), Some(None));
        assert_eq!(ring.try_peek_with(&"a", |v| *v), Some(Some(1)));
        assert_eq!(ring.try_get_with(&"a", |v| *v), Some(Some(1)));
        assert_eq!(ring.try_touch(&"a"), Some(true));
        assert!(ring.try_clear());
        assert!(ring.is_empty());
    }

    // -----------------------------------------------------------------------
    // Iterator tests
    // -----------------------------------------------------------------------

    #[test]
    fn iter_yields_all_occupied_entries() {
        let mut ring = ClockRing::new(5);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);

        let mut pairs: Vec<_> = ring.iter().collect();
        pairs.sort_by_key(|&(k, _)| *k);
        assert_eq!(pairs, vec![(&"a", &1), (&"b", &2), (&"c", &3)]);
    }

    #[test]
    fn iter_skips_empty_slots_after_removal() {
        let mut ring = ClockRing::new(5);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);
        ring.remove(&"b");

        let mut pairs: Vec<_> = ring.iter().collect();
        pairs.sort_by_key(|&(k, _)| *k);
        assert_eq!(pairs, vec![(&"a", &1), (&"c", &3)]);
    }

    #[test]
    fn iter_on_empty_ring() {
        let ring = ClockRing::<&str, i32>::new(5);
        assert_eq!(ring.iter().count(), 0);
    }

    #[test]
    fn iter_on_zero_capacity() {
        let ring = ClockRing::<&str, i32>::new(0);
        assert_eq!(ring.iter().count(), 0);
    }

    #[test]
    fn iter_mut_modifies_values() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        for (_, v) in ring.iter_mut() {
            *v += 100;
        }

        assert_eq!(ring.peek(&"a"), Some(&101));
        assert_eq!(ring.peek(&"b"), Some(&102));
    }

    #[test]
    fn iter_mut_does_not_affect_reference_bits() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        // Reference bits are cleared after insert (only insert sets them)
        // Iterate mutably — reference bits should remain unchanged.
        for (_, v) in ring.iter_mut() {
            *v += 1;
        }
        ring.debug_validate_invariants();
    }

    #[test]
    fn keys_yields_all_keys() {
        let mut ring = ClockRing::new(4);
        ring.insert("x", 10);
        ring.insert("y", 20);
        ring.insert("z", 30);

        let mut keys: Vec<_> = ring.keys().collect();
        keys.sort();
        assert_eq!(keys, vec![&"x", &"y", &"z"]);
    }

    #[test]
    fn values_yields_all_values() {
        let mut ring = ClockRing::new(4);
        ring.insert("x", 10);
        ring.insert("y", 20);
        ring.insert("z", 30);

        let mut vals: Vec<_> = ring.values().collect();
        vals.sort();
        assert_eq!(vals, vec![&10, &20, &30]);
    }

    #[test]
    fn values_mut_modifies_all_values() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        for v in ring.values_mut() {
            *v *= 10;
        }

        let mut vals: Vec<_> = ring.values().copied().collect();
        vals.sort();
        assert_eq!(vals, vec![10, 20]);
    }

    #[test]
    fn into_iter_consumes_ring() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        let mut pairs: Vec<_> = ring.into_iter().collect();
        pairs.sort_by_key(|(k, _)| *k);
        assert_eq!(pairs, vec![("a", 1), ("b", 2)]);
    }

    #[test]
    fn into_iter_empty() {
        let ring = ClockRing::<&str, i32>::new(5);
        assert_eq!(ring.into_iter().count(), 0);
    }

    #[test]
    fn into_iter_after_evictions() {
        let mut ring = ClockRing::new(2);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3); // evicts one

        let pairs: Vec<_> = ring.into_iter().collect();
        assert_eq!(pairs.len(), 2);
    }

    #[test]
    fn into_iter_for_loop() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        let mut sum = 0;
        for (_, v) in ring {
            sum += v;
        }
        assert_eq!(sum, 3);
    }

    #[test]
    fn ref_into_iter_for_loop() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        let mut sum = 0;
        for (_, v) in &ring {
            sum += v;
        }
        assert_eq!(sum, 3);
        assert_eq!(ring.len(), 2); // ring not consumed
    }

    #[test]
    fn mut_ref_into_iter_for_loop() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        for (_, v) in &mut ring {
            *v += 10;
        }
        assert_eq!(ring.peek(&"a"), Some(&11));
        assert_eq!(ring.peek(&"b"), Some(&12));
    }

    #[test]
    fn iter_count_matches_len() {
        let mut ring = ClockRing::new(10);
        for i in 0..7 {
            ring.insert(i, i * 10);
        }
        ring.remove(&3);
        ring.remove(&5);

        assert_eq!(ring.iter().count(), ring.len());
        assert_eq!(ring.keys().count(), ring.len());
        assert_eq!(ring.values().count(), ring.len());
    }

    #[test]
    fn iter_after_clear() {
        let mut ring = ClockRing::new(5);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.clear();

        assert_eq!(ring.iter().count(), 0);
        assert_eq!(ring.keys().count(), 0);
        assert_eq!(ring.values().count(), 0);
    }

    #[cfg(feature = "concurrency")]
    #[test]
    fn concurrent_into_inner_allows_iteration() {
        let cache = ConcurrentClockRing::new(5);
        cache.insert("a", 1);
        cache.insert("b", 2);
        cache.insert("c", 3);

        let ring = cache.into_inner();
        let mut pairs: Vec<_> = ring.iter().collect();
        pairs.sort_by_key(|&(k, _)| *k);
        assert_eq!(pairs, vec![(&"a", &1), (&"b", &2), (&"c", &3)]);
    }

    #[cfg(feature = "concurrency")]
    #[test]
    fn concurrent_into_inner_into_iter() {
        let cache = ConcurrentClockRing::new(3);
        cache.insert("x", 10);
        cache.insert("y", 20);

        let pairs: Vec<_> = cache.into_inner().into_iter().collect();
        assert_eq!(pairs.len(), 2);
    }

    // -----------------------------------------------------------------------
    // Regression: insert must not evict when empty slots exist (len < cap)
    // -----------------------------------------------------------------------

    #[test]
    fn insert_uses_empty_slot_after_remove_no_eviction() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1); // slot 0, hand -> 1
        ring.insert("b", 2); // slot 1, hand -> 2
        ring.insert("c", 3); // slot 2, hand -> 0

        ring.remove(&"b"); // slot 1 now empty, len = 2
        assert_eq!(ring.len(), 2);

        // Ring has room (len=2 < cap=3). Insert must use the empty slot,
        // not evict a live entry.
        let evicted = ring.insert("d", 4);
        assert_eq!(evicted, None, "must not evict when ring has free slots");
        assert_eq!(ring.len(), 3);
        assert!(ring.contains(&"a"), "\"a\" should still be present");
        assert!(ring.contains(&"c"), "\"c\" should still be present");
        assert!(ring.contains(&"d"), "\"d\" should have been inserted");
        ring.debug_validate_invariants();
    }

    #[test]
    fn insert_uses_empty_slot_after_pop_victim_no_eviction() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);

        let victim = ring.pop_victim();
        assert!(victim.is_some());
        assert_eq!(ring.len(), 2);

        // After pop_victim freed a slot, insert must not evict again.
        let evicted = ring.insert("d", 4);
        assert_eq!(evicted, None, "must not evict when ring has free slots");
        assert_eq!(ring.len(), 3);
        assert!(ring.contains(&"d"));
        ring.debug_validate_invariants();
    }

    // -----------------------------------------------------------------------
    // Security hardening: exception safety in insert_swap eviction path.
    //
    // If `K::Clone` panics during the full-ring eviction branch, the ring
    // must not be left with an empty slot while `len == capacity`; a
    // subsequent insert would hit `unreachable!("occupied slot missing
    // under full ring")` and turn a transient allocation failure into a
    // persistent-panic DoS. The fix clones `key` *before* taking the
    // victim slot, so a panic there leaves state unchanged.
    // -----------------------------------------------------------------------

    // Per-test-thread budget for `PanicKey::clone` calls. Tests
    // decrement this to force a clone to panic at a chosen point
    // without depending on allocator failure.
    thread_local! {
        static CLONE_BUDGET: std::cell::Cell<usize> =
            const { std::cell::Cell::new(usize::MAX) };
    }

    #[derive(Debug, PartialEq, Eq, Hash)]
    struct PanicKey(u32);

    impl Clone for PanicKey {
        fn clone(&self) -> Self {
            CLONE_BUDGET.with(|b| {
                let remaining = b.get();
                if remaining == 0 {
                    panic!("PanicKey: clone budget exhausted");
                }
                b.set(remaining - 1);
            });
            Self(self.0)
        }
    }

    #[test]
    fn insert_swap_eviction_panic_in_clone_preserves_invariants() {
        // Build a full ring, ensure the next insert will hit the
        // eviction path. Starve the clone budget to 0 so `key.clone()`
        // inside `insert_swap` panics.
        let mut ring: ClockRing<PanicKey, i32> = ClockRing::new(2);
        CLONE_BUDGET.with(|b| b.set(usize::MAX));
        ring.insert(PanicKey(1), 10);
        ring.insert(PanicKey(2), 20);
        assert_eq!(ring.len(), 2);
        assert_eq!(ring.capacity(), 2);

        // Snapshot live keys as owned `(u32, bool)` pairs so the
        // borrow is released before we re-borrow `ring` mutably.
        let slots_before: Vec<Option<(u32, bool)>> = ring
            .debug_snapshot_slots()
            .into_iter()
            .map(|slot| slot.map(|(k, bit)| (k.0, bit)))
            .collect();

        CLONE_BUDGET.with(|b| b.set(0));
        let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            ring.insert(PanicKey(3), 30);
        }))
        .is_err();
        assert!(panicked, "PanicKey::clone must have fired");

        // Restore the budget so subsequent operations can clone freely.
        CLONE_BUDGET.with(|b| b.set(usize::MAX));

        // Ring state must be internally consistent after the unwind.
        ring.debug_validate_invariants();
        assert_eq!(ring.len(), 2, "len must not drift on clone panic");
        assert_eq!(ring.capacity(), 2);

        // Same two entries are still present — the clone panic must
        // not have evicted or replaced anything.
        let slots_after: Vec<Option<(u32, bool)>> = ring
            .debug_snapshot_slots()
            .into_iter()
            .map(|slot| slot.map(|(k, bit)| (k.0, bit)))
            .collect();
        assert_eq!(slots_before.len(), slots_after.len());
        for (before, after) in slots_before.iter().zip(slots_after.iter()) {
            assert_eq!(
                before.map(|(k, _)| k),
                after.map(|(k, _)| k),
                "slot contents changed after panicking clone"
            );
        }

        // Subsequent insert must not hit the `unreachable!()` branch:
        // this is the regression. Pre-fix, the empty slot created by
        // `slots[idx].take()` combined with `len == capacity` would
        // cause the next eviction sweep to panic.
        let evicted = ring.insert(PanicKey(3), 30);
        assert!(evicted.is_some(), "next insert must evict normally");
        ring.debug_validate_invariants();
        assert_eq!(ring.len(), 2);
    }

    #[test]
    fn insert_swap_eviction_panic_in_clone_does_not_leak_evictee() {
        // Tighter check: the caller-visible value `30` must not appear
        // in the ring, and neither PanicKey(1) nor PanicKey(2) should
        // have been partially removed from the index.
        let mut ring: ClockRing<PanicKey, i32> = ClockRing::new(2);
        CLONE_BUDGET.with(|b| b.set(usize::MAX));
        ring.insert(PanicKey(1), 10);
        ring.insert(PanicKey(2), 20);

        CLONE_BUDGET.with(|b| b.set(0));
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            ring.insert(PanicKey(3), 30);
        }));
        CLONE_BUDGET.with(|b| b.set(usize::MAX));

        ring.debug_validate_invariants();
        assert!(ring.contains(&PanicKey(1)));
        assert!(ring.contains(&PanicKey(2)));
        assert!(!ring.contains(&PanicKey(3)));
    }

    // -----------------------------------------------------------------------
    // Security hardening: exception safety when the evictee's `Hash`/`Eq`
    // panics inside `HashMap::remove`.
    //
    // Pre-fix, `insert_swap` and `pop_victim` would `slots[idx].take()`
    // *before* calling `self.index.remove(&evicted.key)`. A panic in the
    // user-supplied `Hash`/`Eq` on the evictee key then left `len ==
    // capacity` (or, for `pop_victim`, `len` un-decremented) with an
    // empty slot — and the next `insert` fired
    // `unreachable!("occupied slot missing under full ring")`, turning a
    // single bad hash call into a persistent-panic DoS for every thread
    // sharing a `ConcurrentClockRing` handle.
    //
    // The fix removes the index entry *by borrow* before taking the slot.
    // These tests install a key type whose `Hash` panics after a
    // caller-controlled budget and verify the ring survives the unwind.
    // -----------------------------------------------------------------------

    thread_local! {
        static HASH_BUDGET: std::cell::Cell<usize> =
            const { std::cell::Cell::new(usize::MAX) };
    }

    #[derive(Debug, PartialEq, Eq, Clone)]
    struct HashPanicKey(u32);

    impl std::hash::Hash for HashPanicKey {
        fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
            HASH_BUDGET.with(|b| {
                let remaining = b.get();
                if remaining == 0 {
                    panic!("HashPanicKey: hash budget exhausted");
                }
                b.set(remaining - 1);
            });
            self.0.hash(state);
        }
    }

    #[test]
    fn insert_swap_eviction_panic_in_hash_preserves_invariants() {
        // Build a full ring. The next insert must hit the eviction
        // path and call `HashMap::remove` on the evictee's key —
        // which we make panic via a starved hash budget.
        let mut ring: ClockRing<HashPanicKey, i32> = ClockRing::new(2);
        HASH_BUDGET.with(|b| b.set(usize::MAX));
        ring.insert(HashPanicKey(1), 10);
        ring.insert(HashPanicKey(2), 20);
        assert_eq!(ring.len(), 2);

        let slots_before: Vec<Option<u32>> = ring
            .debug_snapshot_slots()
            .into_iter()
            .map(|slot| slot.map(|(k, _)| k.0))
            .collect();

        // Budget: exactly the hash calls required to locate the new
        // key's slot (via `self.index.get(&key)` at the top of
        // insert_swap) and check the full-ring fast path, but starved
        // before `self.index.remove(&evictee.key)` inside the
        // eviction branch. Setting to 1 is enough — the first hash
        // in `index.get(&PanicKey(3))` consumes it, and every
        // subsequent hash call (including the one inside
        // `HashMap::remove`) panics.
        HASH_BUDGET.with(|b| b.set(1));
        let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            ring.insert(HashPanicKey(3), 30);
        }))
        .is_err();
        assert!(panicked, "HashPanicKey::hash must have fired");

        HASH_BUDGET.with(|b| b.set(usize::MAX));

        // The regression: `len` must still equal the number of
        // occupied slots, and `debug_validate_invariants` must pass.
        ring.debug_validate_invariants();
        assert_eq!(ring.len(), 2, "len must not drift on hash panic");
        assert_eq!(ring.capacity(), 2);

        // Same two live entries. `HashPanicKey(3)` must not have
        // been inserted.
        let slots_after: Vec<Option<u32>> = ring
            .debug_snapshot_slots()
            .into_iter()
            .map(|slot| slot.map(|(k, _)| k.0))
            .collect();
        assert_eq!(slots_before, slots_after);

        // And — the regression — the next insert must not fire
        // `unreachable!("occupied slot missing under full ring")`.
        let evicted = ring.insert(HashPanicKey(3), 30);
        assert!(evicted.is_some(), "next insert must evict normally");
        ring.debug_validate_invariants();
    }

    #[test]
    fn pop_victim_panic_in_hash_preserves_invariants() {
        let mut ring: ClockRing<HashPanicKey, i32> = ClockRing::new(3);
        HASH_BUDGET.with(|b| b.set(usize::MAX));
        ring.insert(HashPanicKey(1), 10);
        ring.insert(HashPanicKey(2), 20);
        ring.insert(HashPanicKey(3), 30);
        assert_eq!(ring.len(), 3);

        let snapshot_before = ring.debug_snapshot_slots().len();

        // Starve the hash budget so `self.index.remove(&victim.key)`
        // inside pop_victim panics.
        HASH_BUDGET.with(|b| b.set(0));
        let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            ring.pop_victim();
        }))
        .is_err();
        assert!(panicked, "HashPanicKey::hash must have fired");

        HASH_BUDGET.with(|b| b.set(usize::MAX));

        // `len` must still reflect actual occupancy — pre-fix this
        // drifted when take() ran before the panicking remove.
        ring.debug_validate_invariants();
        assert_eq!(ring.len(), 3, "len must not drift on hash panic");
        assert_eq!(ring.debug_snapshot_slots().len(), snapshot_before);

        // Follow-up pop_victim must succeed normally.
        let evicted = ring.pop_victim();
        assert!(evicted.is_some(), "pop_victim must recover after unwind");
        assert_eq!(ring.len(), 2);
        ring.debug_validate_invariants();
    }

    // -----------------------------------------------------------------------
    // try_clone: matches try_new's fallible-allocation contract.
    // -----------------------------------------------------------------------

    #[test]
    fn try_clone_round_trips_contents() {
        let mut ring: ClockRing<&str, i32> = ClockRing::new(4);
        ring.insert("a", 1);
        ring.insert("b", 2);
        ring.insert("c", 3);
        ring.touch(&"a");

        let clone = ring
            .try_clone()
            .expect("try_clone must succeed on sane capacity");
        assert_eq!(clone.capacity(), ring.capacity());
        assert_eq!(clone.len(), ring.len());
        assert_eq!(clone.peek(&"a"), Some(&1));
        assert_eq!(clone.peek(&"b"), Some(&2));
        assert_eq!(clone.peek(&"c"), Some(&3));
        clone.debug_validate_invariants();

        // Snapshot equality: slot occupancy + ref bits must match.
        assert_eq!(ring.debug_snapshot_slots(), clone.debug_snapshot_slots());
    }

    #[test]
    fn try_clone_is_independent_of_source() {
        let mut ring: ClockRing<&str, i32> = ClockRing::new(3);
        ring.insert("a", 1);
        ring.insert("b", 2);

        let mut clone = ring.try_clone().unwrap();
        clone.insert("c", 3);
        clone.remove(&"a");

        // Mutating the clone must not observably affect the original.
        assert!(ring.contains(&"a"));
        assert!(!ring.contains(&"c"));
        assert_eq!(ring.len(), 2);
    }

    #[test]
    fn clone_trait_matches_try_clone() {
        let mut ring: ClockRing<&str, i32> = ClockRing::new(4);
        ring.insert("x", 10);
        ring.insert("y", 20);

        let via_clone = ring.clone();
        let via_try_clone = ring.try_clone().unwrap();

        assert_eq!(
            via_clone.debug_snapshot_slots(),
            via_try_clone.debug_snapshot_slots()
        );
        assert_eq!(via_clone.len(), via_try_clone.len());
        assert_eq!(via_clone.capacity(), via_try_clone.capacity());
    }

    #[test]
    fn insert_no_eviction_preserves_ref_bits() {
        let mut ring = ClockRing::new(3);
        ring.insert("a", 1); // slot 0, hand -> 1
        ring.insert("b", 2); // slot 1, hand -> 2
        ring.insert("c", 3); // slot 2, hand -> 0

        ring.touch(&"a"); // ref[0] = true

        ring.remove(&"c"); // slot 2 = None, hand stays 0, len = 2

        // Non-full insert: must find the empty slot without clearing
        // "a"'s reference bit.
        let evicted = ring.insert("d", 4);
        assert_eq!(evicted, None, "must not evict when ring has free slots");
        assert_eq!(ring.len(), 3);

        // Now full. Insert triggers eviction. If "a"'s ref bit was
        // preserved, "a" survives; victim is "b" or "d".
        let evicted = ring.insert("e", 5);
        assert!(evicted.is_some());
        let (evicted_key, _) = evicted.unwrap();
        assert!(
            evicted_key == "b" || evicted_key == "d",
            "expected unreferenced victim, got {:?}",
            evicted_key
        );
        assert!(
            ring.contains(&"a"),
            "\"a\" should survive: ref bit preserved"
        );
        ring.debug_validate_invariants();
    }
}

#[cfg(test)]
#[allow(unused_must_use)]
mod property_tests {
    use super::*;
    use proptest::prelude::*;

    // =============================================================================
    // Property Tests - Core Invariants
    // =============================================================================

    proptest! {
        /// Property: len() never exceeds capacity
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_len_within_capacity(
            capacity in 1usize..100,
            ops in prop::collection::vec((0u32..1000, 0u32..100), 0..200)
        ) {
            let mut ring = ClockRing::new(capacity);

            for (key, value) in ops {
                ring.insert(key, value);
                prop_assert!(ring.len() <= ring.capacity());
            }
        }

        /// Property: Index and slot consistency - every indexed key has matching slot
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_index_slot_consistency(
            capacity in 1usize..50,
            ops in prop::collection::vec((0u32..100, 0u32..100), 0..100)
        ) {
            let mut ring = ClockRing::new(capacity);

            for (key, value) in ops {
                ring.insert(key, value);
                ring.debug_validate_invariants();
            }
        }

        /// Property: Get after insert returns the correct value
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_get_after_insert(
            capacity in 1usize..50,
            key in 0u32..100,
            value in 0u32..1000
        ) {
            let mut ring = ClockRing::new(capacity);
            ring.insert(key, value);

            if ring.contains(&key) {
                prop_assert_eq!(ring.peek(&key), Some(&value));
            }
        }

        /// Property: Insert on full ring returns eviction or None
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_insert_eviction_behavior(
            capacity in 1usize..20,
            keys in prop::collection::vec(0u32..50, 0..100)
        ) {
            let mut ring = ClockRing::new(capacity);

            for key in keys {
                let len_before = ring.len();
                let evicted = ring.insert(key, key * 10);

                if len_before < capacity {
                    // Not full - no eviction expected
                    if evicted.is_some() {
                        // Unless we're updating an existing key
                        prop_assert!(len_before == ring.len());
                    }
                } else {
                    // Full - eviction expected unless updating existing key
                    prop_assert!(ring.len() <= capacity);
                }

                ring.debug_validate_invariants();
            }
        }

        /// Property: Remove decreases length and makes key absent
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_remove_behavior(
            capacity in 1usize..50,
            keys in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            // Insert all keys
            for &key in &keys {
                ring.insert(key, key * 10);
            }

            // Remove half
            for &key in &keys[0..keys.len()/2] {
                let len_before = ring.len();
                let removed = ring.remove(&key);

                if removed.is_some() {
                    prop_assert_eq!(ring.len(), len_before - 1);
                    prop_assert!(!ring.contains(&key));
                }

                ring.debug_validate_invariants();
            }
        }

        /// Property: Update doesn't change len, only value
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_update_preserves_len(
            capacity in 1usize..50,
            ops in prop::collection::vec((0u32..50, 0u32..100), 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            for (key, value) in &ops {
                ring.insert(*key, *value);
            }

            let len_before = ring.len();

            // Update existing keys
            for (key, new_value) in ops {
                if ring.contains(&key) {
                    ring.update(&key, new_value + 1000);
                    prop_assert_eq!(ring.len(), len_before);
                }
            }

            ring.debug_validate_invariants();
        }

        /// Property: Referenced entries survive longer than unreferenced
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_second_chance_behavior(
            capacity in 2usize..10
        ) {
            let mut ring = ClockRing::new(capacity);

            // Fill ring
            for i in 0..capacity {
                ring.insert(i as u32, i as u32);
            }

            // Mark first entry as referenced
            ring.touch(&0);

            // Insert new entry - should not evict entry 0
            ring.insert(capacity as u32, capacity as u32);

            // Entry 0 should still be present due to reference bit
            prop_assert!(ring.contains(&0) || ring.len() < capacity);
        }

        /// Property: peek doesn't modify state (can be called multiple times)
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_peek_idempotent(
            capacity in 1usize..50,
            keys in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            for key in keys {
                ring.insert(key, key * 10);
            }

            // Take snapshot
            let snapshot_before = ring.debug_snapshot_slots();

            // Peek all entries multiple times
            for i in 0..100 {
                ring.peek(&i);
            }

            let snapshot_after = ring.debug_snapshot_slots();

            // State should be unchanged
            prop_assert_eq!(snapshot_before, snapshot_after);
        }

        /// Property: Hand position stays within bounds
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_hand_within_bounds(
            capacity in 1usize..50,
            ops in prop::collection::vec((0u32..100, 0u32..100), 0..200)
        ) {
            let mut ring = ClockRing::new(capacity);

            for (key, value) in ops {
                ring.insert(key, value);
                ring.debug_validate_invariants();
            }
        }

        /// Property: pop_victim decreases length
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_pop_victim_decreases_len(
            capacity in 1usize..50,
            keys in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            for key in keys {
                ring.insert(key, key * 10);
            }

            while !ring.is_empty() {
                let len_before = ring.len();
                let evicted = ring.pop_victim();

                if evicted.is_some() {
                    prop_assert_eq!(ring.len(), len_before - 1);
                    let (evicted_key, _) = evicted.unwrap();
                    prop_assert!(!ring.contains(&evicted_key));
                }

                ring.debug_validate_invariants();
            }
        }

        /// Property: Clear empties the ring
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_clear_empties(
            capacity in 1usize..50,
            keys in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            for key in keys {
                ring.insert(key, key * 10);
            }

            ring.clear();

            prop_assert!(ring.is_empty());
            prop_assert_eq!(ring.len(), 0);
            ring.debug_validate_invariants();
        }
    }

    // =============================================================================
    // Property Tests - Sequential Operation Consistency
    // =============================================================================

    #[derive(Debug, Clone)]
    enum Operation {
        Insert(u32, u32),
        Get(u32),
        Peek(u32),
        Touch(u32),
        Update(u32, u32),
        Remove(u32),
        PopVictim,
    }

    fn operation_strategy() -> impl Strategy<Value = Operation> {
        prop_oneof![
            (0u32..50, 0u32..100).prop_map(|(k, v)| Operation::Insert(k, v)),
            (0u32..50).prop_map(Operation::Get),
            (0u32..50).prop_map(Operation::Peek),
            (0u32..50).prop_map(Operation::Touch),
            (0u32..50, 0u32..100).prop_map(|(k, v)| Operation::Update(k, v)),
            (0u32..50).prop_map(Operation::Remove),
            Just(Operation::PopVictim),
        ]
    }

    proptest! {
        /// Property: Arbitrary operation sequences maintain all invariants
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_arbitrary_ops_maintain_invariants(
            capacity in 1usize..30,
            ops in prop::collection::vec(operation_strategy(), 0..200)
        ) {
            let mut ring = ClockRing::new(capacity);

            for op in ops {
                match op {
                    Operation::Insert(k, v) => {
                        ring.insert(k, v);
                    }
                    Operation::Get(k) => {
                        ring.get(&k);
                    }
                    Operation::Peek(k) => {
                        ring.peek(&k);
                    }
                    Operation::Touch(k) => {
                        ring.touch(&k);
                    }
                    Operation::Update(k, v) => {
                        ring.update(&k, v);
                    }
                    Operation::Remove(k) => {
                        ring.remove(&k);
                    }
                    Operation::PopVictim => {
                        ring.pop_victim();
                    }
                }

                // All invariants must hold after every operation
                ring.debug_validate_invariants();
                prop_assert!(ring.len() <= ring.capacity());
            }
        }

        /// Property: Interleaved inserts/removes maintain consistency
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_interleaved_insert_remove(
            capacity in 1usize..30,
            ops in prop::collection::vec((0u32..50, any::<bool>()), 0..200)
        ) {
            let mut ring = ClockRing::new(capacity);

            for (key, should_insert) in ops {
                if should_insert {
                    ring.insert(key, key * 10);
                } else {
                    ring.remove(&key);
                }

                ring.debug_validate_invariants();
                prop_assert!(ring.len() <= ring.capacity());
            }
        }
    }

    // =============================================================================
    // Property Tests - Edge Cases
    // =============================================================================

    proptest! {
        /// Property: Zero capacity ring stays empty
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_zero_capacity_noop(
            ops in prop::collection::vec((0u32..100, 0u32..100), 0..50)
        ) {
            let mut ring = ClockRing::<u32, u32>::new(0);

            for (key, value) in ops {
                ring.insert(key, value);
                prop_assert!(ring.is_empty());
                prop_assert_eq!(ring.len(), 0);
                prop_assert!(!ring.contains(&key));
            }
        }

        /// Property: Capacity 1 ring never exceeds 1 entry
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_capacity_one_single_entry(
            keys in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(1);

            for key in keys {
                ring.insert(key, key * 10);
                prop_assert!(ring.len() <= 1);
                ring.debug_validate_invariants();
            }
        }

        /// Property: Duplicate inserts don't grow the ring
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_duplicate_inserts_no_growth(
            capacity in 1usize..30,
            key in 0u32..50,
            values in prop::collection::vec(0u32..100, 1..50)
        ) {
            let mut ring = ClockRing::new(capacity);

            ring.insert(key, 0);
            let len_after_first = ring.len();

            for value in values {
                ring.insert(key, value);
                prop_assert_eq!(ring.len(), len_after_first);
            }
        }
    }

    // =============================================================================
    // Property Tests - Concurrent Wrapper (if enabled)
    // =============================================================================

    #[cfg(feature = "concurrency")]
    proptest! {
        /// Property: Concurrent wrapper maintains same invariants
        #[cfg_attr(miri, ignore)]
        #[test]
        fn prop_concurrent_maintains_invariants(
            capacity in 1usize..30,
            ops in prop::collection::vec((0u32..50, 0u32..100), 0..100)
        ) {
            let ring = ConcurrentClockRing::new(capacity);

            for (key, value) in ops {
                ring.insert(key, value);
                prop_assert!(ring.len() <= ring.capacity());
            }
        }
    }
}

#[cfg(test)]
#[allow(unused_must_use)]
mod fuzz_tests {
    use super::*;

    /// Fuzz target: arbitrary operation sequences
    ///
    /// This can be used with cargo-fuzz or as a regular test with
    /// generated inputs.
    pub fn fuzz_arbitrary_operations(data: &[u8]) {
        if data.len() < 2 {
            return;
        }

        let capacity = (data[0] as usize % 50).max(1);
        let mut ring = ClockRing::new(capacity);

        let mut idx = 1;
        while idx < data.len() {
            if idx + 2 >= data.len() {
                break;
            }

            let op = data[idx] % 7;
            let key = data[idx + 1] as u32;
            let value = data[idx + 2] as u32;

            match op {
                0 => {
                    ring.insert(key, value);
                },
                1 => {
                    ring.get(&key);
                },
                2 => {
                    ring.peek(&key);
                },
                3 => {
                    ring.touch(&key);
                },
                4 => {
                    ring.update(&key, value);
                },
                5 => {
                    ring.remove(&key);
                },
                6 => {
                    ring.pop_victim();
                },
                _ => unreachable!(),
            }

            // Validate invariants after each operation
            ring.debug_validate_invariants();
            assert!(ring.len() <= ring.capacity());

            idx += 3;
        }
    }

    /// Fuzz target: stress test with many inserts
    pub fn fuzz_insert_stress(data: &[u8]) {
        if data.len() < 4 {
            return;
        }

        let capacity = (data[0] as usize % 100).max(1);
        let mut ring = ClockRing::new(capacity);

        for chunk in data[1..].chunks(2) {
            if chunk.len() < 2 {
                break;
            }
            let key = chunk[0] as u32;
            let value = chunk[1] as u32;
            ring.insert(key, value);

            assert!(ring.len() <= ring.capacity());
        }

        ring.debug_validate_invariants();
    }

    /// Fuzz target: eviction patterns with reference bits
    pub fn fuzz_eviction_patterns(data: &[u8]) {
        if data.len() < 3 {
            return;
        }

        let capacity = (data[0] as usize % 20).max(2);
        let mut ring = ClockRing::new(capacity);

        // Fill the ring
        for i in 0..capacity {
            ring.insert(i as u32, i as u32);
        }

        let mut idx = 1;
        while idx < data.len() {
            if idx + 1 >= data.len() {
                break;
            }

            let key = data[idx] as u32 % capacity as u32;
            let should_touch = data[idx + 1] % 2 == 0;

            if should_touch {
                ring.touch(&key);
            }

            // Insert new entry to trigger eviction
            let new_key = capacity as u32 + (idx as u32);
            ring.insert(new_key, new_key);

            ring.debug_validate_invariants();
            assert!(ring.len() <= ring.capacity());

            idx += 2;
        }
    }

    #[cfg_attr(miri, ignore)]
    #[test]
    fn test_fuzz_arbitrary_operations_smoke() {
        // Smoke test with some sample inputs
        let inputs = vec![
            vec![5, 0, 1, 2, 1, 3, 4, 2, 5, 6],
            vec![10, 6, 7, 8, 5, 9, 10, 0, 1, 2],
            vec![1, 0, 0, 0, 1, 1, 1, 2, 2, 2],
        ];

        for input in inputs {
            fuzz_arbitrary_operations(&input);
        }
    }

    #[cfg_attr(miri, ignore)]
    #[test]
    fn test_fuzz_insert_stress_smoke() {
        let inputs = vec![
            vec![5, 1, 2, 3, 4, 5, 6, 7, 8],
            vec![1, 0, 0, 0, 0, 0, 0],
            vec![20; 100],
        ];

        for input in inputs {
            fuzz_insert_stress(&input);
        }
    }

    #[cfg_attr(miri, ignore)]
    #[test]
    fn test_fuzz_eviction_patterns_smoke() {
        let inputs = vec![
            vec![5, 0, 1, 1, 0, 2, 1, 3, 0],
            vec![3, 0, 0, 1, 1, 2, 0, 0, 1],
            vec![10, 1, 0, 2, 1, 3, 0, 4, 1],
        ];

        for input in inputs {
            fuzz_eviction_patterns(&input);
        }
    }
}