From eb8479eb203ebe8b1eb891e927710b6376b32357 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 5 Apr 2026 07:51:18 +0000 Subject: [PATCH] Heterogeneous bloom filters: delta compression, variable sizing, adaptive heuristic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace fixed 1KB bloom filters with variable-size filters (512B-32KB) that adapt to each node's position in the spanning tree. Core changes: - Internal storage: Vec → Vec for word-level operations - Delta compression: XOR diff with word-level RLE, sequence-based NACK protocol for full retransmit recovery - Size conversion: fold (large→small) and duplicate (small→large) with auto-converting merge for mixed-size filter combination - Native-size storage: peer filters stored at advertised size for full-resolution routing queries, converted only for outgoing filter - Adaptive sizing: outgoing fill ratio drives step-up/step-down between size classes with hysteresis (20%/5% thresholds) - Filter size decoupled from FMP negotiation: announced dynamically in filter updates, bit 7 and TLV field 1 removed from handshake Wire format: FilterAnnounce gains flags byte (delta bit), base_seq field, RLE-compressed payload. New FilterNack message (0x21) for out-of-sequence delta recovery. --- src/bin/fipstop/ui/bloom.rs | 35 ++- src/bloom/codec.rs | 212 ++++++++++++++++++ src/bloom/filter.rs | 235 +++++++++++++++++--- src/bloom/mod.rs | 57 +++-- src/bloom/state.rs | 100 +++++++-- src/bloom/tests.rs | 383 ++++++++++++++++++++++++++++++++- src/node/bloom.rs | 233 ++++++++++++++++---- src/node/handlers/dispatch.rs | 4 + src/node/handlers/handshake.rs | 15 +- src/node/stats.rs | 20 +- src/peer/active.rs | 10 - src/peer/connection.rs | 19 +- src/protocol/filter.rs | 352 ++++++++++++++++++------------ src/protocol/link.rs | 7 +- src/protocol/mod.rs | 8 +- src/protocol/negotiation.rs | 164 +------------- 16 files changed, 1378 insertions(+), 476 deletions(-) create mode 100644 src/bloom/codec.rs diff --git a/src/bin/fipstop/ui/bloom.rs b/src/bin/fipstop/ui/bloom.rs index 532112a..71c657c 100644 --- a/src/bin/fipstop/ui/bloom.rs +++ b/src/bin/fipstop/ui/bloom.rs @@ -12,8 +12,8 @@ pub fn draw(frame: &mut Frame, app: &App, area: Rect) { let data = match app.data.get(&Tab::Bloom) { Some(d) => d, None => { - let msg = - Paragraph::new(" Waiting for data...").style(Style::default().fg(Color::DarkGray)); + let msg = Paragraph::new(" Waiting for data...") + .style(Style::default().fg(Color::DarkGray)); frame.render_widget(msg, area); return; } @@ -22,7 +22,7 @@ pub fn draw(frame: &mut Frame, app: &App, area: Rect) { let chunks = Layout::vertical([ Constraint::Length(7), // Bloom Filter State Constraint::Length(15), // Bloom Announce Stats - Constraint::Min(3), // Peer Filters + Constraint::Min(3), // Peer Filters ]) .split(area); @@ -64,28 +64,20 @@ fn draw_stats(frame: &mut Frame, data: &serde_json::Value, area: Rect) { helpers::section_header("Inbound"), helpers::kv_line("Received", &helpers::nested_u64(data, "stats", "received")), helpers::kv_line("Accepted", &helpers::nested_u64(data, "stats", "accepted")), - helpers::kv_line( - "Decode Error", - &helpers::nested_u64(data, "stats", "decode_error"), - ), + helpers::kv_line("Decode Error", &helpers::nested_u64(data, "stats", "decode_error")), helpers::kv_line("Invalid", &helpers::nested_u64(data, "stats", "invalid")), - helpers::kv_line("Non-V1", &helpers::nested_u64(data, "stats", "non_v1")), - helpers::kv_line( - "Unknown Peer", - &helpers::nested_u64(data, "stats", "unknown_peer"), - ), + helpers::kv_line("Unknown Peer", &helpers::nested_u64(data, "stats", "unknown_peer")), helpers::kv_line("Stale", &helpers::nested_u64(data, "stats", "stale")), Line::from(""), helpers::section_header("Outbound"), helpers::kv_line("Sent", &helpers::nested_u64(data, "stats", "sent")), - helpers::kv_line( - "Debounce Suppressed", - &helpers::nested_u64(data, "stats", "debounce_suppressed"), - ), - helpers::kv_line( - "Send Failed", - &helpers::nested_u64(data, "stats", "send_failed"), - ), + helpers::kv_line("Full Sends", &helpers::nested_u64(data, "stats", "full_sends")), + helpers::kv_line("Deltas Sent", &helpers::nested_u64(data, "stats", "deltas_sent")), + helpers::kv_line("NACKs Sent", &helpers::nested_u64(data, "stats", "nacks_sent")), + helpers::kv_line("NACKs Received", &helpers::nested_u64(data, "stats", "nacks_received")), + helpers::kv_line("Size Changes", &helpers::nested_u64(data, "stats", "size_changes")), + helpers::kv_line("Debounce Suppressed", &helpers::nested_u64(data, "stats", "debounce_suppressed")), + helpers::kv_line("Send Failed", &helpers::nested_u64(data, "stats", "send_failed")), ]; let max_lines = inner.height as usize; @@ -109,7 +101,8 @@ fn draw_peer_filters(frame: &mut Frame, data: &serde_json::Value, area: Rect) { frame.render_widget(block, area); if filters.is_empty() { - let msg = Paragraph::new(" No peers").style(Style::default().fg(Color::DarkGray)); + let msg = + Paragraph::new(" No peers").style(Style::default().fg(Color::DarkGray)); frame.render_widget(msg, inner); return; } diff --git a/src/bloom/codec.rs b/src/bloom/codec.rs new file mode 100644 index 0000000..e809f86 --- /dev/null +++ b/src/bloom/codec.rs @@ -0,0 +1,212 @@ +//! 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. + +/// 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 index e28b0a3..ec6c5b5 100644 --- a/src/bloom/filter.rs +++ b/src/bloom/filter.rs @@ -2,7 +2,10 @@ use std::fmt; -use super::{BloomError, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; +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. @@ -10,10 +13,13 @@ use crate::NodeAddr; /// 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 bytes). - bits: Vec, + /// 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. @@ -32,19 +38,22 @@ impl BloomFilter { 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_bytes = num_bits / 8; + let num_words = num_bits / 64; Ok(Self { - bits: vec![0u8; num_bytes], + words: vec![0u64; num_words], num_bits, hash_count, }) } - /// Create a Bloom filter from raw bytes. + /// 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); @@ -53,8 +62,18 @@ impl BloomFilter { 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 { - bits: bytes, + words, num_bits, hash_count, }) @@ -102,17 +121,19 @@ impl BloomFilter { /// 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 { - 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; + 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(()) } @@ -124,14 +145,154 @@ impl BloomFilter { 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.bits.fill(0); + self.words.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() + self.words.iter().map(|w| w.count_ones() as usize).sum() } /// Estimate the fill ratio (set bits / total bits). @@ -157,12 +318,26 @@ impl BloomFilter { /// Check if the filter is empty. pub fn is_empty(&self) -> bool { - self.bits.iter().all(|&b| b == 0) + self.words.iter().all(|&w| w == 0) } - /// Get the raw bytes. - pub fn as_bytes(&self) -> &[u8] { - &self.bits + /// 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. @@ -172,7 +347,7 @@ impl BloomFilter { /// Get the filter size in bytes. pub fn num_bytes(&self) -> usize { - self.bits.len() + self.words.len() * 8 } /// Get the number of hash functions. @@ -200,15 +375,15 @@ impl BloomFilter { } fn set_bit(&mut self, index: usize) { - let byte_index = index / 8; - let bit_offset = index % 8; - self.bits[byte_index] |= 1 << bit_offset; + 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 byte_index = index / 8; - let bit_offset = index % 8; - (self.bits[byte_index] >> bit_offset) & 1 == 1 + let word_index = index / 64; + let bit_offset = index % 64; + (self.words[word_index] >> bit_offset) & 1 == 1 } } @@ -222,7 +397,7 @@ 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 + && self.words == other.words } } diff --git a/src/bloom/mod.rs b/src/bloom/mod.rs index 663bd99..8b456ba 100644 --- a/src/bloom/mod.rs +++ b/src/bloom/mod.rs @@ -1,19 +1,20 @@ //! Bloom Filter Implementation //! -//! 1KB Bloom filters for reachability in FIPS routing. Each node -//! maintains filters that summarize which destinations are reachable +//! 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. //! -//! ## v1 Parameters +//! 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. //! -//! - 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) +//! ## Parameters //! -//! These parameters are right-sized for typical network occupancy of -//! ~250-800 entries per node. +//! - Hash functions: k=5 (network-wide constant) +//! - Default size: 1 KB (size_class 1) +pub mod codec; mod filter; mod state; @@ -23,9 +24,6 @@ pub use filter::BloomFilter; pub use state::BloomState; /// 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). @@ -33,15 +31,34 @@ 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%. +/// 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; 4] = [512, 1024, 2048, 4096]; +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)] @@ -52,8 +69,20 @@ pub enum BloomError { #[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)] diff --git a/src/bloom/state.rs b/src/bloom/state.rs index 9c39577..e30636c 100644 --- a/src/bloom/state.rs +++ b/src/bloom/state.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; -use super::BloomFilter; +use super::{size_class_to_bits, BloomFilter, MAX_SIZE_CLASS, MIN_SIZE_CLASS, V1_SIZE_CLASS}; use crate::NodeAddr; /// State for managing Bloom filter announcements. @@ -16,6 +16,8 @@ pub struct BloomState { leaf_dependents: HashSet, /// 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). @@ -24,8 +26,14 @@ pub struct BloomState { pending_updates: HashSet, /// Current sequence number for outgoing filters. sequence: u64, - /// Last outgoing filter sent to each peer (for change detection). + /// 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, } impl BloomState { @@ -35,11 +43,15 @@ impl BloomState { own_node_addr, leaf_dependents: HashSet::new(), is_leaf_only: false, + size_class: V1_SIZE_CLASS, update_debounce_ms: 500, last_update_sent: HashMap::new(), pending_updates: HashSet::new(), sequence: 0, last_sent_filters: HashMap::new(), + last_sent_seq: HashMap::new(), + step_up_threshold: 0.20, + step_down_threshold: 0.05, } } @@ -60,6 +72,19 @@ 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 @@ -139,14 +164,43 @@ impl BloomState { self.pending_updates.clear(); } - /// Record the outgoing filter that was sent to a peer. + /// Record the outgoing filter and sequence 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). + 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); } @@ -179,32 +233,24 @@ 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, ) -> BloomFilter { - let mut filter = BloomFilter::new(); + let mut filter = self.base_filter(); - // 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 + // Merge filters from other peers (auto-converting sizes) 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); } } @@ -212,9 +258,27 @@ 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 mut filter = BloomFilter::new(); + 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"); filter.insert(&self.own_node_addr); for dep in &self.leaf_dependents { filter.insert(dep); diff --git a/src/bloom/tests.rs b/src/bloom/tests.rs index 1f8071b..1cba0af 100644 --- a/src/bloom/tests.rs +++ b/src/bloom/tests.rs @@ -107,12 +107,39 @@ fn test_bloom_filter_clear() { } #[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(); +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 result = filter1.merge(&filter2); - assert!(matches!(result, Err(BloomError::InvalidSize { .. }))); + 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] @@ -142,6 +169,22 @@ fn test_bloom_filter_invalid_params() { 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] @@ -207,7 +250,7 @@ fn test_bloom_filter_from_slice() { original.insert(&make_node_addr(42)); let bytes = original.as_bytes(); - let restored = BloomFilter::from_slice(bytes, original.hash_count()).unwrap(); + let restored = BloomFilter::from_slice(&bytes, original.hash_count()).unwrap(); assert_eq!(original, restored); } @@ -228,6 +271,229 @@ fn test_bloom_filter_insert_bytes_contains_bytes() { 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 @@ -261,6 +527,60 @@ fn test_bloom_filter_debug_format() { 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 mut 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] @@ -572,6 +892,57 @@ fn test_bloom_state_mark_changed_peers_excludes_source() { 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] diff --git a/src/node/bloom.rs b/src/node/bloom.rs index 83e8b77..97d0d54 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -1,15 +1,16 @@ //! Bloom filter announce send/receive logic. //! //! Handles building, sending, and receiving FilterAnnounce messages, -//! including debounced propagation to peers. +//! including delta compression with NACK-based recovery and debounced +//! propagation to peers. -use crate::NodeAddr; use crate::bloom::BloomFilter; -use crate::protocol::FilterAnnounce; +use crate::protocol::{FilterAnnounce, FilterNack}; +use crate::NodeAddr; use super::{Node, NodeError}; use std::collections::HashMap; -use tracing::debug; +use tracing::{debug, trace, warn}; impl Node { /// Collect inbound filters from full tree peers for outgoing filter computation. @@ -33,16 +34,29 @@ impl Node { /// Build a FilterAnnounce for a specific peer. /// - /// 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). + /// Returns a delta (XOR diff) if we have a previous filter for this peer + /// at the same size class. Otherwise returns a full send. 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(); - FilterAnnounce::new(filter, 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) } /// Send a FilterAnnounce to a specific peer, respecting debounce. @@ -67,11 +81,26 @@ impl Node { // Build and encode let announce = self.build_filter_announce(peer_addr); - 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), - })?; + 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 { + node_addr: *peer_addr, + reason: format!("FilterAnnounce encode failed: {}", e), + })?; // Send if let Err(e) = self.send_encrypted_link_message(peer_addr, &encoded).await { @@ -80,19 +109,26 @@ impl Node { } self.stats_mut().bloom.sent += 1; + if is_delta { + self.stats_mut().bloom.deltas_sent += 1; + } else { + self.stats_mut().bloom.full_sends += 1; + } // Record send and store the filter for change detection debug!( peer = %self.peer_display_name(peer_addr), seq = announce.sequence, + delta = is_delta, + compressed = stats.compressed_bytes, + runs = stats.run_count, est_entries = format_args!("{:.0}", sent_filter.estimated_count()), - 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); - self.bloom_state.record_sent_filter(*peer_addr, sent_filter); + self.bloom_state + .record_sent_filter(*peer_addr, sent_filter); if let Some(peer) = self.peers.get_mut(peer_addr) { peer.clear_filter_update_needed(); } @@ -134,10 +170,8 @@ impl Node { /// Handle an inbound FilterAnnounce from an authenticated peer. /// - /// 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 + /// Supports both full sends and delta (XOR diff) updates. + /// On out-of-sequence delta, sends a NACK to request full retransmission. pub(super) async fn handle_filter_announce(&mut self, from: &NodeAddr, payload: &[u8]) { self.stats_mut().bloom.received += 1; @@ -156,26 +190,22 @@ impl Node { debug!(from = %self.peer_display_name(from), "FilterAnnounce filter/size_class mismatch"); return; } - if !announce.is_v1_compliant() { - self.stats_mut().bloom.non_v1 += 1; - debug!(from = %self.peer_display_name(from), size_class = announce.size_class, "Non-v1 FilterAnnounce rejected"); - return; - } // Check peer exists - let current_seq = match self.peers.get(from) { - Some(peer) => peer.filter_sequence(), + let peer = match self.peers.get(from) { + Some(p) => p, None => { self.stats_mut().bloom.unknown_peer += 1; 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.stats_mut().bloom.stale += 1; - debug!( + trace!( from = %self.peer_display_name(from), received_seq = announce.sequence, current_seq = current_seq, @@ -184,6 +214,70 @@ 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.stats_mut().bloom.nacks_sent += 1; + 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.stats_mut().bloom.nacks_sent += 1; + 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.stats_mut().bloom.nacks_sent += 1; + return; + } + } + } else { + // Full send: use directly + announce.filter.clone() + }; + self.stats_mut().bloom.accepted += 1; let now_ms = std::time::SystemTime::now() @@ -194,31 +288,94 @@ impl Node { debug!( from = %self.peer_display_name(from), seq = announce.sequence, - est_entries = format_args!("{:.0}", announce.filter.estimated_count()), - set_bits = announce.filter.count_ones(), - fill = format_args!("{:.1}%", announce.filter.fill_ratio() * 100.0), - tree_peer = self.is_tree_peer(from), + delta = announce.is_delta, + est_entries = format_args!("{:.0}", resolved_filter.estimated_count()), + fill = format_args!("{:.1}%", resolved_filter.fill_ratio() * 100.0), "Received FilterAnnounce" ); - // Store on peer + // Store resolved filter on peer if let Some(peer) = self.peers.get_mut(from) { - peer.update_filter(announce.filter, announce.sequence, now_ms); + peer.update_filter(resolved_filter, announce.sequence, now_ms); } - // 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). + // Check which peers' outgoing filters actually changed 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.stats_mut().bloom.nacks_received += 1; + // 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::protocol::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.stats_mut().bloom.size_changes += 1; + 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). /// - /// Sends any pending debounced filter announces. + /// Evaluates adaptive sizing, then sends any pending 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 48ac30b..dd2a62d 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -42,6 +42,10 @@ 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/handlers/handshake.rs b/src/node/handlers/handshake.rs index ada84b4..143dfe2 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1159,8 +1159,6 @@ impl Node { let remote_epoch = connection.remote_epoch(); let peer_profile = connection.peer_profile() .unwrap_or(crate::protocol::NodeProfile::Full); - let agreed_bloom_size_class = connection.agreed_bloom_size_class() - .unwrap_or(crate::bloom::V1_SIZE_CLASS); let peer_node_addr = *verified_identity.node_addr(); let is_outbound = connection.is_outbound(); @@ -1205,7 +1203,6 @@ impl Node { remote_epoch, self.node_profile, peer_profile, - agreed_bloom_size_class, ); new_peer.set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms); @@ -1303,7 +1300,6 @@ impl Node { remote_epoch, self.node_profile, peer_profile, - agreed_bloom_size_class, ); new_peer.set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms); if let Some(ts) = old_announce_ts { @@ -1338,30 +1334,25 @@ impl Node { /// Process an FMP negotiation payload received from a peer. /// -/// Decodes the payload, validates profile pairing, agrees on bloom -/// filter size, and stores the results on the PeerConnection. +/// Decodes the payload, validates profile pairing, and stores the +/// results on the PeerConnection. fn process_fmp_negotiation( our_profile: crate::protocol::NodeProfile, conn: &mut PeerConnection, neg_bytes: &[u8], ) -> Result<(), crate::protocol::ProtocolError> { - let our_payload = NegotiationPayload::fmp(1, 1, our_profile); let their_payload = NegotiationPayload::decode(neg_bytes)?; // Validate profile pairing (at least one Full) let their_profile = their_payload.node_profile()?; NegotiationPayload::validate_profiles(our_profile, their_profile)?; - // Agree on bloom filter size - let agreed_bloom = our_payload.agree_bloom_size(&their_payload)?; - - conn.set_negotiation_results(their_profile, agreed_bloom); + conn.set_negotiation_results(their_profile); debug!( link_id = %conn.link_id(), our_profile = ?our_profile, peer_profile = ?their_profile, - agreed_bloom_size_class = agreed_bloom, "FMP negotiation complete" ); diff --git a/src/node/stats.rs b/src/node/stats.rs index d4e2008..1ffd3c2 100644 --- a/src/node/stats.rs +++ b/src/node/stats.rs @@ -208,7 +208,6 @@ pub struct BloomStats { pub received: u64, pub decode_error: u64, pub invalid: u64, - pub non_v1: u64, pub unknown_peer: u64, pub stale: u64, pub accepted: u64, @@ -216,6 +215,13 @@ pub struct BloomStats { pub sent: u64, pub debounce_suppressed: u64, pub send_failed: u64, + // Delta compression + pub deltas_sent: u64, + pub full_sends: u64, + pub nacks_sent: u64, + pub nacks_received: u64, + // Adaptive sizing + pub size_changes: u64, } impl BloomStats { @@ -224,13 +230,17 @@ impl BloomStats { received: self.received, decode_error: self.decode_error, invalid: self.invalid, - non_v1: self.non_v1, unknown_peer: self.unknown_peer, stale: self.stale, accepted: self.accepted, sent: self.sent, debounce_suppressed: self.debounce_suppressed, send_failed: self.send_failed, + deltas_sent: self.deltas_sent, + full_sends: self.full_sends, + nacks_sent: self.nacks_sent, + nacks_received: self.nacks_received, + size_changes: self.size_changes, } } } @@ -394,13 +404,17 @@ pub struct BloomStatsSnapshot { pub received: u64, pub decode_error: u64, pub invalid: u64, - pub non_v1: u64, pub unknown_peer: u64, pub stale: u64, pub accepted: u64, pub sent: u64, pub debounce_suppressed: u64, pub send_failed: u64, + pub deltas_sent: u64, + pub full_sends: u64, + pub nacks_sent: u64, + pub nacks_received: u64, + pub size_changes: u64, } #[derive(Clone, Debug, Default, Serialize)] diff --git a/src/peer/active.rs b/src/peer/active.rs index 81c4b75..c39a434 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -135,8 +135,6 @@ pub struct ActivePeer { // === Negotiated Profile === /// Peer's node profile (Full, NonRouting, Leaf). peer_profile: NodeProfile, - /// Agreed bloom filter size class for this link. - agreed_bloom_size_class: u8, /// Whether to send sender reports to this peer (our provides_sr AND peer wants_sr). send_sr: bool, /// Whether to send receiver reports to this peer (our provides_rr AND peer wants_rr). @@ -228,7 +226,6 @@ impl ActivePeer { last_seen: authenticated_at, remote_epoch: None, peer_profile: NodeProfile::Full, - agreed_bloom_size_class: crate::bloom::V1_SIZE_CLASS, send_sr: true, send_rr: true, mmp: None, @@ -290,7 +287,6 @@ impl ActivePeer { remote_epoch: Option<[u8; 8]>, our_profile: NodeProfile, peer_profile: NodeProfile, - agreed_bloom_size_class: u8, ) -> Self { // Compute MMP report gating: A sends to B iff A.provides AND B.wants let our_neg = NegotiationPayload::fmp(0, 0, our_profile); @@ -323,7 +319,6 @@ impl ActivePeer { last_seen: authenticated_at, remote_epoch, peer_profile, - agreed_bloom_size_class, send_sr, send_rr, mmp: Some(MmpPeerState::new(mmp_config, is_initiator)), @@ -543,11 +538,6 @@ impl ActivePeer { self.peer_profile } - /// Get agreed bloom filter size class for this link. - pub fn agreed_bloom_size_class(&self) -> u8 { - self.agreed_bloom_size_class - } - /// Whether to send sender reports to this peer. pub fn send_sr(&self) -> bool { self.send_sr diff --git a/src/peer/connection.rs b/src/peer/connection.rs index 1961066..b1410f0 100644 --- a/src/peer/connection.rs +++ b/src/peer/connection.rs @@ -125,9 +125,6 @@ pub struct PeerConnection { // === Negotiation Results === /// Peer's node profile (learned from negotiation payload). peer_profile: Option, - /// Agreed bloom filter size class. - agreed_bloom_size_class: Option, - // === Handshake Resend === /// Wire-format msg1 bytes for resend (initiator only). handshake_msg1: Option>, @@ -169,7 +166,7 @@ impl PeerConnection { source_addr: None, remote_epoch: None, peer_profile: None, - agreed_bloom_size_class: None, + handshake_msg1: None, handshake_msg2: None, resend_count: 0, @@ -200,7 +197,7 @@ impl PeerConnection { source_addr: None, remote_epoch: None, peer_profile: None, - agreed_bloom_size_class: None, + handshake_msg1: None, handshake_msg2: None, resend_count: 0, @@ -230,7 +227,7 @@ impl PeerConnection { source_addr: None, remote_epoch: None, peer_profile: None, - agreed_bloom_size_class: None, + handshake_msg1: None, handshake_msg2: None, resend_count: 0, @@ -264,7 +261,7 @@ impl PeerConnection { source_addr: Some(source_addr), remote_epoch: None, peer_profile: None, - agreed_bloom_size_class: None, + handshake_msg1: None, handshake_msg2: None, resend_count: 0, @@ -405,15 +402,9 @@ impl PeerConnection { self.peer_profile } - /// Get agreed bloom filter size class. - pub fn agreed_bloom_size_class(&self) -> Option { - self.agreed_bloom_size_class - } - /// Store negotiation results from peer's payload. - pub fn set_negotiation_results(&mut self, peer_profile: NodeProfile, bloom_size_class: u8) { + pub fn set_negotiation_results(&mut self, peer_profile: NodeProfile) { self.peer_profile = Some(peer_profile); - self.agreed_bloom_size_class = Some(bloom_size_class); } // === Handshake Resend === diff --git a/src/protocol/filter.rs b/src/protocol/filter.rs index 601b9a9..e4a25f1 100644 --- a/src/protocol/filter.rs +++ b/src/protocol/filter.rs @@ -1,112 +1,118 @@ //! 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::codec::{rle_decode, rle_encode, CompressionStats}; use crate::bloom::BloomFilter; -/// Bloom filter announcement for reachability propagation. +/// 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. /// -/// Sent to peers to advertise which destinations are reachable. +/// ## Wire Format /// -/// ## Wire Format (v1) +/// ```text +/// [0x20][flags:1][sequence:8 LE][base_seq:8 LE][size_class:1][compressed_payload] +/// ``` /// -/// | 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 | +/// - `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. + /// The bloom filter contents (full filter or XOR diff). pub filter: BloomFilter, - /// Sequence number for freshness/dedup. + /// Sequence number for this filter update. pub sequence: u64, - /// Number of hash functions used by the filter. - pub hash_count: u8, + /// 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. - /// v1 protocol requires size_class=1 (1 KB filters). pub size_class: u8, + /// Whether this is a delta (XOR diff) update. + pub is_delta: bool, } impl FilterAnnounce { - /// Create a new FilterAnnounce message with v1 defaults. - pub fn new(filter: BloomFilter, sequence: u64) -> Self { + /// 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 { - hash_count: filter.hash_count(), - size_class: crate::bloom::V1_SIZE_CLASS, filter, sequence, + base_seq: 0, + size_class, + is_delta: false, } } - /// Create with explicit size_class (for testing or future protocol versions). - pub fn with_size_class(filter: BloomFilter, sequence: u64, size_class: u8) -> Self { + /// Create a delta (XOR diff) FilterAnnounce. + pub fn delta( + diff: BloomFilter, + sequence: u64, + base_seq: u64, + size_class: u8, + ) -> Self { Self { - hash_count: filter.hash_count(), - size_class, - filter, + 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 { - 512 << self.size_class + 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.filter.hash_count() == self.hash_count + && (self.size_class as usize) < crate::bloom::SIZE_CLASS_BYTES.len() } - /// Check if this is a v1-compliant filter (size_class=1). - pub fn is_v1_compliant(&self) -> bool { - self.size_class == crate::bloom::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> { + /// 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 filter_bytes = self.filter.as_bytes(); - let size = 1 + Self::MIN_PAYLOAD_SIZE + filter_bytes.len(); + 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()); - // hash_count - buf.push(self.hash_count); + // base_seq (8 LE) + buf.extend_from_slice(&self.base_seq.to_le_bytes()); // size_class buf.push(self.size_class); - // filter_bits - buf.extend_from_slice(filter_bytes); + // compressed payload + buf.extend_from_slice(&compressed); - Ok(buf) + Ok((buf, stats)) } /// 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 { @@ -117,6 +123,11 @@ impl FilterAnnounce { 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] @@ -125,52 +136,93 @@ impl FilterAnnounce { ); pos += 8; - // hash_count - let hash_count = payload[pos]; - pos += 1; + // 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; - // Validate size_class range - if size_class > Self::MAX_SIZE_CLASS { + if (size_class as usize) >= crate::bloom::SIZE_CLASS_BYTES.len() { return Err(ProtocolError::Malformed(format!( "invalid size_class: {size_class} (max {})", - Self::MAX_SIZE_CLASS + crate::bloom::MAX_SIZE_CLASS ))); } - // v1 compliance check - if size_class != crate::bloom::V1_SIZE_CLASS { - return Err(ProtocolError::Malformed(format!( - "unsupported size_class: {size_class} (v1 requires {})", - crate::bloom::V1_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()); } - // Expected filter size from size_class - let expected_filter_bytes = 512usize << size_class; - let remaining = payload.len() - pos; - if remaining != expected_filter_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: Self::MIN_PAYLOAD_SIZE + expected_filter_bytes, + expected: 8, got: payload.len(), }); } - - // Construct BloomFilter from bytes - let filter = crate::bloom::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) + let expected_seq = u64::from_le_bytes( + payload[..8] + .try_into() + .map_err(|_| ProtocolError::Malformed("bad expected_seq".into()))?, + ); + Ok(Self { expected_seq }) } } @@ -186,90 +238,89 @@ mod tests { } #[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() { + 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::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); + 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 (as dispatcher does) + // Decode strips msg_type let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap(); assert_eq!(decoded.sequence, 500); - assert_eq!(decoded.hash_count, 5); + assert_eq!(decoded.base_seq, 0); assert_eq!(decoded.size_class, 1); - assert!(decoded.is_valid()); - assert!(decoded.is_v1_compliant()); - - // Filter contents preserved + 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_decode_rejects_bad_size_class() { - let filter = BloomFilter::new(); - let announce = FilterAnnounce::new(filter, 100); - let mut encoded = announce.encode().unwrap(); + fn test_filter_announce_delta_roundtrip() { + let mut old_filter = BloomFilter::new(); + old_filter.insert(&make_node_addr(1)); - // Corrupt size_class byte (offset: 1 msg_type + 8 seq + 1 hash = 10) - encoded[10] = 5; // invalid size_class > MAX_SIZE_CLASS + let mut new_filter = BloomFilter::new(); + new_filter.insert(&make_node_addr(1)); + new_filter.insert(&make_node_addr(2)); - let result = FilterAnnounce::decode(&encoded[1..]); - assert!(result.is_err()); - assert!( - result - .unwrap_err() - .to_string() - .contains("invalid size_class") - ); + 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_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(); + 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()); - assert!( - result - .unwrap_err() - .to_string() - .contains("unsupported size_class") - ); } #[test] @@ -277,4 +328,21 @@ mod tests { 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 c2d4c7d..f1c7bd2 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -90,8 +90,10 @@ pub enum LinkMessageType { TreeAnnounce = 0x10, // Bloom filter (0x20-0x2F) - /// Bloom filter reachability update. + /// 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. @@ -116,6 +118,7 @@ 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), @@ -138,6 +141,7 @@ 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", @@ -433,6 +437,7 @@ 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 322d7ea..d0c0c24 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -35,12 +35,12 @@ pub use link::{ SESSION_DATAGRAM_HEADER_SIZE, }; pub use tree::TreeAnnounce; -pub use filter::FilterAnnounce; +pub use filter::{FilterAnnounce, FilterNack}; pub use discovery::{LookupRequest, LookupResponse}; pub use negotiation::{ - BloomSizeRange, NegotiationPayload, NodeProfile, TlvEntry, NEGOTIATION_HEADER_SIZE, - FMP_FEAT_BLOOM_SIZE_NEG, FMP_FEAT_PROFILE_MASK, FMP_FEAT_PROVIDES_RR, FMP_FEAT_PROVIDES_SR, - FMP_FEAT_WANTS_RR, FMP_FEAT_WANTS_SR, TLV_BLOOM_SIZE, + NegotiationPayload, NodeProfile, TlvEntry, NEGOTIATION_HEADER_SIZE, + FMP_FEAT_PROFILE_MASK, FMP_FEAT_PROVIDES_RR, FMP_FEAT_PROVIDES_SR, + FMP_FEAT_WANTS_RR, FMP_FEAT_WANTS_SR, }; pub use session::{ CoordsRequired, FspFlags, FspInnerFlags, MtuExceeded, PathBroken, PathMtuNotification, diff --git a/src/protocol/negotiation.rs b/src/protocol/negotiation.rs index 39ce941..440d211 100644 --- a/src/protocol/negotiation.rs +++ b/src/protocol/negotiation.rs @@ -39,14 +39,6 @@ pub const FMP_FEAT_WANTS_SR: u64 = 1 << 5; /// Bit 6: Want MMP receiver reports from peer. pub const FMP_FEAT_WANTS_RR: u64 = 1 << 6; -/// Bit 7: Bloom filter size is negotiable (check TLV). -pub const FMP_FEAT_BLOOM_SIZE_NEG: u64 = 1 << 7; - -// --- TLV field numbers --- - -/// TLV field for bloom filter size classes: `[min_class:1][max_class:1]`. -pub const TLV_BLOOM_SIZE: u16 = 1; - // --- Node profile enum --- /// Node profile advertised during FMP negotiation. @@ -81,15 +73,6 @@ impl TryFrom for NodeProfile { } } -/// Bloom filter size class range from TLV negotiation. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct BloomSizeRange { - /// Minimum supported size class (512 << min_class bytes). - pub min_class: u8, - /// Maximum supported size class (512 << max_class bytes). - pub max_class: u8, -} - /// A TLV entry in the negotiation payload. #[derive(Debug, Clone, PartialEq, Eq)] pub struct TlvEntry { @@ -236,9 +219,7 @@ impl NegotiationPayload { /// Build an FMP negotiation payload for the given node profile. /// - /// Sets the profile bits, MMP wants/provides defaults for the profile, - /// bloom size negotiable bit, and bloom size TLV with the current - /// default size class (min=max=V1_SIZE_CLASS). + /// Sets the profile bits and MMP wants/provides defaults for the profile. pub fn fmp(version_min: u8, version_max: u8, profile: NodeProfile) -> Self { let (provides_sr, provides_rr, wants_sr, wants_rr) = match profile { NodeProfile::Full => (true, true, true, true), @@ -259,12 +240,8 @@ impl NegotiationPayload { if wants_rr { features |= FMP_FEAT_WANTS_RR; } - features |= FMP_FEAT_BLOOM_SIZE_NEG; - - let bloom_size_class = crate::bloom::V1_SIZE_CLASS; Self::new(version_min, version_max, features) - .with_tlv(TLV_BLOOM_SIZE, vec![bloom_size_class, bloom_size_class]) } /// Extract the node profile from the FMP feature bitfield. @@ -293,39 +270,6 @@ impl NegotiationPayload { self.features & FMP_FEAT_WANTS_RR != 0 } - /// Whether bloom filter size is negotiable. - pub fn bloom_size_negotiable(&self) -> bool { - self.features & FMP_FEAT_BLOOM_SIZE_NEG != 0 - } - - /// Extract bloom size range from TLV, if present. - pub fn bloom_size_range(&self) -> Result, ProtocolError> { - for entry in &self.tlv_entries { - if entry.field_num == TLV_BLOOM_SIZE { - if entry.value.len() != 2 { - return Err(ProtocolError::Malformed(format!( - "bloom size TLV: expected 2 bytes, got {}", - entry.value.len() - ))); - } - let min_class = entry.value[0]; - let max_class = entry.value[1]; - if min_class > max_class { - return Err(ProtocolError::Malformed(format!( - "bloom size: min_class ({min_class}) > max_class ({max_class})" - ))); - } - if max_class as usize >= crate::bloom::SIZE_CLASS_BYTES.len() { - return Err(ProtocolError::Malformed(format!( - "bloom size: max_class ({max_class}) exceeds known size classes" - ))); - } - return Ok(Some(BloomSizeRange { min_class, max_class })); - } - } - Ok(None) - } - /// Validate that two profiles form a valid link pairing. /// /// At least one side must be `Full` or the link is rejected. @@ -342,35 +286,6 @@ impl NegotiationPayload { Ok(()) } - /// Agree on a bloom filter size class with a peer. - /// - /// Returns `min(our_max, their_max)`, rejecting if below either - /// side's minimum. Both sides must have the bloom size TLV and the - /// negotiable bit set. - pub fn agree_bloom_size(&self, other: &Self) -> Result { - if !self.bloom_size_negotiable() || !other.bloom_size_negotiable() { - return Err(ProtocolError::Malformed( - "bloom size negotiation: both sides must set negotiable bit".to_string(), - )); - } - - let ours = self.bloom_size_range()?.ok_or_else(|| { - ProtocolError::Malformed("bloom size negotiation: missing TLV (ours)".to_string()) - })?; - - let theirs = other.bloom_size_range()?.ok_or_else(|| { - ProtocolError::Malformed("bloom size negotiation: missing TLV (theirs)".to_string()) - })?; - - let agreed = ours.max_class.min(theirs.max_class); - if agreed < ours.min_class || agreed < theirs.min_class { - return Err(ProtocolError::Malformed(format!( - "bloom size mismatch: ours [{},{}] theirs [{},{}]", - ours.min_class, ours.max_class, theirs.min_class, theirs.max_class - ))); - } - Ok(agreed) - } } #[cfg(test)] @@ -515,11 +430,6 @@ mod tests { assert!(p.provides_rr()); assert!(p.wants_sr()); assert!(p.wants_rr()); - assert!(p.bloom_size_negotiable()); - - let range = p.bloom_size_range().unwrap().unwrap(); - assert_eq!(range.min_class, crate::bloom::V1_SIZE_CLASS); - assert_eq!(range.max_class, crate::bloom::V1_SIZE_CLASS); } #[test] @@ -610,76 +520,4 @@ mod tests { ).is_err()); } - // --- Bloom size agreement tests --- - - #[test] - fn test_bloom_size_agreement_identical() { - let a = NegotiationPayload::fmp(1, 1, NodeProfile::Full); - let b = NegotiationPayload::fmp(1, 1, NodeProfile::Full); - assert_eq!(a.agree_bloom_size(&b).unwrap(), crate::bloom::V1_SIZE_CLASS); - } - - #[test] - fn test_bloom_size_agreement_different_ranges() { - // a supports [0,2], b supports [1,3] - let a = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![0, 2]); - let b = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![1, 3]); - // agreed = min(2,3) = 2, 2 >= 0 and 2 >= 1 → ok - assert_eq!(a.agree_bloom_size(&b).unwrap(), 2); - assert_eq!(b.agree_bloom_size(&a).unwrap(), 2); - } - - #[test] - fn test_bloom_size_agreement_mismatch() { - // a supports [0,0], b supports [2,3] - let a = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![0, 0]); - let b = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![2, 3]); - // agreed = min(0,3) = 0, 0 < 2 → reject - assert!(a.agree_bloom_size(&b).is_err()); - } - - #[test] - fn test_bloom_size_missing_bit() { - let a = NegotiationPayload::fmp(1, 1, NodeProfile::Full); - let b = NegotiationPayload::new(1, 1, 0); // no negotiable bit - assert!(a.agree_bloom_size(&b).is_err()); - } - - #[test] - fn test_bloom_size_missing_tlv() { - let a = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG); // bit set but no TLV - let b = NegotiationPayload::fmp(1, 1, NodeProfile::Full); - assert!(a.agree_bloom_size(&b).is_err()); - } - - #[test] - fn test_bloom_size_tlv_bad_length() { - let p = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![1]); // only 1 byte, need 2 - assert!(p.bloom_size_range().is_err()); - } - - #[test] - fn test_bloom_size_tlv_inverted_range() { - let p = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![3, 1]); // min > max - assert!(p.bloom_size_range().is_err()); - } - - #[test] - fn test_bloom_size_tlv_class_out_of_range() { - let p = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG) - .with_tlv(TLV_BLOOM_SIZE, vec![0, 4]); // max_class=4 exceeds SIZE_CLASS_BYTES - assert!(p.bloom_size_range().is_err()); - } - - #[test] - fn test_bloom_size_no_tlv_returns_none() { - let p = NegotiationPayload::new(1, 1, FMP_FEAT_BLOOM_SIZE_NEG); - assert_eq!(p.bloom_size_range().unwrap(), None); - } }