cl_tds/

lib.rs

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

//! # CL-TDS
//!
//! **Cache-Locked Temporal Decay Sketch** — a fixed-memory, lock-free streaming
//! algorithm for detecting heavy hitters in real-time data streams.
//!
//! CL-TDS answers one question: *"What items are appearing most frequently
//! in this stream, right now?"* — using only **1 MB** of memory, with old
//! data fading automatically.
//!
//! # Getting Started
//!
//! Add to your `Cargo.toml`:
//! ```toml
//! [dependencies]
//! cl-tds = { git = "https://github.com/ddsha441981/cl-tds" }
//! ```
//!
//! Basic usage:
//! ```rust
//! use cl_tds::ClTds;
//!
//! let sketch = ClTds::new();
//!
//! // Record items (accepts any u64 hash)
//! sketch.increment(0xDEAD_BEEF);
//! sketch.increment(0xDEAD_BEEF);
//! sketch.increment(0xCAFE_BABE);
//!
//! // Query frequency estimates
//! assert!(sketch.query(0xDEAD_BEEF) >= 2);
//! assert!(sketch.query(0xCAFE_BABE) >= 1);
//! ```
//!
//! # How To Use
//!
//! CL-TDS is a **library**. You integrate it into your application and feed it
//! your data. The algorithm does not care where data comes from — network packets,
//! log files, database queries, Kafka streams, or anything else.
//!
//! ### Step 1: Hash your data into `u64`
//!
//! CL-TDS operates on `u64` identifiers. Hash your domain objects:
//!
//! ```rust
//! use std::hash::{Hash, Hasher};
//! use std::collections::hash_map::DefaultHasher;
//!
//! fn to_id<T: Hash>(item: &T) -> u64 {
//!     let mut h = DefaultHasher::new();
//!     item.hash(&mut h);
//!     h.finish()
//! }
//!
//! // Works with any type:
//! // to_id(&"192.168.1.1")    → for IP addresses
//! // to_id(&"POST /api/pay")  → for API endpoints
//! // to_id(&"#trending")      → for hashtags
//! // to_id(&12345u64)         → for user IDs
//! ```
//!
//! ### Step 2: Choose your mode
//!
//! ```rust
//! use cl_tds::ClTds;
//!
//! // Manual mode — you control when data decays (good for batch processing)
//! let sketch = ClTds::new();
//! sketch.increment(42);
//! sketch.tick_epoch();  // data halves each tick
//!
//! // Auto mode — data decays based on wall clock (good for real-time)
//! let sketch = ClTds::with_epoch_interval(1000); // decay every 1 second
//! sketch.increment(42);
//! // count automatically halves every second — no manual ticking needed
//! ```
//!
//! ### Step 3: Feed your stream and query
//!
//! ```rust
//! use cl_tds::ClTds;
//!
//! let sketch = ClTds::with_epoch_interval(5000); // decay every 5s
//!
//! // Your data loop (network, logs, events — anything)
//! # let packets: Vec<u64> = vec![1, 2, 1, 1, 3];
//! for packet in packets {
//!     sketch.increment(packet);
//! }
//!
//! // Check if something is a heavy hitter
//! let threshold = 1000;
//! # let suspect: u64 = 1;
//! if sketch.query(suspect) > threshold {
//!     // alert! this item appeared too frequently
//! }
//! ```
//!
//! # Real-World Integration Patterns
//!
//! All patterns use the same core:
//! `sketch.increment(hash)` to record, `sketch.query(hash)` to check.
//! Only the **data source** and **threshold** change per domain.
//!
//! ### Network Security — DDoS Detection
//! ```rust,no_run
//! # use cl_tds::ClTds; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher};
//! # fn hash_str(s: &str) -> u64 { let mut h = DefaultHasher::new(); s.hash(&mut h); h.finish() }
//! let sketch = ClTds::with_epoch_interval(1000); // 1s window
//! // Feed source IPs from packet capture
//! sketch.increment(hash_str("1.2.3.4"));
//! if sketch.query(hash_str("1.2.3.4")) > 10_000 { /* DDoS alert */ }
//! ```
//!
//! ### API Rate Limiting
//! ```rust,no_run
//! # use cl_tds::ClTds;
//! # let user_id: u64 = 1;
//! let limiter = ClTds::with_epoch_interval(60_000); // 1-min window
//! // Feed user IDs from request handler
//! limiter.increment(user_id);
//! if limiter.query(user_id) > 100 { /* throttle: >100 req/min */ }
//! ```
//!
//! ### Log Analysis — Error Surge Detection
//! ```rust,no_run
//! # use cl_tds::ClTds; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher};
//! # fn hash_str(s: &str) -> u64 { let mut h = DefaultHasher::new(); s.hash(&mut h); h.finish() }
//! let monitor = ClTds::with_epoch_interval(10_000); // 10s window
//! // Feed endpoint+status from log pipeline
//! monitor.increment(hash_str("POST /api/checkout 500"));
//! if monitor.query(hash_str("POST /api/checkout 500")) > 50 { /* alert */ }
//! ```
//!
//! ### Gaming — Anti-Cheat
//! ```rust,no_run
//! # use cl_tds::ClTds;
//! # let player_id: u64 = 1;
//! let detector = ClTds::with_epoch_interval(5_000); // 5s window
//! // Feed player action events
//! detector.increment(player_id);
//! if detector.query(player_id) > 500 { /* abnormal speed — possible cheat */ }
//! ```
//!
//! ### Telecom — SIM-Box Fraud
//! ```rust,no_run
//! # use cl_tds::ClTds; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher};
//! # fn hash_str(s: &str) -> u64 { let mut h = DefaultHasher::new(); s.hash(&mut h); h.finish() }
//! let cdr = ClTds::with_epoch_interval(3600_000); // 1-hour window
//! // Feed caller numbers from CDR stream
//! cdr.increment(hash_str("+91-9999000001"));
//! if cdr.query(hash_str("+91-9999000001")) > 1_000 { /* fraud alert */ }
//! ```
//!
//! ### IoT — Faulty Sensor Detection
//! ```rust,no_run
//! # use cl_tds::ClTds;
//! # let sensor_id: u64 = 42;
//! let sensors = ClTds::with_epoch_interval(60_000); // 1-min window
//! // Feed sensor alert events
//! sensors.increment(sensor_id);
//! if sensors.query(sensor_id) > 100 { /* sensor #42 firing too often */ }
//! ```
//!
//! ### Social Media — Trending Hashtags
//! ```rust,no_run
//! # use cl_tds::ClTds; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher};
//! # fn hash_str(s: &str) -> u64 { let mut h = DefaultHasher::new(); s.hash(&mut h); h.finish() }
//! let trends = ClTds::with_epoch_interval(30_000); // 30s window
//! // Feed hashtags from post stream
//! trends.increment(hash_str("#BreakingNews"));
//! if trends.query(hash_str("#BreakingNews")) > 5_000 { /* trending! */ }
//! ```
//!
//! ### Fintech — Card Fraud
//! ```rust,no_run
//! # use cl_tds::ClTds; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher};
//! # fn hash_str(s: &str) -> u64 { let mut h = DefaultHasher::new(); s.hash(&mut h); h.finish() }
//! let fraud = ClTds::with_epoch_interval(3600_000); // 1-hour window
//! // Feed card numbers from transaction stream
//! fraud.increment(hash_str("card-XXXX-4567"));
//! if fraud.query(hash_str("card-XXXX-4567")) > 50 { /* suspicious activity */ }
//! ```
//!
//! **Same pattern works for all 17 proven domains** — DNS monitoring, CDN caching,
//! click fraud, spam detection, search query tracking, smart grid monitoring, and more.
//! See `Applicable Domains` section for the full list.
//!
//! # Key Properties
//!
//! | Property | Value |
//! |----------|-------|
//! | Memory | Fixed 1 MB (4 rows × 65536 cols × 4 bytes) |
//! | Operations | `increment()` O(1), `query()` O(1) |
//! | Thread safety | Lock-free atomic CAS — safe from any number of threads |
//! | Decay | Lazy — O(1) per touch, no background thread |
//! | Dependencies | Zero — pure `std::sync::atomic` |
//! | Error bound (ε) | ≈ 0.0000414 (overcount per stream item) |
//! | Failure probability (δ) | ≤ 1.8% (4 independent rows) |
//! | Persistence | [`ClTds::to_bytes`] / [`ClTds::from_bytes`] |
//!
//! # Multi-Instance Scaling
//!
//! Each sketch uses exactly **1 MB**. A typical CPU L3 cache is **6–12 MB**,
//! so you can run **5 independent sketches** simultaneously — each monitoring
//! a different stream — and all fit in L3 cache without evicting each other.
//!
//! ```rust
//! use cl_tds::ClTds;
//! use std::sync::Arc;
//! use std::thread;
//!
//! // 5 sketches × 1 MB = 5 MB — fits in any modern L3 cache
//! let sketches: Vec<Arc<ClTds>> = (0..5)
//!     .map(|_| Arc::new(ClTds::with_epoch_interval(1000)))
//!     .collect();
//!
//! // Each thread monitors a different domain — zero contention
//! # let sketches_clone: Vec<Arc<ClTds>> = sketches.iter().map(|s| Arc::clone(s)).collect();
//! # let handles: Vec<_> = sketches_clone.into_iter().enumerate().map(|(i, sketch)| {
//! // thread::spawn(move || {
//! //     // Thread 0: Network DDoS monitoring
//! //     // Thread 1: API rate limiting
//! //     // Thread 2: Log error tracking
//! //     // Thread 3: Click fraud detection
//! //     // Thread 4: DNS query monitoring
//! //     sketch.increment(i as u64);
//! // });
//! # thread::spawn(move || { sketch.increment(i as u64); })
//! # }).collect();
//! # for h in handles { h.join().unwrap(); }
//! ```
//!
//! **Benchmarked:** 5 threads running concurrently, each processing its own
//! domain (Network, Crypto, Analytics, Gaming, IP Flood) — all maintaining
//! full throughput with zero speed loss.
//!
//! # Bring Your Own Data
//!
//! CL-TDS accepts any `u64` hash, so you can feed it data from **any source** —
//! CSV files, JSON logs, database streams, Kafka topics, or raw sockets.
//!
//! Read from a file:
//!
//! ```rust,no_run
//! use cl_tds::ClTds;
//! use std::io::{BufRead, BufReader};
//! use std::fs::File;
//! use std::collections::hash_map::DefaultHasher;
//! use std::hash::{Hash, Hasher};
//!
//! fn hash_line(s: &str) -> u64 {
//!     let mut h = DefaultHasher::new();
//!     s.hash(&mut h);
//!     h.finish()
//! }
//!
//! let sketch = ClTds::with_epoch_interval(1000);
//! let file = File::open("traffic.csv").unwrap();
//!
//! for line in BufReader::new(file).lines() {
//!     if let Ok(item) = line {
//!         sketch.increment(hash_line(&item));
//!     }
//! }
//!
//! // Now query any item's frequency
//! println!("Frequency: {}", sketch.query(hash_line("suspicious_item")));
//! ```
//!
//! # Applicable Domains
//!
//! CL-TDS works for any domain that needs: **heavy hitter detection** in a
//! **continuous stream** with **bounded memory** and **temporal decay**.
//!
//! **17 domains benchmarked and proven:** Network security (DDoS), financial
//! trading (HFT), web analytics, gaming anti-cheat, DNS monitoring, API rate
//! limiting, click fraud, CDN caching, telecom fraud, IoT sensors, log analysis,
//! spam detection, social trending, search tracking, smart grid, fintech fraud,
//! and IP flood detection.
//!
//! **Not suitable for:** Exact counting (ledgers/voting), listing unique items,
//! item deletion (GDPR), small datasets (< 100K — use `HashMap`),
//! relationship or sequence detection.
//!
//! # CL-TDS vs HashMap
//!
//! | Unique Items | HashMap | CL-TDS | Savings |
//! |---|---|---|---|
//! | 100K | 1.9 MB | 1 MB | 2x |
//! | 1M | 29.8 MB | 1 MB | **30x** |
//! | 5M | 119 MB | 1 MB | **119x** |
//! | 10M | 238 MB | 1 MB | **238x** |
//!
//! CL-TDS is also **1.8x faster on insert** and **4.8x faster on query**
//! because it fits entirely in L3 cache. Plus HashMap can't do temporal
//! decay — stale data stays forever.
//!
//! # Testing
//!
//! ```bash
//! cargo test                        # 40 unit tests + 14 doc tests
//! cargo run --example basic         # Insert/query demo
//! cargo run --example decay         # Temporal decay visualization
//! cargo run --example multithread   # Concurrent stress test
//! cargo run --example persistence   # Save/restore to disk
//! ```
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::time::Instant;

