sz-orm-core 6.0.0

Core ORM engine: Model trait, ActiveRecord, QueryBuilder, Pool, Transaction, migration, and SQL dialect abstraction
Documentation
//! 布隆过滤器(公共实现)
//!
//! v4.7.0 架构债清零:合并 `cache_warmup_protection::BloomFilter`(自研)与
//! `dist_cache::BloomFilterGuard`(bloomfilter crate)双实现为单一公共模块。
//! 设计取舍:
//!   - 并发安全:内部 `RwLock`,`add`/`might_contain` 均为 `&self`
//!   - 容量拒绝:超过 `capacity` 的 `add` 返回 `CapacityExceeded`(不漏判语义:
//!     拒绝写入而非静默丢位,避免"存在但查不到")
//!   - 误判率:`might_contain` 假阳性 ≤ fpp,不存在一定返回 false
//!
//! 并发正确性:常规并发测试验证"add 后 might_contain 必命中"(不漏判不变量)。
//! 注:loom 模型检查曾尝试引入,但 RUSTFLAGS=--cfg loom 会污染依赖树
//! (crossbeam-queue → concurrent-queue 等无 loom 适配,2026-08-14 评估不可行),
//! 并发验证采用常规多线程测试 + chaos/stress 组合。

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

/// 布隆过滤器错误
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BloomError {
    /// 超出容量(拒绝写入,保持不漏判)
    CapacityExceeded {
        /// 配置容量
        capacity: usize,
        /// 请求写入时的元素数(= 容量 + 1)
        requested: usize,
    },
}

/// 并发安全布隆过滤器
///
/// ```
/// use sz_orm_core::bloom::BloomFilter;
///
/// let mut filter = BloomFilter::new(100, 0.01);
/// filter.add("key-1").unwrap();
/// assert!(filter.might_contain("key-1"));
/// assert!(!filter.might_contain("key-2")); // 不存在一定返回 false(不漏判)
/// ```
pub struct BloomFilter {
    bits: RwLock<Vec<u64>>,
    num_bits: usize,
    num_hashes: usize,
    capacity: usize,
    count: AtomicUsize,
}

impl BloomFilter {
    /// 创建布隆过滤器
    ///
    /// `capacity` 预期元素数量,`fpp` 误判率(0~1,内部收敛到 0.0001~0.5)。
    pub fn new(capacity: usize, fpp: f64) -> Self {
        let capacity = capacity.max(1);
        let fpp = fpp.clamp(0.0001, 0.5);
        let ln2 = std::f64::consts::LN_2;
        let m = (-(capacity as f64) * fpp.ln() / (ln2 * ln2)).ceil() as usize;
        let m = m.max(8);
        let k = ((m as f64 / capacity as f64) * ln2).ceil() as usize;
        let k = k.max(1);
        let num_words = m.div_ceil(64);
        Self {
            bits: RwLock::new(vec![0u64; num_words]),
            num_bits: m,
            num_hashes: k,
            capacity,
            count: AtomicUsize::new(0),
        }
    }

    /// 添加键(容量满时返回 `CapacityExceeded`,拒绝写入保持不漏判)
    pub fn add(&self, key: &str) -> Result<(), BloomError> {
        let count = self.count.load(Ordering::Relaxed);
        if count >= self.capacity {
            return Err(BloomError::CapacityExceeded {
                capacity: self.capacity,
                requested: count + 1,
            });
        }
        let (h1, h2) = self.hash(key);
        let mut bits = self.bits.write().unwrap();
        for i in 0..self.num_hashes {
            let combined = h1.wrapping_add((i as u64).wrapping_mul(h2));
            let idx = (combined as usize) % self.num_bits;
            bits[idx / 64] |= 1u64 << (idx % 64);
        }
        self.count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// 检查键可能存在(不漏判:不存在一定返回 false;假阳性 ≤ fpp)
    pub fn might_contain(&self, key: &str) -> bool {
        let (h1, h2) = self.hash(key);
        let bits = self.bits.read().unwrap();
        for i in 0..self.num_hashes {
            let combined = h1.wrapping_add((i as u64).wrapping_mul(h2));
            let idx = (combined as usize) % self.num_bits;
            if bits[idx / 64] & (1u64 << (idx % 64)) == 0 {
                return false;
            }
        }
        true
    }

    /// 当前元素计数
    pub fn count(&self) -> usize {
        self.count.load(Ordering::Relaxed)
    }

    /// 是否为空
    pub fn is_empty(&self) -> bool {
        self.count() == 0
    }

    /// 容量
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// 清空(保留容量/误判率配置)
    pub fn clear(&self) {
        let mut bits = self.bits.write().unwrap();
        bits.iter_mut().for_each(|w| *w = 0);
        self.count.store(0, Ordering::Relaxed);
    }

    /// 双哈希(xxhash 风格混合:基于 FNV-1a 的 h1 + 二次扰动 h2)
    fn hash(&self, key: &str) -> (u64, u64) {
        let mut h1: u64 = 0xcbf29ce484222325;
        let mut h2: u64 = 0x9e3779b97f4a7c15;
        for b in key.bytes() {
            h1 ^= b as u64;
            h1 = h1.wrapping_mul(0x100000001b3);
            h2 = h2.wrapping_add(b as u64);
            h2 = h2.wrapping_mul(0x100000001b3);
        }
        h2 ^= h2 >> 33;
        h2 = h2.wrapping_mul(0xff51afd7ed558ccd);
        (h1, h2)
    }
}

// ============================================================================
// 并发测试:并发 add 后 might_contain 必命中(不漏判的并发不变量)
// ============================================================================
#[cfg(test)]
mod concurrent_tests {
    use super::*;

    /// 多线程并发写入不同 key,全部完成后 must_contain 必命中。
    /// 验证写锁内原子完成(add 返回即可见)。
    #[test]
    fn bloom_concurrent_add_no_false_negative() {
        use std::sync::Arc;
        use std::thread;

        let filter = Arc::new(BloomFilter::new(4096, 0.01));
        let mut handles = vec![];
        for t in 0..8 {
            let f = Arc::clone(&filter);
            handles.push(thread::spawn(move || {
                for i in 0..200 {
                    f.add(&format!("thread-{t}-key-{i}")).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        // 并发不变量:写入完成后必须命中(不存在才允许 false)
        for t in 0..8 {
            for i in 0..200 {
                assert!(
                    filter.might_contain(&format!("thread-{t}-key-{i}")),
                    "add 后 must_contain 必须命中(并发不漏判): thread-{t}-key-{i}"
                );
            }
        }
        assert_eq!(filter.count(), 1600);
    }

    /// 并发读写混合:might_contain 不得 panic(RwLock 正确性冒烟)
    #[test]
    fn bloom_concurrent_read_write_smoke() {
        use std::sync::Arc;
        use std::thread;

        let filter = Arc::new(BloomFilter::new(2048, 0.01));
        let mut handles = vec![];
        for t in 0..4 {
            let f = Arc::clone(&filter);
            handles.push(thread::spawn(move || {
                for i in 0..100 {
                    let key = format!("t{t}-k{i}");
                    let _ = f.add(&key);
                    let _ = f.might_contain(&key);
                    let _ = f.count();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        assert!(filter.count() > 0);
    }
}