Heterogeneous bloom filters: delta compression, variable sizing, adaptive heuristic

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<u8> → Vec<u64> 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.
This commit is contained in:
Johnathan Corgan
2026-04-11 08:16:01 +00:00
parent 8162d3c237
commit eb8479eb20
16 changed files with 1378 additions and 476 deletions
+14 -21
View File
@@ -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;
}
+212
View File
@@ -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<u8>, 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<Vec<u64>, 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<u64> = (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
}
}
+205 -30
View File
@@ -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<u8>,
/// Bit array storage (packed as 64-bit words, little-endian bit order).
words: Vec<u64>,
/// 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<u8>, hash_count: u8) -> Result<Self, BloomError> {
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<BloomFilter, BloomError> {
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<u64> = 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<BloomFilter, BloomError> {
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<BloomFilter, BloomError> {
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<BloomFilter, BloomError> {
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<BloomFilter, BloomError> {
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<BloomFilter, BloomError> {
if self.num_bits != other.num_bits {
return Err(BloomError::InvalidSize {
expected: self.num_bits,
got: other.num_bits,
});
}
let words: Vec<u64> = 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<u8> {
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
}
}
+43 -14
View File
@@ -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<u8> {
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)]
+82 -18
View File
@@ -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<NodeAddr>,
/// 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<NodeAddr>,
/// 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<NodeAddr, BloomFilter>,
/// Sequence number of the last filter sent to each peer.
last_sent_seq: HashMap<NodeAddr, u64>,
/// 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<u64> {
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<NodeAddr, BloomFilter>,
) -> 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<u8> {
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);
+377 -6
View File
@@ -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]
+195 -38
View File
@@ -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<NodeAddr> = 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<NodeAddr> = 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;
}
}
+4
View File
@@ -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;
+3 -12
View File
@@ -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"
);
+17 -3
View File
@@ -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)]
-10
View File
@@ -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
+5 -14
View File
@@ -125,9 +125,6 @@ pub struct PeerConnection {
// === Negotiation Results ===
/// Peer's node profile (learned from negotiation payload).
peer_profile: Option<NodeProfile>,
/// Agreed bloom filter size class.
agreed_bloom_size_class: Option<u8>,
// === Handshake Resend ===
/// Wire-format msg1 bytes for resend (initiator only).
handshake_msg1: Option<Vec<u8>>,
@@ -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<u8> {
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 ===
+210 -142
View File
@@ -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<Vec<u8>, ProtocolError> {
/// The filter words are RLE-compressed.
pub fn encode(&self) -> Result<(Vec<u8>, 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<Self, ProtocolError> {
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<u8> {
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<Self, ProtocolError> {
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());
}
}
+6 -1
View File
@@ -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,
+4 -4
View File
@@ -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,
+1 -163
View File
@@ -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<u8> 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<Option<BloomSizeRange>, 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<u8, ProtocolError> {
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);
}
}