/// Matrix width — columns per row. 2^16 = 65536.
pub const WIDTH: usize = 1 << 16;

/// Matrix depth — number of independent hash rows.
/// δ = e^{-DEPTH} ≈ 1.8% with DEPTH=4.
pub const DEPTH: usize = 4;

/// Bits allocated to timestamp in each cell.
const TS_BITS: u32 = 8;

/// Bits allocated to frequency counter in each cell.
const COUNT_BITS: u32 = 24;

/// Bitmask for extracting counter (lower 24 bits).
const COUNT_MASK: u32 = (1 << COUNT_BITS) - 1; // 0x00FF_FFFF

/// Maximum counter value before saturation.
pub const MAX_COUNT: u32 = COUNT_MASK; // 16,777,215

/// Maximum meaningful decay shifts (beyond this, count = 0).
const MAX_DECAY: u32 = COUNT_BITS;

/// Timestamp mask (lower 8 bits of epoch).
const TS_MASK: u32 = (1 << TS_BITS) - 1; // 0xFF

/// Packs a timestamp (upper 8 bits) and a counter (lower 24 bits) into one `u32` cell.
#[inline(always)]
pub fn pack(timestamp: u32, count: u32) -> u32 {
    ((timestamp & TS_MASK) << COUNT_BITS) | (count & COUNT_MASK)
}

