diff --git a/src/bloom/codec.rs b/src/bloom/codec.rs deleted file mode 100644 index bf31533..0000000 --- a/src/bloom/codec.rs +++ /dev/null @@ -1,224 +0,0 @@ -//! Word-level RLE compression for bloom filter data. -//! -//! Encodes a sequence of `u64` words using run-length encoding. -//! Each run is encoded as `[count:2 LE][word:8 LE]` (10 bytes per run). -//! Sparse data (XOR diffs with mostly zero words) compresses well. -//! -//! ## Delta vs Full Strategy -//! -//! The sender tracks `last_sent_filter` per peer. When a new filter is -//! ready, the sender XORs it with the last-sent filter to produce a diff. -//! The diff is mostly zero words (only changed bits set), which RLE -//! compresses efficiently. If no previous filter exists (first send, -//! size class change, or NACK recovery), a full filter is sent instead. -//! -//! The same RLE codec handles both cases — full filters at ~25% fill -//! still benefit from zero-word runs between set regions. -//! -//! ## NACK Recovery -//! -//! If the receiver detects a sequence gap (missed delta), it sends a -//! NACK. The sender responds with a full filter, resetting the delta -//! baseline for that peer. - -/// Statistics from a compression operation. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct CompressionStats { - /// Number of u64 words in the uncompressed input. - pub raw_words: usize, - /// Number of bytes in the compressed output. - pub compressed_bytes: usize, - /// Number of distinct runs in the encoding. - pub run_count: usize, -} - -/// RLE-encode a slice of u64 words. -/// -/// Returns the compressed bytes and compression statistics. -/// Each run is encoded as `[count:2 LE][word:8 LE]`. -/// Maximum run length is `u16::MAX` (65535); longer runs are split. -pub fn rle_encode(words: &[u64]) -> (Vec, CompressionStats) { - let mut buf = Vec::new(); - let mut run_count = 0usize; - - let mut i = 0; - while i < words.len() { - let value = words[i]; - let mut count = 1u16; - while i + (count as usize) < words.len() - && words[i + count as usize] == value - && count < u16::MAX - { - count += 1; - } - - buf.extend_from_slice(&count.to_le_bytes()); - buf.extend_from_slice(&value.to_le_bytes()); - run_count += 1; - i += count as usize; - } - - let stats = CompressionStats { - raw_words: words.len(), - compressed_bytes: buf.len(), - run_count, - }; - (buf, stats) -} - -/// RLE-decode compressed bytes back to u64 words. -/// -/// `expected_words` is the expected number of output words (for validation). -/// Returns an error if the data is truncated or the decoded length doesn't -/// match the expected count. -pub fn rle_decode(data: &[u8], expected_words: usize) -> Result, RleError> { - let mut words = Vec::with_capacity(expected_words); - let mut pos = 0; - - while pos + 10 <= data.len() { - let count = u16::from_le_bytes(data[pos..pos + 2].try_into().unwrap()) as usize; - let value = u64::from_le_bytes(data[pos + 2..pos + 10].try_into().unwrap()); - pos += 10; - - if words.len() + count > expected_words { - return Err(RleError::DecodedTooLarge { - expected: expected_words, - got: words.len() + count, - }); - } - - words.extend(std::iter::repeat_n(value, count)); - } - - if pos != data.len() { - return Err(RleError::TruncatedInput { - remaining: data.len() - pos, - }); - } - - if words.len() != expected_words { - return Err(RleError::DecodedSizeMismatch { - expected: expected_words, - got: words.len(), - }); - } - - Ok(words) -} - -/// Errors from RLE decoding. -#[derive(Debug, thiserror::Error)] -pub enum RleError { - #[error("truncated RLE input: {remaining} trailing bytes")] - TruncatedInput { remaining: usize }, - - #[error("decoded size mismatch: expected {expected} words, got {got}")] - DecodedSizeMismatch { expected: usize, got: usize }, - - #[error("decoded data too large: expected {expected} words, got {got}")] - DecodedTooLarge { expected: usize, got: usize }, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_rle_round_trip_all_zero() { - let words = vec![0u64; 128]; // 1KB filter - let (encoded, stats) = rle_encode(&words); - - // All zeros = 1 run of 128 zero words = 10 bytes - assert_eq!(stats.run_count, 1); - assert_eq!(stats.compressed_bytes, 10); - assert_eq!(stats.raw_words, 128); - - let decoded = rle_decode(&encoded, 128).unwrap(); - assert_eq!(decoded, words); - } - - #[test] - fn test_rle_round_trip_random() { - // Non-uniform data: each word different - let words: Vec = (0..64).map(|i| i * 0x0123456789ABCDEF).collect(); - let (encoded, stats) = rle_encode(&words); - - assert_eq!(stats.raw_words, 64); - assert_eq!(stats.run_count, 64); // no compression - assert_eq!(stats.compressed_bytes, 64 * 10); - - let decoded = rle_decode(&encoded, 64).unwrap(); - assert_eq!(decoded, words); - } - - #[test] - fn test_rle_round_trip_sparse_diff() { - // Simulate XOR diff: mostly zeros with a few set words - let mut words = vec![0u64; 128]; - words[10] = 0xFF00FF00; - words[50] = 0xDEADBEEF; - words[127] = 0x1; - - let (encoded, stats) = rle_encode(&words); - - // Should compress well: ~7 runs - assert!(stats.run_count < 10); - assert!(stats.compressed_bytes < 128 * 8); // much smaller than raw - - let decoded = rle_decode(&encoded, 128).unwrap(); - assert_eq!(decoded, words); - } - - #[test] - fn test_rle_empty_input() { - let (encoded, stats) = rle_encode(&[]); - assert_eq!(stats.raw_words, 0); - assert_eq!(stats.compressed_bytes, 0); - assert_eq!(stats.run_count, 0); - - let decoded = rle_decode(&encoded, 0).unwrap(); - assert!(decoded.is_empty()); - } - - #[test] - fn test_rle_decode_truncated() { - // 5 bytes is not a complete run (need 10) - let bad_data = vec![0u8; 5]; - let result = rle_decode(&bad_data, 1); - assert!(matches!(result, Err(RleError::TruncatedInput { .. }))); - } - - #[test] - fn test_rle_decode_size_mismatch() { - let words = vec![0u64; 10]; - let (encoded, _) = rle_encode(&words); - - // Expect wrong number of words - let result = rle_decode(&encoded, 20); - assert!(matches!(result, Err(RleError::DecodedSizeMismatch { .. }))); - } - - #[test] - fn test_rle_decode_too_large() { - let words = vec![0u64; 100]; - let (encoded, _) = rle_encode(&words); - - // Expect fewer words than what's encoded - let result = rle_decode(&encoded, 50); - assert!(matches!(result, Err(RleError::DecodedTooLarge { .. }))); - } - - #[test] - fn test_rle_compression_stats() { - // 3 runs: 10 zeros, 5 ones, 5 zeros - let mut words = vec![0u64; 20]; - for w in &mut words[10..15] { - *w = u64::MAX; - } - - let (_, stats) = rle_encode(&words); - assert_eq!(stats.raw_words, 20); - assert_eq!(stats.run_count, 3); - assert_eq!(stats.compressed_bytes, 30); // 3 * 10 - } -} diff --git a/src/bloom/filter.rs b/src/bloom/filter.rs deleted file mode 100644 index 4b15623..0000000 --- a/src/bloom/filter.rs +++ /dev/null @@ -1,441 +0,0 @@ -//! Generic Bloom filter data structure. - -use std::fmt; - -use tracing::trace; - -use super::{ - BloomError, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT, MAX_SIZE_CLASS, MIN_SIZE_CLASS, - SIZE_CLASS_BYTES, -}; -use crate::NodeAddr; - -/// A Bloom filter for probabilistic set membership. -/// -/// Used in FIPS to track which destinations are reachable through a peer. -/// The filter uses double hashing to generate k hash functions from two -/// base hashes derived from the input. -/// -/// Internal storage uses 64-bit words for efficient bitwise operations -/// and word-level RLE compression. -#[derive(Clone)] -pub struct BloomFilter { - /// Bit array storage (packed as 64-bit words, little-endian bit order). - words: Vec, - /// Number of bits in the filter. - num_bits: usize, - /// Number of hash functions to use. - hash_count: u8, -} - -impl BloomFilter { - /// Create a new empty Bloom filter with default parameters. - pub fn new() -> Self { - Self::with_params(DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT) - .expect("default params are valid") - } - - /// Create a Bloom filter with custom parameters. - pub fn with_params(num_bits: usize, hash_count: u8) -> Result { - if num_bits == 0 || !num_bits.is_multiple_of(8) { - return Err(BloomError::SizeNotByteAligned(num_bits)); - } - if !num_bits.is_multiple_of(64) { - return Err(BloomError::SizeNotWordAligned(num_bits)); - } - if hash_count == 0 { - return Err(BloomError::ZeroHashCount); - } - - let num_words = num_bits / 64; - Ok(Self { - words: vec![0u64; num_words], - num_bits, - hash_count, - }) - } - - /// Create a Bloom filter from raw bytes (little-endian byte order). - pub fn from_bytes(bytes: Vec, hash_count: u8) -> Result { - if hash_count == 0 { - return Err(BloomError::ZeroHashCount); - } - if bytes.is_empty() { - return Err(BloomError::SizeNotByteAligned(0)); - } - let num_bits = bytes.len() * 8; - if !num_bits.is_multiple_of(64) { - return Err(BloomError::SizeNotWordAligned(num_bits)); - } - - let num_words = num_bits / 64; - let mut words = Vec::with_capacity(num_words); - for chunk in bytes.chunks_exact(8) { - words.push(u64::from_le_bytes(chunk.try_into().unwrap())); - } - - Ok(Self { - words, - num_bits, - hash_count, - }) - } - - /// Create a Bloom filter from a byte slice. - pub fn from_slice(bytes: &[u8], hash_count: u8) -> Result { - Self::from_bytes(bytes.to_vec(), hash_count) - } - - /// Insert a NodeAddr into the filter. - pub fn insert(&mut self, node_addr: &NodeAddr) { - for i in 0..self.hash_count { - let bit_index = self.hash(node_addr.as_bytes(), i); - self.set_bit(bit_index); - } - } - - /// Insert raw bytes into the filter. - pub fn insert_bytes(&mut self, data: &[u8]) { - for i in 0..self.hash_count { - let bit_index = self.hash(data, i); - self.set_bit(bit_index); - } - } - - /// Check if the filter might contain a NodeAddr. - /// - /// Returns `true` if the item might be in the set (possible false positive). - /// Returns `false` if the item is definitely not in the set. - pub fn contains(&self, node_addr: &NodeAddr) -> bool { - self.contains_bytes(node_addr.as_bytes()) - } - - /// Check if the filter might contain raw bytes. - pub fn contains_bytes(&self, data: &[u8]) -> bool { - for i in 0..self.hash_count { - let bit_index = self.hash(data, i); - if !self.get_bit(bit_index) { - return false; - } - } - true - } - - /// Merge another filter into this one (OR operation). - /// - /// If the other filter is a different size, it is converted to this - /// filter's size first (fold if larger, duplicate if smaller). - /// After merge, this filter contains all elements from both filters. - pub fn merge(&mut self, other: &BloomFilter) -> Result<(), BloomError> { - if self.num_bits == other.num_bits { - for (a, b) in self.words.iter_mut().zip(other.words.iter()) { - *a |= b; - } - } else { - let converted = other.convert_to(self.num_bits)?; - for (a, b) in self.words.iter_mut().zip(converted.words.iter()) { - *a |= b; - } - } - Ok(()) - } - - /// Create a new filter that is the union of this and another. - pub fn union(&self, other: &BloomFilter) -> Result { - let mut result = self.clone(); - result.merge(other)?; - Ok(result) - } - - /// Fold the filter in half (large → small). - /// - /// ORs the top half of words with the bottom half, halving the filter - /// size. No false negatives are introduced, but the fill ratio roughly - /// doubles. - pub fn fold(&self) -> Result { - let min_bits = SIZE_CLASS_BYTES[MIN_SIZE_CLASS as usize] * 8; - if self.num_bits <= min_bits { - return Err(BloomError::CannotFold(self.num_bits)); - } - - let half = self.words.len() / 2; - let words: Vec = self.words[..half] - .iter() - .zip(self.words[half..].iter()) - .map(|(a, b)| a | b) - .collect(); - - Ok(BloomFilter { - words, - num_bits: self.num_bits / 2, - hash_count: self.hash_count, - }) - } - - /// Fold repeatedly to reach the target size in bits. - pub fn fold_to(&self, target_bits: usize) -> Result { - if target_bits >= self.num_bits { - return Err(BloomError::InvalidTargetSize(target_bits)); - } - if !target_bits.is_power_of_two() - || target_bits < SIZE_CLASS_BYTES[MIN_SIZE_CLASS as usize] * 8 - { - return Err(BloomError::InvalidTargetSize(target_bits)); - } - - let mut result = self.fold()?; - while result.num_bits > target_bits { - result = result.fold()?; - } - Ok(result) - } - - /// Duplicate the filter (small → large). - /// - /// Concatenates the filter with itself, doubling the size. - /// The duplicated filter is compatible with the larger hash space: - /// `h(x) mod 2m` maps to either `h(x) mod m` or `h(x) mod m + m`, - /// and both positions have the bit set. - pub fn duplicate(&self) -> Result { - let max_bits = SIZE_CLASS_BYTES[MAX_SIZE_CLASS as usize] * 8; - if self.num_bits >= max_bits { - return Err(BloomError::CannotDuplicate(self.num_bits)); - } - - let mut words = Vec::with_capacity(self.words.len() * 2); - words.extend_from_slice(&self.words); - words.extend_from_slice(&self.words); - - Ok(BloomFilter { - words, - num_bits: self.num_bits * 2, - hash_count: self.hash_count, - }) - } - - /// Duplicate repeatedly to reach the target size in bits. - pub fn duplicate_to(&self, target_bits: usize) -> Result { - if target_bits <= self.num_bits { - return Err(BloomError::InvalidTargetSize(target_bits)); - } - if !target_bits.is_power_of_two() - || target_bits > SIZE_CLASS_BYTES[MAX_SIZE_CLASS as usize] * 8 - { - return Err(BloomError::InvalidTargetSize(target_bits)); - } - - let mut result = self.duplicate()?; - while result.num_bits < target_bits { - result = result.duplicate()?; - } - Ok(result) - } - - /// Convert the filter to a different size. - /// - /// Folds (if target is smaller) or duplicates (if target is larger). - /// Returns a clone if the target matches the current size. - pub fn convert_to(&self, target_bits: usize) -> Result { - if target_bits == self.num_bits { - return Ok(self.clone()); - } - if target_bits < self.num_bits { - self.fold_to(target_bits) - } else { - self.duplicate_to(target_bits) - } - } - - /// Compute the XOR diff between this filter and another. - /// - /// The result contains only the bits that differ between the two filters. - /// Used for delta compression: `old.xor_diff(&new)` produces a diff that - /// can be applied to `old` to reconstruct `new`. - pub fn xor_diff(&self, other: &BloomFilter) -> Result { - if self.num_bits != other.num_bits { - return Err(BloomError::InvalidSize { - expected: self.num_bits, - got: other.num_bits, - }); - } - - let words: Vec = self - .words - .iter() - .zip(other.words.iter()) - .map(|(a, b)| a ^ b) - .collect(); - - Ok(BloomFilter { - words, - num_bits: self.num_bits, - hash_count: self.hash_count, - }) - } - - /// Apply a XOR diff to this filter in place. - /// - /// This is the inverse of `xor_diff()`: if `diff = old.xor_diff(&new)`, - /// then `old.apply_diff(&diff)` transforms `old` into `new`. - pub fn apply_diff(&mut self, diff: &BloomFilter) -> Result<(), BloomError> { - if self.num_bits != diff.num_bits { - return Err(BloomError::InvalidSize { - expected: self.num_bits, - got: diff.num_bits, - }); - } - - for (a, b) in self.words.iter_mut().zip(diff.words.iter()) { - *a ^= b; - } - Ok(()) - } - - /// Clear all bits in the filter. - pub fn clear(&mut self) { - self.words.fill(0); - } - - /// Count the number of set bits (population count). - pub fn count_ones(&self) -> usize { - self.words.iter().map(|w| w.count_ones() as usize).sum() - } - - /// Estimate the fill ratio (set bits / total bits). - pub fn fill_ratio(&self) -> f64 { - self.count_ones() as f64 / self.num_bits as f64 - } - - /// Estimate the number of elements in the filter. - /// - /// Uses the formula: n = -(m/k) * ln(1 - X/m) - /// where m = num_bits, k = hash_count, X = count_ones - /// - /// Returns `None` when the filter's FPR exceeds `max_fpr` (antipoison - /// cap) or the filter is saturated (`count_ones() >= num_bits`). Pass - /// `f64::INFINITY` for `max_fpr` to disable the cap — useful in - /// Debug/log contexts where no policy is in scope. The saturated - /// branch is always honored regardless of `max_fpr`, preventing the - /// `f64::INFINITY` return that the previous signature produced. - pub fn estimated_count(&self, max_fpr: f64) -> Option { - let m = self.num_bits as f64; - let k = self.hash_count as f64; - let x = self.count_ones() as f64; - - if x >= m { - return None; - } - - let fill = x / m; - let fpr = fill.powi(self.hash_count as i32); - if fpr > max_fpr { - trace!(fill, fpr, max_fpr, "estimated_count: filter above cap"); - return None; - } - - Some(-(m / k) * (1.0 - fill).ln()) - } - - /// Check if the filter is empty. - pub fn is_empty(&self) -> bool { - self.words.iter().all(|&w| w == 0) - } - - /// Get the filter contents as bytes (little-endian byte order). - pub fn as_bytes(&self) -> Vec { - let mut bytes = Vec::with_capacity(self.words.len() * 8); - for &word in &self.words { - bytes.extend_from_slice(&word.to_le_bytes()); - } - bytes - } - - /// Get the internal word storage. - pub fn as_words(&self) -> &[u64] { - &self.words - } - - /// Get the number of 64-bit words in the filter. - pub fn num_words(&self) -> usize { - self.words.len() - } - - /// Get the filter size in bits. - pub fn num_bits(&self) -> usize { - self.num_bits - } - - /// Get the filter size in bytes. - pub fn num_bytes(&self) -> usize { - self.words.len() * 8 - } - - /// Get the number of hash functions. - pub fn hash_count(&self) -> u8 { - self.hash_count - } - - /// Compute a hash index for the given data and hash function number. - /// - /// Uses double hashing: h(x,i) = (h1(x) + i*h2(x)) mod m - fn hash(&self, data: &[u8], k: u8) -> usize { - // Use first 16 bytes of SHA-256 for h1 and h2 - use sha2::{Digest, Sha256}; - let mut hasher = Sha256::new(); - hasher.update(data); - let hash = hasher.finalize(); - - // h1 from first 8 bytes - let h1 = u64::from_le_bytes(hash[0..8].try_into().unwrap()); - // h2 from next 8 bytes - let h2 = u64::from_le_bytes(hash[8..16].try_into().unwrap()); - - let combined = h1.wrapping_add((k as u64).wrapping_mul(h2)); - (combined as usize) % self.num_bits - } - - fn set_bit(&mut self, index: usize) { - let word_index = index / 64; - let bit_offset = index % 64; - self.words[word_index] |= 1 << bit_offset; - } - - fn get_bit(&self, index: usize) -> bool { - let word_index = index / 64; - let bit_offset = index % 64; - (self.words[word_index] >> bit_offset) & 1 == 1 - } -} - -impl Default for BloomFilter { - fn default() -> Self { - Self::new() - } -} - -impl PartialEq for BloomFilter { - fn eq(&self, other: &Self) -> bool { - self.num_bits == other.num_bits - && self.hash_count == other.hash_count - && self.words == other.words - } -} - -impl Eq for BloomFilter {} - -impl fmt::Debug for BloomFilter { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("BloomFilter") - .field("bits", &self.num_bits) - .field("hash_count", &self.hash_count) - .field("fill_ratio", &format!("{:.2}%", self.fill_ratio() * 100.0)) - .field( - "est_count", - &match self.estimated_count(f64::INFINITY) { - Some(n) => format!("{:.0}", n), - None => "saturated".to_string(), - }, - ) - .finish() - } -} diff --git a/src/bloom/mod.rs b/src/bloom/mod.rs deleted file mode 100644 index 8b456ba..0000000 --- a/src/bloom/mod.rs +++ /dev/null @@ -1,89 +0,0 @@ -//! Bloom Filter Implementation -//! -//! Variable-size Bloom filters for reachability in FIPS routing. Each -//! node maintains filters that summarize which destinations are reachable -//! through each peer, enabling efficient routing decisions without -//! global network knowledge. -//! -//! Filter sizes range from 512 bytes (size_class 0) to 32 KB -//! (size_class 6), in power-of-two steps. Nodes choose their own -//! size class based on subtree load and adapt dynamically. -//! -//! ## Parameters -//! -//! - Hash functions: k=5 (network-wide constant) -//! - Default size: 1 KB (size_class 1) - -pub mod codec; -mod filter; -mod state; - -use thiserror::Error; - -pub use filter::BloomFilter; -pub use state::BloomState; - -/// Default filter size in bits (1KB = 8,192 bits). -pub const DEFAULT_FILTER_SIZE_BITS: usize = 8192; - -/// Default filter size in bytes (1KB). -pub const DEFAULT_FILTER_SIZE_BYTES: usize = DEFAULT_FILTER_SIZE_BITS / 8; - -/// Default number of hash functions. -/// -/// k=5 is a network-wide constant. Optimal at ~7.2 bits per element. -pub const DEFAULT_HASH_COUNT: u8 = 5; - -/// Size class for v1 protocol (1 KB filters). -pub const V1_SIZE_CLASS: u8 = 1; - -/// Minimum size class (512 bytes). -pub const MIN_SIZE_CLASS: u8 = 0; - -/// Maximum size class (32 KB). -pub const MAX_SIZE_CLASS: u8 = 6; - -/// Filter sizes by size_class: bytes = 512 << size_class -pub const SIZE_CLASS_BYTES: [usize; 7] = [512, 1024, 2048, 4096, 8192, 16384, 32768]; - -/// Convert a size class to filter size in bits. -pub fn size_class_to_bits(size_class: u8) -> usize { - SIZE_CLASS_BYTES[size_class as usize] * 8 -} - -/// Convert a filter size in bits to its size class, if valid. -pub fn bits_to_size_class(num_bits: usize) -> Option { - let num_bytes = num_bits / 8; - SIZE_CLASS_BYTES - .iter() - .position(|&s| s == num_bytes) - .map(|i| i as u8) -} - -/// Errors related to Bloom filter operations. -#[derive(Debug, Error)] -pub enum BloomError { - #[error("invalid filter size: expected {expected} bits, got {got}")] - InvalidSize { expected: usize, got: usize }, - - #[error("filter size must be a multiple of 8, got {0}")] - SizeNotByteAligned(usize), - - #[error("filter size must be a multiple of 64, got {0}")] - SizeNotWordAligned(usize), - - #[error("hash count must be positive")] - ZeroHashCount, - - #[error("cannot fold: filter is already at minimum size ({0} bits)")] - CannotFold(usize), - - #[error("cannot duplicate: filter is already at maximum size ({0} bits)")] - CannotDuplicate(usize), - - #[error("target size {0} bits is not a valid power-of-two filter size")] - InvalidTargetSize(usize), -} - -#[cfg(test)] -mod tests; diff --git a/src/bloom/tests.rs b/src/bloom/tests.rs deleted file mode 100644 index 02a1261..0000000 --- a/src/bloom/tests.rs +++ /dev/null @@ -1,1064 +0,0 @@ -use super::*; -use crate::NodeAddr; -use std::collections::HashMap; - -fn make_node_addr(val: u8) -> NodeAddr { - let mut bytes = [0u8; 16]; - bytes[0] = val; - NodeAddr::from_bytes(bytes) -} - -// ===== BloomFilter Tests ===== - -#[test] -fn test_bloom_filter_new() { - let filter = BloomFilter::new(); - assert_eq!(filter.num_bits(), DEFAULT_FILTER_SIZE_BITS); - assert_eq!(filter.hash_count(), DEFAULT_HASH_COUNT); - assert_eq!(filter.count_ones(), 0); - assert!(filter.is_empty()); -} - -#[test] -fn test_bloom_filter_insert_contains() { - let mut filter = BloomFilter::new(); - let node1 = make_node_addr(1); - let node2 = make_node_addr(2); - - assert!(!filter.contains(&node1)); - assert!(!filter.contains(&node2)); - - filter.insert(&node1); - - assert!(filter.contains(&node1)); - // node2 might have false positive, but very unlikely with single insert - assert!(!filter.is_empty()); -} - -#[test] -fn test_bloom_filter_multiple_inserts() { - let mut filter = BloomFilter::new(); - - for i in 0..100 { - let node = make_node_addr(i); - filter.insert(&node); - } - - // All inserted items should be found - for i in 0..100 { - let node = make_node_addr(i); - assert!(filter.contains(&node), "Node {} not found", i); - } - - // Fill ratio should be reasonable - let fill = filter.fill_ratio(); - assert!(fill > 0.0 && fill < 0.5, "Unexpected fill ratio: {}", fill); -} - -#[test] -fn test_bloom_filter_merge() { - let mut filter1 = BloomFilter::new(); - let mut filter2 = BloomFilter::new(); - - let node1 = make_node_addr(1); - let node2 = make_node_addr(2); - - filter1.insert(&node1); - filter2.insert(&node2); - - filter1.merge(&filter2).unwrap(); - - assert!(filter1.contains(&node1)); - assert!(filter1.contains(&node2)); -} - -#[test] -fn test_bloom_filter_union() { - let mut filter1 = BloomFilter::new(); - let mut filter2 = BloomFilter::new(); - - let node1 = make_node_addr(1); - let node2 = make_node_addr(2); - - filter1.insert(&node1); - filter2.insert(&node2); - - let union = filter1.union(&filter2).unwrap(); - - assert!(union.contains(&node1)); - assert!(union.contains(&node2)); - // Original filters unchanged - assert!(!filter1.contains(&node2)); - assert!(!filter2.contains(&node1)); -} - -#[test] -fn test_bloom_filter_clear() { - let mut filter = BloomFilter::new(); - let node = make_node_addr(1); - - filter.insert(&node); - assert!(!filter.is_empty()); - - filter.clear(); - assert!(filter.is_empty()); - assert_eq!(filter.count_ones(), 0); - assert!(!filter.contains(&node)); -} - -#[test] -fn test_bloom_filter_merge_cross_size_fold() { - // Merge a 2KB filter into a 1KB filter (fold the larger) - let mut filter1 = BloomFilter::with_params(1024 * 8, 5).unwrap(); - let mut filter2 = BloomFilter::with_params(2048 * 8, 5).unwrap(); - - let node1 = make_node_addr(1); - let node2 = make_node_addr(2); - filter1.insert(&node1); - filter2.insert(&node2); - - filter1.merge(&filter2).unwrap(); - - assert!(filter1.contains(&node1)); - assert!(filter1.contains(&node2)); - assert_eq!(filter1.num_bits(), 1024 * 8); // size unchanged -} - -#[test] -fn test_bloom_filter_merge_cross_size_duplicate() { - // Merge a 512B filter into a 2KB filter (duplicate the smaller) - let mut filter1 = BloomFilter::with_params(2048 * 8, 5).unwrap(); - let mut filter2 = BloomFilter::with_params(512 * 8, 5).unwrap(); - - let node1 = make_node_addr(1); - let node2 = make_node_addr(2); - filter1.insert(&node1); - filter2.insert(&node2); - - filter1.merge(&filter2).unwrap(); - - assert!(filter1.contains(&node1)); - assert!(filter1.contains(&node2)); - assert_eq!(filter1.num_bits(), 2048 * 8); -} - -#[test] -fn test_bloom_filter_custom_params() { - let filter = BloomFilter::with_params(1024, 5).unwrap(); - assert_eq!(filter.num_bits(), 1024); - assert_eq!(filter.num_bytes(), 128); - assert_eq!(filter.hash_count(), 5); -} - -#[test] -fn test_bloom_filter_invalid_params() { - // Not byte-aligned (1001 is not divisible by 8) - assert!(matches!( - BloomFilter::with_params(1001, 7), - Err(BloomError::SizeNotByteAligned(1001)) - )); - - // Zero size - assert!(matches!( - BloomFilter::with_params(0, 7), - Err(BloomError::SizeNotByteAligned(0)) - )); - - // Zero hash count - assert!(matches!( - BloomFilter::with_params(1024, 0), - Err(BloomError::ZeroHashCount) - )); - - // Byte-aligned but not word-aligned (24 bits = 3 bytes, not 8) - assert!(matches!( - BloomFilter::with_params(24, 5), - Err(BloomError::SizeNotWordAligned(24)) - )); -} - -#[test] -fn test_bloom_filter_from_bytes_not_word_aligned() { - // 5 bytes = 40 bits, not a multiple of 64 - let result = BloomFilter::from_bytes(vec![0u8; 5], 5); - assert!(matches!(result, Err(BloomError::SizeNotWordAligned(40)))); - - // 8 bytes = 64 bits, should succeed - assert!(BloomFilter::from_bytes(vec![0u8; 8], 5).is_ok()); -} - -#[test] -fn test_bloom_filter_from_bytes() { - let original = BloomFilter::new(); - let bytes = original.as_bytes().to_vec(); - - let restored = BloomFilter::from_bytes(bytes, original.hash_count()).unwrap(); - - assert_eq!(original, restored); -} - -#[test] -fn test_bloom_filter_estimated_count() { - let mut filter = BloomFilter::new(); - - // Empty filter - assert_eq!(filter.estimated_count(f64::INFINITY), Some(0.0)); - - // Insert some items - for i in 0..50 { - filter.insert(&make_node_addr(i)); - } - - // Estimate should be reasonably close to 50 - let estimate = filter.estimated_count(f64::INFINITY).unwrap(); - assert!( - estimate > 30.0 && estimate < 100.0, - "Unexpected estimate: {}", - estimate - ); -} - -#[test] -fn test_bloom_filter_equality() { - let mut filter1 = BloomFilter::new(); - let mut filter2 = BloomFilter::new(); - - assert_eq!(filter1, filter2); - - filter1.insert(&make_node_addr(1)); - assert_ne!(filter1, filter2); - - filter2.insert(&make_node_addr(1)); - assert_eq!(filter1, filter2); -} - -#[test] -fn test_bloom_filter_from_bytes_empty() { - let result = BloomFilter::from_bytes(vec![], 5); - assert!(matches!(result, Err(BloomError::SizeNotByteAligned(0)))); -} - -#[test] -fn test_bloom_filter_from_bytes_zero_hash_count() { - let result = BloomFilter::from_bytes(vec![0u8; 128], 0); - assert!(matches!(result, Err(BloomError::ZeroHashCount))); -} - -#[test] -fn test_bloom_filter_from_slice() { - let mut original = BloomFilter::new(); - original.insert(&make_node_addr(42)); - let bytes = original.as_bytes(); - - let restored = BloomFilter::from_slice(&bytes, original.hash_count()).unwrap(); - assert_eq!(original, restored); -} - -#[test] -fn test_bloom_filter_insert_bytes_contains_bytes() { - let mut filter = BloomFilter::new(); - let data1 = b"hello world"; - let data2 = b"goodbye"; - - assert!(!filter.contains_bytes(data1)); - - filter.insert_bytes(data1); - assert!(filter.contains_bytes(data1)); - assert!(!filter.contains_bytes(data2)); - - filter.insert_bytes(data2); - assert!(filter.contains_bytes(data1)); - assert!(filter.contains_bytes(data2)); -} - -#[test] -fn test_bloom_filter_as_bytes_round_trip() { - let mut original = BloomFilter::new(); - for i in 0..50 { - original.insert(&make_node_addr(i)); - } - - let bytes = original.as_bytes(); - let restored = BloomFilter::from_bytes(bytes, original.hash_count()).unwrap(); - assert_eq!(original, restored); - - // Verify all inserted elements are still found - for i in 0..50 { - assert!(restored.contains(&make_node_addr(i))); - } -} - -#[test] -fn test_bloom_filter_as_words() { - let filter = BloomFilter::new(); - // Default 8192 bits = 128 words - assert_eq!(filter.as_words().len(), 128); - assert_eq!(filter.num_words(), 128); - assert!(filter.as_words().iter().all(|&w| w == 0)); - - // Small filter: 64 bits = 1 word - let small = BloomFilter::with_params(64, 3).unwrap(); - assert_eq!(small.as_words().len(), 1); - assert_eq!(small.num_words(), 1); -} - -#[test] -fn test_bloom_filter_xor_diff_and_apply() { - let mut filter_a = BloomFilter::new(); - let mut filter_b = BloomFilter::new(); - - // Insert different elements into each - for i in 0..20 { - filter_a.insert(&make_node_addr(i)); - } - for i in 10..30 { - filter_b.insert(&make_node_addr(i)); - } - - // Compute diff: applying diff to A should yield B - let diff = filter_a.xor_diff(&filter_b).unwrap(); - - let mut reconstructed = filter_a.clone(); - reconstructed.apply_diff(&diff).unwrap(); - assert_eq!(reconstructed, filter_b); -} - -#[test] -fn test_bloom_filter_xor_diff_identical() { - let mut filter = BloomFilter::new(); - for i in 0..10 { - filter.insert(&make_node_addr(i)); - } - - // XOR of identical filters should be all zeros - let diff = filter.xor_diff(&filter).unwrap(); - assert!(diff.is_empty()); - assert_eq!(diff.count_ones(), 0); -} - -#[test] -fn test_bloom_filter_xor_diff_size_mismatch() { - let filter_a = BloomFilter::with_params(1024, 5).unwrap(); - let filter_b = BloomFilter::with_params(2048, 5).unwrap(); - - assert!(matches!( - filter_a.xor_diff(&filter_b), - Err(BloomError::InvalidSize { .. }) - )); -} - -#[test] -fn test_bloom_filter_apply_diff_size_mismatch() { - let mut filter = BloomFilter::new(); - let diff = BloomFilter::with_params(1024, 5).unwrap(); - - assert!(matches!( - filter.apply_diff(&diff), - Err(BloomError::InvalidSize { .. }) - )); -} - -// ===== Fold/Duplicate/Convert Tests ===== - -#[test] -fn test_bloom_filter_fold() { - // 2KB filter → fold to 1KB - let mut filter = BloomFilter::with_params(2048 * 8, 5).unwrap(); - for i in 0..50 { - filter.insert(&make_node_addr(i)); - } - - let folded = filter.fold().unwrap(); - assert_eq!(folded.num_bits(), 1024 * 8); - - // All inserted elements must still be found (no false negatives) - for i in 0..50 { - assert!( - folded.contains(&make_node_addr(i)), - "Node {} not found after fold", - i - ); - } - - // Fill ratio should roughly double - let original_fill = filter.fill_ratio(); - let folded_fill = folded.fill_ratio(); - assert!( - folded_fill > original_fill * 1.5, - "Fill ratio didn't increase enough" - ); -} - -#[test] -fn test_bloom_filter_fold_to() { - // 4KB → fold to 512B (3 folds) - let mut filter = BloomFilter::with_params(4096 * 8, 5).unwrap(); - for i in 0..20 { - filter.insert(&make_node_addr(i)); - } - - let folded = filter.fold_to(512 * 8).unwrap(); - assert_eq!(folded.num_bits(), 512 * 8); - - for i in 0..20 { - assert!(folded.contains(&make_node_addr(i))); - } -} - -#[test] -fn test_bloom_filter_fold_at_minimum() { - let filter = BloomFilter::with_params(512 * 8, 5).unwrap(); - assert!(matches!(filter.fold(), Err(BloomError::CannotFold(_)))); -} - -#[test] -fn test_bloom_filter_duplicate() { - let mut filter = BloomFilter::with_params(1024 * 8, 5).unwrap(); - for i in 0..50 { - filter.insert(&make_node_addr(i)); - } - - let duped = filter.duplicate().unwrap(); - assert_eq!(duped.num_bits(), 2048 * 8); - - // All elements still found at the larger size - for i in 0..50 { - assert!( - duped.contains(&make_node_addr(i)), - "Node {} not found after duplicate", - i - ); - } -} - -#[test] -fn test_bloom_filter_duplicate_to() { - let mut filter = BloomFilter::with_params(512 * 8, 5).unwrap(); - for i in 0..10 { - filter.insert(&make_node_addr(i)); - } - - let duped = filter.duplicate_to(4096 * 8).unwrap(); - assert_eq!(duped.num_bits(), 4096 * 8); - - for i in 0..10 { - assert!(duped.contains(&make_node_addr(i))); - } -} - -#[test] -fn test_bloom_filter_duplicate_at_maximum() { - let filter = BloomFilter::with_params(32768 * 8, 5).unwrap(); - assert!(matches!( - filter.duplicate(), - Err(BloomError::CannotDuplicate(_)) - )); -} - -#[test] -fn test_bloom_filter_duplicate_then_fold_round_trip() { - let mut filter = BloomFilter::with_params(1024 * 8, 5).unwrap(); - for i in 0..30 { - filter.insert(&make_node_addr(i)); - } - - // Duplicate to 2KB then fold back to 1KB should yield equivalent filter - let duped = filter.duplicate().unwrap(); - let folded_back = duped.fold().unwrap(); - - // The round-trip should be identical because duplication places - // identical copies in both halves, and folding ORs them back - assert_eq!(filter, folded_back); -} - -#[test] -fn test_bloom_filter_convert_to() { - let mut filter = BloomFilter::with_params(1024 * 8, 5).unwrap(); - for i in 0..20 { - filter.insert(&make_node_addr(i)); - } - - // Same size → clone - let same = filter.convert_to(1024 * 8).unwrap(); - assert_eq!(filter, same); - - // Larger → duplicate - let larger = filter.convert_to(4096 * 8).unwrap(); - assert_eq!(larger.num_bits(), 4096 * 8); - for i in 0..20 { - assert!(larger.contains(&make_node_addr(i))); - } - - // Smaller → fold - let smaller = filter.convert_to(512 * 8).unwrap(); - assert_eq!(smaller.num_bits(), 512 * 8); - for i in 0..20 { - assert!(smaller.contains(&make_node_addr(i))); - } -} - -#[test] -fn test_bloom_filter_convert_to_invalid() { - let filter = BloomFilter::with_params(1024 * 8, 5).unwrap(); - - // Not a power of two - assert!(matches!( - filter.convert_to(1000 * 8), - Err(BloomError::InvalidTargetSize(_)) - )); -} - -#[test] -fn test_bloom_filter_estimated_count_saturated() { - // Create a small filter with all bits set - let bytes = vec![0xFF; 8]; // all bits set - let filter = BloomFilter::from_bytes(bytes, 3).unwrap(); - - // Saturated filter returns None regardless of cap (defense in depth). - // Previously returned f64::INFINITY. - assert_eq!(filter.estimated_count(f64::INFINITY), None); - assert_eq!(filter.estimated_count(0.05), None); -} - -#[test] -fn test_bloom_filter_estimated_count_fpr_cap_boundary() { - // Cap boundary: FPR = fill^k = 0.05 at k=5 ⇒ fill ≈ 0.5493 - // 1KB filter (8192 bits). 560 bytes of 0xFF = 4480 bits set = - // fill 0.5469, FPR ≈ 0.04877 — just below cap. - // 564 bytes of 0xFF = 4512 bits set = fill 0.5508, FPR ≈ 0.05060 — - // just above cap. - - let mut below = vec![0x00u8; 1024]; - below[..560].fill(0xFF); - let below_filter = BloomFilter::from_bytes(below, DEFAULT_HASH_COUNT).unwrap(); - assert!( - below_filter.estimated_count(0.05).is_some(), - "fill 0.5469 (FPR ≈ 0.049) must be accepted by cap 0.05" - ); - - let mut above = vec![0x00u8; 1024]; - above[..564].fill(0xFF); - let above_filter = BloomFilter::from_bytes(above, DEFAULT_HASH_COUNT).unwrap(); - assert_eq!( - above_filter.estimated_count(0.05), - None, - "fill 0.5508 (FPR ≈ 0.051) must be rejected by cap 0.05" - ); - - // Same above-cap filter with a looser cap is accepted. - assert!( - above_filter.estimated_count(0.10).is_some(), - "fill 0.5508 (FPR ≈ 0.051) must be accepted by cap 0.10" - ); -} - -#[test] -fn test_bloom_filter_default() { - let default: BloomFilter = Default::default(); - let explicit = BloomFilter::new(); - assert_eq!(default, explicit); -} - -#[test] -fn test_bloom_filter_debug_format() { - let mut filter = BloomFilter::new(); - let debug = format!("{:?}", filter); - assert!(debug.contains("BloomFilter")); - assert!(debug.contains("8192")); - assert!(debug.contains("hash_count")); - - // With some entries - for i in 0..10 { - filter.insert(&make_node_addr(i)); - } - let debug = format!("{:?}", filter); - assert!(debug.contains("fill_ratio")); - assert!(debug.contains("est_count")); -} - -// ===== Mixed-Size Integration Tests ===== - -#[test] -fn test_mixed_size_outgoing_filter_construction() { - // Node at 1KB (size_class 1) with peers at different sizes - let my_node = make_node_addr(0); - let state = BloomState::new(my_node); - // state defaults to size_class 1 (1KB) - - let peer_a = make_node_addr(10); - let peer_b = make_node_addr(20); - let peer_c = make_node_addr(30); - - // Peer A: 512B filter - let mut filter_a = BloomFilter::with_params(512 * 8, 5).unwrap(); - filter_a.insert(&make_node_addr(100)); - - // Peer B: 2KB filter - let mut filter_b = BloomFilter::with_params(2048 * 8, 5).unwrap(); - filter_b.insert(&make_node_addr(200)); - - // Peer C: 4KB filter - let mut filter_c = BloomFilter::with_params(4096 * 8, 5).unwrap(); - filter_c.insert(&make_node_addr(250)); - - let mut peer_filters = HashMap::new(); - peer_filters.insert(peer_a, filter_a); - peer_filters.insert(peer_b, filter_b); - peer_filters.insert(peer_c, filter_c); - - // Outgoing filter for peer_a should be 1KB (our size) - // and should contain entries from peers B and C (converted) - let outgoing = state.compute_outgoing_filter(&peer_a, &peer_filters); - assert_eq!(outgoing.num_bits(), 1024 * 8); // our size class - assert!(outgoing.contains(&my_node)); - assert!(outgoing.contains(&make_node_addr(200))); // from B (folded 2KB→1KB) - assert!(outgoing.contains(&make_node_addr(250))); // from C (folded 4KB→1KB) -} - -#[test] -fn test_native_size_routing_queries() { - // Peer filters stored at native size work for contains() queries - let mut filter_2kb = BloomFilter::with_params(2048 * 8, 5).unwrap(); - let target = make_node_addr(42); - filter_2kb.insert(&target); - - // Query at native 2KB resolution - assert!(filter_2kb.contains(&target)); - - // After folding to 1KB, still found (but higher FPR) - let folded = filter_2kb.fold().unwrap(); - assert!(folded.contains(&target)); -} - -// ===== BloomState Tests ===== - -#[test] -fn test_bloom_state_new() { - let node = make_node_addr(0); - let state = BloomState::new(node); - - assert_eq!(state.own_node_addr(), &node); - assert!(!state.is_leaf_only()); - assert_eq!(state.sequence(), 0); - assert_eq!(state.leaf_dependent_count(), 0); -} - -#[test] -fn test_bloom_state_leaf_only() { - let node = make_node_addr(0); - let state = BloomState::leaf_only(node); - - assert!(state.is_leaf_only()); -} - -#[test] -fn test_bloom_state_leaf_dependents() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let leaf1 = make_node_addr(1); - let leaf2 = make_node_addr(2); - - state.add_leaf_dependent(leaf1); - state.add_leaf_dependent(leaf2); - assert_eq!(state.leaf_dependent_count(), 2); - - assert!(state.remove_leaf_dependent(&leaf1)); - assert_eq!(state.leaf_dependent_count(), 1); - - assert!(!state.remove_leaf_dependent(&leaf1)); // already removed -} - -#[test] -fn test_bloom_state_debounce() { - let node = make_node_addr(0); - let peer = make_node_addr(1); - let mut state = BloomState::new(node); - state.set_update_debounce_ms(500); - - state.mark_update_needed(peer); - - // Should send initially - assert!(state.should_send_update(&peer, 1000)); - - // Record send - state.record_update_sent(peer, 1000); - state.mark_update_needed(peer); - - // Should not send immediately (within debounce) - assert!(!state.should_send_update(&peer, 1200)); - - // Should send after debounce period - assert!(state.should_send_update(&peer, 1600)); -} - -#[test] -fn test_bloom_state_sequence() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - assert_eq!(state.sequence(), 0); - assert_eq!(state.next_sequence(), 1); - assert_eq!(state.next_sequence(), 2); - assert_eq!(state.sequence(), 2); -} - -#[test] -fn test_bloom_state_pending_updates() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer1 = make_node_addr(1); - let peer2 = make_node_addr(2); - - assert!(!state.needs_update(&peer1)); - - state.mark_update_needed(peer1); - assert!(state.needs_update(&peer1)); - assert!(!state.needs_update(&peer2)); - - state.mark_all_updates_needed(vec![peer1, peer2]); - assert!(state.needs_update(&peer1)); - assert!(state.needs_update(&peer2)); - - state.clear_pending_updates(); - assert!(!state.needs_update(&peer1)); - assert!(!state.needs_update(&peer2)); -} - -#[test] -fn test_bloom_state_base_filter() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let leaf = make_node_addr(1); - state.add_leaf_dependent(leaf); - - let filter = state.base_filter(); - - assert!(filter.contains(&node)); - assert!(filter.contains(&leaf)); - assert!(!filter.contains(&make_node_addr(99))); -} - -#[test] -fn test_bloom_state_compute_outgoing_filter() { - let my_node = make_node_addr(0); - let mut state = BloomState::new(my_node); - - let leaf = make_node_addr(1); - state.add_leaf_dependent(leaf); - - let peer1 = make_node_addr(10); - let peer2 = make_node_addr(20); - - // Create peer filters - let mut filter1 = BloomFilter::new(); - filter1.insert(&make_node_addr(100)); - filter1.insert(&make_node_addr(101)); - - let mut filter2 = BloomFilter::new(); - filter2.insert(&make_node_addr(200)); - - let mut peer_filters = HashMap::new(); - peer_filters.insert(peer1, filter1); - peer_filters.insert(peer2, filter2); - - // Filter for peer1 should exclude peer1's contributions - let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); - assert!(outgoing1.contains(&my_node)); // self - assert!(outgoing1.contains(&leaf)); // leaf dependent - assert!(outgoing1.contains(&make_node_addr(200))); // from peer2 - // peer1's nodes may or may not be present (depends on split brain) - - // Filter for peer2 should exclude peer2's contributions - let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); - assert!(outgoing2.contains(&my_node)); - assert!(outgoing2.contains(&leaf)); - assert!(outgoing2.contains(&make_node_addr(100))); // from peer1 - assert!(outgoing2.contains(&make_node_addr(101))); // from peer1 -} - -#[test] -fn test_bloom_state_leaf_dependents_accessor() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let leaf1 = make_node_addr(1); - let leaf2 = make_node_addr(2); - - state.add_leaf_dependent(leaf1); - state.add_leaf_dependent(leaf2); - - let deps = state.leaf_dependents(); - assert!(deps.contains(&leaf1)); - assert!(deps.contains(&leaf2)); - assert!(!deps.contains(&make_node_addr(99))); - assert_eq!(deps.len(), 2); -} - -#[test] -fn test_bloom_state_record_sent_filter() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer = make_node_addr(1); - let mut filter = BloomFilter::new(); - filter.insert(&make_node_addr(42)); - - // Record a sent filter, then mark_changed_peers should detect no change - // when the outgoing filter matches what was recorded - state.record_sent_filter(peer, filter); - - // Compute what would be sent to peer (just our own node, no peer filters) - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer]; - state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); - - // Outgoing filter (just self) differs from recorded (self + node 42), - // so peer should be marked for update - assert!(state.needs_update(&peer)); -} - -#[test] -fn test_bloom_state_remove_peer_state() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer = make_node_addr(1); - - // Populate all three internal maps for this peer - state.mark_update_needed(peer); - state.record_update_sent(peer, 1000); - state.mark_update_needed(peer); // re-mark after send - let filter = BloomFilter::new(); - state.record_sent_filter(peer, filter); - - assert!(state.needs_update(&peer)); - - // Remove all peer state - state.remove_peer_state(&peer); - - // Pending updates cleared - assert!(!state.needs_update(&peer)); - - // Debounce state cleared — should be able to send immediately - state.mark_update_needed(peer); - assert!(state.should_send_update(&peer, 0)); - - // Sent filter cleared — mark_changed_peers should treat as "never sent" - state.clear_pending_updates(); - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer]; - state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); - assert!(state.needs_update(&peer)); // never sent → must send -} - -#[test] -fn test_bloom_state_mark_changed_peers_never_sent() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer1 = make_node_addr(1); - let peer2 = make_node_addr(2); - - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer1, peer2]; - - // No filters ever sent — all peers should be marked - state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); - - assert!(state.needs_update(&peer1)); - assert!(state.needs_update(&peer2)); -} - -#[test] -fn test_bloom_state_mark_changed_peers_unchanged() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer1 = make_node_addr(1); - let peer2 = make_node_addr(2); - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer1, peer2]; - - // Compute and record what would be sent to each peer - let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); - let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); - state.record_sent_filter(peer1, outgoing1); - state.record_sent_filter(peer2, outgoing2); - - // Nothing changed — no peers should be marked - state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); - - assert!(!state.needs_update(&peer1)); - assert!(!state.needs_update(&peer2)); -} - -#[test] -fn test_bloom_state_mark_changed_peers_one_changed() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer1 = make_node_addr(1); - let peer2 = make_node_addr(2); - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer1, peer2]; - - // Record current outgoing filters for both peers - let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); - let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); - state.record_sent_filter(peer1, outgoing1); - state.record_sent_filter(peer2, outgoing2); - - // Now peer1 sends us a filter with new entries - let mut inbound_from_peer1 = BloomFilter::new(); - inbound_from_peer1.insert(&make_node_addr(100)); - let mut updated_peer_filters = HashMap::new(); - updated_peer_filters.insert(peer1, inbound_from_peer1); - - // mark_changed_peers triggered by receiving from peer1 - state.mark_changed_peers(&peer1, &peer_addrs, &updated_peer_filters); - - // peer1 is excluded (it's the source), peer2's outgoing changed - // (now includes peer1's entries via split-horizon) - assert!(!state.needs_update(&peer1)); - assert!(state.needs_update(&peer2)); -} - -#[test] -fn test_bloom_state_mark_changed_peers_excludes_source() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - - let peer1 = make_node_addr(1); - let peer_filters = HashMap::new(); - let peer_addrs = vec![peer1]; - - // peer1 is both the source and the only peer — should be skipped - state.mark_changed_peers(&peer1, &peer_addrs, &peer_filters); - - assert!(!state.needs_update(&peer1)); -} - -// ===== Adaptive Sizing Tests ===== - -#[test] -fn test_adaptive_sizing_step_up() { - let node = make_node_addr(0); - let state = BloomState::new(node); // defaults: size_class=1, up=0.20, down=0.05 - - // Above threshold → step up - assert_eq!(state.evaluate_size_change(0.25), Some(2)); -} - -#[test] -fn test_adaptive_sizing_step_down() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - state.set_size_class(2); - - // Below threshold → step down - assert_eq!(state.evaluate_size_change(0.03), Some(1)); -} - -#[test] -fn test_adaptive_sizing_deadband() { - let node = make_node_addr(0); - let state = BloomState::new(node); - - // In deadband → no change - assert_eq!(state.evaluate_size_change(0.10), None); - assert_eq!(state.evaluate_size_change(0.15), None); -} - -#[test] -fn test_adaptive_sizing_at_max() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - state.set_size_class(crate::bloom::MAX_SIZE_CLASS); - - // Above threshold but at max → no change - assert_eq!(state.evaluate_size_change(0.30), None); -} - -#[test] -fn test_adaptive_sizing_at_min() { - let node = make_node_addr(0); - let mut state = BloomState::new(node); - state.set_size_class(crate::bloom::MIN_SIZE_CLASS); - - // Below threshold but at min → no change - assert_eq!(state.evaluate_size_change(0.02), None); -} - -// === Non-routing dependent tests === - -#[test] -fn test_non_routing_peer_included_as_dependent() { - // When a non-routing peer connects, F adds it as a dependent. - // The outgoing filter should include the non-routing peer's identity. - let my_node = make_node_addr(0); - let mut state = BloomState::new(my_node); - - let non_routing_peer = make_node_addr(5); - state.add_leaf_dependent(non_routing_peer); - - let outgoing = state.compute_outgoing_filter(&make_node_addr(99), &HashMap::new()); - assert!(outgoing.contains(&my_node)); - assert!(outgoing.contains(&non_routing_peer)); -} - -#[test] -fn test_non_routing_dependent_removed_on_disconnect() { - let my_node = make_node_addr(0); - let mut state = BloomState::new(my_node); - - let non_routing_peer = make_node_addr(5); - state.add_leaf_dependent(non_routing_peer); - assert!(state.leaf_dependents().contains(&non_routing_peer)); - - state.remove_leaf_dependent(&non_routing_peer); - assert!(!state.leaf_dependents().contains(&non_routing_peer)); - - // Filter no longer contains the peer - let outgoing = state.compute_outgoing_filter(&make_node_addr(99), &HashMap::new()); - assert!(!outgoing.contains(&non_routing_peer)); -} - -#[test] -fn test_non_routing_filter_not_merged_into_outgoing() { - // Even if a non-routing peer somehow has an inbound filter, - // it should not be included in peer_filters passed to - // compute_outgoing_filter (enforced at the Node level). - // Here we verify that excluding a peer's filter from the map - // means their entries don't appear in the outgoing filter. - let my_node = make_node_addr(0); - let state = BloomState::new(my_node); - - let full_peer = make_node_addr(10); - let non_routing_peer = make_node_addr(20); - - let mut full_filter = BloomFilter::new(); - full_filter.insert(&make_node_addr(100)); - - let mut nr_filter = BloomFilter::new(); - nr_filter.insert(&make_node_addr(200)); - - // Only include the full peer's filter (simulating the Node-level exclusion) - let mut peer_filters = HashMap::new(); - peer_filters.insert(full_peer, full_filter); - // nr_filter deliberately NOT included - - let outgoing = state.compute_outgoing_filter(&make_node_addr(99), &peer_filters); - assert!(outgoing.contains(&my_node)); - assert!(outgoing.contains(&make_node_addr(100))); // from full peer - assert!(!outgoing.contains(&make_node_addr(200))); // non-routing excluded - - // But if non-routing peer is a dependent, its identity IS in the filter - let mut state2 = BloomState::new(my_node); - state2.add_leaf_dependent(non_routing_peer); - let outgoing2 = state2.compute_outgoing_filter(&make_node_addr(99), &peer_filters); - assert!(outgoing2.contains(&non_routing_peer)); // identity present - assert!(!outgoing2.contains(&make_node_addr(200))); // but not their filter entries -} diff --git a/src/lib.rs b/src/lib.rs index d7d5d33..e784c5a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,7 +9,6 @@ // distance to extracting the pure cores into a `no_std` crate later. extern crate alloc; -pub mod bloom; pub mod cache; pub mod config; pub mod control; @@ -47,8 +46,8 @@ pub use discovery::{BootstrapHandoffResult, EstablishedTraversal}; // Re-export tree types (relocated from tree:: to proto::stp) pub use proto::stp::{CoordEntry, ParentDeclaration, TreeCoordinate, TreeError, TreeState}; -// Re-export bloom filter types -pub use bloom::{BloomError, BloomFilter, BloomState}; +// Re-export bloom filter types (relocated from bloom:: to proto::bloom) +pub use proto::bloom::{BloomError, BloomFilter, BloomState}; // Re-export transport types pub use transport::udp::UdpTransport; @@ -60,13 +59,16 @@ pub use transport::{ // Re-export protocol types pub use protocol::{ - FilterAnnounce, LinkMessageType, ProtocolError, SessionAck, SessionDatagram, SessionFlags, - SessionMessageType, SessionSetup, + LinkMessageType, ProtocolError, SessionAck, SessionDatagram, SessionFlags, SessionMessageType, + SessionSetup, }; // Re-export STP wire types (relocated from protocol:: to proto::stp) pub use proto::stp::TreeAnnounce; +// Re-export bloom wire types (relocated from protocol:: to proto::bloom) +pub use proto::bloom::FilterAnnounce; + // Re-export discovery wire types (relocated from protocol:: to proto::discovery) pub use proto::discovery::{LookupRequest, LookupResponse}; diff --git a/src/node/bloom.rs b/src/node/bloom.rs index 7d1c907..17e67de 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -1,30 +1,26 @@ //! Bloom filter announce send/receive logic. //! //! Handles building, sending, and receiving FilterAnnounce messages, -//! including delta compression with NACK-based recovery and debounced -//! propagation to peers. +//! including debounced propagation to peers. use crate::NodeAddr; -use crate::bloom::BloomFilter; -use crate::protocol::{FilterAnnounce, FilterNack}; +use crate::proto::bloom::BloomFilter; +use crate::proto::bloom::FilterAnnounce; use super::reject::BloomReject; use super::{Node, NodeError}; -use std::collections::HashMap; -use tracing::{debug, trace, warn}; +use std::collections::BTreeMap; +use tracing::{debug, warn}; impl Node { - /// Collect inbound filters from full tree peers for outgoing filter computation. + /// Collect inbound filters from all peers for outgoing filter computation. /// /// Returns a map of (peer_node_addr -> filter) for peers that - /// have sent us a FilterAnnounce. Non-routing and leaf peers are - /// excluded (they don't send filters; their identity is covered - /// via leaf_dependents). - pub(super) fn peer_inbound_filters(&self) -> HashMap { - let mut filters = HashMap::new(); + /// have sent us a FilterAnnounce. + pub(super) fn peer_inbound_filters(&self) -> BTreeMap { + let mut filters = BTreeMap::new(); for (addr, peer) in &self.peers { if self.is_tree_peer(addr) - && peer.peer_profile() == crate::proto::fmp::NodeProfile::Full && let Some(filter) = peer.inbound_filter() { filters.insert(*addr, filter.clone()); @@ -35,29 +31,16 @@ impl Node { /// Build a FilterAnnounce for a specific peer. /// - /// Returns a delta (XOR diff) if we have a previous filter for this peer - /// at the same size class. Otherwise returns a full send. + /// The outgoing filter excludes the destination peer's own filter + /// to prevent routing loops (don't tell a peer about destinations + /// reachable only through them). fn build_filter_announce(&mut self, exclude_peer: &NodeAddr) -> FilterAnnounce { let peer_filters = self.peer_inbound_filters(); let filter = self .bloom_state .compute_outgoing_filter(exclude_peer, &peer_filters); let sequence = self.bloom_state.next_sequence(); - let size_class = self.bloom_state.size_class(); - - // Try delta if we have a previous filter for this peer at the same size - if let Some(last_filter) = self.bloom_state.last_sent_filter(exclude_peer) - && last_filter.num_bits() == filter.num_bits() - && let (Some(base_seq), Ok(diff)) = ( - self.bloom_state.last_sent_seq(exclude_peer), - last_filter.xor_diff(&filter), - ) - { - return FilterAnnounce::delta(diff, sequence, base_seq, size_class); - } - - // Full send - FilterAnnounce::full(filter, sequence, size_class) + FilterAnnounce::new(filter, sequence) } /// Send a FilterAnnounce to a specific peer, respecting debounce. @@ -82,22 +65,8 @@ impl Node { // Build and encode let announce = self.build_filter_announce(peer_addr); - let is_delta = announce.is_delta; - let sent_filter = if is_delta { - // For deltas, reconstruct the actual filter for change detection: - // apply the diff to the last-sent filter - let mut reconstructed = self - .bloom_state - .last_sent_filter(peer_addr) - .cloned() - .unwrap_or_default(); - let _ = reconstructed.apply_diff(&announce.filter); - reconstructed - } else { - announce.filter.clone() - }; - - let (encoded, stats) = announce.encode().map_err(|e| NodeError::SendFailed { + let sent_filter = announce.filter.clone(); + let encoded = announce.encode().map_err(|e| NodeError::SendFailed { node_addr: *peer_addr, reason: format!("FilterAnnounce encode failed: {}", e), })?; @@ -109,19 +78,6 @@ impl Node { } self.metrics().bloom.sent.inc(); - if is_delta { - self.metrics().bloom.deltas_sent.inc(); - } else { - self.metrics().bloom.full_sends.inc(); - } - self.metrics() - .bloom - .total_compressed_bytes - .add(stats.compressed_bytes as u64); - self.metrics() - .bloom - .total_raw_bytes - .add((stats.raw_words * 8) as u64); // Self-plausibility check: WARN if our own outgoing filter is // above the antipoison cap. Independent detection signal if @@ -154,15 +110,13 @@ impl Node { debug!( peer = %self.peer_display_name(peer_addr), seq = announce.sequence, - delta = is_delta, - compressed = stats.compressed_bytes, - runs = stats.run_count, est_entries = match sent_filter.estimated_count(max_fpr) { Some(n) => format!("{:.0}", n), None => "—".to_string(), }, set_bits = sent_filter.count_ones(), fill = format_args!("{:.1}%", sent_filter.fill_ratio() * 100.0), + tree_peer = self.is_tree_peer(peer_addr), "Sent FilterAnnounce" ); self.bloom_state.record_update_sent(*peer_addr, now_ms); @@ -175,14 +129,7 @@ impl Node { } /// Send pending rate-limited filter announces whose debounce has expired. - /// - /// Non-routing nodes do not send filters (they receive only). pub(super) async fn send_pending_filter_announces(&mut self) { - // Non-routing and leaf nodes don't send bloom filters - if self.node_profile() != crate::proto::fmp::NodeProfile::Full { - return; - } - let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) @@ -208,8 +155,10 @@ impl Node { /// Handle an inbound FilterAnnounce from an authenticated peer. /// - /// Supports both full sends and delta (XOR diff) updates. - /// On out-of-sequence delta, sends a NACK to request full retransmission. + /// 1. Decode and validate the message + /// 2. Check sequence freshness (reject stale/replay) + /// 3. Store the filter on the peer + /// 4. Mark other peers for outgoing filter update pub(super) async fn handle_filter_announce(&mut self, from: &NodeAddr, payload: &[u8]) { self.metrics().bloom.received.inc(); @@ -228,22 +177,26 @@ impl Node { debug!(from = %self.peer_display_name(from), "FilterAnnounce filter/size_class mismatch"); return; } + if !announce.is_v1_compliant() { + self.metrics().bloom.record_reject(BloomReject::NonV1); + debug!(from = %self.peer_display_name(from), size_class = announce.size_class, "Non-v1 FilterAnnounce rejected"); + return; + } // Check peer exists - let peer = match self.peers.get(from) { - Some(p) => p, + let current_seq = match self.peers.get(from) { + Some(peer) => peer.filter_sequence(), None => { self.metrics().bloom.record_reject(BloomReject::UnknownPeer); debug!(from = %self.peer_display_name(from), "FilterAnnounce from unknown peer"); return; } }; - let current_seq = peer.filter_sequence(); // Reject stale/replay if announce.sequence <= current_seq { self.metrics().bloom.record_reject(BloomReject::Stale); - trace!( + debug!( from = %self.peer_display_name(from), received_seq = announce.sequence, current_seq = current_seq, @@ -252,75 +205,15 @@ impl Node { return; } - // Handle delta vs full - let resolved_filter = if announce.is_delta { - // Delta: apply XOR diff to stored inbound filter - let expected_base = current_seq; - if announce.base_seq != expected_base { - // Out-of-sequence delta — send NACK - debug!( - from = %self.peer_display_name(from), - expected_base = expected_base, - got_base = announce.base_seq, - "Out-of-sequence delta, sending NACK" - ); - let nack = FilterNack { - expected_seq: expected_base, - }; - let nack_encoded = nack.encode(); - let _ = self.send_encrypted_link_message(from, &nack_encoded).await; - self.metrics().bloom.nacks_sent.inc(); - return; - } - - // Apply diff to current inbound filter - match self.peers.get(from).and_then(|p| p.inbound_filter()) { - Some(current) => { - let mut result = current.clone(); - if let Err(e) = result.apply_diff(&announce.filter) { - warn!( - from = %self.peer_display_name(from), - error = %e, - "Failed to apply filter delta" - ); - // Send NACK to request full retransmit - let nack = FilterNack { - expected_seq: current_seq, - }; - let _ = self.send_encrypted_link_message(from, &nack.encode()).await; - self.metrics().bloom.nacks_sent.inc(); - return; - } - result - } - None => { - // No stored filter to apply delta to — NACK - debug!( - from = %self.peer_display_name(from), - "Delta received but no stored filter, sending NACK" - ); - let nack = FilterNack { expected_seq: 0 }; - let _ = self.send_encrypted_link_message(from, &nack.encode()).await; - self.metrics().bloom.nacks_sent.inc(); - return; - } - } - } else { - // Full send: use directly - announce.filter.clone() - }; - - // Antipoison FPR cap. Reject announces whose resolved filter FPR - // exceeds node.bloom.max_inbound_fpr. Silent on the wire (no NACK) - // — the peer's prior accepted filter and filter_sequence stay - // untouched so the peer is not permanently silenced and a single - // corrupted frame cannot be weaponized to wipe a victim's - // contribution to aggregation. Applied to the resolved filter so - // deltas that reconstruct to an over-cap state are caught as well - // as full sends. + // Antipoison FPR cap. Reject announces whose FPR exceeds + // node.bloom.max_inbound_fpr. Silent on the wire (no NACK) — + // the peer's prior accepted filter and filter_sequence stay + // untouched so the peer is not permanently silenced and an + // on-path attacker cannot weaponize a single corrupted frame + // to wipe a victim's contribution to aggregation. let max_fpr = self.config().node.bloom.max_inbound_fpr; - let fill = resolved_filter.fill_ratio(); - let fpr = fill.powi(resolved_filter.hash_count() as i32); + let fill = announce.filter.fill_ratio(); + let fpr = fill.powi(announce.filter.hash_count() as i32); if fpr > max_fpr { self.metrics() .bloom @@ -346,99 +239,34 @@ impl Node { debug!( from = %self.peer_display_name(from), seq = announce.sequence, - delta = announce.is_delta, - est_entries = match resolved_filter.estimated_count(max_fpr) { + est_entries = match announce.filter.estimated_count(max_fpr) { Some(n) => format!("{:.0}", n), None => "—".to_string(), }, - set_bits = resolved_filter.count_ones(), - fill = format_args!("{:.1}%", resolved_filter.fill_ratio() * 100.0), + set_bits = announce.filter.count_ones(), + fill = format_args!("{:.1}%", announce.filter.fill_ratio() * 100.0), tree_peer = self.is_tree_peer(from), "Received FilterAnnounce" ); - // Store resolved filter on peer + // Store on peer if let Some(peer) = self.peers.get_mut(from) { - peer.update_filter(resolved_filter, announce.sequence, now_ms); + peer.update_filter(announce.filter, announce.sequence, now_ms); } - // Check which peers' outgoing filters actually changed + // Check which peers' outgoing filters actually changed. + // All peers receive filters, but only tree peers' inbound filters + // are merged into outgoing computation (tree-only propagation). let peer_addrs: Vec = self.peers.keys().copied().collect(); let peer_filters = self.peer_inbound_filters(); self.bloom_state .mark_changed_peers(from, &peer_addrs, &peer_filters); } - /// Handle an inbound FilterNack from a peer. - /// - /// Clears the last-sent filter for that peer, forcing a full re-send - /// on the next tick. - pub(super) async fn handle_filter_nack(&mut self, from: &NodeAddr, payload: &[u8]) { - let nack = match FilterNack::decode(payload) { - Ok(n) => n, - Err(e) => { - debug!(from = %self.peer_display_name(from), error = %e, "Malformed FilterNack"); - return; - } - }; - - debug!( - from = %self.peer_display_name(from), - expected_seq = nack.expected_seq, - "Received FilterNack, scheduling full re-send" - ); - - self.metrics().bloom.nacks_received.inc(); - // Clear sent state for this peer → next send will be full - self.bloom_state.clear_sent_filter(from); - self.bloom_state.mark_update_needed(*from); - } - - /// Evaluate adaptive filter sizing and adjust if needed. - /// - /// Checks the outgoing fill ratio for a representative peer and - /// steps up or down the size class if thresholds are crossed. - /// On size change, clears all sent filters (forcing full re-sends) - /// and marks all peers for update. - fn check_adaptive_sizing(&mut self) { - // Only Full nodes participate in filter sizing - if self.node_profile() != crate::proto::fmp::NodeProfile::Full { - return; - } - - // Use an arbitrary peer to compute a representative outgoing filter - let representative_peer = match self.peers.keys().next() { - Some(addr) => *addr, - None => return, - }; - - let peer_filters = self.peer_inbound_filters(); - let outgoing = self - .bloom_state - .compute_outgoing_filter(&representative_peer, &peer_filters); - let fill = outgoing.fill_ratio(); - - if let Some(new_class) = self.bloom_state.evaluate_size_change(fill) { - let old_class = self.bloom_state.size_class(); - debug!( - old_class = old_class, - new_class = new_class, - fill = format_args!("{:.1}%", fill * 100.0), - "Adaptive bloom filter resize" - ); - self.bloom_state.set_size_class(new_class); - self.bloom_state.clear_all_sent_filters(); - self.metrics().bloom.size_changes.inc(); - let all_peers: Vec = self.peers.keys().copied().collect(); - self.bloom_state.mark_all_updates_needed(all_peers); - } - } - /// Check bloom filter state on tick (called from event loop). /// - /// Evaluates adaptive sizing, then sends any pending filter announces. + /// Sends any pending debounced filter announces. pub(super) async fn check_bloom_state(&mut self) { - self.check_adaptive_sizing(); self.send_pending_filter_announces().await; } } diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 39f4f9a..0247568 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -42,10 +42,6 @@ impl Node { // FilterAnnounce self.handle_filter_announce(from, payload).await; } - 0x21 => { - // FilterNack - self.handle_filter_nack(from, payload).await; - } 0x30 => { // LookupRequest self.handle_lookup_request(from, payload).await; diff --git a/src/node/mod.rs b/src/node/mod.rs index 823176f..60d4c71 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -53,10 +53,10 @@ use self::wire::{ ESTABLISHED_HEADER_SIZE, FLAG_CE, FLAG_KEY_EPOCH, build_encrypted, build_established_header, prepend_inner_header, }; -use crate::bloom::{BloomFilter, BloomState}; use crate::cache::CoordCache; use crate::node::session::SessionEntry; use crate::peer::{ActivePeer, PeerConnection}; +use crate::proto::bloom::{BloomFilter, BloomState}; use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter}; use crate::proto::fmp::Fmp; use crate::proto::fmp::NodeProfile; diff --git a/src/node/tests/bloom.rs b/src/node/tests/bloom.rs index 48d090a..268fc23 100644 --- a/src/node/tests/bloom.rs +++ b/src/node/tests/bloom.rs @@ -455,8 +455,8 @@ async fn test_bloom_filter_split_horizon() { /// counted once, not the double-count fingerprint. #[test] fn compute_mesh_size_counts_each_peer_filter_once() { - use crate::bloom::BloomFilter; use crate::peer::ActivePeer; + use crate::proto::bloom::BloomFilter; use crate::proto::stp::ParentDeclaration; let mut node = make_node(); @@ -550,8 +550,8 @@ fn compute_mesh_size_counts_each_peer_filter_once() { /// `estimated_mesh_size` carries. #[test] fn compute_mesh_size_unions_overlapping_filters() { - use crate::bloom::BloomFilter; use crate::peer::ActivePeer; + use crate::proto::bloom::BloomFilter; let mut node = make_node(); @@ -632,8 +632,8 @@ fn compute_mesh_size_unions_overlapping_filters() { /// removes the parent, and asserts the estimate does not collapse. #[test] fn compute_mesh_size_stable_across_parent_drop_with_cross_link() { - use crate::bloom::BloomFilter; use crate::peer::ActivePeer; + use crate::proto::bloom::BloomFilter; let mut node = make_node(); diff --git a/src/node/tests/bloom_metrics.rs b/src/node/tests/bloom_metrics.rs deleted file mode 100644 index 14c9441..0000000 --- a/src/node/tests/bloom_metrics.rs +++ /dev/null @@ -1,102 +0,0 @@ -//! Registry-counter coverage tests for the bloom-v2 metric counters that -//! the mesh-lab suites do not reliably exercise. -//! -//! The send-path bloom counters (`deltas_sent`, `full_sends`, -//! `total_compressed_bytes`, `total_raw_bytes`) fire on every filter -//! announce and are covered by the steady-state suites. The three -//! condition-dependent counters (`nacks_sent`, `nacks_received`, -//! `size_changes`) only fire on out-of-sequence deltas, inbound NACKs, -//! and adaptive resizes — none of which occur in the stable, lossless -//! mesh-lab scenarios. These tests drive each of those paths directly and -//! assert the registry counter increments. - -use super::*; -use crate::bloom::{BloomFilter, V1_SIZE_CLASS}; -use crate::peer::ActivePeer; -use crate::protocol::{FilterAnnounce, FilterNack}; - -/// Inject a synthetic active peer with a known NodeAddr; returns it. -fn inject_peer(node: &mut Node) -> NodeAddr { - let peer_identity = make_peer_identity(); - let peer_addr = *peer_identity.node_addr(); - let peer = ActivePeer::new(peer_identity, LinkId::new(1), 0); - node.peers.insert(peer_addr, peer); - peer_addr -} - -/// Encode a FilterAnnounce to the payload format handle_filter_announce -/// expects (msg_type byte stripped). -fn encode_announce(announce: &FilterAnnounce) -> Vec { - let (mut full, _stats) = announce.encode().unwrap(); - full.remove(0); // strip msg_type byte - full -} - -/// An out-of-sequence delta to a peer with no stored filter makes the node -/// send a NACK, bumping `nacks_sent`. -#[tokio::test] -async fn test_bloom_nacks_sent_counter() { - let mut node = make_node(); - let peer_addr = inject_peer(&mut node); - - // Fresh peer: filter_sequence == 0. A delta whose base_seq does not - // match the expected base (0) is out-of-sequence → NACK. - let announce = FilterAnnounce::delta(BloomFilter::new(), 2, 5, V1_SIZE_CLASS); - let payload = encode_announce(&announce); - - node.handle_filter_announce(&peer_addr, &payload).await; - - assert_eq!( - node.metrics().bloom.nacks_sent.get(), - 1, - "registry nacks_sent must increment on out-of-sequence delta" - ); -} - -/// An inbound FilterNack bumps `nacks_received`. -#[tokio::test] -async fn test_bloom_nacks_received_counter() { - let mut node = make_node(); - let peer_addr = inject_peer(&mut node); - - let mut payload = FilterNack { expected_seq: 7 }.encode(); - payload.remove(0); // strip msg_type byte (decode expects the seq only) - - node.handle_filter_nack(&peer_addr, &payload).await; - - assert_eq!( - node.metrics().bloom.nacks_received.get(), - 1, - "registry nacks_received must increment on inbound NACK" - ); -} - -/// A fresh Full node starts at V1_SIZE_CLASS with a nearly empty outgoing -/// filter (just its own addr), so the first adaptive-sizing pass steps the -/// size class down, bumping `size_changes`. -#[tokio::test] -async fn test_bloom_size_changes_counter() { - let mut node = make_node(); - // check_adaptive_sizing needs at least one peer for the representative - // outgoing-filter computation. - let _peer = inject_peer(&mut node); - - assert_eq!( - node.bloom_state.size_class(), - V1_SIZE_CLASS, - "fresh node starts at the v1 size class" - ); - - node.check_bloom_state().await; - - assert_eq!( - node.metrics().bloom.size_changes.get(), - 1, - "registry size_changes must increment on adaptive resize" - ); - assert_eq!( - node.bloom_state.size_class(), - V1_SIZE_CLASS - 1, - "near-empty outgoing filter steps the size class down" - ); -} diff --git a/src/node/tests/bloom_poison.rs b/src/node/tests/bloom_poison.rs index 711bbd5..d29cf67 100644 --- a/src/node/tests/bloom_poison.rs +++ b/src/node/tests/bloom_poison.rs @@ -7,9 +7,9 @@ //! bloom.rs. use super::*; -use crate::bloom::{BloomFilter, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; use crate::peer::ActivePeer; -use crate::protocol::FilterAnnounce; +use crate::proto::bloom::FilterAnnounce; +use crate::proto::bloom::{BloomFilter, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; /// Inject a synthetic active peer into the node with a known NodeAddr. /// Returns the peer's NodeAddr. @@ -24,7 +24,7 @@ fn inject_peer(node: &mut Node) -> NodeAddr { /// Encode a FilterAnnounce to the payload format handle_filter_announce /// expects (msg_type byte stripped). fn encode_payload(announce: &FilterAnnounce) -> Vec { - let (mut full, _stats) = announce.encode().unwrap(); + let mut full = announce.encode().unwrap(); full.remove(0); // strip msg_type byte full } @@ -40,7 +40,7 @@ async fn test_m1_rejects_all_ones_filter_announce() { DEFAULT_HASH_COUNT, ) .unwrap(); - let announce = FilterAnnounce::full(all_ones, 1, 1); + let announce = FilterAnnounce::new(all_ones, 1); let payload = encode_payload(&announce); let before_fill_exceeded = node.metrics().bloom.fill_exceeded.get(); @@ -88,7 +88,7 @@ async fn test_m1_accepts_sub_cap_filter() { bytes[0] = i; filter.insert(&NodeAddr::from_bytes(bytes)); } - let announce = FilterAnnounce::full(filter, 1, 1); + let announce = FilterAnnounce::new(filter, 1); let payload = encode_payload(&announce); let before_fill_exceeded = node.metrics().bloom.fill_exceeded.get(); @@ -135,7 +135,7 @@ async fn test_m1_sequence_not_advanced_allows_recovery() { DEFAULT_HASH_COUNT, ) .unwrap(); - let bad_announce = FilterAnnounce::full(bad, 1, 1); + let bad_announce = FilterAnnounce::new(bad, 1); node.handle_filter_announce(&peer_addr, &encode_payload(&bad_announce)) .await; assert_eq!( @@ -152,7 +152,7 @@ async fn test_m1_sequence_not_advanced_allows_recovery() { bytes[0] = i; good.insert(&NodeAddr::from_bytes(bytes)); } - let good_announce = FilterAnnounce::full(good, 1, 1); + let good_announce = FilterAnnounce::new(good, 1); node.handle_filter_announce(&peer_addr, &encode_payload(&good_announce)) .await; diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index 30ef768..65dc61a 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -1160,8 +1160,8 @@ async fn test_open_discovery_sweep_queues_eligible_skips_filtered() { /// that `initiate_lookup` ran fresh on each attempt. #[tokio::test] async fn test_check_pending_lookups_default_sequence_unreachable() { - use crate::bloom::BloomFilter; use crate::peer::ActivePeer; + use crate::proto::bloom::BloomFilter; use crate::proto::discovery::PendingLookup; use crate::transport::LinkId; use std::sync::mpsc; diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 9c39db6..2bc5d4f 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -8,7 +8,6 @@ mod acl; #[cfg(target_os = "linux")] mod ble; mod bloom; -mod bloom_metrics; mod bloom_poison; mod bootstrap; mod decrypt_failure; diff --git a/src/node/tests/routing.rs b/src/node/tests/routing.rs index 48fa688..5123581 100644 --- a/src/node/tests/routing.rs +++ b/src/node/tests/routing.rs @@ -4,7 +4,7 @@ //! filter priority, greedy tree routing, and tie-breaking. use super::*; -use crate::bloom::BloomFilter; +use crate::proto::bloom::BloomFilter; use crate::proto::stp::{ParentDeclaration, TreeCoordinate}; use spanning_tree::{ TestNode, cleanup_nodes, drain_all_packets, generate_random_edges, initiate_handshake, diff --git a/src/peer/active.rs b/src/peer/active.rs index ad756ca..e336d5c 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -3,10 +3,10 @@ //! Represents a fully authenticated peer after successful Noise handshake. //! ActivePeer holds tree state, Bloom filter, and routing information. -use crate::bloom::BloomFilter; use crate::mmp::MmpConfig; use crate::node::REKEY_JITTER_SECS; use crate::noise::{HandshakeState as NoiseHandshakeState, NoiseError, NoiseSession}; +use crate::proto::bloom::BloomFilter; use crate::proto::fmp::{NegotiationPayload, NodeProfile}; use crate::proto::mmp::MmpPeerState; use crate::proto::stp::{ParentDeclaration, TreeCoordinate}; diff --git a/src/proto/bloom/core.rs b/src/proto/bloom/core.rs new file mode 100644 index 0000000..2876faf --- /dev/null +++ b/src/proto/bloom/core.rs @@ -0,0 +1,259 @@ +//! Generic Bloom filter data structure. + +use core::fmt; + +use super::{BloomError, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; +use crate::NodeAddr; + +/// A Bloom filter for probabilistic set membership. +/// +/// Used in FIPS to track which destinations are reachable through a peer. +/// The filter uses double hashing to generate k hash functions from two +/// base hashes derived from the input. +#[derive(Clone)] +pub struct BloomFilter { + /// Bit array storage (packed as bytes). + bits: Vec, + /// Number of bits in the filter. + num_bits: usize, + /// Number of hash functions to use. + hash_count: u8, +} + +impl BloomFilter { + /// Create a new empty Bloom filter with default parameters. + pub fn new() -> Self { + Self::with_params(DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT) + .expect("default params are valid") + } + + /// Create a Bloom filter with custom parameters. + pub fn with_params(num_bits: usize, hash_count: u8) -> Result { + if num_bits == 0 || !num_bits.is_multiple_of(8) { + return Err(BloomError::SizeNotByteAligned(num_bits)); + } + if hash_count == 0 { + return Err(BloomError::ZeroHashCount); + } + + let num_bytes = num_bits / 8; + Ok(Self { + bits: vec![0u8; num_bytes], + num_bits, + hash_count, + }) + } + + /// Create a Bloom filter from raw bytes. + pub fn from_bytes(bytes: Vec, hash_count: u8) -> Result { + if hash_count == 0 { + return Err(BloomError::ZeroHashCount); + } + if bytes.is_empty() { + return Err(BloomError::SizeNotByteAligned(0)); + } + let num_bits = bytes.len() * 8; + Ok(Self { + bits: bytes, + num_bits, + hash_count, + }) + } + + /// Create a Bloom filter from a byte slice. + pub fn from_slice(bytes: &[u8], hash_count: u8) -> Result { + Self::from_bytes(bytes.to_vec(), hash_count) + } + + /// Insert a NodeAddr into the filter. + pub fn insert(&mut self, node_addr: &NodeAddr) { + for i in 0..self.hash_count { + let bit_index = self.hash(node_addr.as_bytes(), i); + self.set_bit(bit_index); + } + } + + /// Insert raw bytes into the filter. + pub fn insert_bytes(&mut self, data: &[u8]) { + for i in 0..self.hash_count { + let bit_index = self.hash(data, i); + self.set_bit(bit_index); + } + } + + /// Check if the filter might contain a NodeAddr. + /// + /// Returns `true` if the item might be in the set (possible false positive). + /// Returns `false` if the item is definitely not in the set. + pub fn contains(&self, node_addr: &NodeAddr) -> bool { + self.contains_bytes(node_addr.as_bytes()) + } + + /// Check if the filter might contain raw bytes. + pub fn contains_bytes(&self, data: &[u8]) -> bool { + for i in 0..self.hash_count { + let bit_index = self.hash(data, i); + if !self.get_bit(bit_index) { + return false; + } + } + true + } + + /// Merge another filter into this one (OR operation). + /// + /// After merge, this filter contains all elements from both filters. + pub fn merge(&mut self, other: &BloomFilter) -> Result<(), BloomError> { + if self.num_bits != other.num_bits { + return Err(BloomError::InvalidSize { + expected: self.num_bits, + got: other.num_bits, + }); + } + + for (a, b) in self.bits.iter_mut().zip(other.bits.iter()) { + *a |= b; + } + Ok(()) + } + + /// Create a new filter that is the union of this and another. + pub fn union(&self, other: &BloomFilter) -> Result { + let mut result = self.clone(); + result.merge(other)?; + Ok(result) + } + + /// Clear all bits in the filter. + pub fn clear(&mut self) { + self.bits.fill(0); + } + + /// Count the number of set bits (population count). + pub fn count_ones(&self) -> usize { + self.bits.iter().map(|b| b.count_ones() as usize).sum() + } + + /// Estimate the fill ratio (set bits / total bits). + pub fn fill_ratio(&self) -> f64 { + self.count_ones() as f64 / self.num_bits as f64 + } + + /// Estimate the number of elements in the filter. + /// + /// Uses the formula: n = -(m/k) * ln(1 - X/m) + /// where m = num_bits, k = hash_count, X = count_ones + /// + /// Returns `None` when the filter's FPR exceeds `max_fpr` (antipoison + /// cap) or the filter is saturated (`count_ones() >= num_bits`). Pass + /// `f64::INFINITY` for `max_fpr` to disable the cap — useful in + /// Debug/log contexts where no policy is in scope. The saturated + /// branch is always honored regardless of `max_fpr`, preventing the + /// `f64::INFINITY` return that the previous signature produced. + pub fn estimated_count(&self, max_fpr: f64) -> Option { + let m = self.num_bits as f64; + let k = self.hash_count as f64; + let x = self.count_ones() as f64; + + if x >= m { + return None; + } + + let fill = x / m; + let fpr = fill.powi(self.hash_count as i32); + if fpr > max_fpr { + return None; + } + + Some(-(m / k) * (1.0 - fill).ln()) + } + + /// Check if the filter is empty. + pub fn is_empty(&self) -> bool { + self.bits.iter().all(|&b| b == 0) + } + + /// Get the raw bytes. + pub fn as_bytes(&self) -> &[u8] { + &self.bits + } + + /// Get the filter size in bits. + pub fn num_bits(&self) -> usize { + self.num_bits + } + + /// Get the filter size in bytes. + pub fn num_bytes(&self) -> usize { + self.bits.len() + } + + /// Get the number of hash functions. + pub fn hash_count(&self) -> u8 { + self.hash_count + } + + /// Compute a hash index for the given data and hash function number. + /// + /// Uses double hashing: h(x,i) = (h1(x) + i*h2(x)) mod m + fn hash(&self, data: &[u8], k: u8) -> usize { + // Use first 16 bytes of SHA-256 for h1 and h2 + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(data); + let hash = hasher.finalize(); + + // h1 from first 8 bytes + let h1 = u64::from_le_bytes(hash[0..8].try_into().unwrap()); + // h2 from next 8 bytes + let h2 = u64::from_le_bytes(hash[8..16].try_into().unwrap()); + + let combined = h1.wrapping_add((k as u64).wrapping_mul(h2)); + (combined as usize) % self.num_bits + } + + fn set_bit(&mut self, index: usize) { + let byte_index = index / 8; + let bit_offset = index % 8; + self.bits[byte_index] |= 1 << bit_offset; + } + + fn get_bit(&self, index: usize) -> bool { + let byte_index = index / 8; + let bit_offset = index % 8; + (self.bits[byte_index] >> bit_offset) & 1 == 1 + } +} + +impl Default for BloomFilter { + fn default() -> Self { + Self::new() + } +} + +impl PartialEq for BloomFilter { + fn eq(&self, other: &Self) -> bool { + self.num_bits == other.num_bits + && self.hash_count == other.hash_count + && self.bits == other.bits + } +} + +impl Eq for BloomFilter {} + +impl fmt::Debug for BloomFilter { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BloomFilter") + .field("bits", &self.num_bits) + .field("hash_count", &self.hash_count) + .field("fill_ratio", &format!("{:.2}%", self.fill_ratio() * 100.0)) + .field( + "est_count", + &match self.estimated_count(f64::INFINITY) { + Some(n) => format!("{:.0}", n), + None => "saturated".to_string(), + }, + ) + .finish() + } +} diff --git a/src/proto/bloom/limits.rs b/src/proto/bloom/limits.rs new file mode 100644 index 0000000..bef6c27 --- /dev/null +++ b/src/proto/bloom/limits.rs @@ -0,0 +1,28 @@ +//! v1 bloom filter sizing constants (the tunables). + +/// Default filter size in bits (1KB = 8,192 bits). +/// +/// Sized for ~800-1,600 entries. FPR ~0.05% at 400 entries, ~0.9% at 800. +/// This is v1 protocol default (size_class=1). +pub const DEFAULT_FILTER_SIZE_BITS: usize = 8192; + +/// Default filter size in bytes (1KB). +/// +/// Retained for completeness of the v1 tunable set; no current consumer. +#[allow(dead_code)] +pub const DEFAULT_FILTER_SIZE_BYTES: usize = DEFAULT_FILTER_SIZE_BITS / 8; + +/// Default number of hash functions. +/// +/// k=5 is optimal at ~1,200 entries and a good compromise for 800-1,600. +/// At 400 entries: FPR ~0.05%. At 800 entries: FPR ~0.9%. +pub const DEFAULT_HASH_COUNT: u8 = 5; + +/// Size class for v1 protocol (1 KB filters). +pub const V1_SIZE_CLASS: u8 = 1; + +/// Filter sizes by size_class: bytes = 512 << size_class +/// +/// Retained for completeness of the v1 tunable set; no current consumer. +#[allow(dead_code)] +pub const SIZE_CLASS_BYTES: [usize; 4] = [512, 1024, 2048, 4096]; diff --git a/src/proto/bloom/mod.rs b/src/proto/bloom/mod.rs new file mode 100644 index 0000000..0a00c9f --- /dev/null +++ b/src/proto/bloom/mod.rs @@ -0,0 +1,47 @@ +//! Sans-IO bloom filter subsystem. +//! +//! 1KB Bloom filters for reachability in FIPS routing. Each node maintains +//! filters that summarize which destinations are reachable through each peer, +//! enabling efficient routing decisions without global network knowledge. +//! +//! ## v1 Parameters +//! +//! - Size: 1 KB (8,192 bits) - sized for actual ~400-800 entry occupancy +//! - Hash functions: k=5 - optimal at ~1,200 entries, good for 800-1,600 +//! - Bandwidth: 1 KB/announce (75% reduction from original 4KB design) +//! +//! - `core.rs` — the pure `BloomFilter` data structure. +//! - `state.rs` — `BloomState` (per-peer inbound store + outgoing filter +//! computation + the send-debounce decision). +//! - `limits.rs` — the v1 sizing constants. +//! - `wire.rs` — `FilterAnnounce` + `encode`/`decode` (the std-tethered file). +//! It imports the shared `ProtocolError` and `LinkMessageType` downward from +//! `crate::protocol`. + +mod core; +mod limits; +mod state; +mod wire; + +#[cfg(test)] +mod tests; + +use thiserror::Error; + +pub use core::BloomFilter; +pub use limits::{DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT, V1_SIZE_CLASS}; +pub use state::BloomState; +pub use wire::FilterAnnounce; + +/// Errors related to Bloom filter operations. +#[derive(Debug, Error)] +pub enum BloomError { + #[error("invalid filter size: expected {expected} bits, got {got}")] + InvalidSize { expected: usize, got: usize }, + + #[error("filter size must be a multiple of 8, got {0}")] + SizeNotByteAligned(usize), + + #[error("hash count must be positive")] + ZeroHashCount, +} diff --git a/src/bloom/state.rs b/src/proto/bloom/state.rs similarity index 62% rename from src/bloom/state.rs rename to src/proto/bloom/state.rs index 546d4f8..abbd451 100644 --- a/src/bloom/state.rs +++ b/src/proto/bloom/state.rs @@ -1,8 +1,8 @@ //! FIPS-specific Bloom filter announcement state management. -use std::collections::{HashMap, HashSet}; +use alloc::collections::{BTreeMap, BTreeSet}; -use super::{BloomFilter, MAX_SIZE_CLASS, MIN_SIZE_CLASS, V1_SIZE_CLASS, size_class_to_bits}; +use super::BloomFilter; use crate::NodeAddr; /// State for managing Bloom filter announcements. @@ -13,27 +13,19 @@ pub struct BloomState { /// This node's NodeAddr (always included in outgoing filters). own_node_addr: NodeAddr, /// Leaf-only nodes we speak for (included in our filter). - leaf_dependents: HashSet, + leaf_dependents: BTreeSet, /// Whether this node operates in leaf-only mode. is_leaf_only: bool, - /// This node's filter size class. - size_class: u8, /// Rate limiting: minimum interval between outgoing updates (milliseconds). update_debounce_ms: u64, /// Timestamp of last update sent (per peer, in milliseconds). - last_update_sent: HashMap, + last_update_sent: BTreeMap, /// Peers that need a filter update. - pending_updates: HashSet, + pending_updates: BTreeSet, /// Current sequence number for outgoing filters. sequence: u64, - /// Last outgoing filter sent to each peer (for change detection and delta computation). - last_sent_filters: HashMap, - /// Sequence number of the last filter sent to each peer. - last_sent_seq: HashMap, - /// Fill ratio threshold above which to step up to a larger size class. - step_up_threshold: f64, - /// Fill ratio threshold below which to step down to a smaller size class. - step_down_threshold: f64, + /// Last outgoing filter sent to each peer (for change detection). + last_sent_filters: BTreeMap, } impl BloomState { @@ -41,17 +33,13 @@ impl BloomState { pub fn new(own_node_addr: NodeAddr) -> Self { Self { own_node_addr, - leaf_dependents: HashSet::new(), + leaf_dependents: BTreeSet::new(), is_leaf_only: false, - size_class: V1_SIZE_CLASS, update_debounce_ms: 500, - last_update_sent: HashMap::new(), - pending_updates: HashSet::new(), + last_update_sent: BTreeMap::new(), + pending_updates: BTreeSet::new(), sequence: 0, - last_sent_filters: HashMap::new(), - last_sent_seq: HashMap::new(), - step_up_threshold: 0.20, - step_down_threshold: 0.05, + last_sent_filters: BTreeMap::new(), } } @@ -72,19 +60,6 @@ impl BloomState { self.is_leaf_only } - /// Get the current filter size class. - pub fn size_class(&self) -> u8 { - self.size_class - } - - /// Set the filter size class. - /// - /// This does NOT trigger re-sends; the caller must clear sent filters - /// and mark all peers for update. - pub fn set_size_class(&mut self, size_class: u8) { - self.size_class = size_class; - } - /// Get the current sequence number. pub fn sequence(&self) -> u64 { self.sequence @@ -117,7 +92,7 @@ impl BloomState { } /// Get the set of leaf dependents. - pub fn leaf_dependents(&self) -> &HashSet { + pub fn leaf_dependents(&self) -> &BTreeSet { &self.leaf_dependents } @@ -164,43 +139,24 @@ impl BloomState { self.pending_updates.clear(); } - /// Record the outgoing filter and sequence that was sent to a peer. + /// Record the outgoing filter that was sent to a peer. pub fn record_sent_filter(&mut self, peer_id: NodeAddr, filter: BloomFilter) { - let seq = self.sequence; self.last_sent_filters.insert(peer_id, filter); - self.last_sent_seq.insert(peer_id, seq); } - /// Get the last filter sent to a peer (for delta computation). + /// Read back the last outgoing filter actually sent to a peer, if any. + /// + /// Returns the filter recorded by [`record_sent_filter`](Self::record_sent_filter) + /// — i.e. what the peer currently holds for us — or `None` when no announce + /// has been sent to that peer yet (or the node is root, with no parent to + /// send to). pub fn last_sent_filter(&self, peer_id: &NodeAddr) -> Option<&BloomFilter> { self.last_sent_filters.get(peer_id) } - /// Get the sequence number of the last filter sent to a peer. - pub fn last_sent_seq(&self, peer_id: &NodeAddr) -> Option { - self.last_sent_seq.get(peer_id).copied() - } - - /// Clear the sent filter for a specific peer (e.g., on NACK). - /// - /// Forces the next send to be a full filter. - pub fn clear_sent_filter(&mut self, peer_id: &NodeAddr) { - self.last_sent_filters.remove(peer_id); - self.last_sent_seq.remove(peer_id); - } - - /// Clear all sent filters (e.g., on size class change). - /// - /// Forces full sends to all peers. - pub fn clear_all_sent_filters(&mut self) { - self.last_sent_filters.clear(); - self.last_sent_seq.clear(); - } - /// Remove stored filter state for a peer that was removed. pub fn remove_peer_state(&mut self, peer_id: &NodeAddr) { self.last_sent_filters.remove(peer_id); - self.last_sent_seq.remove(peer_id); self.last_update_sent.remove(peer_id); self.pending_updates.remove(peer_id); } @@ -214,7 +170,7 @@ impl BloomState { &mut self, exclude_from: &NodeAddr, peer_addrs: &[NodeAddr], - peer_filters: &HashMap, + peer_filters: &BTreeMap, ) { for peer_addr in peer_addrs { if peer_addr == exclude_from { @@ -233,24 +189,32 @@ impl BloomState { /// Compute the outgoing filter for a specific peer. /// - /// The filter is created at this node's size class. Peer filters of - /// different sizes are automatically converted (folded or duplicated) - /// during the merge operation. - /// /// The filter includes: /// - This node's own ID /// - All leaf dependents /// - Entries from other peers' inbound filters (excluding the destination peer) + /// + /// The `peer_filters` map contains inbound filters from each peer. + /// The filter for `exclude_peer` is excluded to prevent routing loops. pub fn compute_outgoing_filter( &self, exclude_peer: &NodeAddr, - peer_filters: &HashMap, + peer_filters: &BTreeMap, ) -> BloomFilter { - let mut filter = self.base_filter(); + let mut filter = BloomFilter::new(); - // Merge filters from other peers (auto-converting sizes) + // Always include ourselves + filter.insert(&self.own_node_addr); + + // Include leaf dependents + for dep in &self.leaf_dependents { + filter.insert(dep); + } + + // Merge filters from other peers for (peer_id, peer_filter) in peer_filters { if peer_id != exclude_peer { + // Ignore merge errors (size mismatches) - just skip that filter let _ = filter.merge(peer_filter); } } @@ -258,27 +222,9 @@ impl BloomState { filter } - /// Evaluate whether the filter size class should change. - /// - /// Returns `Some(new_class)` if the outgoing fill ratio crosses a - /// threshold, `None` if no change is needed. - pub fn evaluate_size_change(&self, fill_ratio: f64) -> Option { - if fill_ratio > self.step_up_threshold && self.size_class < MAX_SIZE_CLASS { - Some(self.size_class + 1) - } else if fill_ratio < self.step_down_threshold && self.size_class > MIN_SIZE_CLASS { - Some(self.size_class - 1) - } else { - None - } - } - /// Create a base filter containing just this node and its dependents. - /// - /// The filter is created at this node's size class. pub fn base_filter(&self) -> BloomFilter { - let num_bits = size_class_to_bits(self.size_class); - let mut filter = BloomFilter::with_params(num_bits, super::DEFAULT_HASH_COUNT) - .expect("size_class produces valid params"); + let mut filter = BloomFilter::new(); filter.insert(&self.own_node_addr); for dep in &self.leaf_dependents { filter.insert(dep); diff --git a/src/proto/bloom/tests/core.rs b/src/proto/bloom/tests/core.rs new file mode 100644 index 0000000..b081427 --- /dev/null +++ b/src/proto/bloom/tests/core.rs @@ -0,0 +1,290 @@ +//! Tests for the `BloomFilter` data structure. + +use crate::proto::bloom::{BloomError, BloomFilter, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; +use crate::testutil::make_node_addr; + +#[test] +fn test_bloom_filter_new() { + let filter = BloomFilter::new(); + assert_eq!(filter.num_bits(), DEFAULT_FILTER_SIZE_BITS); + assert_eq!(filter.hash_count(), DEFAULT_HASH_COUNT); + assert_eq!(filter.count_ones(), 0); + assert!(filter.is_empty()); +} + +#[test] +fn test_bloom_filter_insert_contains() { + let mut filter = BloomFilter::new(); + let node1 = make_node_addr(1); + let node2 = make_node_addr(2); + + assert!(!filter.contains(&node1)); + assert!(!filter.contains(&node2)); + + filter.insert(&node1); + + assert!(filter.contains(&node1)); + // node2 might have false positive, but very unlikely with single insert + assert!(!filter.is_empty()); +} + +#[test] +fn test_bloom_filter_multiple_inserts() { + let mut filter = BloomFilter::new(); + + for i in 0..100 { + let node = make_node_addr(i); + filter.insert(&node); + } + + // All inserted items should be found + for i in 0..100 { + let node = make_node_addr(i); + assert!(filter.contains(&node), "Node {} not found", i); + } + + // Fill ratio should be reasonable + let fill = filter.fill_ratio(); + assert!(fill > 0.0 && fill < 0.5, "Unexpected fill ratio: {}", fill); +} + +#[test] +fn test_bloom_filter_merge() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + let node1 = make_node_addr(1); + let node2 = make_node_addr(2); + + filter1.insert(&node1); + filter2.insert(&node2); + + filter1.merge(&filter2).unwrap(); + + assert!(filter1.contains(&node1)); + assert!(filter1.contains(&node2)); +} + +#[test] +fn test_bloom_filter_union() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + let node1 = make_node_addr(1); + let node2 = make_node_addr(2); + + filter1.insert(&node1); + filter2.insert(&node2); + + let union = filter1.union(&filter2).unwrap(); + + assert!(union.contains(&node1)); + assert!(union.contains(&node2)); + // Original filters unchanged + assert!(!filter1.contains(&node2)); + assert!(!filter2.contains(&node1)); +} + +#[test] +fn test_bloom_filter_clear() { + let mut filter = BloomFilter::new(); + let node = make_node_addr(1); + + filter.insert(&node); + assert!(!filter.is_empty()); + + filter.clear(); + assert!(filter.is_empty()); + assert_eq!(filter.count_ones(), 0); + assert!(!filter.contains(&node)); +} + +#[test] +fn test_bloom_filter_merge_size_mismatch() { + let mut filter1 = BloomFilter::with_params(1024, 7).unwrap(); + let filter2 = BloomFilter::with_params(2048, 7).unwrap(); + + let result = filter1.merge(&filter2); + assert!(matches!(result, Err(BloomError::InvalidSize { .. }))); +} + +#[test] +fn test_bloom_filter_custom_params() { + let filter = BloomFilter::with_params(1024, 5).unwrap(); + assert_eq!(filter.num_bits(), 1024); + assert_eq!(filter.num_bytes(), 128); + assert_eq!(filter.hash_count(), 5); +} + +#[test] +fn test_bloom_filter_invalid_params() { + // Not byte-aligned (1001 is not divisible by 8) + assert!(matches!( + BloomFilter::with_params(1001, 7), + Err(BloomError::SizeNotByteAligned(1001)) + )); + + // Zero size + assert!(matches!( + BloomFilter::with_params(0, 7), + Err(BloomError::SizeNotByteAligned(0)) + )); + + // Zero hash count + assert!(matches!( + BloomFilter::with_params(1024, 0), + Err(BloomError::ZeroHashCount) + )); +} + +#[test] +fn test_bloom_filter_from_bytes() { + let original = BloomFilter::new(); + let bytes = original.as_bytes().to_vec(); + + let restored = BloomFilter::from_bytes(bytes, original.hash_count()).unwrap(); + + assert_eq!(original, restored); +} + +#[test] +fn test_bloom_filter_estimated_count() { + let mut filter = BloomFilter::new(); + + // Empty filter + assert_eq!(filter.estimated_count(f64::INFINITY), Some(0.0)); + + // Insert some items + for i in 0..50 { + filter.insert(&make_node_addr(i)); + } + + // Estimate should be reasonably close to 50 + let estimate = filter.estimated_count(f64::INFINITY).unwrap(); + assert!( + estimate > 30.0 && estimate < 100.0, + "Unexpected estimate: {}", + estimate + ); +} + +#[test] +fn test_bloom_filter_equality() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + assert_eq!(filter1, filter2); + + filter1.insert(&make_node_addr(1)); + assert_ne!(filter1, filter2); + + filter2.insert(&make_node_addr(1)); + assert_eq!(filter1, filter2); +} + +#[test] +fn test_bloom_filter_from_bytes_empty() { + let result = BloomFilter::from_bytes(vec![], 5); + assert!(matches!(result, Err(BloomError::SizeNotByteAligned(0)))); +} + +#[test] +fn test_bloom_filter_from_bytes_zero_hash_count() { + let result = BloomFilter::from_bytes(vec![0u8; 128], 0); + assert!(matches!(result, Err(BloomError::ZeroHashCount))); +} + +#[test] +fn test_bloom_filter_from_slice() { + let mut original = BloomFilter::new(); + original.insert(&make_node_addr(42)); + let bytes = original.as_bytes(); + + let restored = BloomFilter::from_slice(bytes, original.hash_count()).unwrap(); + assert_eq!(original, restored); +} + +#[test] +fn test_bloom_filter_insert_bytes_contains_bytes() { + let mut filter = BloomFilter::new(); + let data1 = b"hello world"; + let data2 = b"goodbye"; + + assert!(!filter.contains_bytes(data1)); + + filter.insert_bytes(data1); + assert!(filter.contains_bytes(data1)); + assert!(!filter.contains_bytes(data2)); + + filter.insert_bytes(data2); + assert!(filter.contains_bytes(data1)); + assert!(filter.contains_bytes(data2)); +} + +#[test] +fn test_bloom_filter_estimated_count_saturated() { + // Create a small filter with all bits set + let bytes = vec![0xFF; 8]; // all bits set + let filter = BloomFilter::from_bytes(bytes, 3).unwrap(); + + // Saturated filter returns None regardless of cap (defense in depth). + // Previously returned f64::INFINITY. + assert_eq!(filter.estimated_count(f64::INFINITY), None); + assert_eq!(filter.estimated_count(0.05), None); +} + +#[test] +fn test_bloom_filter_estimated_count_fpr_cap_boundary() { + // Cap boundary: FPR = fill^k = 0.05 at k=5 ⇒ fill ≈ 0.5493 + // 1KB filter (8192 bits). 560 bytes of 0xFF = 4480 bits set = + // fill 0.5469, FPR ≈ 0.04877 — just below cap. + // 564 bytes of 0xFF = 4512 bits set = fill 0.5508, FPR ≈ 0.05060 — + // just above cap. + + let mut below = vec![0x00u8; 1024]; + below[..560].fill(0xFF); + let below_filter = BloomFilter::from_bytes(below, DEFAULT_HASH_COUNT).unwrap(); + assert!( + below_filter.estimated_count(0.05).is_some(), + "fill 0.5469 (FPR ≈ 0.049) must be accepted by cap 0.05" + ); + + let mut above = vec![0x00u8; 1024]; + above[..564].fill(0xFF); + let above_filter = BloomFilter::from_bytes(above, DEFAULT_HASH_COUNT).unwrap(); + assert_eq!( + above_filter.estimated_count(0.05), + None, + "fill 0.5508 (FPR ≈ 0.051) must be rejected by cap 0.05" + ); + + // Same above-cap filter with a looser cap is accepted. + assert!( + above_filter.estimated_count(0.10).is_some(), + "fill 0.5508 (FPR ≈ 0.051) must be accepted by cap 0.10" + ); +} + +#[test] +fn test_bloom_filter_default() { + let default: BloomFilter = Default::default(); + let explicit = BloomFilter::new(); + assert_eq!(default, explicit); +} + +#[test] +fn test_bloom_filter_debug_format() { + let mut filter = BloomFilter::new(); + let debug = format!("{:?}", filter); + assert!(debug.contains("BloomFilter")); + assert!(debug.contains("8192")); + assert!(debug.contains("hash_count")); + + // With some entries + for i in 0..10 { + filter.insert(&make_node_addr(i)); + } + let debug = format!("{:?}", filter); + assert!(debug.contains("fill_ratio")); + assert!(debug.contains("est_count")); +} diff --git a/src/proto/bloom/tests/mod.rs b/src/proto/bloom/tests/mod.rs new file mode 100644 index 0000000..35c691d --- /dev/null +++ b/src/proto/bloom/tests/mod.rs @@ -0,0 +1,6 @@ +//! Bloom subsystem unit tests, extracted from the co-located `#[cfg(test)]` +//! blocks in the sibling source modules. + +mod core; +mod state; +mod wire; diff --git a/src/proto/bloom/tests/state.rs b/src/proto/bloom/tests/state.rs new file mode 100644 index 0000000..6c74746 --- /dev/null +++ b/src/proto/bloom/tests/state.rs @@ -0,0 +1,315 @@ +//! Tests for `BloomState` (announcement state management). + +use std::collections::BTreeMap; + +use crate::proto::bloom::{BloomFilter, BloomState}; +use crate::testutil::make_node_addr; + +#[test] +fn test_bloom_state_new() { + let node = make_node_addr(0); + let state = BloomState::new(node); + + assert_eq!(state.own_node_addr(), &node); + assert!(!state.is_leaf_only()); + assert_eq!(state.sequence(), 0); + assert_eq!(state.leaf_dependent_count(), 0); +} + +#[test] +fn test_bloom_state_leaf_only() { + let node = make_node_addr(0); + let state = BloomState::leaf_only(node); + + assert!(state.is_leaf_only()); +} + +#[test] +fn test_bloom_state_leaf_dependents() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let leaf1 = make_node_addr(1); + let leaf2 = make_node_addr(2); + + state.add_leaf_dependent(leaf1); + state.add_leaf_dependent(leaf2); + assert_eq!(state.leaf_dependent_count(), 2); + + assert!(state.remove_leaf_dependent(&leaf1)); + assert_eq!(state.leaf_dependent_count(), 1); + + assert!(!state.remove_leaf_dependent(&leaf1)); // already removed +} + +#[test] +fn test_bloom_state_debounce() { + let node = make_node_addr(0); + let peer = make_node_addr(1); + let mut state = BloomState::new(node); + state.set_update_debounce_ms(500); + + state.mark_update_needed(peer); + + // Should send initially + assert!(state.should_send_update(&peer, 1000)); + + // Record send + state.record_update_sent(peer, 1000); + state.mark_update_needed(peer); + + // Should not send immediately (within debounce) + assert!(!state.should_send_update(&peer, 1200)); + + // Should send after debounce period + assert!(state.should_send_update(&peer, 1600)); +} + +#[test] +fn test_bloom_state_sequence() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + assert_eq!(state.sequence(), 0); + assert_eq!(state.next_sequence(), 1); + assert_eq!(state.next_sequence(), 2); + assert_eq!(state.sequence(), 2); +} + +#[test] +fn test_bloom_state_pending_updates() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_addr(1); + let peer2 = make_node_addr(2); + + assert!(!state.needs_update(&peer1)); + + state.mark_update_needed(peer1); + assert!(state.needs_update(&peer1)); + assert!(!state.needs_update(&peer2)); + + state.mark_all_updates_needed(vec![peer1, peer2]); + assert!(state.needs_update(&peer1)); + assert!(state.needs_update(&peer2)); + + state.clear_pending_updates(); + assert!(!state.needs_update(&peer1)); + assert!(!state.needs_update(&peer2)); +} + +#[test] +fn test_bloom_state_base_filter() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let leaf = make_node_addr(1); + state.add_leaf_dependent(leaf); + + let filter = state.base_filter(); + + assert!(filter.contains(&node)); + assert!(filter.contains(&leaf)); + assert!(!filter.contains(&make_node_addr(99))); +} + +#[test] +fn test_bloom_state_compute_outgoing_filter() { + let my_node = make_node_addr(0); + let mut state = BloomState::new(my_node); + + let leaf = make_node_addr(1); + state.add_leaf_dependent(leaf); + + let peer1 = make_node_addr(10); + let peer2 = make_node_addr(20); + + // Create peer filters + let mut filter1 = BloomFilter::new(); + filter1.insert(&make_node_addr(100)); + filter1.insert(&make_node_addr(101)); + + let mut filter2 = BloomFilter::new(); + filter2.insert(&make_node_addr(200)); + + let mut peer_filters = BTreeMap::new(); + peer_filters.insert(peer1, filter1); + peer_filters.insert(peer2, filter2); + + // Filter for peer1 should exclude peer1's contributions + let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); + assert!(outgoing1.contains(&my_node)); // self + assert!(outgoing1.contains(&leaf)); // leaf dependent + assert!(outgoing1.contains(&make_node_addr(200))); // from peer2 + // peer1's nodes may or may not be present (depends on split brain) + + // Filter for peer2 should exclude peer2's contributions + let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); + assert!(outgoing2.contains(&my_node)); + assert!(outgoing2.contains(&leaf)); + assert!(outgoing2.contains(&make_node_addr(100))); // from peer1 + assert!(outgoing2.contains(&make_node_addr(101))); // from peer1 +} + +#[test] +fn test_bloom_state_leaf_dependents_accessor() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let leaf1 = make_node_addr(1); + let leaf2 = make_node_addr(2); + + state.add_leaf_dependent(leaf1); + state.add_leaf_dependent(leaf2); + + let deps = state.leaf_dependents(); + assert!(deps.contains(&leaf1)); + assert!(deps.contains(&leaf2)); + assert!(!deps.contains(&make_node_addr(99))); + assert_eq!(deps.len(), 2); +} + +#[test] +fn test_bloom_state_record_sent_filter() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer = make_node_addr(1); + let mut filter = BloomFilter::new(); + filter.insert(&make_node_addr(42)); + + // Record a sent filter, then mark_changed_peers should detect no change + // when the outgoing filter matches what was recorded + state.record_sent_filter(peer, filter); + + // Compute what would be sent to peer (just our own node, no peer filters) + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer]; + state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); + + // Outgoing filter (just self) differs from recorded (self + node 42), + // so peer should be marked for update + assert!(state.needs_update(&peer)); +} + +#[test] +fn test_bloom_state_remove_peer_state() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer = make_node_addr(1); + + // Populate all three internal maps for this peer + state.mark_update_needed(peer); + state.record_update_sent(peer, 1000); + state.mark_update_needed(peer); // re-mark after send + let filter = BloomFilter::new(); + state.record_sent_filter(peer, filter); + + assert!(state.needs_update(&peer)); + + // Remove all peer state + state.remove_peer_state(&peer); + + // Pending updates cleared + assert!(!state.needs_update(&peer)); + + // Debounce state cleared — should be able to send immediately + state.mark_update_needed(peer); + assert!(state.should_send_update(&peer, 0)); + + // Sent filter cleared — mark_changed_peers should treat as "never sent" + state.clear_pending_updates(); + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer]; + state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); + assert!(state.needs_update(&peer)); // never sent → must send +} + +#[test] +fn test_bloom_state_mark_changed_peers_never_sent() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_addr(1); + let peer2 = make_node_addr(2); + + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer1, peer2]; + + // No filters ever sent — all peers should be marked + state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); + + assert!(state.needs_update(&peer1)); + assert!(state.needs_update(&peer2)); +} + +#[test] +fn test_bloom_state_mark_changed_peers_unchanged() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_addr(1); + let peer2 = make_node_addr(2); + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer1, peer2]; + + // Compute and record what would be sent to each peer + let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); + let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); + state.record_sent_filter(peer1, outgoing1); + state.record_sent_filter(peer2, outgoing2); + + // Nothing changed — no peers should be marked + state.mark_changed_peers(&make_node_addr(99), &peer_addrs, &peer_filters); + + assert!(!state.needs_update(&peer1)); + assert!(!state.needs_update(&peer2)); +} + +#[test] +fn test_bloom_state_mark_changed_peers_one_changed() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_addr(1); + let peer2 = make_node_addr(2); + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer1, peer2]; + + // Record current outgoing filters for both peers + let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); + let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); + state.record_sent_filter(peer1, outgoing1); + state.record_sent_filter(peer2, outgoing2); + + // Now peer1 sends us a filter with new entries + let mut inbound_from_peer1 = BloomFilter::new(); + inbound_from_peer1.insert(&make_node_addr(100)); + let mut updated_peer_filters = BTreeMap::new(); + updated_peer_filters.insert(peer1, inbound_from_peer1); + + // mark_changed_peers triggered by receiving from peer1 + state.mark_changed_peers(&peer1, &peer_addrs, &updated_peer_filters); + + // peer1 is excluded (it's the source), peer2's outgoing changed + // (now includes peer1's entries via split-horizon) + assert!(!state.needs_update(&peer1)); + assert!(state.needs_update(&peer2)); +} + +#[test] +fn test_bloom_state_mark_changed_peers_excludes_source() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_addr(1); + let peer_filters = BTreeMap::new(); + let peer_addrs = vec![peer1]; + + // peer1 is both the source and the only peer — should be skipped + state.mark_changed_peers(&peer1, &peer_addrs, &peer_filters); + + assert!(!state.needs_update(&peer1)); +} diff --git a/src/proto/bloom/tests/wire.rs b/src/proto/bloom/tests/wire.rs new file mode 100644 index 0000000..f6a5e58 --- /dev/null +++ b/src/proto/bloom/tests/wire.rs @@ -0,0 +1,99 @@ +//! Tests for the bloom wire codec (`FilterAnnounce`). + +use crate::proto::bloom::BloomFilter; +use crate::proto::bloom::FilterAnnounce; +use crate::protocol::LinkMessageType; +use crate::testutil::make_node_addr; + +#[test] +fn test_filter_announce_size_class() { + let filter = BloomFilter::new(); + let announce = FilterAnnounce::new(filter.clone(), 100); + + // v1 defaults + assert_eq!(announce.size_class, 1); + assert_eq!(announce.hash_count, 5); + assert!(announce.is_v1_compliant()); + assert!(announce.is_valid()); + assert_eq!(announce.filter_size_bytes(), 1024); +} + +#[test] +fn test_filter_announce_with_size_class() { + let filter = BloomFilter::with_params(2048 * 8, 7).unwrap(); + let announce = FilterAnnounce::with_size_class(filter, 100, 2); + + assert_eq!(announce.size_class, 2); + assert_eq!(announce.hash_count, 7); + assert!(!announce.is_v1_compliant()); + assert!(announce.is_valid()); + assert_eq!(announce.filter_size_bytes(), 2048); +} + +#[test] +fn test_filter_announce_encode_decode_roundtrip() { + let mut filter = BloomFilter::new(); + filter.insert(&make_node_addr(42)); + filter.insert(&make_node_addr(99)); + let announce = FilterAnnounce::new(filter, 500); + + let encoded = announce.encode().unwrap(); + // msg_type(1) + sequence(8) + hash_count(1) + size_class(1) + filter(1024) + assert_eq!(encoded.len(), 1035); + assert_eq!(encoded[0], LinkMessageType::FilterAnnounce.to_byte()); + + // Decode strips msg_type (as dispatcher does) + let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.sequence, 500); + assert_eq!(decoded.hash_count, 5); + assert_eq!(decoded.size_class, 1); + assert!(decoded.is_valid()); + assert!(decoded.is_v1_compliant()); + + // Filter contents preserved + assert!(decoded.filter.contains(&make_node_addr(42))); + assert!(decoded.filter.contains(&make_node_addr(99))); + assert!(!decoded.filter.contains(&make_node_addr(1))); +} + +#[test] +fn test_filter_announce_decode_rejects_bad_size_class() { + let filter = BloomFilter::new(); + let announce = FilterAnnounce::new(filter, 100); + let mut encoded = announce.encode().unwrap(); + + // Corrupt size_class byte (offset: 1 msg_type + 8 seq + 1 hash = 10) + encoded[10] = 5; // invalid size_class > MAX_SIZE_CLASS + + let result = FilterAnnounce::decode(&encoded[1..]); + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .to_string() + .contains("invalid size_class") + ); +} + +#[test] +fn test_filter_announce_decode_rejects_non_v1_size_class() { + // Build a size_class=0 payload manually (valid range but not v1) + let filter = BloomFilter::with_params(512 * 8, 5).unwrap(); + let announce = FilterAnnounce::with_size_class(filter, 100, 0); + let encoded = announce.encode().unwrap(); + + let result = FilterAnnounce::decode(&encoded[1..]); + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .to_string() + .contains("unsupported size_class") + ); +} + +#[test] +fn test_filter_announce_decode_rejects_truncated() { + let result = FilterAnnounce::decode(&[0u8; 5]); + assert!(result.is_err()); +} diff --git a/src/proto/bloom/wire.rs b/src/proto/bloom/wire.rs new file mode 100644 index 0000000..5946a4c --- /dev/null +++ b/src/proto/bloom/wire.rs @@ -0,0 +1,175 @@ +//! FilterAnnounce message: bloom filter reachability propagation. + +use super::BloomFilter; +use crate::protocol::LinkMessageType; +use crate::protocol::ProtocolError; + +/// Bloom filter announcement for reachability propagation. +/// +/// Sent to peers to advertise which destinations are reachable. +/// +/// ## Wire Format (v1) +/// +/// | Offset | Field | Size | Notes | +/// |--------|-------------|----------|----------------------------------| +/// | 0 | msg_type | 1 byte | 0x20 | +/// | 1 | sequence | 8 bytes | LE u64 | +/// | 9 | hash_count | 1 byte | Number of hash functions | +/// | 10 | size_class | 1 byte | Filter size: 512 << size_class | +/// | 11 | filter_bits | variable | 512 << size_class bytes | +#[derive(Clone, Debug)] +pub struct FilterAnnounce { + /// The bloom filter contents. + pub filter: BloomFilter, + /// Sequence number for freshness/dedup. + pub sequence: u64, + /// Number of hash functions used by the filter. + pub hash_count: u8, + /// Size class: filter size in bytes = 512 << size_class. + /// v1 protocol requires size_class=1 (1 KB filters). + pub size_class: u8, +} + +impl FilterAnnounce { + /// Create a new FilterAnnounce message with v1 defaults. + pub fn new(filter: BloomFilter, sequence: u64) -> Self { + Self { + hash_count: filter.hash_count(), + size_class: super::V1_SIZE_CLASS, + filter, + sequence, + } + } + + /// Create with explicit size_class (for testing or future protocol versions). + pub fn with_size_class(filter: BloomFilter, sequence: u64, size_class: u8) -> Self { + Self { + hash_count: filter.hash_count(), + size_class, + filter, + sequence, + } + } + + /// Get the expected filter size in bytes for this size_class. + pub fn filter_size_bytes(&self) -> usize { + 512 << self.size_class + } + + /// Validate the filter matches the declared size_class. + pub fn is_valid(&self) -> bool { + self.filter.num_bytes() == self.filter_size_bytes() + && self.filter.hash_count() == self.hash_count + } + + /// Check if this is a v1-compliant filter (size_class=1). + pub fn is_v1_compliant(&self) -> bool { + self.size_class == super::V1_SIZE_CLASS + } + + /// Minimum payload size after msg_type is stripped: + /// sequence(8) + hash_count(1) + size_class(1) = 10 + const MIN_PAYLOAD_SIZE: usize = 10; + + /// Maximum allowed size_class value. + const MAX_SIZE_CLASS: u8 = 3; + + /// Encode as link-layer plaintext (includes msg_type byte). + /// + /// ```text + /// [0x20][sequence:8 LE][hash_count:1][size_class:1][filter_bits:variable] + /// ``` + pub fn encode(&self) -> Result, ProtocolError> { + if !self.is_valid() { + return Err(ProtocolError::Malformed( + "filter size does not match size_class".into(), + )); + } + + let filter_bytes = self.filter.as_bytes(); + let size = 1 + Self::MIN_PAYLOAD_SIZE + filter_bytes.len(); + let mut buf = Vec::with_capacity(size); + + // msg_type + buf.push(LinkMessageType::FilterAnnounce.to_byte()); + // sequence (8 LE) + buf.extend_from_slice(&self.sequence.to_le_bytes()); + // hash_count + buf.push(self.hash_count); + // size_class + buf.push(self.size_class); + // filter_bits + buf.extend_from_slice(filter_bytes); + + Ok(buf) + } + + /// Decode from link-layer payload (after msg_type byte stripped by dispatcher). + /// + /// The payload starts with the sequence field. + pub fn decode(payload: &[u8]) -> Result { + if payload.len() < Self::MIN_PAYLOAD_SIZE { + return Err(ProtocolError::MessageTooShort { + expected: Self::MIN_PAYLOAD_SIZE, + got: payload.len(), + }); + } + + let mut pos = 0; + + // sequence (8 LE) + let sequence = u64::from_le_bytes( + payload[pos..pos + 8] + .try_into() + .map_err(|_| ProtocolError::Malformed("bad sequence".into()))?, + ); + pos += 8; + + // hash_count + let hash_count = payload[pos]; + pos += 1; + + // size_class + let size_class = payload[pos]; + pos += 1; + + // Validate size_class range + if size_class > Self::MAX_SIZE_CLASS { + return Err(ProtocolError::Malformed(format!( + "invalid size_class: {size_class} (max {})", + Self::MAX_SIZE_CLASS + ))); + } + + // v1 compliance check + if size_class != super::V1_SIZE_CLASS { + return Err(ProtocolError::Malformed(format!( + "unsupported size_class: {size_class} (v1 requires {})", + super::V1_SIZE_CLASS + ))); + } + + // Expected filter size from size_class + let expected_filter_bytes = 512usize << size_class; + let remaining = payload.len() - pos; + if remaining != expected_filter_bytes { + return Err(ProtocolError::MessageTooShort { + expected: Self::MIN_PAYLOAD_SIZE + expected_filter_bytes, + got: payload.len(), + }); + } + + // Construct BloomFilter from bytes + let filter = BloomFilter::from_slice(&payload[pos..], hash_count) + .map_err(|e| ProtocolError::Malformed(format!("invalid bloom filter: {e}")))?; + + let announce = Self { + filter, + sequence, + hash_count, + size_class, + }; + + Ok(announce) + } +} diff --git a/src/proto/mod.rs b/src/proto/mod.rs index 1afa3c1..44c404b 100644 --- a/src/proto/mod.rs +++ b/src/proto/mod.rs @@ -3,6 +3,7 @@ //! A module here has been migrated out of the async node shell; the async //! I/O adapters remain in `node::handlers`. +pub(crate) mod bloom; pub(crate) mod discovery; pub(crate) mod fmp; pub(crate) mod mmp; diff --git a/src/protocol/filter.rs b/src/protocol/filter.rs deleted file mode 100644 index 30d86d2..0000000 --- a/src/protocol/filter.rs +++ /dev/null @@ -1,340 +0,0 @@ -//! FilterAnnounce message: bloom filter reachability propagation. -//! -//! Supports both full sends and delta (XOR diff) updates with RLE compression. - -use super::error::ProtocolError; -use super::link::LinkMessageType; -use crate::bloom::BloomFilter; -use crate::bloom::codec::{CompressionStats, rle_decode, rle_encode}; - -/// Flag bit: this is a delta (XOR diff) update, not a full filter. -const FLAG_DELTA: u8 = 0x01; - -/// FilterAnnounce message for bloom filter reachability propagation. -/// -/// ## Wire Format -/// -/// ```text -/// [0x20][flags:1][sequence:8 LE][base_seq:8 LE][size_class:1][compressed_payload] -/// ``` -/// -/// - `flags` bit 0: is_delta (0 = full filter, 1 = XOR diff) -/// - `sequence`: current filter sequence number -/// - `base_seq`: for deltas, the sequence this diff is relative to (0 for full) -/// - `size_class`: filter size in bytes = 512 << size_class (0-6) -/// - `compressed_payload`: RLE-compressed u64 words -#[derive(Clone, Debug)] -pub struct FilterAnnounce { - /// The bloom filter contents (full filter or XOR diff). - pub filter: BloomFilter, - /// Sequence number for this filter update. - pub sequence: u64, - /// For deltas: the sequence number this diff is relative to. - /// For full sends: 0. - pub base_seq: u64, - /// Size class: filter size in bytes = 512 << size_class. - pub size_class: u8, - /// Whether this is a delta (XOR diff) update. - pub is_delta: bool, -} - -impl FilterAnnounce { - /// Minimum payload size after msg_type is stripped: - /// flags(1) + sequence(8) + base_seq(8) + size_class(1) = 18 - const MIN_PAYLOAD_SIZE: usize = 18; - - /// Create a full (non-delta) FilterAnnounce. - pub fn full(filter: BloomFilter, sequence: u64, size_class: u8) -> Self { - Self { - filter, - sequence, - base_seq: 0, - size_class, - is_delta: false, - } - } - - /// Create a delta (XOR diff) FilterAnnounce. - pub fn delta(diff: BloomFilter, sequence: u64, base_seq: u64, size_class: u8) -> Self { - Self { - filter: diff, - sequence, - base_seq, - size_class, - is_delta: true, - } - } - - /// Get the expected filter size in bytes for this size_class. - pub fn filter_size_bytes(&self) -> usize { - 512usize << self.size_class - } - - /// Validate the filter matches the declared size_class. - pub fn is_valid(&self) -> bool { - self.filter.num_bytes() == self.filter_size_bytes() - && (self.size_class as usize) < crate::bloom::SIZE_CLASS_BYTES.len() - } - - /// Encode as link-layer plaintext (includes msg_type byte). - /// - /// The filter words are RLE-compressed. - pub fn encode(&self) -> Result<(Vec, CompressionStats), ProtocolError> { - if !self.is_valid() { - return Err(ProtocolError::Malformed( - "filter size does not match size_class".into(), - )); - } - - let (compressed, stats) = rle_encode(self.filter.as_words()); - let size = 1 + Self::MIN_PAYLOAD_SIZE + compressed.len(); - let mut buf = Vec::with_capacity(size); - - // msg_type - buf.push(LinkMessageType::FilterAnnounce.to_byte()); - // flags - let flags = if self.is_delta { FLAG_DELTA } else { 0 }; - buf.push(flags); - // sequence (8 LE) - buf.extend_from_slice(&self.sequence.to_le_bytes()); - // base_seq (8 LE) - buf.extend_from_slice(&self.base_seq.to_le_bytes()); - // size_class - buf.push(self.size_class); - // compressed payload - buf.extend_from_slice(&compressed); - - Ok((buf, stats)) - } - - /// Decode from link-layer payload (after msg_type byte stripped by dispatcher). - pub fn decode(payload: &[u8]) -> Result { - if payload.len() < Self::MIN_PAYLOAD_SIZE { - return Err(ProtocolError::MessageTooShort { - expected: Self::MIN_PAYLOAD_SIZE, - got: payload.len(), - }); - } - - let mut pos = 0; - - // flags - let flags = payload[pos]; - let is_delta = flags & FLAG_DELTA != 0; - pos += 1; - - // sequence (8 LE) - let sequence = u64::from_le_bytes( - payload[pos..pos + 8] - .try_into() - .map_err(|_| ProtocolError::Malformed("bad sequence".into()))?, - ); - pos += 8; - - // base_seq (8 LE) - let base_seq = u64::from_le_bytes( - payload[pos..pos + 8] - .try_into() - .map_err(|_| ProtocolError::Malformed("bad base_seq".into()))?, - ); - pos += 8; - - // size_class - let size_class = payload[pos]; - pos += 1; - - if (size_class as usize) >= crate::bloom::SIZE_CLASS_BYTES.len() { - return Err(ProtocolError::Malformed(format!( - "invalid size_class: {size_class} (max {})", - crate::bloom::MAX_SIZE_CLASS - ))); - } - - // Decompress RLE payload - let expected_bytes = 512usize << size_class; - let expected_words = expected_bytes / 8; - let compressed_data = &payload[pos..]; - - let words = rle_decode(compressed_data, expected_words) - .map_err(|e| ProtocolError::Malformed(format!("RLE decode error: {e}")))?; - - // Convert words to bytes for BloomFilter construction - let mut bytes = Vec::with_capacity(expected_bytes); - for &word in &words { - bytes.extend_from_slice(&word.to_le_bytes()); - } - - let filter = BloomFilter::from_bytes(bytes, crate::bloom::DEFAULT_HASH_COUNT) - .map_err(|e| ProtocolError::Malformed(format!("invalid bloom filter: {e}")))?; - - Ok(Self { - filter, - sequence, - base_seq, - size_class, - is_delta, - }) - } -} - -/// FilterNack message: request full filter retransmission. -/// -/// Sent when a node receives an out-of-sequence delta update. -/// -/// ## Wire Format -/// -/// ```text -/// [0x21][expected_seq:8 LE] -/// ``` -#[derive(Clone, Debug)] -pub struct FilterNack { - /// The sequence number the receiver expected. - pub expected_seq: u64, -} - -impl FilterNack { - /// Encode as link-layer plaintext (includes msg_type byte). - pub fn encode(&self) -> Vec { - let mut buf = Vec::with_capacity(9); - buf.push(LinkMessageType::FilterNack.to_byte()); - buf.extend_from_slice(&self.expected_seq.to_le_bytes()); - buf - } - - /// Decode from link-layer payload (after msg_type byte stripped by dispatcher). - pub fn decode(payload: &[u8]) -> Result { - if payload.len() < 8 { - return Err(ProtocolError::MessageTooShort { - expected: 8, - got: payload.len(), - }); - } - let expected_seq = u64::from_le_bytes( - payload[..8] - .try_into() - .map_err(|_| ProtocolError::Malformed("bad expected_seq".into()))?, - ); - Ok(Self { expected_seq }) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::NodeAddr; - - fn make_node_addr(val: u8) -> NodeAddr { - let mut bytes = [0u8; 16]; - bytes[0] = val; - NodeAddr::from_bytes(bytes) - } - - #[test] - fn test_filter_announce_full_roundtrip() { - let mut filter = BloomFilter::new(); - filter.insert(&make_node_addr(42)); - filter.insert(&make_node_addr(99)); - - let announce = FilterAnnounce::full(filter, 500, 1); - assert!(announce.is_valid()); - assert!(!announce.is_delta); - - let (encoded, stats) = announce.encode().unwrap(); - assert!(stats.compressed_bytes > 0); - assert_eq!(encoded[0], LinkMessageType::FilterAnnounce.to_byte()); - - // Decode strips msg_type - let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.sequence, 500); - assert_eq!(decoded.base_seq, 0); - assert_eq!(decoded.size_class, 1); - assert!(!decoded.is_delta); - assert!(decoded.filter.contains(&make_node_addr(42))); - assert!(decoded.filter.contains(&make_node_addr(99))); - assert!(!decoded.filter.contains(&make_node_addr(1))); - } - - #[test] - fn test_filter_announce_delta_roundtrip() { - let mut old_filter = BloomFilter::new(); - old_filter.insert(&make_node_addr(1)); - - let mut new_filter = BloomFilter::new(); - new_filter.insert(&make_node_addr(1)); - new_filter.insert(&make_node_addr(2)); - - let diff = old_filter.xor_diff(&new_filter).unwrap(); - let announce = FilterAnnounce::delta(diff.clone(), 5, 4, 1); - assert!(announce.is_delta); - - let (encoded, _) = announce.encode().unwrap(); - let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap(); - - assert_eq!(decoded.sequence, 5); - assert_eq!(decoded.base_seq, 4); - assert!(decoded.is_delta); - assert_eq!(decoded.filter, diff); - } - - #[test] - fn test_filter_announce_empty_filter_compresses_well() { - let filter = BloomFilter::new(); // all zeros - let announce = FilterAnnounce::full(filter, 1, 1); - let (encoded, stats) = announce.encode().unwrap(); - - // 1KB of zeros should compress to ~10 bytes of RLE + 19 bytes header - assert!(encoded.len() < 50, "encoded size: {}", encoded.len()); - assert_eq!(stats.run_count, 1); - } - - #[test] - fn test_filter_announce_various_size_classes() { - for size_class in 0..=6u8 { - let num_bits = crate::bloom::size_class_to_bits(size_class); - let filter = BloomFilter::with_params(num_bits, 5).unwrap(); - let announce = FilterAnnounce::full(filter, 1, size_class); - assert!(announce.is_valid()); - - let (encoded, _) = announce.encode().unwrap(); - let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.size_class, size_class); - assert_eq!(decoded.filter.num_bits(), num_bits); - } - } - - #[test] - fn test_filter_announce_decode_rejects_bad_size_class() { - let filter = BloomFilter::new(); - let announce = FilterAnnounce::full(filter, 1, 1); - let (mut encoded, _) = announce.encode().unwrap(); - - // Corrupt size_class byte (offset: 1 msg_type + 1 flags + 8 seq + 8 base_seq = 18) - encoded[18] = 7; // invalid - - let result = FilterAnnounce::decode(&encoded[1..]); - assert!(result.is_err()); - } - - #[test] - fn test_filter_announce_decode_rejects_truncated() { - let result = FilterAnnounce::decode(&[0u8; 5]); - assert!(result.is_err()); - } - - #[test] - fn test_filter_nack_roundtrip() { - let nack = FilterNack { expected_seq: 42 }; - let encoded = nack.encode(); - assert_eq!(encoded.len(), 9); - assert_eq!(encoded[0], LinkMessageType::FilterNack.to_byte()); - - let decoded = FilterNack::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.expected_seq, 42); - } - - #[test] - fn test_filter_nack_decode_truncated() { - let result = FilterNack::decode(&[0u8; 3]); - assert!(result.is_err()); - } -} diff --git a/src/protocol/link.rs b/src/protocol/link.rs index 77105c1..834d726 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -34,8 +34,6 @@ pub enum LinkMessageType { // Bloom filter (0x20-0x2F) /// Bloom filter reachability update (full or delta). FilterAnnounce = 0x20, - /// Request full filter retransmission (NACK for out-of-sequence delta). - FilterNack = 0x21, // Discovery (0x30-0x3F) /// Request to discover a node's coordinates. @@ -60,7 +58,6 @@ impl LinkMessageType { 0x02 => Some(LinkMessageType::ReceiverReport), 0x10 => Some(LinkMessageType::TreeAnnounce), 0x20 => Some(LinkMessageType::FilterAnnounce), - 0x21 => Some(LinkMessageType::FilterNack), 0x30 => Some(LinkMessageType::LookupRequest), 0x31 => Some(LinkMessageType::LookupResponse), 0x50 => Some(LinkMessageType::Disconnect), @@ -83,7 +80,6 @@ impl fmt::Display for LinkMessageType { LinkMessageType::ReceiverReport => "ReceiverReport", LinkMessageType::TreeAnnounce => "TreeAnnounce", LinkMessageType::FilterAnnounce => "FilterAnnounce", - LinkMessageType::FilterNack => "FilterNack", LinkMessageType::LookupRequest => "LookupRequest", LinkMessageType::LookupResponse => "LookupResponse", LinkMessageType::Disconnect => "Disconnect", @@ -262,7 +258,6 @@ mod tests { let types = [ LinkMessageType::TreeAnnounce, LinkMessageType::FilterAnnounce, - LinkMessageType::FilterNack, LinkMessageType::LookupRequest, LinkMessageType::LookupResponse, LinkMessageType::SessionDatagram, diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 3b6538f..6637111 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -21,13 +21,11 @@ //! layer, encrypted end-to-end independently of per-hop link encryption. mod error; -mod filter; mod link; pub(crate) mod session; // Re-export all public types at protocol:: level pub use error::ProtocolError; -pub use filter::{FilterAnnounce, FilterNack}; pub use link::{ LinkMessageType, SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, SessionDatagramRef, };