laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement FilterAnnounce send/receive, remove TTL/K-hop scoping

Browse Source 
Add bloom filter reachability announcement protocol:
- FilterAnnounce encode/decode (wire format 0x20, 1035 bytes)
- node/bloom.rs: send/receive with debounce, split-horizon loop prevention
- Handler wiring: dispatch, tick, peer promotion/removal, cross-connection
- Five integration tests: 10-node, star, chain, ring, 100-node convergence

Remove TTL/K-hop mechanism from code and design docs after discovering
that per-entry TTL scoping is fundamentally incompatible with flat bloom
filter merge + regeneration architecture. Each node re-originates filters
with fresh TTL, making propagation unbounded regardless of TTL value.
Split-horizon remains the primary loop prevention mechanism.

Document spanning tree known limitations (v1) in spanning-tree-dynamics.md.

316 tests pass, clean build, zero warnings.
This commit is contained in:
Johnathan Corgan
2026-02-11 03:16:25 +00:00
parent 7bc5b21c3a
commit 5d7af5b478
11 changed files with 758 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/design/fips-architecture.md
+4 -6
View File
@@ -156,7 +156,6 @@ Peer
│ // Bloom filter (inbound—what's reachable through them)
├── inbound_filter: BloomFilter
├── filter_sequence: u64
├── filter_ttl: u8
├── filter_received_at: Timestamp
├── pending_filter_update: bool // we owe them an update
│
@@ -246,9 +245,8 @@ BloomState
Stored on Peer:
- `inbound_filter`: what they advertise to us (4KB Bloom filter)
- `inbound_filter`: what they advertise to us (1KB Bloom filter)
- `filter_sequence`: freshness/dedup
- `filter_ttl`: remaining propagation hops
- `filter_received_at`: for staleness detection
### Computed (On-Demand)
@@ -259,11 +257,11 @@ Outgoing filter to peer Q is computed, not stored:
outbound_filter(Q) =
own_node_addr
∪ leaf_dependents
∪ { entries from peer[P].inbound_filter for all P ≠ Q where filter_ttl > 0 }
∪ { entries from peer[P].inbound_filter for all P ≠ Q }
```
TTL is decremented on contributed entries. Recomputation is cheap (4KB filter,
7 hashes) so on-demand is preferred over cache invalidation complexity.
Recomputation is cheap (1KB filter, 5 hashes) so on-demand is preferred over
cache invalidation complexity.
---
docs/design/fips-gossip-protocol.md
+15 -20
View File
@@ -187,7 +187,6 @@ node's filter indicates which destinations are reachable through it.
```text
FilterAnnounce {
sequence: u64, // For freshness/deduplication
ttl: u8, // Remaining propagation hops
filter: BloomFilter, // Variable size based on size_class
}
@@ -220,9 +219,6 @@ Receivers can fold larger filters down to their preferred size.
**sequence**: Monotonic counter for this node's filter. Allows receivers to
detect stale or duplicate announcements.
**ttl**: Remaining propagation depth. Starts at K (typically 2), decremented
each hop. At TTL=0, entries are not propagated further.
**hash_count**: Number of hash functions used. v1 uses k=5, which is optimal
for 800-1,600 entries in a 1 KB filter.
@@ -244,10 +240,12 @@ A node's outgoing filter to peer Q contains:
1. This node's own node_addr
2. Node_ids of leaf-only dependents (nodes using this node as sole peer)
3. Entries from filters received from other peers (not Q) with TTL > 0
3. Entries merged from filters received from all other peers (not Q)
This creates K-hop reachability scope. With K=2, entries propagate ~4 hops
before TTL exhaustion.
This split-horizon merge (excluding the destination peer's own filter from
the computation) prevents a node's entries from being echoed back to it,
providing loop prevention. Filters propagate transitively through the
network without any hop limit.
### 3.4 Exchange Rules
@@ -273,10 +271,10 @@ Rate limiting:
```text
1. Store: peer_filters[P] = received.filter
2. If received.ttl > 0:
- Include entries in next announcement to other peers
- Decrement TTL for propagated entries
3. Recompute own outgoing filters if changed
2. Recompute outgoing filters for all other peers:
- For each peer Q (Q != P):
outgoing[Q] = merge(self_filter, peer_filters[all peers except Q])
- If outgoing[Q] changed, send FilterAnnounce to Q
```
### 3.5 Filter Expiration
@@ -435,7 +433,6 @@ handshake completion.
| ROOT_REFRESH_INTERVAL | 30 min | Root regenerates timestamp |
| ROOT_TIMEOUT | 60 min | Root declaration considered stale |
| TREE_ENTRY_TTL | 5-10 min | Individual entry expiration |
| FILTER_TTL_HOPS | 2 | Bloom filter propagation depth |
| ANNOUNCE_MIN_INTERVAL | 500 ms | Rate limit for announcements |
| LOOKUP_TTL | 8 | Discovery request propagation limit |
| LOOKUP_TIMEOUT | 5 sec | Time to wait for response |
@@ -575,10 +572,9 @@ Propagates Bloom filter reachability information.
│ ├────────┼──────────────────┼───────────┼───────────────────────────────┤ │
│ │ 0 │ msg_type │ 1 byte │ 0x11 │ │
│ │ 1 │ sequence │ 8 bytes │ u64 LE, monotonic counter │ │
│ │ 9 │ ttl │ 1 byte │ Remaining propagation hops │ │
│ │ 10 │ hash_count │ 1 byte │ Number of hash functions (5) │ │
│ │ 11 │ size_class │ 1 byte │ Filter size: 512 << class │ │
│ │ 12 │ filter_bits │ variable │ 512 << size_class bytes │ │
│ │ 9 │ hash_count │ 1 byte │ Number of hash functions (5) │ │
│ │ 10 │ size_class │ 1 byte │ Filter size: 512 << class │ │
│ │ 11 │ filter_bits │ variable │ 512 << size_class bytes │ │
│ └────────┴──────────────────┴───────────┴───────────────────────────────┘ │
│ │
│ Size classes (powers of 2 for foldability): │
@@ -587,8 +583,8 @@ Propagates Bloom filter reachability information.
│ 2 = 2,048 bytes (16,384 bits) - Reserved for future │
│ 3 = 4,096 bytes (32,768 bits) - Reserved for future │
│ │
│ v1 total payload: 1 + 8 + 1 + 1 + 1 + 1024 = 1036 bytes │
│ With link overhead: 1065 bytes │
│ v1 total payload: 1 + 8 + 1 + 1 + 1024 = 1035 bytes │
│ With link overhead: 1064 bytes │
│ │
├─────────────────────────────────────────────────────────────────────────────┤
│ BLOOM FILTER STRUCTURE │
@@ -621,12 +617,11 @@ Propagates Bloom filter reachability information.
PLAINTEXT BYTES:
11 ← msg_type = FilterAnnounce
2A 00 00 00 00 00 00 00 ← sequence = 42
02 ← ttl = 2 (will propagate 2 more hops)
05 ← hash_count = 5
01 ← size_class = 1 (1 KB filter)
[1024 bytes of filter bits] ← Bloom filter
Total: 1036 bytes
Total: 1035 bytes
```
### A.3 LookupRequest (0x12)
docs/design/fips-routing.md
+20 -30
View File
@@ -11,8 +11,8 @@ For spanning tree dynamics and convergence, see [spanning-tree-dynamics.md](span
FIPS routing combines three mechanisms:
1. **Bloom filters**: Fast reachability lookup for nearby destinations (within
K-hop scope)
1. **Bloom filters**: Fast reachability lookup for destinations reachable
through peers
2. **Discovery protocol**: Query-based lookup for distant destinations
3. **Greedy tree routing**: Coordinate-based forwarding using spanning tree
position
@@ -34,7 +34,7 @@ coordinates handle the latter.
| Scale | Nodes | Bloom Filter Role |
|-------|-------|-------------------|
| Small private network | 100-1,000 | Covers entire network |
| Modest public network | ~1,000,000 | Covers K-hop neighborhood |
| Modest public network | ~1,000,000 | Covers transitive peer neighborhood |
| Internet-scale | Billions | Out of scope (requires different architecture) |
The primary design target is networks up to ~1M nodes.
@@ -64,11 +64,10 @@ traffic tunnels through that peer.
### Parameters
| Parameter | Value | Rationale |
|-----------|-------|-----------|
| Filter size | 1 KB (8,192 bits) | Sized for expected occupancy with margin |
| Hash functions | 5 | Optimal for 800-1,600 entries at this size |
| Scope (K) | 2 | Effective ~4-hop reach with TTL propagation |
| Parameter | Value | Rationale |
|----------------|-------------------|--------------------------------------------|
| Filter size | 1 KB (8,192 bits) | Sized for expected occupancy with margin |
| Hash functions | 5 | Optimal for 800-1,600 entries at this size |
### Mathematical Foundation
@@ -99,21 +98,15 @@ For 1% FPR: m ≈ 9.6n bits. For 5% FPR: m ≈ 6.2n bits.
### Expected Filter Occupancy
Filter occupancy depends on K-hop scope and node degree, **not** total network
size. The TTL mechanism bounds entries regardless of network scale.
**Nodes within h hops in a tree (branching factor b = d-1):**
```text
nodes_within_h_hops = (b^(h+1) - 1) / (b - 1)
```
For d=8 (b=7), K=2: each peer's 2-hop neighborhood ≈ 57 nodes.
Filter occupancy depends on network topology and node degree. In practice,
filters reach a natural equilibrium determined by the network's structure —
merging peer filters transitively means each filter converges to represent
the node's reachable neighborhood.
**Outgoing filter to peer Q contains:**
- Self (1 entry)
- Entries from (d-1) other peers' filters, with overlap
- Entries from (d-1) other peers' filters (excluding Q), with overlap
**Expected occupancy by node degree:**
@@ -203,17 +196,14 @@ A node's outgoing filter to peer Q contains:
1. This node's own Node ID
2. Node IDs of leaf-only dependents
3. Entries from filters received from other peers (not Q) with TTL > 0
3. Entries merged from filters received from all other peers (not Q)
This creates K-hop reachability scope through TTL-based propagation.
### K-Hop Scope Emergence
With TTL starting at K=2:
- Entries propagate ~2K hops before stopping
- Each node's filter contains destinations within ~4-hop effective range
- Bounded by O(d^2K) entries regardless of total network size
Filters propagate transitively through the network. Each node merges all
inbound peer filters (excluding the destination peer) into its outgoing
filter — this split-horizon approach prevents a node's own entries from
being echoed back to it, providing loop prevention. Propagation is
unbounded; filters naturally converge as the Bloom filter's fixed size
limits information density.
### Expiration
@@ -611,7 +601,7 @@ enum RouteState {
When nodes join/leave:
- Bloom filter updates propagate (bounded by K-hop scope)
- Bloom filter updates propagate through affected peers
- Affected sessions may need re-establishment
- Discovery queries for newly-joined nodes
docs/design/spanning-tree-dynamics.md
+97
View File
@@ -21,6 +21,7 @@ The protocol is based on Yggdrasil v0.5's CRDT gossip design.
8. [Cost Metrics and Parent Selection](#8-cost-metrics-and-parent-selection)
9. [Steady State Behavior](#9-steady-state-behavior)
10. [Worked Examples](#10-worked-examples)
11. [Known Limitations (v1 Implementation)](#known-limitations-v1-implementation)
---
@@ -1059,6 +1060,102 @@ initial exchange).
---
## Known Limitations (v1 Implementation)
The following limitations exist in the current implementation relative to the
design described in this document. They are documented here to guide future
work.
### Known Limitation: Root Timeout Not Enforced
The design specifies a 60-minute root timeout (§1 timing parameters, §6
partition detection) after which nodes should treat the root as departed and
re-elect. The current implementation does not track root entry timestamps or
perform staleness checks.
**Impact**: If the root node disappears permanently without a graceful
disconnect, remaining nodes retain stale root state indefinitely. Nodes that
lose their direct parent will re-elect locally (via `handle_parent_lost()`),
but nodes with an intact path to a now-departed root will not detect the
failure.
**Required fix**: Track the timestamp of the most recent root declaration in
`TreeState`. In `check_tree_state()` (called every 1s from the RX loop),
compare against `root_timeout` (default 60 min). On expiration, treat it as
root loss — increment sequence number, become own root, and re-announce.
### Known Limitation: No TTL on Tree Entries
The design specifies a 5-10 minute TTL on tree entries (§8 timing
parameters). Peer entries in `TreeState` are never expired; they persist until
explicitly removed by peer disconnection.
**Impact**: Stale ancestry information from departed nodes remains in
`TreeState`, potentially affecting coordinate computation. In practice, this
is partially mitigated by parent loss handling, but entries for non-parent
peers that depart without a graceful disconnect will linger.
**Required fix**: Add a `last_seen` timestamp to peer entries in `TreeState`.
In `check_tree_state()`, expire entries older than `tree_entry_ttl`. When
entries expire, re-evaluate parent selection if the expired entry was the
current parent.
### Known Limitation: No Partition Detection
The design describes partition detection via gossip staleness (§6) where
nodes detect isolation when root announcements stop arriving and the root
entry eventually expires, triggering independent partition operation.
**Impact**: Without root timeout enforcement (see above), partitioned nodes
cannot detect that they've lost connectivity to the root. They continue with
stale coordinates rather than forming an independent partition with a local
root. This affects only the case where the path to root is broken at some
intermediate point — direct parent loss is handled correctly.
**Required fix**: Depends on root timeout implementation. Once root timeout
is enforced, partition detection follows naturally: a node whose root entry
expires and has no peer with a fresher root declaration is partitioned. It
becomes its own root and announces, allowing the partition to converge
independently.
### Known Limitation: Limited Stability Mechanisms
The implementation includes basic hysteresis (`PARENT_SWITCH_THRESHOLD = 1`
depth difference required to switch parents), but the temporal stability
mechanisms described in the design are not implemented:
- No hold timer on parent changes (minimum time before next switch)
- No sequence number advancement rate limiting
- No announcement suppression during transient topology changes
- No minimum stable state duration before re-announcing
**Impact**: Rapid topology changes (e.g., a flapping link) could cause
excessive announcement traffic and repeated coordinate recomputation.
Currently mitigated by per-peer rate limiting on TreeAnnounce sends, but
the source node is not throttled.
**Proposed fix**: Add a hold-down timer (e.g., 5-10s) after each parent
change during which further parent switches are suppressed unless the current
parent is lost entirely. Track announcement rate and suppress if exceeding a
threshold.
### Known Limitation: Integration Test Gaps
Unit tests for `TreeState` and `TreeCoordinate` are comprehensive, and basic
integration tests verify TreeAnnounce exchange and parent ancestry
propagation. However, the following failure scenarios lack test coverage:
- Root node failure and network-wide re-election
- Network partition formation and independent operation
- Partition healing and root convergence
- Stale entry cleanup (depends on TTL implementation)
- Parent flapping under rapid topology changes
These tests are blocked on or related to the limitations above and should be
added as each limitation is resolved.
---
## Summary
The gossip-based spanning tree protocol achieves distributed coordination
src/bloom.rs
+9 -33
View File
@@ -1,6 +1,6 @@
//! Bloom Filter Implementation
//!
//! 1KB Bloom filters for K-hop reachability in FIPS routing. Each node
//! 1KB Bloom filters for reachability in FIPS routing. Each node
//! maintains filters that summarize which destinations are reachable
//! through each peer, enabling efficient routing decisions without
//! global network knowledge.
@@ -11,9 +11,8 @@
//! - Hash functions: k=5 - optimal for 800-1,600 entries at 1KB
//! - Bandwidth: 1 KB/announce (75% reduction from original 4KB design)
//!
//! The original 4KB/k=7 parameters were oversized because the d^(2K) estimate
//! overcounted by assuming mesh connectivity vs tree structure with TTL-bounded
//! propagation. Actual filter occupancy is ~250-800 entries for typical nodes.
//! These parameters are right-sized for typical network occupancy of
//! ~250-800 entries per node.
use crate::NodeAddr;
use std::collections::{HashMap, HashSet};
@@ -431,7 +430,7 @@ impl BloomState {
pub fn compute_outgoing_filter(
&self,
exclude_peer: &NodeAddr,
peer_filters: &HashMap<NodeAddr, (BloomFilter, u8)>, // (filter, ttl)
peer_filters: &HashMap<NodeAddr, BloomFilter>,
) -> BloomFilter {
let mut filter = BloomFilter::new();
@@ -443,9 +442,9 @@ impl BloomState {
filter.insert(dep);
}
// Merge filters from other peers (with TTL > 0)
for (peer_id, (peer_filter, ttl)) in peer_filters {
if peer_id != exclude_peer && *ttl > 0 {
// Merge filters from other peers
for (peer_id, peer_filter) in peer_filters {
if peer_id != exclude_peer {
// Ignore merge errors (size mismatches) - just skip that filter
let _ = filter.merge(peer_filter);
}
@@ -788,8 +787,8 @@ mod tests {
filter2.insert(&make_node_addr(200));
let mut peer_filters = HashMap::new();
peer_filters.insert(peer1, (filter1, 2)); // TTL 2
peer_filters.insert(peer2, (filter2, 1)); // TTL 1
peer_filters.insert(peer1, filter1);
peer_filters.insert(peer2, filter2);
// Filter for peer1 should exclude peer1's contributions
let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters);
@@ -806,27 +805,4 @@ mod tests {
assert!(outgoing2.contains(&make_node_addr(101))); // from peer1
}
#[test]
fn test_bloom_state_ttl_filtering() {
let my_node = make_node_addr(0);
let state = BloomState::new(my_node);
let peer1 = make_node_addr(10);
let peer2 = make_node_addr(20);
let mut filter1 = BloomFilter::new();
filter1.insert(&make_node_addr(100));
let mut filter2 = BloomFilter::new();
filter2.insert(&make_node_addr(200));
let mut peer_filters = HashMap::new();
peer_filters.insert(peer1, (filter1, 1)); // TTL 1 - included
peer_filters.insert(peer2, (filter2, 0)); // TTL 0 - excluded
let outgoing = state.compute_outgoing_filter(&make_node_addr(99), &peer_filters);
assert!(outgoing.contains(&make_node_addr(100))); // TTL 1
assert!(!outgoing.contains(&make_node_addr(200))); // TTL 0 excluded
}
}
src/node/bloom.rs
+184
View File
@@ -0,0 +1,184 @@
//! Bloom filter announce send/receive logic.
//!
//! Handles building, sending, and receiving FilterAnnounce messages,
//! including debounced propagation to peers.
use crate::bloom::BloomFilter;
use crate::protocol::FilterAnnounce;
use crate::NodeAddr;
use super::{Node, NodeError};
use std::collections::HashMap;
use tracing::{debug, info};
impl Node {
/// Collect inbound filters from all peers for outgoing filter computation.
///
/// Returns a map of (peer_node_addr -> filter) for peers that
/// have sent us a FilterAnnounce.
fn peer_inbound_filters(&self) -> HashMap<NodeAddr, BloomFilter> {
let mut filters = HashMap::new();
for (addr, peer) in &self.peers {
if let Some(filter) = peer.inbound_filter() {
filters.insert(*addr, filter.clone());
}
}
filters
}
/// 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).
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)
}
/// Send a FilterAnnounce to a specific peer, respecting debounce.
///
/// If the peer is rate-limited, the update stays pending for
/// delivery on the next tick cycle.
pub(super) async fn send_filter_announce_to_peer(
&mut self,
peer_addr: &NodeAddr,
) -> Result<(), NodeError> {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// Check debounce
if !self.bloom_state.should_send_update(peer_addr, now_ms) {
// Either not pending or rate-limited; will retry on tick
return Ok(());
}
// Build and encode
let announce = self.build_filter_announce(peer_addr);
let encoded = announce.encode().map_err(|e| NodeError::SendFailed {
node_addr: *peer_addr,
reason: format!("FilterAnnounce encode failed: {}", e),
})?;
// Send
self.send_encrypted_link_message(peer_addr, &encoded).await?;
// Record send
self.bloom_state.record_update_sent(*peer_addr, now_ms);
if let Some(peer) = self.peers.get_mut(peer_addr) {
peer.clear_filter_update_needed();
}
debug!(peer = %peer_addr, seq = announce.sequence, "Sent FilterAnnounce");
Ok(())
}
/// Send pending rate-limited filter announces whose debounce has expired.
pub(super) async fn send_pending_filter_announces(&mut self) {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let ready: Vec<NodeAddr> = self
.peers
.keys()
.filter(|addr| self.bloom_state.should_send_update(addr, now_ms))
.copied()
.collect();
for peer_addr in ready {
if let Err(e) = self.send_filter_announce_to_peer(&peer_addr).await {
debug!(
peer = %peer_addr,
error = %e,
"Failed to send pending FilterAnnounce"
);
}
}
}
/// 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
pub(super) async fn handle_filter_announce(&mut self, from: &NodeAddr, payload: &[u8]) {
let announce = match FilterAnnounce::decode(payload) {
Ok(a) => a,
Err(e) => {
debug!(from = %from, error = %e, "Malformed FilterAnnounce");
return;
}
};
// Validate
if !announce.is_valid() {
debug!(from = %from, "FilterAnnounce filter/size_class mismatch");
return;
}
if !announce.is_v1_compliant() {
debug!(from = %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(),
None => {
debug!(from = %from, "FilterAnnounce from unknown peer");
return;
}
};
// Reject stale/replay
if announce.sequence <= current_seq {
debug!(
from = %from,
received_seq = announce.sequence,
current_seq = current_seq,
"Stale FilterAnnounce rejected"
);
return;
}
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// Store on peer
if let Some(peer) = self.peers.get_mut(from) {
peer.update_filter(announce.filter, announce.sequence, now_ms);
}
info!(
from = %from,
seq = announce.sequence,
"Received FilterAnnounce"
);
// Our outgoing filter changed — mark all other peers for update
let other_peers: Vec<NodeAddr> = self
.peers
.keys()
.filter(|addr| *addr != from)
.copied()
.collect();
self.bloom_state.mark_all_updates_needed(other_peers);
}
/// Check bloom filter state on tick (called from event loop).
///
/// Sends any pending debounced filter announces.
pub(super) async fn check_bloom_state(&mut self) {
self.send_pending_filter_announces().await;
}
}
src/node/handlers.rs
+16 -1
View File
@@ -44,6 +44,7 @@ impl Node {
.unwrap_or(0);
self.process_pending_retries(now_ms).await;
self.check_tree_state().await;
self.check_bloom_state().await;
}
}
}
@@ -320,6 +321,8 @@ impl Node {
if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await {
debug!(peer = %node_addr, error = %e, "Failed to send initial TreeAnnounce");
}
// Schedule filter announce (sent on next tick via debounce)
self.bloom_state.mark_update_needed(node_addr);
}
PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => {
// Clean up the losing connection's link
@@ -333,6 +336,8 @@ impl Node {
if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await {
debug!(peer = %node_addr, error = %e, "Failed to send initial TreeAnnounce");
}
// Schedule filter announce (sent on next tick via debounce)
self.bloom_state.mark_update_needed(node_addr);
}
PromotionResult::CrossConnectionLost { winner_link_id } => {
// This connection lost — clean up its link
@@ -534,6 +539,8 @@ impl Node {
if let Err(e) = self.send_tree_announce_to_peer(&peer_node_addr).await {
debug!(peer = %peer_node_addr, error = %e, "Failed to send TreeAnnounce after cross-connection resolution");
}
// Schedule filter announce (sent on next tick via debounce)
self.bloom_state.mark_update_needed(peer_node_addr);
return;
}
@@ -553,6 +560,8 @@ impl Node {
if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await {
debug!(peer = %node_addr, error = %e, "Failed to send initial TreeAnnounce");
}
// Schedule filter announce (sent on next tick via debounce)
self.bloom_state.mark_update_needed(node_addr);
}
PromotionResult::CrossConnectionWon { loser_link_id, node_addr } => {
// Clean up the losing connection's link
@@ -571,6 +580,8 @@ impl Node {
if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await {
debug!(peer = %node_addr, error = %e, "Failed to send initial TreeAnnounce");
}
// Schedule filter announce (sent on next tick via debounce)
self.bloom_state.mark_update_needed(node_addr);
}
PromotionResult::CrossConnectionLost { winner_link_id } => {
// This connection lost — clean up its link
@@ -801,7 +812,7 @@ impl Node {
}
0x20 => {
// FilterAnnounce
debug!("Received FilterAnnounce (not yet implemented)");
self.handle_filter_announce(from, payload).await;
}
0x30 => {
// LookupRequest
@@ -885,6 +896,10 @@ impl Node {
}
}
// Bloom filter cleanup: our outgoing filter changed (lost a peer's filter)
let remaining_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(remaining_peers);
info!(
node_addr = %node_addr,
link_id = %link_id,
src/node/mod.rs
+1
View File
@@ -4,6 +4,7 @@
//! holds all state required for mesh routing: identity, tree state,
//! Bloom filters, coordinate caches, transports, links, and peers.
mod bloom;
mod handlers;
mod lifecycle;
mod retry;
src/node/tests.rs
+236 -1
View File
@@ -1720,9 +1720,10 @@ async fn drain_all_packets(nodes: &mut [TestNode], verbose: bool) -> usize {
// Wait for rate limit window (500ms) to fully expire
tokio::time::sleep(Duration::from_millis(550)).await;
// Flush pending rate-limited tree announces on all nodes
// Flush pending rate-limited tree and filter announces on all nodes
for tn in nodes.iter_mut() {
tn.node.send_pending_tree_announces().await;
tn.node.send_pending_filter_announces().await;
}
// Allow flushed packets to arrive
@@ -2113,3 +2114,237 @@ async fn test_spanning_tree_disconnected() {
verify_tree_convergence_components(&nodes, &[vec![0, 1, 2], vec![3, 4, 5]]);
cleanup_nodes(&mut nodes).await;
}
// ===== Bloom Filter Integration Tests =====
/// Verify that all peer pairs have exchanged bloom filters and each
/// peer's inbound filter contains the peer's own node_addr.
///
/// Also verifies propagation: for each node, check that destinations
/// reachable through a peer's filter include the peer's direct neighbors.
fn verify_bloom_filter_exchange(nodes: &[TestNode], edges: &[(usize, usize)]) {
// Build adjacency for hop distance computation
let n = nodes.len();
let mut adj = vec![vec![]; n];
for &(i, j) in edges {
adj[i].push(j);
adj[j].push(i);
}
// Every peer pair must have exchanged filters
for &(i, j) in edges {
let j_addr = *nodes[j].node.node_addr();
let i_addr = *nodes[i].node.node_addr();
// Node i should have a filter from node j
let peer_j = nodes[i]
.node
.get_peer(&j_addr)
.unwrap_or_else(|| panic!("Node {} should have peer {}", i, j));
let filter_from_j = peer_j.inbound_filter().unwrap_or_else(|| {
panic!(
"Node {} should have inbound filter from node {} (addr={})",
i, j, j_addr
)
});
// The filter from j must contain j's own node_addr
assert!(
filter_from_j.contains(&j_addr),
"Node {}'s filter from node {} should contain node {}'s addr",
i,
j,
j
);
// Node j should have a filter from node i
let peer_i = nodes[j]
.node
.get_peer(&i_addr)
.unwrap_or_else(|| panic!("Node {} should have peer {}", j, i));
let filter_from_i = peer_i.inbound_filter().unwrap_or_else(|| {
panic!(
"Node {} should have inbound filter from node {} (addr={})",
j, i, i_addr
)
});
// The filter from i must contain i's own node_addr
assert!(
filter_from_i.contains(&i_addr),
"Node {}'s filter from node {} should contain node {}'s addr",
j,
i,
i
);
}
// Verify propagation: each node's filter from a peer should
// contain addresses of the peer's direct neighbors (which were
// merged into the peer's outgoing filter).
for &(i, j) in edges {
let j_addr = *nodes[j].node.node_addr();
let peer_j = nodes[i].node.get_peer(&j_addr).unwrap();
let filter = peer_j.inbound_filter().unwrap();
// All of j's direct neighbors (except i) should be in j's filter to i
for &neighbor_idx in &adj[j] {
if neighbor_idx == i {
continue; // j excludes i's direction from i's filter
}
let neighbor_addr = *nodes[neighbor_idx].node.node_addr();
assert!(
filter.contains(&neighbor_addr),
"Node {}'s filter from node {} should contain node {}'s neighbor {} (addr={})",
i,
j,
j,
neighbor_idx,
neighbor_addr
);
}
}
}
/// 10-node random graph: tree + bloom filter convergence.
#[tokio::test]
async fn test_bloom_filter_10_nodes() {
let edges = generate_random_edges(10, 20, 123);
let mut nodes = run_tree_test(10, &edges, false).await;
verify_tree_convergence(&nodes);
verify_bloom_filter_exchange(&nodes, &edges);
cleanup_nodes(&mut nodes).await;
}
/// 5-node star: hub node's filter should contain all spokes.
#[tokio::test]
async fn test_bloom_filter_star() {
let edges: Vec<(usize, usize)> = vec![(0, 1), (0, 2), (0, 3), (0, 4)];
let mut nodes = run_tree_test(5, &edges, false).await;
verify_tree_convergence(&nodes);
verify_bloom_filter_exchange(&nodes, &edges);
// Hub (node 0) sends each spoke a filter containing the other spokes
let hub_addr = *nodes[0].node.node_addr();
for spoke in 1..5 {
let peer = nodes[spoke].node.get_peer(&hub_addr).unwrap();
let filter = peer.inbound_filter().unwrap();
// Filter from hub should contain all OTHER spokes
for other in 1..5 {
if other == spoke {
continue;
}
let other_addr = *nodes[other].node.node_addr();
assert!(
filter.contains(&other_addr),
"Spoke {}'s filter from hub should contain spoke {} (addr={})",
spoke,
other,
other_addr
);
}
}
cleanup_nodes(&mut nodes).await;
}
/// 8-node chain: verify full propagation.
///
/// Chain: 0-1-2-3-4-5-6-7. Each node's outgoing filter is the merge
/// of its own address plus all peer inbound filters (excluding the
/// destination peer). This means entries propagate through the entire
/// chain: node 1 merges node 2's filter, which contains node 3's
/// entries, and so on. Both endpoints should see all other nodes.
#[tokio::test]
async fn test_bloom_filter_chain_propagation() {
let edges: Vec<(usize, usize)> =
vec![(0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, 6), (6, 7)];
let mut nodes = run_tree_test(8, &edges, false).await;
verify_tree_convergence(&nodes);
verify_bloom_filter_exchange(&nodes, &edges);
let addrs: Vec<NodeAddr> = nodes.iter().map(|tn| *tn.node.node_addr()).collect();
// Node 0's filter from node 1 should contain node 1 and its
// immediate neighbor node 2 (node 1 directly merges node 2's filter).
let peer_1 = nodes[0].node.get_peer(&addrs[1]).unwrap();
let filter = peer_1.inbound_filter().unwrap();
assert!(filter.contains(&addrs[1]), "Should contain node 1 (self)");
assert!(
filter.contains(&addrs[2]),
"Should contain node 2 (1-hop neighbor of node 1)"
);
// Entries propagate through the full chain because each
// intermediate node merges its peer's filter into its outgoing
// filter. Verify all nodes are reachable from the endpoints.
for i in 2..8 {
assert!(
filter.contains(&addrs[i]),
"Node 0's filter from node 1 should contain node {} \
(chain merge propagation)",
i
);
}
// Verify symmetric: node 7's filter from node 6 should contain all
for i in 0..6 {
let peer_6 = nodes[7].node.get_peer(&addrs[6]).unwrap();
let filter_6 = peer_6.inbound_filter().unwrap();
assert!(
filter_6.contains(&addrs[i]),
"Node 7's filter from node 6 should contain node {} \
(chain merge propagation)",
i
);
}
cleanup_nodes(&mut nodes).await;
}
/// 5-node ring: every node should see all others (all within 2-hop reach).
#[tokio::test]
async fn test_bloom_filter_ring() {
let edges: Vec<(usize, usize)> = vec![(0, 1), (1, 2), (2, 3), (3, 4), (4, 0)];
let mut nodes = run_tree_test(5, &edges, false).await;
verify_tree_convergence(&nodes);
verify_bloom_filter_exchange(&nodes, &edges);
// In a 5-node ring, each node has 2 peers. Through each peer,
// the other 3 nodes are at most 2 hops away. So every node should
// be reachable via at least one peer's filter.
for i in 0..5 {
for j in 0..5 {
if i == j {
continue;
}
let target_addr = *nodes[j].node.node_addr();
let reachable = nodes[i]
.node
.peers()
.any(|peer| peer.may_reach(&target_addr));
assert!(
reachable,
"Node {} should see node {} as reachable via at least one peer's filter",
i, j
);
}
}
cleanup_nodes(&mut nodes).await;
}
/// 100-node random graph: bloom filter exchange at scale.
#[tokio::test]
async fn test_bloom_filter_convergence_100_nodes() {
const NUM_NODES: usize = 100;
const TARGET_EDGES: usize = 250;
const SEED: u64 = 42;
let edges = generate_random_edges(NUM_NODES, TARGET_EDGES, SEED);
let mut nodes = run_tree_test(NUM_NODES, &edges, false).await;
verify_tree_convergence(&nodes);
verify_bloom_filter_exchange(&nodes, &edges);
cleanup_nodes(&mut nodes).await;
}
src/peer/active.rs
+1 -13
View File
@@ -105,8 +105,6 @@ pub struct ActivePeer {
inbound_filter: Option<BloomFilter>,
/// Their filter's sequence number.
filter_sequence: u64,
/// Remaining propagation hops on their filter.
filter_ttl: u8,
/// When we received their last filter (Unix milliseconds).
filter_received_at: u64,
/// Whether we owe them a filter update.
@@ -142,7 +140,6 @@ impl ActivePeer {
pending_tree_announce: false,
inbound_filter: None,
filter_sequence: 0,
filter_ttl: 0,
filter_received_at: 0,
pending_filter_update: true, // Send filter on new connection
link_stats: LinkStats::new(),
@@ -197,7 +194,6 @@ impl ActivePeer {
pending_tree_announce: false,
inbound_filter: None,
filter_sequence: 0,
filter_ttl: 0,
filter_received_at: 0,
pending_filter_update: true,
link_stats,
@@ -362,11 +358,6 @@ impl ActivePeer {
self.filter_sequence
}
/// Get the filter TTL.
pub fn filter_ttl(&self) -> u8 {
self.filter_ttl
}
/// Check if this peer's filter is stale.
pub fn filter_is_stale(&self, current_time_ms: u64, stale_threshold_ms: u64) -> bool {
if self.filter_received_at == 0 {
@@ -512,12 +503,10 @@ impl ActivePeer {
&mut self,
filter: BloomFilter,
sequence: u64,
ttl: u8,
current_time_ms: u64,
) {
self.inbound_filter = Some(filter);
self.filter_sequence = sequence;
self.filter_ttl = ttl;
self.filter_received_at = current_time_ms;
self.last_seen = current_time_ms;
}
@@ -526,7 +515,6 @@ impl ActivePeer {
pub fn clear_filter(&mut self) {
self.inbound_filter = None;
self.filter_sequence = 0;
self.filter_ttl = 0;
self.filter_received_at = 0;
}
@@ -645,7 +633,7 @@ mod tests {
let mut filter = BloomFilter::new();
filter.insert(&target);
peer.update_filter(filter, 1, 2, 1500);
peer.update_filter(filter, 1, 1500);
assert!(peer.may_reach(&target));
assert!(!peer.filter_is_stale(1800, 500));
src/protocol.rs
+175 -55
View File
@@ -558,7 +558,6 @@ impl TreeAnnounce {
/// Bloom filter announcement for reachability propagation.
///
/// Sent to peers to advertise which destinations are reachable.
/// The TTL controls propagation depth (decremented at each hop).
///
/// ## Wire Format (v1)
///
@@ -566,16 +565,13 @@ impl TreeAnnounce {
/// |--------|-------------|----------|----------------------------------|
/// | 0 | msg_type | 1 byte | 0x20 |
/// | 1 | sequence | 8 bytes | LE u64 |
/// | 9 | ttl | 1 byte | Remaining hops |
/// | 10 | hash_count | 1 byte | Number of hash functions |
/// | 11 | size_class | 1 byte | Filter size: 512 << size_class |
/// | 12 | filter_bits | variable | 512 << size_class bytes |
/// | 9 | hash_count | 1 byte | Number of hash functions |
/// | 10 | size_class | 1 byte | Filter size: 512 << size_class |
/// | 11 | filter_bits | variable | 512 << size_class bytes |
#[derive(Clone, Debug)]
pub struct FilterAnnounce {
/// The bloom filter contents.
pub filter: BloomFilter,
/// Remaining propagation hops (decremented at each forward).
pub ttl: u8,
/// Sequence number for freshness/dedup.
pub sequence: u64,
/// Number of hash functions used by the filter.
@@ -587,12 +583,11 @@ pub struct FilterAnnounce {
impl FilterAnnounce {
/// Create a new FilterAnnounce message with v1 defaults.
pub fn new(filter: BloomFilter, ttl: u8, sequence: u64) -> Self {
pub fn new(filter: BloomFilter, sequence: u64) -> Self {
Self {
hash_count: filter.hash_count(),
size_class: crate::bloom::V1_SIZE_CLASS,
filter,
ttl,
sequence,
}
}
@@ -600,7 +595,6 @@ impl FilterAnnounce {
/// Create with explicit size_class (for testing or future protocol versions).
pub fn with_size_class(
filter: BloomFilter,
ttl: u8,
sequence: u64,
size_class: u8,
) -> Self {
@@ -608,30 +602,10 @@ impl FilterAnnounce {
hash_count: filter.hash_count(),
size_class,
filter,
ttl,
sequence,
}
}
/// Check if this filter can be forwarded (TTL > 0).
pub fn can_forward(&self) -> bool {
self.ttl > 0
}
/// Create a forwarded version with decremented TTL.
pub fn forwarded(&self) -> Option<Self> {
if self.ttl == 0 {
return None;
}
Some(Self {
filter: self.filter.clone(),
ttl: self.ttl - 1,
sequence: self.sequence,
hash_count: self.hash_count,
size_class: self.size_class,
})
}
/// Get the expected filter size in bytes for this size_class.
pub fn filter_size_bytes(&self) -> usize {
512 << self.size_class
@@ -647,6 +621,114 @@ impl FilterAnnounce {
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> {
if !self.is_valid() {
return Err(ProtocolError::Malformed(
"filter size does not match size_class".into(),
));
}
let filter_bytes = self.filter.as_bytes();
let size = 1 + Self::MIN_PAYLOAD_SIZE + filter_bytes.len();
let mut buf = Vec::with_capacity(size);
// msg_type
buf.push(LinkMessageType::FilterAnnounce.to_byte());
// sequence (8 LE)
buf.extend_from_slice(&self.sequence.to_le_bytes());
// hash_count
buf.push(self.hash_count);
// size_class
buf.push(self.size_class);
// filter_bits
buf.extend_from_slice(filter_bytes);
Ok(buf)
}
/// Decode from link-layer payload (after msg_type byte stripped by dispatcher).
///
/// The payload starts with the sequence field.
pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
if payload.len() < Self::MIN_PAYLOAD_SIZE {
return Err(ProtocolError::MessageTooShort {
expected: Self::MIN_PAYLOAD_SIZE,
got: payload.len(),
});
}
let mut pos = 0;
// sequence (8 LE)
let sequence = u64::from_le_bytes(
payload[pos..pos + 8]
.try_into()
.map_err(|_| ProtocolError::Malformed("bad sequence".into()))?,
);
pos += 8;
// hash_count
let hash_count = payload[pos];
pos += 1;
// size_class
let size_class = payload[pos];
pos += 1;
// Validate size_class range
if size_class > Self::MAX_SIZE_CLASS {
return Err(ProtocolError::Malformed(format!(
"invalid size_class: {size_class} (max {})",
Self::MAX_SIZE_CLASS
)));
}
// v1 compliance check
if size_class != crate::bloom::V1_SIZE_CLASS {
return Err(ProtocolError::Malformed(format!(
"unsupported size_class: {size_class} (v1 requires {})",
crate::bloom::V1_SIZE_CLASS
)));
}
// Expected filter size from size_class
let expected_filter_bytes = 512usize << size_class;
let remaining = payload.len() - pos;
if remaining != expected_filter_bytes {
return Err(ProtocolError::MessageTooShort {
expected: Self::MIN_PAYLOAD_SIZE + expected_filter_bytes,
got: payload.len(),
});
}
// Construct BloomFilter from bytes
let filter =
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)
}
}
// ============ Discovery Messages ============
@@ -1452,28 +1534,10 @@ mod tests {
// ===== FilterAnnounce Tests =====
#[test]
fn test_filter_announce_forward() {
let filter = BloomFilter::new();
let announce = FilterAnnounce::new(filter, 2, 100);
assert!(announce.can_forward());
let forwarded = announce.forwarded().unwrap();
assert_eq!(forwarded.ttl, 1);
assert_eq!(forwarded.sequence, 100);
let forwarded2 = forwarded.forwarded().unwrap();
assert_eq!(forwarded2.ttl, 0);
assert!(!forwarded2.can_forward());
assert!(forwarded2.forwarded().is_none());
}
#[test]
fn test_filter_announce_size_class() {
let filter = BloomFilter::new();
let announce = FilterAnnounce::new(filter.clone(), 2, 100);
let announce = FilterAnnounce::new(filter.clone(), 100);
// v1 defaults
assert_eq!(announce.size_class, 1);
@@ -1481,17 +1545,12 @@ mod tests {
assert!(announce.is_v1_compliant());
assert!(announce.is_valid());
assert_eq!(announce.filter_size_bytes(), 1024);
// Forwarded preserves size_class
let forwarded = announce.forwarded().unwrap();
assert_eq!(forwarded.size_class, 1);
assert_eq!(forwarded.hash_count, 5);
}
#[test]
fn test_filter_announce_with_size_class() {
let filter = BloomFilter::with_params(2048 * 8, 7).unwrap();
let announce = FilterAnnounce::with_size_class(filter, 2, 100, 2);
let announce = FilterAnnounce::with_size_class(filter, 100, 2);
assert_eq!(announce.size_class, 2);
assert_eq!(announce.hash_count, 7);
@@ -1500,6 +1559,67 @@ mod tests {
assert_eq!(announce.filter_size_bytes(), 2048);
}
#[test]
fn test_filter_announce_encode_decode_roundtrip() {
let mut filter = BloomFilter::new();
filter.insert(&make_node_addr(42));
filter.insert(&make_node_addr(99));
let announce = FilterAnnounce::new(filter, 500);
let encoded = announce.encode().unwrap();
// msg_type(1) + sequence(8) + hash_count(1) + size_class(1) + filter(1024)
assert_eq!(encoded.len(), 1035);
assert_eq!(encoded[0], LinkMessageType::FilterAnnounce.to_byte());
// Decode strips msg_type (as dispatcher does)
let decoded = FilterAnnounce::decode(&encoded[1..]).unwrap();
assert_eq!(decoded.sequence, 500);
assert_eq!(decoded.hash_count, 5);
assert_eq!(decoded.size_class, 1);
assert!(decoded.is_valid());
assert!(decoded.is_v1_compliant());
// Filter contents preserved
assert!(decoded.filter.contains(&make_node_addr(42)));
assert!(decoded.filter.contains(&make_node_addr(99)));
assert!(!decoded.filter.contains(&make_node_addr(1)));
}
#[test]
fn test_filter_announce_decode_rejects_bad_size_class() {
let filter = BloomFilter::new();
let announce = FilterAnnounce::new(filter, 100);
let mut encoded = announce.encode().unwrap();
// Corrupt size_class byte (offset: 1 msg_type + 8 seq + 1 hash = 10)
encoded[10] = 5; // invalid size_class > MAX_SIZE_CLASS
let result = FilterAnnounce::decode(&encoded[1..]);
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("invalid size_class"));
}
#[test]
fn test_filter_announce_decode_rejects_non_v1_size_class() {
// Build a size_class=0 payload manually (valid range but not v1)
let filter = BloomFilter::with_params(512 * 8, 5).unwrap();
let announce = FilterAnnounce::with_size_class(filter, 100, 0);
let encoded = announce.encode().unwrap();
let result = FilterAnnounce::decode(&encoded[1..]);
assert!(result.is_err());
assert!(result
.unwrap_err()
.to_string()
.contains("unsupported size_class"));
}
#[test]
fn test_filter_announce_decode_rejects_truncated() {
let result = FilterAnnounce::decode(&[0u8; 5]);
assert!(result.is_err());
}
// ===== SessionSetup Tests =====
#[test]