/// Extracts the timestamp and counter from a packed `u32` cell.
#[inline(always)]
pub fn unpack(cell: u32) -> (u32, u32) {
    let timestamp = cell >> COUNT_BITS;
    let count = cell & COUNT_MASK;
    (timestamp, count)
}

/// Calculates how many epochs have passed since a cell was last touched.
/// Handles 8-bit timestamp wraparound. Result is capped at 24
/// because shifting a 24-bit counter more than 24 times always gives zero.
#[inline(always)]
pub fn decay_steps(cell_ts: u32, current_epoch: u64) -> u32 {
    let epoch_low = (current_epoch & TS_MASK as u64) as u32;
    let diff = epoch_low.wrapping_sub(cell_ts) & TS_MASK;
    diff.min(MAX_DECAY)
}

/// Halves the count value `steps` times (right-shift by `steps`).
#[inline(always)]
fn apply_decay(count: u32, steps: u32) -> u32 {
    if steps >= MAX_DECAY {
        0
    } else {
        count >> steps
    }
}

/// Fixed hash constants used only in deterministic/testing mode.
const DEFAULT_HASH_A: [u64; DEPTH] = [
    0x9e3779b97f4a7c15, // golden ratio derivative
    0x517cc1b727220a95, // FxHash constant
    0x6c62272e07bb0142, // splitmix64 constant
    0xbf58476d1ce4e5b9, // splitmix64 constant 2
];

const DEFAULT_HASH_B: [u64; DEPTH] = [
    0xd2a98b26625eee7b,
    0x94d049bb133111eb,
    0xc4ceb9fe1a85ec53,
    0xe7037ed1a0b428db,
];

/// Generates unique random hash seeds using system entropy.
/// Each sketch instance gets different seeds, making it resistant to targeted attacks.
fn random_hash_params() -> ([u64; DEPTH], [u64; DEPTH]) {
    use std::collections::hash_map::RandomState;
    use std::hash::{BuildHasher, Hasher};

    let mut a = [0u64; DEPTH];
    let mut b = [0u64; DEPTH];

    for i in 0..DEPTH {
        let state = RandomState::new();
        let mut hasher = state.build_hasher();
        hasher.write_usize(i);
        a[i] = hasher.finish() | 1; // ensure odd (better distribution)

        let state2 = RandomState::new();
        let mut hasher2 = state2.build_hasher();
        hasher2.write_usize(i + DEPTH);
        b[i] = hasher2.finish();
    }

    (a, b)
}

/// Maps an item ID to a column index (0..65535) for the given row.
/// Uses the universal hash formula: `h(x) = (A·x + B) >> 48`.
#[inline(always)]
fn compute_hash(id: u64, row: usize, hash_a: &[u64; DEPTH], hash_b: &[u64; DEPTH]) -> usize {
    let h = hash_a[row].wrapping_mul(id).wrapping_add(hash_b[row]);
    (h >> 48) as usize // top 16 bits
}

/// One row of the sketch matrix (65536 atomic counters, cache-line aligned).
#[repr(align(64))]
pub struct Row([AtomicU32; WIDTH]);

/// The main sketch data structure. Uses 1 MB of memory to track item
/// frequencies in a stream, with old data fading out automatically.
///
/// All operations are lock-free and safe to call from multiple threads.
///
/// # Guarantees
///
/// - **Error bound:** estimated count can overcount by at most ε·N (ε ≈ 0.00004)
/// - **False positive rate:** at most 1.8% chance of misidentifying a non-heavy-hitter
/// - **Lazy decay:** produces the exact same result as decaying every cell every epoch
///
/// # Mathematical Formulation
///
/// ```text
/// Theorem 1 (Error):   E[query(x)] ≤ f_t(x) + ε · N_effective(t),  ε = e/65536
/// Theorem 2 (FP):      P[query(x) > φ·N_eff] ≤ δ = e^{-4} ≈ 1.8%
/// Theorem 3 (Decay):   V_lazy(C) = V_full(C)  (exact equality at touch time)
/// ```
pub struct ClTds {
    /// 4 independent hash rows, each 256 KB = total 1 MB.
    rows: Box<[Row; DEPTH]>,
    /// Global monotonic epoch counter (manual mode).
    epoch: AtomicU64,
    /// Creation timestamp (auto mode). None = manual mode.
    created_at: Option<Instant>,
    /// Epoch interval in milliseconds (auto mode).
    epoch_interval_ms: u64,
    /// Per-instance hash multiplicative constants (adversarial resistance).
    hash_a: [u64; DEPTH],
    /// Per-instance hash additive constants (adversarial resistance).
    hash_b: [u64; DEPTH],
}

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

impl ClTds {
    /// Creates a new sketch in manual mode. You control decay by calling
    /// [`tick_epoch()`](Self::tick_epoch) yourself. Hash seeds are randomized.
    pub fn new() -> Self {
        let (a, b) = random_hash_params();
        Self::alloc(None, 0, a, b)
    }

