Packages
ferricstore
0.1.0
0.11.14
0.11.12
0.11.11
0.11.10
0.11.9
0.11.8
0.11.7
0.11.6
0.11.5
0.11.4
0.11.3
0.11.2
0.11.1
0.11.0
0.10.3
0.10.2
0.10.1
0.10.0
0.9.1
0.9.0
0.8.0
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.0
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.2.0
0.1.0
FerricFlow durable workflows and queues with native-protocol storage, Raft durability, and Bitcask persistence.
Current section
Files
Jump to
Current section
Files
native/ferricstore_bitcask/src/bloom.rs
//! Memory-mapped Bloom filter exposed as Rustler NIFs.
//!
//! Each Bloom filter is stored as a file on disk and memory-mapped via `mmap`.
//! The file layout is:
//!
//! ```text
//! [header: 32 bytes][bit array: ceil(num_bits / 8) bytes]
//! ```
//!
//! Header (32 bytes, little-endian):
//! - bytes 0..7: magic number (0x424C4F4F4D465F31 = "BLOOMF_1")
//! - bytes 8..15: num_bits (u64)
//! - bytes 16..19: num_hashes (u32)
//! - bytes 20..23: reserved (u32, zero)
//! - bytes 24..31: count (u64) — number of elements inserted
//!
//! The bit array is directly accessible via the mmap pointer for nanosecond-level
//! bit reads. Writes flush to disk via `msync`.
//!
//! ## Hash functions
//!
//! Uses the Kirsch-Mitzenmacker (2006) enhanced double-hashing technique:
//! `h_i(x) = (h1(x) + i * h2(x)) mod m`
//!
//! where h1 and h2 are derived from MD5(element) split into two 64-bit halves.
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use rustler::schedule::consume_timeslice;
use rustler::{Binary, Encoder, Env, NifResult, ResourceArc, Term};
/// How often (in items) to call `consume_timeslice` and let the BEAM
/// decide whether we should yield. 64 matches the interval used in lib.rs.
const YIELD_CHECK_INTERVAL: usize = 64;
const MAGIC: u64 = 0x424C_4F4F_4D46_5F31; // "BLOOMF_1"
const HEADER_SIZE: usize = 32;
// ---------------------------------------------------------------------------
// BloomFilter core
// ---------------------------------------------------------------------------
/// A memory-mapped Bloom filter.
///
/// The mmap region covers the entire file (header + bit array). All mutations
/// go through the mmap pointer; `msync` is called after writes to ensure
/// durability.
pub struct BloomFilter {
/// Memory-mapped file contents (header + bit array).
mmap: *mut u8,
/// Length of the mmap region in bytes.
mmap_len: usize,
/// Number of bits in the filter (m).
num_bits: u64,
/// Number of hash functions (k).
num_hashes: u32,
/// Path to the backing file.
path: PathBuf,
}
// SAFETY: The mmap pointer is protected by a Mutex in BloomResource. All access
// is serialized — only one thread touches the pointer at a time. Reads through
// mmap are safe for concurrent access (the kernel guarantees coherent reads from
// a shared mapping), and we serialize writes via the Mutex.
unsafe impl Send for BloomFilter {}
unsafe impl Sync for BloomFilter {}
impl BloomFilter {
/// Create a new bloom filter file at `path` with the given parameters.
///
/// The file is created, sized, and memory-mapped. All bits start at zero.
pub fn create(path: &Path, num_bits: u64, num_hashes: u32) -> Result<Self, String> {
if num_bits == 0 {
return Err("num_bits must be > 0".into());
}
if num_hashes == 0 {
return Err("num_hashes must be > 0".into());
}
// Ensure parent directory exists.
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).map_err(|e| format!("mkdir: {e}"))?;
}
let byte_count = num_bits.div_ceil(8) as usize;
let file_size = HEADER_SIZE + byte_count;
// Write the file with header + zeroed bit array.
let mut file = File::create(path).map_err(|e| format!("create: {e}"))?;
let mut header = [0u8; HEADER_SIZE];
header[0..8].copy_from_slice(&MAGIC.to_le_bytes());
header[8..16].copy_from_slice(&num_bits.to_le_bytes());
header[16..20].copy_from_slice(&num_hashes.to_le_bytes());
// bytes 20..24 reserved (zero)
// bytes 24..32 count = 0
file.write_all(&header)
.map_err(|e| format!("write header: {e}"))?;
// Write zeroed bit array.
let zeros = vec![0u8; byte_count];
file.write_all(&zeros)
.map_err(|e| format!("write bits: {e}"))?;
file.sync_all().map_err(|e| format!("fsync: {e}"))?;
drop(file);
// Now mmap the file.
Self::open(path, file_size)
}
/// Open an existing bloom filter file and memory-map it.
pub fn open_existing(path: &Path) -> Result<Self, String> {
let meta = fs::metadata(path).map_err(|e| format!("stat: {e}"))?;
let file_size = meta.len() as usize;
if file_size < HEADER_SIZE {
return Err("file too small for bloom header".into());
}
Self::open(path, file_size)
}
fn open(path: &Path, file_size: usize) -> Result<Self, String> {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(|e| format!("open: {e}"))?;
let mmap = unsafe {
libc::mmap(
std::ptr::null_mut(),
file_size,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_SHARED,
std::os::unix::io::AsRawFd::as_raw_fd(&file),
0,
)
};
if mmap == libc::MAP_FAILED {
return Err(format!("mmap failed: {}", std::io::Error::last_os_error()));
}
let ptr = mmap as *mut u8;
// Validate header.
let magic = u64::from_le_bytes(unsafe {
let mut buf = [0u8; 8];
std::ptr::copy_nonoverlapping(ptr, buf.as_mut_ptr(), 8);
buf
});
if magic != MAGIC {
unsafe {
libc::munmap(mmap, file_size);
}
return Err("invalid bloom file magic".into());
}
let num_bits = u64::from_le_bytes(unsafe {
let mut buf = [0u8; 8];
std::ptr::copy_nonoverlapping(ptr.add(8), buf.as_mut_ptr(), 8);
buf
});
let num_hashes = u32::from_le_bytes(unsafe {
let mut buf = [0u8; 4];
std::ptr::copy_nonoverlapping(ptr.add(16), buf.as_mut_ptr(), 4);
buf
});
let expected_size = HEADER_SIZE + num_bits.div_ceil(8) as usize;
if file_size < expected_size {
unsafe {
libc::munmap(mmap, file_size);
}
return Err(format!(
"file too small: expected {expected_size}, got {file_size}"
));
}
// M-4 fix: hint the kernel that access is random (hash-determined bit
// positions). Without MADV_RANDOM, the kernel's readahead heuristic may
// prefetch sequential pages that will never be used, wasting page cache
// and evicting hot data file pages under memory pressure.
unsafe {
libc::madvise(mmap, file_size, libc::MADV_RANDOM);
}
Ok(Self {
mmap: ptr,
mmap_len: file_size,
num_bits,
num_hashes,
path: path.to_path_buf(),
})
}
/// Read the insertion count from the header.
fn count(&self) -> u64 {
u64::from_le_bytes(unsafe {
let mut buf = [0u8; 8];
std::ptr::copy_nonoverlapping(self.mmap.add(24), buf.as_mut_ptr(), 8);
buf
})
}
/// Write the insertion count to the header.
fn set_count(&self, count: u64) {
unsafe {
std::ptr::copy_nonoverlapping(count.to_le_bytes().as_ptr(), self.mmap.add(24), 8);
}
}
/// Compute hash positions for an element using Kirsch-Mitzenmacker double hashing.
///
/// H-4 fix: replaced MD5 with xxh3 (5-10x faster for non-cryptographic hashing).
/// Uses two xxh3 hashes with different seeds to produce h1 and h2.
///
/// M-3 fix: returns an iterator instead of allocating a Vec on every call.
fn hash_positions(&self, element: &[u8]) -> impl Iterator<Item = u64> {
let h1 = xxhash_rust::xxh3::xxh3_64_with_seed(element, 0);
let h2 = xxhash_rust::xxh3::xxh3_64_with_seed(element, 0x9E37_79B9_7F4A_7C15);
let num_bits = self.num_bits;
(0..self.num_hashes as u64).map(move |i| h1.wrapping_add(i.wrapping_mul(h2)) % num_bits)
}
/// Get a bit from the mmap'd bit array.
#[inline]
fn get_bit(&self, position: u64) -> bool {
let byte_index = (position / 8) as usize;
let bit_offset = (position % 8) as u8;
let byte = unsafe { *self.mmap.add(HEADER_SIZE + byte_index) };
(byte & (1 << bit_offset)) != 0
}
/// Set a bit in the mmap'd bit array. Returns true if the bit was previously unset.
#[inline]
fn set_bit(&self, position: u64) -> bool {
let byte_index = (position / 8) as usize;
let bit_offset = (position % 8) as u8;
let ptr = unsafe { self.mmap.add(HEADER_SIZE + byte_index) };
let byte = unsafe { *ptr };
let mask = 1u8 << bit_offset;
if (byte & mask) != 0 {
false
} else {
unsafe {
*ptr = byte | mask;
}
true
}
}
/// Add an element. Returns true if the element was new (at least one new bit set).
pub fn add(&self, element: &[u8]) -> bool {
let mut any_new = false;
for pos in self.hash_positions(element) {
if self.set_bit(pos) {
any_new = true;
}
}
if any_new {
self.set_count(self.count() + 1);
}
any_new
}
/// Check if an element may exist.
pub fn exists(&self, element: &[u8]) -> bool {
self.hash_positions(element).all(|pos| self.get_bit(pos))
}
/// Flush changes to disk synchronously (MS_SYNC — blocks until pages are on disk).
pub fn msync(&self) -> Result<(), String> {
let ret =
unsafe { libc::msync(self.mmap as *mut libc::c_void, self.mmap_len, libc::MS_SYNC) };
if ret != 0 {
Err(format!("msync failed: {}", std::io::Error::last_os_error()))
} else {
Ok(())
}
}
/// Mark dirty pages for background writeback (MS_ASYNC — non-blocking).
///
/// M-2 fix: `bloom_add()` now uses this instead of the blocking `msync()`.
/// `MS_ASYNC` tells the kernel the pages are dirty and should be written back
/// at its convenience, without blocking the caller. This drops per-add cost
/// from ~100-500 us (NVMe fsync) to ~1 us.
///
/// Durability is still guaranteed by:
/// - `bloom_madd()` calls `msync()` (MS_SYNC) after all additions
/// - `Drop` impl calls `msync()` (MS_SYNC) before `munmap`
/// - Explicit `msync()` calls from Elixir
pub fn msync_async(&self) -> Result<(), String> {
let ret = unsafe {
libc::msync(
self.mmap as *mut libc::c_void,
self.mmap_len,
libc::MS_ASYNC,
)
};
if ret != 0 {
Err(format!(
"msync_async failed: {}",
std::io::Error::last_os_error()
))
} else {
Ok(())
}
}
/// Delete the bloom filter: munmap + unlink the file.
///
/// L-3 fix: use `ManuallyDrop` to avoid an unnecessary `path.clone()`
/// while still preventing Drop from double-munmapping.
pub fn delete(self) -> Result<(), String> {
let mut this = std::mem::ManuallyDrop::new(self);
unsafe {
libc::munmap(this.mmap as *mut libc::c_void, this.mmap_len);
}
// Null out the pointer so that if ManuallyDrop is ever dropped
// (it won't be, but belt-and-suspenders), the Drop impl is a no-op.
this.mmap = std::ptr::null_mut();
fs::remove_file(&this.path).map_err(|e| format!("unlink: {e}"))?;
Ok(())
}
/// Return filter info.
pub fn info(&self) -> (u64, u64, u32) {
(self.num_bits, self.count(), self.num_hashes)
}
/// Return the file path.
pub fn path(&self) -> &Path {
&self.path
}
}
impl Drop for BloomFilter {
fn drop(&mut self) {
if !self.mmap.is_null() {
// Best-effort msync before munmap.
unsafe {
libc::msync(self.mmap as *mut libc::c_void, self.mmap_len, libc::MS_SYNC);
libc::munmap(self.mmap as *mut libc::c_void, self.mmap_len);
}
}
}
}
// H-4 fix: removed duplicate hand-rolled MD5 implementation.
// Bloom filter now uses xxh3 (non-cryptographic, 5-10x faster than MD5)
// via the hash_positions() method above.
// ---------------------------------------------------------------------------
// NIF resource
// ---------------------------------------------------------------------------
/// Wraps `BloomFilter` in a `Mutex` for `Send + Sync` required by `ResourceArc`.
pub struct BloomResource {
pub filter: Mutex<BloomFilter>,
}
// ---------------------------------------------------------------------------
// NIF atoms
// ---------------------------------------------------------------------------
mod atoms {
rustler::atoms! {
ok,
error,
}
}
// ---------------------------------------------------------------------------
// NIF functions
// ---------------------------------------------------------------------------
/// Create a new bloom filter file at the given path.
/// Returns `{:ok, resource}` or `{:error, reason}`.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_create(env: Env, path: String, num_bits: u64, num_hashes: u32) -> NifResult<Term> {
match BloomFilter::create(Path::new(&path), num_bits, num_hashes) {
Ok(filter) => {
let resource = ResourceArc::new(BloomResource {
filter: Mutex::new(filter),
});
Ok((atoms::ok(), resource).encode(env))
}
Err(e) => Ok((atoms::error(), e).encode(env)),
}
}
/// Open an existing bloom filter file.
/// Returns `{:ok, resource}` or `{:error, reason}`.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_open(env: Env, path: String) -> NifResult<Term> {
match BloomFilter::open_existing(Path::new(&path)) {
Ok(filter) => {
let resource = ResourceArc::new(BloomResource {
filter: Mutex::new(filter),
});
Ok((atoms::ok(), resource).encode(env))
}
Err(e) => Ok((atoms::error(), e).encode(env)),
}
}
/// Add an element to a bloom filter.
/// Returns 1 if the element was new, 0 if it probably already existed.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_add<'a>(
env: Env<'a>,
resource: ResourceArc<BloomResource>,
element: Binary<'a>,
) -> NifResult<Term<'a>> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
let is_new = filter.add(element.as_slice());
// M-2 fix: use MS_ASYNC (non-blocking) for individual adds instead of
// MS_SYNC which blocks until pages are flushed to disk (~100-500 us per call).
// Durability is guaranteed by msync(MS_SYNC) in bloom_madd, Drop, and explicit flush.
let _ = filter.msync_async();
Ok(u32::from(is_new).encode(env))
}
/// Add multiple elements to a bloom filter.
/// Returns a list of 1/0 for each element (1 = new, 0 = already existed).
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_madd<'a>(
env: Env<'a>,
resource: ResourceArc<BloomResource>,
elements: Vec<Binary<'a>>,
) -> NifResult<Term<'a>> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
let mut results: Vec<u32> = Vec::with_capacity(elements.len());
for (i, e) in elements.iter().enumerate() {
results.push(u32::from(filter.add(e.as_slice())));
if i % YIELD_CHECK_INTERVAL == 0 && i > 0 {
let _ = consume_timeslice(env, 1);
}
}
let _ = filter.msync();
Ok(results.encode(env))
}
/// Check if an element may exist in the bloom filter.
/// Returns 1 if possibly present, 0 if definitely not present.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_exists<'a>(
env: Env<'a>,
resource: ResourceArc<BloomResource>,
element: Binary<'a>,
) -> NifResult<Term<'a>> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
let found = filter.exists(element.as_slice());
Ok(u32::from(found).encode(env))
}
/// Check if multiple elements may exist.
/// Returns a list of 1/0 for each element.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_mexists<'a>(
env: Env<'a>,
resource: ResourceArc<BloomResource>,
elements: Vec<Binary<'a>>,
) -> NifResult<Term<'a>> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
let mut results: Vec<u32> = Vec::with_capacity(elements.len());
for (i, e) in elements.iter().enumerate() {
results.push(u32::from(filter.exists(e.as_slice())));
if i % YIELD_CHECK_INTERVAL == 0 && i > 0 {
let _ = consume_timeslice(env, 1);
}
}
Ok(results.encode(env))
}
/// Return the estimated cardinality (number of items inserted).
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_card(env: Env, resource: ResourceArc<BloomResource>) -> NifResult<Term> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
Ok(filter.count().encode(env))
}
/// Return filter info as a flat list:
/// ["Capacity", capacity, "Size", size, "Number of filters", 1,
/// "Number of items inserted", count, "Expansion rate", 0,
/// "Error rate", error_rate, "Number of hash functions", k, "Number of bits", m]
///
/// Returns `{:ok, {num_bits, count, num_hashes}}`.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_info(env: Env, resource: ResourceArc<BloomResource>) -> NifResult<Term> {
let filter = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
let (num_bits, count, num_hashes) = filter.info();
Ok((atoms::ok(), (num_bits, count, num_hashes as u64)).encode(env))
}
/// Delete a bloom filter: munmap + unlink the file.
/// Returns `:ok` or `{:error, reason}`.
#[rustler::nif(schedule = "Normal")]
#[allow(clippy::needless_pass_by_value, clippy::unnecessary_wraps)]
pub fn bloom_delete(env: Env, resource: ResourceArc<BloomResource>) -> NifResult<Term> {
// Take ownership of the filter out of the Mutex.
let mut filter_guard = resource.filter.lock().map_err(|_| rustler::Error::BadArg)?;
// We need to take ownership. Create a dummy filter and swap.
// After this, the Mutex holds a null/empty filter and we have the real one.
let path = filter_guard.path.clone();
let mmap = filter_guard.mmap;
let mmap_len = filter_guard.mmap_len;
// Null out the guard so Drop doesn't double-munmap.
filter_guard.mmap = std::ptr::null_mut();
filter_guard.mmap_len = 0;
drop(filter_guard);
// munmap
if !mmap.is_null() {
unsafe {
libc::munmap(mmap as *mut libc::c_void, mmap_len);
}
}
// unlink
match fs::remove_file(&path) {
Ok(()) => Ok(atoms::ok().encode(env)),
Err(e) => Ok((atoms::error(), format!("unlink: {e}")).encode(env)),
}
}
// ---------------------------------------------------------------------------
// Rust unit tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bloom_create_add_exists() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("test.bloom");
let filter = BloomFilter::create(&path, 1000, 7).unwrap();
assert!(!filter.exists(b"hello"));
assert!(filter.add(b"hello"));
assert!(filter.exists(b"hello"));
assert!(!filter.add(b"hello")); // duplicate
assert_eq!(filter.count(), 1);
}
#[test]
fn bloom_persistence() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("persist.bloom");
{
let filter = BloomFilter::create(&path, 1000, 7).unwrap();
filter.add(b"persisted");
filter.msync().unwrap();
}
// Reopen and verify
let filter = BloomFilter::open_existing(&path).unwrap();
assert!(filter.exists(b"persisted"));
assert_eq!(filter.count(), 1);
}
#[test]
fn bloom_delete_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("delete_me.bloom");
let filter = BloomFilter::create(&path, 100, 3).unwrap();
filter.add(b"test");
assert!(path.exists());
filter.delete().unwrap();
assert!(!path.exists());
}
#[test]
fn bloom_large_filter() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("large.bloom");
let filter = BloomFilter::create(&path, 1_000_000, 7).unwrap();
for i in 0..1000 {
filter.add(format!("item_{i}").as_bytes());
}
assert_eq!(filter.count(), 1000);
// Verify all items exist
for i in 0..1000 {
assert!(filter.exists(format!("item_{i}").as_bytes()));
}
}
#[test]
fn bloom_false_positive_rate() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("fpr.bloom");
// Parameters for ~1% FPR with 1000 items
let num_bits = 9586; // -1000 * ln(0.01) / (ln(2))^2
let num_hashes = 7; // (9586/1000) * ln(2)
let filter = BloomFilter::create(&path, num_bits, num_hashes).unwrap();
for i in 0..1000 {
filter.add(format!("added_{i}").as_bytes());
}
let test_count = 10000;
let fp_count = (0..test_count)
.filter(|i| filter.exists(format!("not_added_{i}").as_bytes()))
.count();
let fp_rate = fp_count as f64 / test_count as f64;
assert!(fp_rate < 0.02, "False positive rate {fp_rate} exceeds 2%");
}
// -----------------------------------------------------------------------
// Edge-case tests
// -----------------------------------------------------------------------
#[test]
fn create_with_zero_bits_returns_error() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("zero.bloom");
let result = BloomFilter::create(&path, 0, 7);
assert!(result.is_err(), "creating with 0 bits must fail");
match result {
Err(msg) => assert!(msg.contains("num_bits must be > 0"), "wrong error: {msg}"),
Ok(_) => panic!("expected error"),
}
}
#[test]
fn create_with_one_bit_one_hash() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("onebit.bloom");
let filter = BloomFilter::create(&path, 1, 1).unwrap();
assert!(filter.add(b"x"));
assert!(filter.exists(b"x"));
// With 1 bit, everything should now be a false positive
assert!(filter.exists(b"y"));
}
#[test]
fn add_empty_key() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("empty_key.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
assert!(filter.add(b""));
assert!(filter.exists(b""));
assert_eq!(filter.count(), 1);
}
#[test]
fn add_large_key_1mb() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("bigkey.bloom");
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
let big_key = vec![0xABu8; 1_024 * 1_024]; // 1 MB
assert!(filter.add(&big_key));
assert!(filter.exists(&big_key));
}
#[test]
fn exists_on_empty_filter_returns_false() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("empty.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
assert!(!filter.exists(b"anything"));
assert!(!filter.exists(b"something_else"));
}
#[test]
fn add_same_key_twice_is_idempotent() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("idem.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
assert!(filter.add(b"dup")); // first add: new bit(s) set
assert!(!filter.add(b"dup")); // second add: no new bits
assert_eq!(filter.count(), 1); // count only incremented once
assert!(filter.exists(b"dup"));
}
#[test]
fn false_positive_rate_1m_adds_at_001_target() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("fpr_1m.bloom");
// For n=100_000 items at 1% FPR:
// m = -n*ln(p) / (ln2)^2 = 958_506
// k = (m/n)*ln2 = 6.64 -> 7
let num_bits = 958_506u64;
let num_hashes = 7u32;
let filter = BloomFilter::create(&path, num_bits, num_hashes).unwrap();
// Add 100K items
for i in 0..100_000 {
filter.add(format!("insert_{i}").as_bytes());
}
// Test 100K items that were NOT added
let test_count = 100_000;
let fp_count = (0..test_count)
.filter(|i| filter.exists(format!("probe_{i}").as_bytes()))
.count();
let fp_rate = fp_count as f64 / test_count as f64;
assert!(
fp_rate < 0.02,
"FPR {fp_rate:.4} exceeds 2% (target was 1%)"
);
}
#[test]
fn concurrent_reads_no_crash() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("concurrent.bloom");
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
for i in 0..1_000 {
filter.add(format!("k{i}").as_bytes());
}
// Concurrent reads via multiple threads
let filter_ptr = std::ptr::addr_of!(filter);
let handles: Vec<_> = (0..4)
.map(|t| {
let p = filter_ptr as usize;
std::thread::spawn(move || {
let f = unsafe { &*(p as *const BloomFilter) };
for i in 0..1_000 {
let _ = f.exists(format!("k{i}").as_bytes());
}
let _ = f.exists(format!("thread_{t}").as_bytes());
})
})
.collect();
for h in handles {
h.join().expect("thread panicked");
}
}
#[test]
fn serialize_deserialize_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("roundtrip.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
for i in 0..100 {
filter.add(format!("item_{i}").as_bytes());
}
filter.msync().unwrap();
drop(filter);
// Reopen = deserialize
let filter2 = BloomFilter::open_existing(&path).unwrap();
for i in 0..100 {
assert!(
filter2.exists(format!("item_{i}").as_bytes()),
"item_{i} missing after reopen"
);
}
assert_eq!(filter2.count(), 100);
}
#[test]
fn deserialize_garbage_returns_error() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("garbage.bloom");
// Write random garbage
std::fs::write(&path, b"this is not a bloom filter file").unwrap();
let result = BloomFilter::open_existing(&path);
assert!(result.is_err());
}
#[test]
fn delete_removes_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("todelete.bloom");
let filter = BloomFilter::create(&path, 500, 3).unwrap();
filter.add(b"x");
assert!(path.exists());
filter.delete().unwrap();
assert!(!path.exists());
}
#[test]
fn open_nonexistent_file_returns_error() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("does_not_exist.bloom");
let result = BloomFilter::open_existing(&path);
assert!(result.is_err());
}
#[test]
fn bit_array_boundary_last_position() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("boundary.bloom");
// Use exactly 64 bits (8 bytes) to test boundary
let filter = BloomFilter::create(&path, 64, 1).unwrap();
// Manually test that setting the last bit position works
assert!(!filter.get_bit(63));
filter.set_bit(63);
assert!(filter.get_bit(63));
}
#[test]
fn hash_distribution_all_positions_hit() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("distribution.bloom");
let num_bits = 1000u64;
let filter = BloomFilter::create(&path, num_bits, 1).unwrap();
// Add 10K keys and verify all hash positions are used
let mut positions_hit = std::collections::HashSet::new();
for i in 0..10_000u64 {
let key = format!("dist_key_{i}");
let positions = filter.hash_positions(key.as_bytes());
for pos in positions {
positions_hit.insert(pos);
}
}
// With 10K keys, 1 hash function, and 1000 positions, expect most filled
// Birthday problem: P(hit all 1000) with 10K samples is very high
let coverage = positions_hit.len() as f64 / num_bits as f64;
assert!(
coverage > 0.99,
"hash distribution coverage {coverage:.3} (hit {}/{}), expected > 99%",
positions_hit.len(),
num_bits
);
}
#[test]
fn create_with_zero_hashes_returns_error() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("zerohash.bloom");
let result = BloomFilter::create(&path, 1000, 0);
assert!(result.is_err());
}
// ==================================================================
// Deep NIF edge cases — targeting mmap/FFI safety pitfalls
// ==================================================================
#[test]
fn bloom_add_key_with_embedded_nulls() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("nullkey.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
let key = b"a\0b\0c";
assert!(filter.add(key));
assert!(filter.exists(key));
// Distinct from truncated variant
assert!(!filter.exists(b"a") || filter.count() == 1);
}
#[test]
fn bloom_concurrent_add_and_exists_100_threads() {
use std::sync::{Arc, Mutex};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("concurrent_rw.bloom");
let filter = Arc::new(Mutex::new(
BloomFilter::create(&path, 1_000_000, 7).unwrap(),
));
let handles: Vec<_> = (0..100)
.map(|t| {
let f = Arc::clone(&filter);
std::thread::spawn(move || {
let guard = f.lock().unwrap();
if t % 2 == 0 {
guard.add(format!("elem_{t}").as_bytes());
} else {
let _ = guard.exists(format!("elem_{t}").as_bytes());
}
})
})
.collect();
for h in handles {
h.join()
.expect("thread panicked during bloom concurrent access");
}
}
#[test]
fn bloom_create_in_nonexistent_nested_dir() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("deep").join("nested").join("filter.bloom");
let filter = BloomFilter::create(&path, 1000, 3).unwrap();
filter.add(b"nested_test");
assert!(filter.exists(b"nested_test"));
assert!(path.exists());
}
#[test]
fn bloom_open_corrupted_header_bad_magic() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("badmagic.bloom");
// Write a file with wrong magic bytes
std::fs::write(&path, [0xFF; 64]).unwrap();
let result = BloomFilter::open_existing(&path);
assert!(result.is_err(), "bad magic must return error");
}
#[test]
fn bloom_open_zero_length_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("empty.bloom");
std::fs::write(&path, []).unwrap();
let result = BloomFilter::open_existing(&path);
assert!(result.is_err(), "zero-length file must return error");
}
#[test]
fn bloom_madd_many_elements() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("madd.bloom");
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
let elements: Vec<Vec<u8>> = (0..500).map(|i| format!("madd_{i}").into_bytes()).collect();
for e in &elements {
filter.add(e);
}
// Verify all present
for e in &elements {
assert!(filter.exists(e), "element missing after madd");
}
assert_eq!(filter.count(), 500);
}
#[test]
fn bloom_persistence_after_many_operations() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("persist_heavy.bloom");
{
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
for i in 0..1000 {
filter.add(format!("persistent_{i}").as_bytes());
}
filter.msync().unwrap();
}
let filter = BloomFilter::open_existing(&path).unwrap();
assert_eq!(filter.count(), 1000);
for i in 0..1000 {
assert!(
filter.exists(format!("persistent_{i}").as_bytes()),
"item {i} missing after reopen"
);
}
}
#[test]
fn bloom_file_truncated_after_create_returns_error_on_open() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("truncated.bloom");
{
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
filter.add(b"data");
filter.msync().unwrap();
}
// Truncate to just the header (32 bytes), removing the bit array
let data = std::fs::read(&path).unwrap();
std::fs::write(&path, &data[..HEADER_SIZE.min(data.len())]).unwrap();
let result = BloomFilter::open_existing(&path);
assert!(result.is_err(), "truncated bloom file must fail to open");
}
// ------------------------------------------------------------------
// H-4: xxh3 replaces MD5 — verify hash quality and correctness
// ------------------------------------------------------------------
#[test]
fn h4_xxh3_no_false_negatives_10k_items() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("h4_xxh3.bloom");
let filter = BloomFilter::create(&path, 1_000_000, 7).unwrap();
for i in 0..10_000 {
filter.add(format!("h4_item_{i}").as_bytes());
}
assert_eq!(filter.count(), 10_000);
// Verify zero false negatives
for i in 0..10_000 {
assert!(
filter.exists(format!("h4_item_{i}").as_bytes()),
"false negative for h4_item_{i}"
);
}
}
#[test]
fn h4_xxh3_fpr_within_expected_bounds() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("h4_fpr.bloom");
// Parameters for ~1% FPR with 10K items
let num_bits = 95_851;
let num_hashes = 7;
let filter = BloomFilter::create(&path, num_bits, num_hashes).unwrap();
for i in 0..10_000 {
filter.add(format!("added_{i}").as_bytes());
}
let test_count = 50_000;
let fp_count = (0..test_count)
.filter(|i| filter.exists(format!("not_added_{i}").as_bytes()))
.count();
let fp_rate = fp_count as f64 / test_count as f64;
assert!(
fp_rate < 0.02,
"False positive rate {fp_rate:.4} exceeds 2% (target was 1%)"
);
}
#[test]
fn h4_xxh3_persistence_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("h4_persist.bloom");
{
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
for i in 0..500 {
filter.add(format!("persist_{i}").as_bytes());
}
filter.msync().unwrap();
}
// Reopen and verify all items found
let filter = BloomFilter::open_existing(&path).unwrap();
assert_eq!(filter.count(), 500);
for i in 0..500 {
assert!(
filter.exists(format!("persist_{i}").as_bytes()),
"item persist_{i} missing after reopen"
);
}
}
// ------------------------------------------------------------------
// M-2: msync_async uses MS_ASYNC (non-blocking) for individual adds
// ------------------------------------------------------------------
#[test]
fn m2_msync_async_does_not_block() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("m2_async.bloom");
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
// msync_async should succeed without error
filter.add(b"test_item");
assert!(filter.msync_async().is_ok(), "msync_async must not fail");
// Data should still be readable in-memory after msync_async
assert!(filter.exists(b"test_item"));
}
#[test]
fn m2_msync_async_bulk_adds_then_sync() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("m2_bulk.bloom");
let filter = BloomFilter::create(&path, 100_000, 7).unwrap();
// Simulate the bloom_add NIF pattern: add + msync_async per item
for i in 0..1000 {
filter.add(format!("m2_item_{i}").as_bytes());
filter.msync_async().unwrap();
}
// All items must still be present in memory
for i in 0..1000 {
assert!(
filter.exists(format!("m2_item_{i}").as_bytes()),
"m2_item_{i} missing after bulk add with msync_async"
);
}
assert_eq!(filter.count(), 1000);
// Explicit msync (MS_SYNC) for durability, then reopen
filter.msync().unwrap();
drop(filter);
let filter2 = BloomFilter::open_existing(&path).unwrap();
assert_eq!(filter2.count(), 1000);
for i in 0..1000 {
assert!(
filter2.exists(format!("m2_item_{i}").as_bytes()),
"m2_item_{i} missing after reopen"
);
}
}
// ------------------------------------------------------------------
// M-4: madvise(MADV_RANDOM) on mmap regions
// ------------------------------------------------------------------
#[test]
fn m4_madvise_random_does_not_break_operations() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("m4_madvise.bloom");
let filter = BloomFilter::create(&path, 1_000_000, 7).unwrap();
// madvise is called during open; verify all operations still work
for i in 0..1000 {
filter.add(format!("m4_{i}").as_bytes());
}
for i in 0..1000 {
assert!(
filter.exists(format!("m4_{i}").as_bytes()),
"m4_{i} missing after madvise"
);
}
// Verify false negative rate is unchanged
let fp_count = (0..10_000)
.filter(|i| filter.exists(format!("m4_probe_{i}").as_bytes()))
.count();
let fpr = fp_count as f64 / 10_000.0;
assert!(fpr < 0.02, "FPR {fpr:.4} too high after madvise");
}
// ------------------------------------------------------------------
// L-3: ManuallyDrop-based delete avoids path.clone()
// ------------------------------------------------------------------
#[test]
fn l3_delete_removes_file_and_does_not_leak() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("l3_del.bloom");
let filter = BloomFilter::create(&path, 10_000, 7).unwrap();
for i in 0..100 {
filter.add(format!("l3_item_{i}").as_bytes());
}
filter.msync().unwrap();
assert!(path.exists(), "file must exist before delete");
filter.delete().unwrap();
assert!(!path.exists(), "file must not exist after delete");
}
#[test]
fn l3_delete_on_large_filter() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("l3_large.bloom");
let filter = BloomFilter::create(&path, 1_000_000, 7).unwrap();
for i in 0..1_000 {
filter.add(format!("l3_large_{i}").as_bytes());
}
let file_size = std::fs::metadata(&path).unwrap().len();
assert!(file_size > 100_000, "large bloom file must be > 100KB");
filter.delete().unwrap();
assert!(!path.exists(), "large bloom file must be deleted");
}
}