    /// Creates a sketch with fixed hash seeds. Two sketches created this
    /// way will produce identical results for identical inputs.
    ///
    /// Only use this in tests — it is NOT safe against targeted attacks.
    pub fn new_deterministic() -> Self {
        Self::alloc(None, 0, DEFAULT_HASH_A, DEFAULT_HASH_B)
    }

    /// Creates a sketch in auto mode. The decay clock ticks automatically
    /// based on real time — no background thread needed.
    ///
    /// `interval_ms` sets how often data halves. For example, `1000` means
    /// every second, all counts are halved. After 24 seconds, old data is gone.
    pub fn with_epoch_interval(interval_ms: u64) -> Self {
        assert!(interval_ms > 0, "epoch interval must be > 0");
        let (a, b) = random_hash_params();
        Self::alloc(Some(Instant::now()), interval_ms, a, b)
    }

    /// Allocates the 1 MB matrix on the heap, zeroed.
    fn alloc(
        created_at: Option<Instant>,
        epoch_interval_ms: u64,
        hash_a: [u64; DEPTH],
        hash_b: [u64; DEPTH],
    ) -> Self {
        // SAFETY: AtomicU32 with zeroed bytes = AtomicU32::new(0).
        // Guaranteed by Rust's atomic type representation.
        let rows = unsafe {
            let layout = std::alloc::Layout::new::<[Row; DEPTH]>();
            let ptr = std::alloc::alloc_zeroed(layout) as *mut [Row; DEPTH];
            if ptr.is_null() {
                std::alloc::handle_alloc_error(layout);
            }
            Box::from_raw(ptr)
        };
        ClTds {
            rows,
            epoch: AtomicU64::new(0),
            created_at,
            epoch_interval_ms,
            hash_a,
            hash_b,
        }
    }

    /// Returns the current epoch number. In manual mode, this reflects how
    /// many times `tick_epoch()` was called. In auto mode, it's derived from
    /// the wall clock.
    #[inline(always)]
    fn current_epoch(&self) -> u64 {
        match self.created_at {
            Some(t) => t.elapsed().as_millis() as u64 / self.epoch_interval_ms,
            None => self.epoch.load(Ordering::Relaxed),
        }
    }

    /// Hashes `id` to a column index for the given `row`.
    #[inline(always)]
    fn hash(&self, id: u64, row: usize) -> usize {
        compute_hash(id, row, &self.hash_a, &self.hash_b)
    }

    /// Records one occurrence of `id` in the stream.
    ///
    /// This is the main write operation. It hashes `id` to 4 cells (one per row),
    /// decays any stale values, and increments the counter — all atomically.
    ///
    /// Safe to call from multiple threads without any locking.
    ///
    /// Decay is lazy and exact: `(x >> a) >> b = x >> (a + b)` — no approximation.
    pub fn increment(&self, id: u64) {
        let epoch = self.current_epoch();
        let epoch_low = (epoch & TS_MASK as u64) as u32;

        for row_idx in 0..DEPTH {
            let col = self.hash(id, row_idx);
            let cell = &self.rows[row_idx].0[col];

            loop {
                let old = cell.load(Ordering::Relaxed);
                let (old_ts, old_count) = unpack(old);
                let steps = decay_steps(old_ts, epoch);
                let decayed = apply_decay(old_count, steps);
                let new_count = (decayed + 1).min(MAX_COUNT);
                let new_val = pack(epoch_low, new_count);
                match cell.compare_exchange_weak(old, new_val, Ordering::Relaxed, Ordering::Relaxed)
                {
                    Ok(_) => break,
                    Err(_) => continue,
                }
            }
        }
    }

    /// Returns the estimated frequency of `id` in the current time window.
    ///
    /// Looks up `id` in all 4 rows, decays stale values, and returns the
    /// minimum count. The minimum filters out noise from hash collisions,
    /// giving you the tightest possible estimate.
    ///
    /// ```text
    /// Guarantee: E[query(x)] ≤ f_t(x) + ε · N_effective(t)
    ///   where ε = e / 65536 ≈ 0.0000414
    /// False positive: P ≤ e^{-4} ≈ 1.8%
    /// ```
    pub fn query(&self, id: u64) -> u32 {
        let epoch = self.current_epoch();
        let mut min_count = u32::MAX;

        for row_idx in 0..DEPTH {
            let col = self.hash(id, row_idx);
            let cell = &self.rows[row_idx].0[col];
            let val = cell.load(Ordering::Relaxed);
            let (ts, count) = unpack(val);
            let steps = decay_steps(ts, epoch);
            let decayed = apply_decay(count, steps);

            min_count = min_count.min(decayed);
        }

        min_count
    }

    /// Advances the decay clock by one step (manual mode only).
    ///
    /// Each tick halves all counters the next time they're touched.
    /// Call this on a timer — for example, once per second.
    pub fn tick_epoch(&self) {
        self.epoch.fetch_add(1, Ordering::Relaxed);
    }

    /// Returns the current epoch number.
    pub fn epoch(&self) -> u64 {
        self.current_epoch()
    }

    /// Returns `true` if the sketch was created with [`with_epoch_interval`](Self::with_epoch_interval).
    pub fn is_auto_epoch(&self) -> bool {
        self.created_at.is_some()
    }

    /// Returns the total memory used by the matrix, in bytes. Always `1,048,576` (1 MB).
    pub fn memory_bytes(&self) -> usize {
        DEPTH * WIDTH * std::mem::size_of::<AtomicU32>()
    }

    /// Returns the algorithm's tuning parameters: `(ε, δ, width, depth)`.
    ///
    /// - `ε` — max overcount per stream item (~0.00004)
    /// - `δ` — probability of a false positive (~1.8%)
    /// - `width` — columns per row (65536)
    /// - `depth` — number of rows (4)
    pub fn algorithm_parameters() -> (f64, f64, usize, usize) {
        let epsilon = std::f64::consts::E / WIDTH as f64;
        let delta = (-(DEPTH as f64)).exp();
        (epsilon, delta, WIDTH, DEPTH)
    }

    /// Tells you the worst-case overcount for a stream of `n_effective` items.
    ///
    /// For example, if you've seen 1 million items, the max error is about 41.
    pub fn error_bound(n_effective: u64) -> f64 {
        let epsilon = std::f64::consts::E / WIDTH as f64;
        epsilon * n_effective as f64
    }

    /// Saves the entire sketch state to a byte vector.
    ///
    /// Write this to disk for crash recovery. The output is about 1 MB
    /// (matrix data + a small header with epoch and hash seeds).
    pub fn to_bytes(&self) -> Vec<u8> {
        let epoch = self.current_epoch();
        let matrix_size = DEPTH * WIDTH * 4;
        let header_size = 8 + (DEPTH * 8 * 2); // epoch + hash params
        let mut buf = Vec::with_capacity(header_size + matrix_size);

        buf.extend_from_slice(&epoch.to_le_bytes());
        for i in 0..DEPTH {
            buf.extend_from_slice(&self.hash_a[i].to_le_bytes());
        }
        for i in 0..DEPTH {
            buf.extend_from_slice(&self.hash_b[i].to_le_bytes());
        }
        for row in 0..DEPTH {
            for col in 0..WIDTH {
                let val = self.rows[row].0[col].load(Ordering::Relaxed);
                buf.extend_from_slice(&val.to_le_bytes());
            }
        }
        buf
    }

    /// Restores a sketch from a byte slice previously created by [`to_bytes()`](Self::to_bytes).
    ///
    /// Returns `None` if the data is corrupted or the wrong size.
    pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let matrix_size = DEPTH * WIDTH * 4;
        let header_size = 8 + (DEPTH * 8 * 2);
        if bytes.len() != header_size + matrix_size {
            return None;
        }

        let mut pos = 0;
        let epoch = u64::from_le_bytes(bytes[pos..pos + 8].try_into().ok()?);
        pos += 8;

        let mut hash_a = [0u64; DEPTH];
        for item in hash_a.iter_mut() {
            *item = u64::from_le_bytes(bytes[pos..pos + 8].try_into().ok()?);
            pos += 8;
        }
        let mut hash_b = [0u64; DEPTH];
        for item in hash_b.iter_mut() {
            *item = u64::from_le_bytes(bytes[pos..pos + 8].try_into().ok()?);
            pos += 8;
        }

        let sketch = Self::alloc(None, 0, hash_a, hash_b);
        sketch.epoch.store(epoch, Ordering::Relaxed);

        for row in 0..DEPTH {
            for col in 0..WIDTH {
                let val = u32::from_le_bytes(bytes[pos..pos + 4].try_into().ok()?);
                sketch.rows[row].0[col].store(val, Ordering::Relaxed);
                pos += 4;
            }
        }
        Some(sketch)
    }
}

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

    // === Bit Packing Tests ===

    #[test]
    fn pack_unpack_roundtrip() {
        for ts in [0, 1, 127, 255] {
            for count in [0, 1, 1000, MAX_COUNT] {
                let packed = pack(ts, count);
                let (got_ts, got_count) = unpack(packed);
                assert_eq!(got_ts, ts, "ts mismatch");
                assert_eq!(got_count, count, "count mismatch");
            }
        }
    }

    #[test]
    fn pack_truncates_overflow() {
        // Timestamp overflow (>255) should be masked
        let (ts, _) = unpack(pack(256, 0));
        assert_eq!(ts, 0); // 256 & 0xFF = 0

        // Count overflow should be masked
        let (_, count) = unpack(pack(0, MAX_COUNT + 1));
        assert_eq!(count, 0); // (MAX_COUNT+1) & COUNT_MASK wraps
    }

    #[test]
    fn decay_steps_same_epoch() {
        assert_eq!(decay_steps(5, 5), 0);
        assert_eq!(decay_steps(0, 0), 0);
        assert_eq!(decay_steps(255, 255), 0);
    }

    #[test]
    fn decay_steps_simple_gap() {
        assert_eq!(decay_steps(0, 1), 1);
        assert_eq!(decay_steps(0, 5), 5);
        assert_eq!(decay_steps(10, 15), 5);
    }

    #[test]
    fn decay_steps_wraparound() {
        // 8-bit wrap: epoch=2, cell_ts=250 → gap = (2-250) & 0xFF = 8
        assert_eq!(decay_steps(250, 2), 8);
    }

    #[test]
    fn decay_steps_capped_at_count_bits() {
        // Gap of 100 should be capped at 24
        assert_eq!(decay_steps(0, 100), 24);
    }

    // === Hash Tests ===

    #[test]
    fn hash_produces_valid_indices() {
        for row in 0..DEPTH {
            for id in [0, 1, u64::MAX, 0xDEADBEEF, 42] {
                let idx = compute_hash(id, row, &DEFAULT_HASH_A, &DEFAULT_HASH_B);
                assert!(idx < WIDTH, "hash out of range: {}", idx);
            }
        }
    }

    #[test]
    fn hash_different_rows_differ() {
        // Same ID should map to different columns in different rows
        // (with very high probability for most IDs)
        let id = 0xDEADBEEF_u64;
        let indices: Vec<usize> = (0..DEPTH)
            .map(|r| compute_hash(id, r, &DEFAULT_HASH_A, &DEFAULT_HASH_B))
            .collect();
        // At least 2 of 4 should differ (probabilistic but near-certain)
        let unique: std::collections::HashSet<_> = indices.iter().collect();
        assert!(unique.len() >= 2, "hash functions not independent enough");
    }

    #[test]
    fn hash_distribution_uniformity() {
        // Insert 100K random-ish IDs, check bucket distribution
        let mut buckets = vec![0u32; WIDTH];
        for id in 0..100_000u64 {
            let idx = compute_hash(
                id.wrapping_mul(0x12345),
                0,
                &DEFAULT_HASH_A,
                &DEFAULT_HASH_B,
            );
            buckets[idx] += 1;
        }
        let max = *buckets.iter().max().unwrap();
        let expected = 100_000.0 / WIDTH as f64;
        // Max bucket should be < 10x expected (very loose bound)
        assert!(
            max as f64 <= expected * 10.0,
            "distribution too skewed: max={}, expected={:.1}",
            max,
            expected
        );
    }

    // === Core Algorithm Tests ===

    #[test]
    fn basic_increment_and_query() {
        let sketch = ClTds::new_deterministic();
        let id = 42u64;

        // Insert 100 times
        for _ in 0..100 {
            sketch.increment(id);
        }

        // Query should return exactly 100 (no collisions expected for single item)
        let count = sketch.query(id);
        assert_eq!(count, 100, "expected 100, got {}", count);
    }

    #[test]
    fn query_unseen_item_returns_zero() {
        let sketch = ClTds::new_deterministic();
        sketch.increment(1);
        sketch.increment(2);

        // Item 999 was never inserted
        let count = sketch.query(999);
        assert_eq!(count, 0, "unseen item should have count 0, got {}", count);
    }

    #[test]
    fn multiple_items_independent() {
        let sketch = ClTds::new_deterministic();

        sketch.increment(100);
        sketch.increment(100);
        sketch.increment(100);

        sketch.increment(200);

        assert_eq!(sketch.query(100), 3);
        assert_eq!(sketch.query(200), 1);
    }

    #[test]
    fn counter_saturation() {
        let sketch = ClTds::new_deterministic();
        let id = 1u64;

        // Manually set a cell near MAX_COUNT to test saturation
        let col = sketch.hash(id, 0);
        let near_max = pack(0, MAX_COUNT - 1);
        sketch.rows[0].0[col].store(near_max, Ordering::Relaxed);

        // Increment should saturate at MAX_COUNT, not overflow
        sketch.increment(id);
        let val = sketch.rows[0].0[col].load(Ordering::Relaxed);
        let (_, count) = unpack(val);
        assert_eq!(count, MAX_COUNT);
    }

    // === Decay Tests ===

    #[test]
    fn decay_halves_count() {
        let sketch = ClTds::new_deterministic();
        let id = 1u64;

        // Insert 1000 times at epoch 0
        for _ in 0..1000 {
            sketch.increment(id);
        }
        assert_eq!(sketch.query(id), 1000);

        // Advance 1 epoch → count should halve
        sketch.tick_epoch();
        assert_eq!(sketch.query(id), 500); // 1000 >> 1

        // Advance 1 more → halve again
        sketch.tick_epoch();
        assert_eq!(sketch.query(id), 250); // 1000 >> 2
    }

    #[test]
    fn decay_multiple_epochs() {
        let sketch = ClTds::new_deterministic();
        let id = 7u64;

        for _ in 0..1024 {
            sketch.increment(id);
        }

        // Advance 5 epochs at once
        for _ in 0..5 {
            sketch.tick_epoch();
        }

        // 1024 >> 5 = 32
        assert_eq!(sketch.query(id), 32);
    }

    #[test]
    fn decay_to_zero() {
        let sketch = ClTds::new_deterministic();
        let id = 1u64;

        sketch.increment(id);
        assert_eq!(sketch.query(id), 1);

        // Advance 1 epoch → 1 >> 1 = 0
        sketch.tick_epoch();
        assert_eq!(sketch.query(id), 0);
    }

    #[test]
    fn decay_with_interleaved_inserts() {
        let sketch = ClTds::new_deterministic();
        let id = 1u64;

        // Epoch 0: insert 100
        for _ in 0..100 {
            sketch.increment(id);
        }

        // Epoch 1: insert 50 more
        sketch.tick_epoch();
        for _ in 0..50 {
            sketch.increment(id);
        }

        // Query: 100 >> 1 + 50 = 50 + 50 = 100
        // (first 100 decayed by 1 epoch, plus 50 new)
        assert_eq!(sketch.query(id), 100);
    }

    // === Claim 1 Verification: Error Bound ===

    #[test]
    fn claim1_no_undercounting() {
        // CMS guarantee: query(x) ≥ true_decayed_count (no undercount)
        let sketch = ClTds::new_deterministic();
        let target = 42u64;

        for _ in 0..500 {
            sketch.increment(target);
        }
        // Add noise from other items
        for id in 1000..2000u64 {
            sketch.increment(id);
        }

        let est = sketch.query(target);
        assert!(
            est >= 500,
            "Claim 1 violation: undercounting! got {} < 500",
            est
        );
    }

    #[test]
    fn claim1_overcount_bounded() {
        // E[query(x)] ≤ f(x) + ε·N where ε = e/w ≈ 0.0000414
        let sketch = ClTds::new_deterministic();

        let heavy_hitter = 42u64;
        let true_count = 1000u32;

        for _ in 0..true_count {
            sketch.increment(heavy_hitter);
        }

        // Insert N total other items (noise)
        let n_noise = 100_000u64;
        for id in 0..n_noise {
            sketch.increment(id + 10_000);
        }

        let n_total = true_count as f64 + n_noise as f64;
        let epsilon = std::f64::consts::E / WIDTH as f64;
        let max_overcount = epsilon * n_total;

        let est = sketch.query(heavy_hitter);
        let overcount = est as f64 - true_count as f64;

        // Allow 10x the expected error (statistical slack)
        assert!(
            overcount <= max_overcount * 10.0,
            "Claim 1 error too large: overcount={:.0}, bound={:.0}",
            overcount,
            max_overcount * 10.0
        );
    }

    // === Claim 2 Verification: False Positive Bound ===

    #[test]
    fn claim2_false_positive_rate() {
        // Insert some heavy hitters, then query many non-existent items.
        // False positive rate should be ≤ ~2% (δ ≤ e^{-4}).
        let sketch = ClTds::new_deterministic();

        // Insert 100 heavy hitters
        for id in 0..100u64 {
            for _ in 0..1000 {
                sketch.increment(id);
            }
        }

        let _n_total = 100 * 1000;
        let threshold = 50u32; // items above this = "heavy hitter"

        // Query 10,000 non-existent items
        let n_queries = 10_000u64;
        let mut false_positives = 0u64;

        for id in 1_000_000..1_000_000 + n_queries {
            if sketch.query(id) > threshold {
                false_positives += 1;
            }
        }

        let fp_rate = false_positives as f64 / n_queries as f64;

        // δ ≤ e^{-4} ≈ 1.8%, allow 5% margin
        assert!(
            fp_rate <= 0.05,
            "Claim 2 violation: false positive rate {:.2}% > 5%",
            fp_rate * 100.0
        );
    }

    // === Claim 3 Verification: Lazy = Full Decay ===

    #[test]
    fn claim3_lazy_equals_full_decay() {
        // Verify: decaying step-by-step = decaying all-at-once
        // This is the core mathematical claim: 2^{-a} · 2^{-b} = 2^{-(a+b)}
        for initial in [1, 7, 100, 1000, 65535, MAX_COUNT] {
            for total_steps in 0..=24u32 {
                // Full decay: shift 1 bit at a time
                let mut full = initial;
                for _ in 0..total_steps {
                    full >>= 1;
                }

                // Lazy decay: shift all at once
                let lazy = apply_decay(initial, total_steps);

                assert_eq!(
                    full, lazy,
                    "Claim 3 violation! initial={}, steps={}: full={}, lazy={}",
                    initial, total_steps, full, lazy
                );
            }
        }
    }

    #[test]
    fn claim3_lazy_decay_in_sketch() {
        // Verify lazy decay produces same result as manual step-by-step
        let sketch_lazy = ClTds::new_deterministic();
        let id = 99u64;

        // Insert 1024 at epoch 0
        for _ in 0..1024 {
            sketch_lazy.increment(id);
        }

        // Advance 5 epochs (lazy: decay applied at next query)
        for _ in 0..5 {
            sketch_lazy.tick_epoch();
        }
        let lazy_result = sketch_lazy.query(id);

        // Manual full decay: 1024 >> 5 = 32
        let full_result = 1024u32 >> 5;

        assert_eq!(
            lazy_result, full_result,
            "Claim 3 in-sketch: lazy={}, full={}",
            lazy_result, full_result
        );
    }

    // === Thread Safety Test ===

    #[test]
    fn concurrent_increments() {
        use std::sync::Arc;

        let sketch = Arc::new(ClTds::new_deterministic());
        let id = 42u64;
        let threads = 4;
        let per_thread = 10_000;

        std::thread::scope(|s| {
            for _ in 0..threads {
                let sk = Arc::clone(&sketch);
                s.spawn(move || {
                    for _ in 0..per_thread {
                        sk.increment(id);
                    }
                });
            }
        });

        let total = sketch.query(id);
        let expected = (threads * per_thread) as u32;

        assert_eq!(
            total, expected,
            "Thread safety: got {}, expected {}",
            total, expected
        );
    }

    // === Memory Size Verification ===

    #[test]
    fn matrix_size_is_1mb() {
        let sketch = ClTds::new_deterministic();
        let size = sketch.memory_bytes();
        assert_eq!(
            size, 1_048_576,
            "Matrix should be exactly 1 MB, got {} bytes",
            size
        );
    }

    #[test]
    fn epoch_advances() {
        let sketch = ClTds::new_deterministic();
        assert_eq!(sketch.epoch(), 0);
        sketch.tick_epoch();
        assert_eq!(sketch.epoch(), 1);
        sketch.tick_epoch();
        sketch.tick_epoch();
        assert_eq!(sketch.epoch(), 3);
    }

    // === Claim 2: Zipf Real-World Stress Test ===

    #[test]
    fn claim2_zipf_stress_test() {
        // Real network traffic follows Zipf distribution:
        //   top 1% of IPs = ~80% of all packets
        // This tests Claim 2 under realistic conditions.
        let sketch = ClTds::new_deterministic();

        let n_heavy = 10u64; // 10 heavy hitters (top ~1%)
        let n_mice = 990u64; // 990 normal IPs
        let per_heavy = 8_000u64; // 80K total from heavy hitters
        let per_mouse = 20u64; // 19.8K total from mice
                               // Total ≈ 100K packets, heavy = 80%

        // Insert heavy hitters
        for id in 1..=n_heavy {
            for _ in 0..per_heavy {
                sketch.increment(id);
            }
        }

        // Insert normal traffic
        for id in 0..n_mice {
            for _ in 0..per_mouse {
                sketch.increment(id + 100_000);
            }
        }

        // All heavy hitters should be detected
        let threshold = 1000u32;
        for id in 1..=n_heavy {
            let count = sketch.query(id);
            assert!(
                count >= per_heavy as u32,
                "Zipf: heavy hitter {} not detected: count={} < {}",
                id,
                count,
                per_heavy
            );
        }

        // False positive check on 50K unseen IPs
        let n_test = 50_000u64;
        let mut fp = 0u64;
        for id in 1_000_000..1_000_000 + n_test {
            if sketch.query(id) > threshold {
                fp += 1;
            }
        }
        let fp_rate = fp as f64 / n_test as f64;
        assert!(
            fp_rate <= 0.02,
            "Zipf Claim 2: false positive rate {:.3}% exceeds 2%",
            fp_rate * 100.0
        );
    }

    // === Auto Epoch Tests ===

    #[test]
    fn manual_mode_by_default() {
        let sketch = ClTds::new_deterministic();
        assert!(!sketch.is_auto_epoch());
        assert_eq!(sketch.epoch(), 0);
    }

    #[test]
    fn auto_mode_via_constructor() {
        let sketch = ClTds::with_epoch_interval(1000);
        assert!(sketch.is_auto_epoch());
        // Epoch starts at 0 (just created)
        assert_eq!(sketch.epoch(), 0);
    }

    #[test]
    fn auto_epoch_advances_with_time() {
        // 1ms per epoch → should advance quickly
        let sketch = ClTds::with_epoch_interval(1);

        // Insert some data
        for _ in 0..100 {
            sketch.increment(42);
        }

        // Sleep 15ms → epoch should be ≥10
        std::thread::sleep(std::time::Duration::from_millis(15));
        let epoch = sketch.epoch();
        assert!(
            epoch >= 10,
            "auto epoch should advance with time, got {}",
            epoch
        );
    }

    #[test]
    fn auto_epoch_decay_works() {
        // 10ms per epoch
        let sketch = ClTds::with_epoch_interval(10);

        for _ in 0..1024 {
            sketch.increment(7);
        }
        assert_eq!(sketch.query(7), 1024);

        // Wait 50ms → ~5 epochs → 1024 >> 5 = 32
        std::thread::sleep(std::time::Duration::from_millis(55));
        let count = sketch.query(7);
        // Allow some tolerance (timing isn't exact)
        assert!(
            count <= 64 && count >= 16,
            "auto decay after ~5 epochs: expected ~32, got {}",
            count
        );
    }

    #[test]
    #[should_panic(expected = "epoch interval must be > 0")]
    fn auto_epoch_zero_interval_panics() {
        let _sketch = ClTds::with_epoch_interval(0);
    }

    // === Adversarial Hash Resistance Tests ===

    #[test]
    fn two_sketches_different_hashes() {
        // Two new() sketches should have different hash mappings
        let s1 = ClTds::new();
        let s2 = ClTds::new();

        let id = 42u64;
        let idx1: Vec<usize> = (0..DEPTH).map(|r| s1.hash(id, r)).collect();
        let idx2: Vec<usize> = (0..DEPTH).map(|r| s2.hash(id, r)).collect();

        // With random hashes, extremely unlikely to be identical
        assert_ne!(
            idx1, idx2,
            "Two sketches should have different hash mappings"
        );
    }

    #[test]
    fn random_hashes_produce_valid_results() {
        // new() with random hashes should still work correctly
        let sketch = ClTds::new(); // random hashes

        for _ in 0..100 {
            sketch.increment(42);
        }
        let count = sketch.query(42);
        assert_eq!(count, 100, "random hashes: expected 100, got {}", count);
        assert_eq!(sketch.query(999999), 0, "unseen item should be 0");
    }

    #[test]
    fn deterministic_hashes_reproducible() {
        // Two deterministic sketches should produce identical results
        let s1 = ClTds::new_deterministic();
        let s2 = ClTds::new_deterministic();

        for _ in 0..500 {
            s1.increment(42);
            s2.increment(42);
        }

        assert_eq!(s1.query(42), s2.query(42));
    }

    // === Algorithm Parameters Tests ===

    #[test]
    fn algorithm_parameters_correct() {
        let (epsilon, delta, w, d) = ClTds::algorithm_parameters();

        assert_eq!(w, 65536);
        assert_eq!(d, 4);
        assert!((epsilon - std::f64::consts::E / 65536.0).abs() < 1e-10);
        assert!((delta - (-4.0_f64).exp()).abs() < 1e-10);

        // Human-readable checks
        assert!(epsilon < 0.0001, "ε should be tiny");
        assert!(delta < 0.02, "δ should be < 2%");
    }

    #[test]
    fn error_bound_scales_with_stream() {
        let err_1m = ClTds::error_bound(1_000_000);
        let err_10m = ClTds::error_bound(10_000_000);

        // Error should scale linearly with N
        assert!((err_10m / err_1m - 10.0).abs() < 0.001);

        // Error for 1M stream: e/65536 * 1M ≈ 41.4
        assert!(err_1m < 50.0, "error bound for 1M stream should be < 50");
    }

    // === Phase 4: Complete Algorithm Validation ===

    #[test]
    fn adversarial_collision_min_filter_holds() {
        // Even if an attacker floods specific IDs to pollute buckets,
        // the 4-row min-filter should prevent false positives for
        // unrelated queries.
        let sketch = ClTds::new_deterministic();

        // Attacker floods 1000 different IDs, each 1000 times
        for id in 0..1000u64 {
            for _ in 0..1000 {
                sketch.increment(id);
            }
        }

        // Query 10,000 IDs that were NEVER inserted
        // Min-filter should keep most at 0 or very low
        let mut high_false = 0u64;
        for id in 500_000..510_000u64 {
            let count = sketch.query(id);
            if count > 100 {
                high_false += 1;
            }
        }

        let fp_rate = high_false as f64 / 10_000.0;
        assert!(
            fp_rate < 0.02,
            "Adversarial: false positive rate {:.2}% exceeds 2%",
            fp_rate * 100.0
        );
    }

    #[test]
    fn decay_accuracy_over_many_epochs() {
        // Verify decay stays mathematically exact over 20 epochs
        let sketch = ClTds::new_deterministic();
        let id = 1u64;
        let initial = 1_000_000u32;

        for _ in 0..initial {
            sketch.increment(id);
        }

        // Verify exact halving for each epoch (up to 20)
        for epoch in 1..=20u32 {
            sketch.tick_epoch();
            let expected = initial >> epoch;
            let actual = sketch.query(id);
            assert_eq!(
                actual, expected,
                "Epoch {}: expected {}, got {}",
                epoch, expected, actual
            );
        }
    }

    #[test]
    fn concurrent_increment_with_decay() {
        // Multiple threads inserting WHILE epoch advances
        use std::sync::Arc;

        let sketch = Arc::new(ClTds::new_deterministic());
        let id = 99u64;
        let threads = 4;
        let per_thread = 5_000;

        std::thread::scope(|s| {
            // 4 writer threads
            for _ in 0..threads {
                let sk = Arc::clone(&sketch);
                s.spawn(move || {
                    for _ in 0..per_thread {
                        sk.increment(id);
                    }
                });
            }

            // 1 epoch ticker thread (advances 3 times during inserts)
            let sk = Arc::clone(&sketch);
            s.spawn(move || {
                std::thread::sleep(std::time::Duration::from_millis(1));
                sk.tick_epoch();
                std::thread::sleep(std::time::Duration::from_millis(1));
                sk.tick_epoch();
                std::thread::sleep(std::time::Duration::from_millis(1));
                sk.tick_epoch();
            });
        });

        // With 3 decay epochs during 20K inserts, count should be
        // significantly less than 20K but still substantial
        let count = sketch.query(id);
        assert!(
            count > 0 && count <= (threads * per_thread) as u32,
            "Concurrent+decay: count {} out of reasonable range",
            count
        );
    }

    #[test]
    fn full_integration_test() {
        // Combines: random hashes + auto epoch + Zipf + decay + query
        let sketch = ClTds::new(); // random hashes (adversarial-resistant)

        // Phase 1: Insert heavy hitters
        for id in 1..=5u64 {
            for _ in 0..10_000 {
                sketch.increment(id);
            }
        }

        // Phase 2: Insert normal traffic
        for id in 100..1100u64 {
            for _ in 0..10 {
                sketch.increment(id);
            }
        }

        // Phase 3: Verify detection
        for id in 1..=5u64 {
            assert!(
                sketch.query(id) >= 10_000,
                "Integration: heavy hitter {} not detected",
                id
            );
        }

        // Phase 4: Decay and verify forgetting
        for _ in 0..10 {
            sketch.tick_epoch();
        }
        for id in 1..=5u64 {
            let decayed = sketch.query(id);
            // 10000 >> 10 = 9
            assert!(
                decayed <= 15,
                "Integration: heavy hitter {} not forgotten after decay: {}",
                id,
                decayed
            );
        }

        // Phase 5: New traffic after decay should be detected fresh
        for _ in 0..500 {
            sketch.increment(42);
        }
        assert_eq!(
            sketch.query(42),
            500,
            "New traffic after decay should be exact"
        );

        // Phase 6: Verify algorithm parameters are accessible
        let (eps, delta, w, d) = ClTds::algorithm_parameters();
        assert!(eps > 0.0 && delta > 0.0 && w == WIDTH && d == DEPTH);
    }
}