laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

proto/stp: sans-IO spanning-tree state machine

Browse Source 
Migrate the full non-async spanning-tree surface into proto/stp/, mirroring the
discovery/routing/fmp/mmp conversions. The classification ladder (parent-switch /
self-root / loop-drop / ancestry-update / periodic-rebroadcast / parent-lost) moves
out of the async node handlers into a pure Stp classify layer returning a
TreeDecision the shell drives, with effect ordering and per-arm invalidation
preserved verbatim. src/tree/ relocates wholesale: TreeState + ParentDeclaration data
+ coordinates into proto/stp/{state,coordinate}, the flap-dampening / hold-down
cluster into a FlapDampener in limits.rs, and the wire codec into wire.rs. The clock
is injected as u64 (wall-clock secs for the escaping declaration timestamp, monotonic
ms for the dampening timers via mmp::mono_ms); declaration crypto is field-partitioned
so sign/verify/hash run in the shell while the in-core modules carry data +
signing_bytes only. Peer maps/sets move to BTree; core/state/coordinate/limits are
core+alloc clean, with wire.rs the one std-tethered file. Behavior-neutral:
characterization tests added for the handler decision arms; convergence suite and
ci-local (36/36) green.
This commit is contained in:
Johnathan Corgan
2026-07-07 17:07:33 +00:00
parent 50a595a0ed
commit a67801099d
33 changed files with 1968 additions and 1139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/mmp.rs
+24 -10
View File
@@ -7,6 +7,7 @@
use crate::NodeAddr;
use crate::node::Node;
use crate::node::reject::{MmpReject, RejectReason, TreeReject};
use crate::node::tree::sign_declaration;
use crate::proto::mmp::{
BackoffUpdate, LinkReportKind, LinkReportSnapshot, MmpAction, MmpSessionState,
PathMtuNotification, PeerLivenessSnapshot, ReceiverReport, RrLog, SendResult, SenderReport,
@@ -185,25 +186,36 @@ impl Node {
// Trigger re-evaluation so the node doesn't wait for the next
// periodic tick or TreeAnnounce.
if first_rtt {
let peer_costs: std::collections::HashMap<crate::NodeAddr, f64> = self
let peer_costs: std::collections::BTreeMap<crate::NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, p)| p.has_srtt())
.map(|(a, p)| (*a, p.link_cost()))
.collect();
if let Some(new_parent) = self.tree_state.evaluate_parent(&peer_costs) {
// Wall-clock seconds for the escaping declaration timestamp;
// monotonic ms for the flap-dampening / hold-down timers.
let now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mono_now_ms = crate::mmp::mono_ms();
if let Some(new_parent) = self.tree_state.evaluate_parent(
&peer_costs,
&std::collections::BTreeSet::new(),
mono_now_ms,
) {
let new_seq = self.tree_state.my_declaration().sequence() + 1;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp);
let flap_dampened =
self.tree_state
.set_parent(new_parent, new_seq, now_secs, mono_now_ms);
self.tree_state.recompute_coords();
// Clone identity once: sign_declaration borrows &mut tree_state while
// the identity() accessor borrows all of &self, so an owned copy avoids
// the split-borrow conflict on this infrequent parent-switch path.
let our_identity = self.identity().clone();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after first-RTT parent eval");
self.metrics()
.tree
@@ -232,10 +244,12 @@ impl Node {
let all_peers: Vec<crate::NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
} else if !self.tree_state.is_root() && self.tree_state.should_be_root() {
self.tree_state.become_root();
self.tree_state.become_root(now_secs);
// Clone identity once (see the parent-switch branch above for why).
let our_identity = self.identity().clone();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign self-root declaration after first-RTT");
self.metrics()
.tree
src/node/handlers/session.rs
+6 -3
View File
@@ -52,8 +52,8 @@ struct PipelinedSend<'a> {
timestamp: u32,
fsp_flags: u8,
inner_plaintext: &'a [u8],
my_coords: Option<&'a crate::tree::TreeCoordinate>,
dest_coords: Option<&'a crate::tree::TreeCoordinate>,
my_coords: Option<&'a crate::proto::stp::TreeCoordinate>,
dest_coords: Option<&'a crate::proto::stp::TreeCoordinate>,
}
impl Node {
@@ -2071,7 +2071,10 @@ impl Node {
/// Returns our own coordinates as a fallback (the SessionSetup will
/// carry src_coords for return path routing; empty dest_coords
/// would fail wire encoding since TreeCoordinate requires ≥1 entry).
pub(in crate::node) fn get_dest_coords(&self, dest: &NodeAddr) -> crate::tree::TreeCoordinate {
pub(in crate::node) fn get_dest_coords(
&self,
dest: &NodeAddr,
) -> crate::proto::stp::TreeCoordinate {
let now_ms = Self::now_ms();
if let Some(coords) = self.coord_cache.get(dest, now_ms) {
return coords.clone();
src/node/mod.rs
+20 -11
View File
@@ -48,6 +48,7 @@ use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateL
use crate::proto::fmp::Fmp;
use crate::proto::mmp::Mmp;
use crate::proto::routing::{self, Router, RoutingErrorRateLimiter};
use crate::proto::stp::TreeState;
#[cfg(unix)]
use crate::transport::ethernet::EthernetTransport;
use crate::transport::nym::NymTransport;
@@ -58,14 +59,13 @@ use crate::transport::{
ConnectionState, Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError,
TransportHandle, TransportId,
};
use crate::tree::TreeState;
use crate::upper::hosts::HostMap;
use crate::upper::icmp_rate_limit::IcmpRateLimiter;
use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
use crate::utils::index::IndexAllocator;
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity, TreeCoordinate};
use rand::Rng;
use std::collections::{HashMap, HashSet, VecDeque};
use std::collections::{BTreeSet, HashMap, HashSet, VecDeque};
use std::fmt;
use std::sync::Arc;
use std::thread::JoinHandle;
@@ -562,7 +562,11 @@ impl Node {
};
// Initialize tree state with signed self-declaration
let mut tree_state = TreeState::new(node_addr);
let tree_now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mut tree_state = TreeState::new(node_addr, tree_now_secs);
tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
tree_state.set_hold_down(config.node.tree.hold_down_secs);
tree_state.set_flap_dampening(
@@ -570,8 +574,7 @@ impl Node {
config.node.tree.flap_window_secs,
config.node.tree.flap_dampening_secs,
);
tree_state
.sign_declaration(&identity)
tree::sign_declaration(tree_state.my_declaration_mut(), &identity)
.expect("signing own declaration should never fail");
let coord_cache = CoordCache::new(
@@ -723,7 +726,11 @@ impl Node {
};
// Initialize tree state with signed self-declaration
let mut tree_state = TreeState::new(node_addr);
let tree_now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mut tree_state = TreeState::new(node_addr, tree_now_secs);
tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
tree_state.set_hold_down(config.node.tree.hold_down_secs);
tree_state.set_flap_dampening(
@@ -731,8 +738,7 @@ impl Node {
config.node.tree.flap_window_secs,
config.node.tree.flap_dampening_secs,
);
tree_state
.sign_declaration(&identity)
tree::sign_declaration(tree_state.my_declaration_mut(), &identity)
.expect("signing own declaration should never fail");
let mut bloom_state = BloomState::new(node_addr);
@@ -1564,7 +1570,7 @@ impl Node {
/// Resolution order: this node when it is root, then the root as a live
/// authenticated peer (cryptographically attested npub), then the
/// identity-cache, else `None`.
pub(crate) fn resolve_root_npub(&self, tree: &crate::tree::TreeState) -> Option<String> {
pub(crate) fn resolve_root_npub(&self, tree: &crate::proto::stp::TreeState) -> Option<String> {
if tree.is_root() {
return Some(self.npub());
}
@@ -2641,8 +2647,11 @@ impl Node {
return self.peers.get(&next_hop);
}
// 4. Greedy tree routing fallback
let next_hop_id = self.tree_state.find_next_hop(&dest_coords)?;
// 4. Greedy tree routing fallback. No peers are excluded from transit
// on this branch; the non-full/leaf skip is a next-only refinement.
let next_hop_id = self
.tree_state
.find_next_hop(&dest_coords, &BTreeSet::new())?;
self.peers.get(&next_hop_id).filter(|p| p.can_send())
}
src/node/session_wire.rs
+1 -1
View File
@@ -32,8 +32,8 @@
//! | 0x2 | - | Handshake msg2 | SessionAck (Noise XK msg2) |
//! | 0x3 | - | Handshake msg3 | SessionMsg3 (Noise XK msg3) |
use crate::proto::stp::TreeCoordinate;
use crate::protocol::{ProtocolError, decode_optional_coords};
use crate::tree::TreeCoordinate;
// ============================================================================
// Constants
src/node/tests/bloom.rs
+4 -4
View File
@@ -457,7 +457,7 @@ async fn test_bloom_filter_split_horizon() {
fn compute_mesh_size_counts_each_peer_filter_once() {
use crate::bloom::BloomFilter;
use crate::peer::ActivePeer;
use crate::tree::ParentDeclaration;
use crate::proto::stp::ParentDeclaration;
let mut node = make_node();
let my_addr = *node.tree_state().my_node_addr();
@@ -498,8 +498,8 @@ fn compute_mesh_size_counts_each_peer_filter_once() {
// Seed parent ancestry first so recompute_coords can extend it and
// flip is_root() to false; child ancestry is for completeness.
let parent_ancestry = crate::tree::TreeCoordinate::root_with_meta(parent_addr, 1, 1);
let child_ancestry = crate::tree::TreeCoordinate::root_with_meta(child_addr, 1, 1);
let parent_ancestry = crate::proto::stp::TreeCoordinate::root_with_meta(parent_addr, 1, 1);
let child_ancestry = crate::proto::stp::TreeCoordinate::root_with_meta(child_addr, 1, 1);
// Inject the stale-cache scenario: peer_declaration(P) still names
// US (M) as P's parent (the pre-switch advert that the cache hasn't
// refreshed yet). Q is a legitimate child also naming M as parent.
@@ -513,7 +513,7 @@ fn compute_mesh_size_counts_each_peer_filter_once() {
.update_peer(child_decl, child_ancestry);
// Switch our parent to P and recompute coords so root flips off self.
node.tree_state_mut().set_parent(parent_addr, 2, 1);
node.tree_state_mut().set_parent(parent_addr, 2, 1, 1);
node.tree_state_mut().recompute_coords();
assert!(
!node.tree_state().is_root(),
src/node/tests/discovery.rs
+2 -2
View File
@@ -6,7 +6,7 @@
use super::*;
use crate::proto::discovery::{LookupRequest, LookupResponse, RecentRequest};
use crate::tree::TreeCoordinate;
use crate::proto::stp::TreeCoordinate;
use spanning_tree::{
cleanup_nodes, generate_random_edges, lock_large_network_test, process_available_packets,
run_tree_test, verify_tree_convergence,
@@ -1082,7 +1082,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() {
// as its parent. `is_tree_peer` checks both directions — the child
// direction (peer.parent_id == self.node_addr) is what we exercise.
let our_addr = *node.node_addr();
let peer_decl = crate::tree::ParentDeclaration::new(peer_addr, our_addr, 1, 0);
let peer_decl = crate::proto::stp::ParentDeclaration::new(peer_addr, our_addr, 1, 0);
let peer_coords = TreeCoordinate::from_addrs(vec![peer_addr, our_addr]).unwrap();
node.tree_state_mut().update_peer(peer_decl, peer_coords);
assert!(node.is_tree_peer(&peer_addr), "peer must be a tree peer");
src/node/tests/forwarding.rs
+1 -1
View File
@@ -6,8 +6,8 @@
use super::*;
use crate::node::session_wire::{FSP_FLAG_CP, build_fsp_header};
use crate::proto::stp::TreeCoordinate;
use crate::protocol::{SessionAck, SessionDatagram, SessionSetup, encode_coords};
use crate::tree::TreeCoordinate;
use spanning_tree::{
TestNode, cleanup_nodes, process_available_packets, run_tree_test, verify_tree_convergence,
};
src/node/tests/mmp_chartests.rs
+1 -1
View File
@@ -33,7 +33,7 @@ use crate::node::session::{EndToEndState, SessionEntry};
use crate::noise::HandshakeState;
use crate::peer::ActivePeer;
use crate::proto::mmp::{MmpMode, ReceiverReport};
use crate::tree::{ParentDeclaration, TreeCoordinate};
use crate::proto::stp::{ParentDeclaration, TreeCoordinate};
// ===========================================================================
// Helpers
src/node/tests/routing.rs
+5 -5
View File
@@ -5,7 +5,7 @@
use super::*;
use crate::bloom::BloomFilter;
use crate::tree::{ParentDeclaration, TreeCoordinate};
use crate::proto::stp::{ParentDeclaration, TreeCoordinate};
use spanning_tree::{
TestNode, cleanup_nodes, drain_all_packets, generate_random_edges, initiate_handshake,
lock_large_network_test, make_test_node, run_tree_test, verify_tree_convergence,
@@ -826,7 +826,7 @@ async fn test_routing_stops_after_peer_removal() {
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let all_coords: Vec<(NodeAddr, crate::tree::TreeCoordinate)> = nodes
let all_coords: Vec<(NodeAddr, crate::proto::stp::TreeCoordinate)> = nodes
.iter()
.map(|tn| {
(
@@ -1007,7 +1007,7 @@ async fn test_routing_source_only_coords_100_nodes() {
.unwrap_or(0);
// Collect all coords for injection
let all_coords: Vec<(NodeAddr, crate::tree::TreeCoordinate)> = nodes
let all_coords: Vec<(NodeAddr, crate::proto::stp::TreeCoordinate)> = nodes
.iter()
.map(|tn| {
(
@@ -1361,7 +1361,7 @@ fn test_parent_loss_reparent_invalidates_coord_cache() {
TreeCoordinate::from_addrs(vec![alt, root]).unwrap(),
);
// Adopt `parent`; our coords become [my_addr, parent, root], root = `root`.
node.tree_state_mut().set_parent(parent, 1, 1000);
node.tree_state_mut().set_parent(parent, 1, 1000, 1000);
node.tree_state_mut().recompute_coords();
assert!(!node.tree_state().is_root());
assert_eq!(node.tree_state().root(), &root);
@@ -1416,7 +1416,7 @@ fn test_parent_loss_selfroot_invalidates_coord_cache() {
ParentDeclaration::new(parent, old_root, 1, 1000),
TreeCoordinate::from_addrs(vec![parent, old_root]).unwrap(),
);
node.tree_state_mut().set_parent(parent, 1, 1000);
node.tree_state_mut().set_parent(parent, 1, 1000, 1000);
node.tree_state_mut().recompute_coords();
assert!(!node.tree_state().is_root());
src/node/tests/session.rs
+1 -1
View File
@@ -18,7 +18,7 @@ fn populate_all_coord_caches(nodes: &mut [TestNode]) {
.unwrap()
.as_millis() as u64;
let all_coords: Vec<(NodeAddr, crate::tree::TreeCoordinate)> = nodes
let all_coords: Vec<(NodeAddr, crate::proto::stp::TreeCoordinate)> = nodes
.iter()
.map(|tn| {
(
src/node/tests/spanning_tree.rs
+373 -9
View File
@@ -5,9 +5,10 @@
//! reused by bloom filter tests.
use super::*;
use crate::protocol::TreeAnnounce;
use crate::node::tree::sign_declaration;
use crate::proto::stp::TreeAnnounce;
use crate::proto::stp::{CoordEntry, ParentDeclaration, TreeCoordinate};
use crate::transport::loopback::{LoopbackRegistry, LoopbackTransport, new_registry};
use crate::tree::{CoordEntry, ParentDeclaration, TreeCoordinate};
static LARGE_NETWORK_TEST_LOCK: std::sync::LazyLock<tokio::sync::Mutex<()>> =
std::sync::LazyLock::new(|| tokio::sync::Mutex::new(()));
@@ -891,7 +892,7 @@ async fn test_rejects_tree_announce_with_inconsistent_root() {
// sequence/timestamp so the announce would be acceptable on freshness
// grounds if its ancestry semantics were valid.
let mut declaration = ParentDeclaration::new(a_addr, fake_parent, 99, 12345);
declaration.sign(nodes[0].node.identity()).unwrap();
sign_declaration(&mut declaration, nodes[0].node.identity()).unwrap();
let announce = TreeAnnounce::new(
declaration,
@@ -970,18 +971,15 @@ async fn test_tree_announce_repushed_on_root_disagreement() {
// it had never processed the root's attaching announce. The child's
// advertised root is now itself, disagreeing with the root's view, and its
// own periodic re-evaluation cannot recover it (single peer).
nodes[child_idx].node.tree_state_mut().become_root();
nodes[child_idx].node.tree_state_mut().become_root(1000);
nodes[child_idx]
.node
.tree_state_mut()
.remove_peer(&root_addr);
{
let identity = nodes[child_idx].node.identity().clone();
nodes[child_idx]
.node
.tree_state_mut()
.sign_declaration(&identity)
.unwrap();
let decl_mut = nodes[child_idx].node.tree_state_mut().my_declaration_mut();
sign_declaration(decl_mut, &identity).unwrap();
}
assert!(nodes[child_idx].node.tree_state().is_root());
@@ -1029,3 +1027,369 @@ async fn test_tree_announce_repushed_on_root_disagreement() {
cleanup_nodes(&mut nodes).await;
}
// ===== Direct handler characterization tests =====
//
// These drive `handle_tree_announce` directly to pin the individual
// classification and validation arms that the aggregate convergence suite
// only exercises indirectly: the four validation rejects (addr-mismatch,
// sig-fail, stale, unknown-peer) and the self-root / loop-drop /
// same-parent ancestry-update transitions. `run_tree_test(2, ..)` supplies
// two handshaked peers (so the sender's pubkey is known); the transition
// tests then force the receiver's local tree state into the precise shape
// each arm requires — the same `tree_state_mut()` seam
// `test_tree_announce_repushed_on_root_disagreement` uses above.
//
// `make_node_addr(0)` is the all-zero address, strictly smaller than every
// randomly-generated real node address, so it is used as a synthetic global
// root that keeps forced ancestries valid (advertised root = path minimum).
/// A TreeAnnounce whose declared `node_addr` does not match the sending peer
/// must be rejected (addr-mismatch) before any state mutation. The addr-match
/// gate precedes signature verification, so a harvested-but-unrelated
/// signature is enough to reach it.
#[tokio::test]
async fn test_tree_announce_rejects_addr_mismatch() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let a_addr = *nodes[0].node.node_addr();
let sig = nodes[0].node.identity().sign(&[0u8; 48]).to_byte_array();
let bogus = make_node_addr(200);
let declaration = ParentDeclaration::with_signature(bogus, bogus, 5, 2000, sig);
let announce = TreeAnnounce::new(
declaration,
TreeCoordinate::from_addrs(vec![bogus]).unwrap(),
);
let encoded = announce.encode().unwrap();
let mismatch_before = nodes[1].node.metrics().tree.addr_mismatch.get();
let accepted_before = nodes[1].node.metrics().tree.accepted.get();
let root_before = *nodes[1].node.tree_state().root();
// Sender is the known peer a_addr, but the declaration claims `bogus`.
nodes[1]
.node
.handle_tree_announce(&a_addr, &encoded[1..])
.await;
assert_eq!(
nodes[1].node.metrics().tree.addr_mismatch.get(),
mismatch_before + 1
);
assert_eq!(nodes[1].node.metrics().tree.accepted.get(), accepted_before);
assert_eq!(*nodes[1].node.tree_state().root(), root_before);
cleanup_nodes(&mut nodes).await;
}
/// A TreeAnnounce whose declared node_addr matches the sender but whose
/// signature does not verify under the sender's pubkey must be rejected
/// (sig-fail) without mutating tree state.
#[tokio::test]
async fn test_tree_announce_rejects_bad_signature() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let a_addr = *nodes[0].node.node_addr();
// A valid A-signature, but over a *different* declaration, so verifying it
// against the forged declaration's signing bytes fails.
let mut signed_other = ParentDeclaration::new(a_addr, a_addr, 99, 88);
sign_declaration(&mut signed_other, nodes[0].node.identity()).unwrap();
let sig = *signed_other.signature().unwrap();
let forged = ParentDeclaration::with_signature(a_addr, a_addr, 5, 2000, sig);
let announce = TreeAnnounce::new(forged, TreeCoordinate::from_addrs(vec![a_addr]).unwrap());
let encoded = announce.encode().unwrap();
let sig_failed_before = nodes[1].node.metrics().tree.sig_failed.get();
let accepted_before = nodes[1].node.metrics().tree.accepted.get();
nodes[1]
.node
.handle_tree_announce(&a_addr, &encoded[1..])
.await;
assert_eq!(
nodes[1].node.metrics().tree.sig_failed.get(),
sig_failed_before + 1
);
assert_eq!(nodes[1].node.metrics().tree.accepted.get(), accepted_before);
cleanup_nodes(&mut nodes).await;
}
/// Replaying a peer's already-known declaration verbatim (same sequence) is
/// not fresher, so `update_peer` reports no change and the announce is counted
/// stale and ignored rather than accepted.
#[tokio::test]
async fn test_tree_announce_stale_ignored() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let a_addr = *nodes[0].node.node_addr();
let stored_decl = nodes[1]
.node
.tree_state()
.peer_declaration(&a_addr)
.expect("node 1 should hold A's declaration after convergence")
.clone();
let stored_coords = nodes[1]
.node
.tree_state()
.peer_coords(&a_addr)
.expect("node 1 should hold A's coordinates after convergence")
.clone();
let announce = TreeAnnounce::new(stored_decl, stored_coords);
let encoded = announce.encode().unwrap();
let stale_before = nodes[1].node.metrics().tree.stale.get();
let accepted_before = nodes[1].node.metrics().tree.accepted.get();
nodes[1]
.node
.handle_tree_announce(&a_addr, &encoded[1..])
.await;
assert_eq!(nodes[1].node.metrics().tree.stale.get(), stale_before + 1);
assert_eq!(nodes[1].node.metrics().tree.accepted.get(), accepted_before);
cleanup_nodes(&mut nodes).await;
}
/// A TreeAnnounce from a node that is not a known peer must be rejected
/// (unknown-peer) at the pubkey-lookup gate.
#[tokio::test]
async fn test_tree_announce_rejects_unknown_peer() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let unknown = make_node_addr(201);
let sig = nodes[0].node.identity().sign(&[0u8; 48]).to_byte_array();
let declaration = ParentDeclaration::with_signature(unknown, unknown, 1, 1000, sig);
let announce = TreeAnnounce::new(
declaration,
TreeCoordinate::from_addrs(vec![unknown]).unwrap(),
);
let encoded = announce.encode().unwrap();
let unknown_before = nodes[1].node.metrics().tree.unknown_peer.get();
nodes[1]
.node
.handle_tree_announce(&unknown, &encoded[1..])
.await;
assert_eq!(
nodes[1].node.metrics().tree.unknown_peer.get(),
unknown_before + 1
);
cleanup_nodes(&mut nodes).await;
}
/// A non-root node whose only visible root is larger than its own address must
/// self-promote to root. P (the smaller-addr node) is forced into a child of
/// its larger peer L rooted at a synthetic smaller root; when L then announces
/// itself as its own (larger) root, P's smallest visible root becomes L, so P
/// promotes itself.
#[tokio::test]
async fn test_tree_announce_self_root_promotion() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
// P must be the smaller-addr node (the one that should win root); L larger.
let (p_idx, l_idx) = if nodes[0].node.node_addr() < nodes[1].node.node_addr() {
(0, 1)
} else {
(1, 0)
};
let p_addr = *nodes[p_idx].node.node_addr();
let l_addr = *nodes[l_idx].node.node_addr();
let fake_root = make_node_addr(0);
// Force P into a non-root child of L rooted at fake_root: coords
// [P, L, fake_root]. P > fake_root, so recompute keeps it attached.
{
let ts = nodes[p_idx].node.tree_state_mut();
ts.remove_peer(&l_addr);
ts.update_peer(
ParentDeclaration::new(l_addr, fake_root, 1, 1000),
TreeCoordinate::from_addrs(vec![l_addr, fake_root]).unwrap(),
);
ts.set_parent(l_addr, 1, 1000, 1000);
ts.recompute_coords();
}
{
let identity = nodes[p_idx].node.identity().clone();
let decl_mut = nodes[p_idx].node.tree_state_mut().my_declaration_mut();
sign_declaration(decl_mut, &identity).unwrap();
}
assert!(!nodes[p_idx].node.tree_state().is_root());
assert_eq!(*nodes[p_idx].node.tree_state().root(), fake_root);
// L announces a fresh self-root (root = L > P).
let mut decl = ParentDeclaration::self_root(l_addr, 5, 2000);
sign_declaration(&mut decl, nodes[l_idx].node.identity()).unwrap();
let announce = TreeAnnounce::new(decl, TreeCoordinate::from_addrs(vec![l_addr]).unwrap());
let encoded = announce.encode().unwrap();
let switched_before = nodes[p_idx].node.metrics().tree.parent_switched.get();
nodes[p_idx]
.node
.handle_tree_announce(&l_addr, &encoded[1..])
.await;
assert!(
nodes[p_idx].node.tree_state().is_root(),
"P should self-promote to root when its only visible root is larger"
);
assert_eq!(*nodes[p_idx].node.tree_state().root(), p_addr);
assert_eq!(
nodes[p_idx].node.metrics().tree.parent_switched.get(),
switched_before + 1
);
cleanup_nodes(&mut nodes).await;
}
/// When our current parent's freshly announced ancestry comes to contain us, a
/// loop has formed and the parent must be dropped. P is forced into a child of
/// Q rooted at a synthetic root; Q then announces an ancestry [Q, P, root] that
/// runs back through P, so P detects the loop and (having no alternative) falls
/// back to self-root.
#[tokio::test]
async fn test_tree_announce_loop_detection_drops_parent() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let (p_idx, q_idx) = (0, 1);
let p_addr = *nodes[p_idx].node.node_addr();
let q_addr = *nodes[q_idx].node.node_addr();
let root = make_node_addr(0);
// Force P into a child of Q rooted at `root`: coords [P, Q, root].
{
let ts = nodes[p_idx].node.tree_state_mut();
ts.remove_peer(&q_addr);
ts.update_peer(
ParentDeclaration::new(q_addr, root, 1, 1000),
TreeCoordinate::from_addrs(vec![q_addr, root]).unwrap(),
);
ts.set_parent(q_addr, 1, 1000, 1000);
ts.recompute_coords();
}
{
let identity = nodes[p_idx].node.identity().clone();
let decl_mut = nodes[p_idx].node.tree_state_mut().my_declaration_mut();
sign_declaration(decl_mut, &identity).unwrap();
}
assert!(!nodes[p_idx].node.tree_state().is_root());
assert_eq!(
nodes[p_idx].node.tree_state().my_declaration().parent_id(),
&q_addr
);
// Q announces an ancestry that now runs through P (declaring P as its own
// parent): [Q, P, root]. Adopting it would form a loop.
let mut decl = ParentDeclaration::new(q_addr, p_addr, 5, 2000);
sign_declaration(&mut decl, nodes[q_idx].node.identity()).unwrap();
let announce = TreeAnnounce::new(
decl,
TreeCoordinate::from_addrs(vec![q_addr, p_addr, root]).unwrap(),
);
let encoded = announce.encode().unwrap();
let loop_before = nodes[p_idx].node.metrics().tree.loop_detected.get();
nodes[p_idx]
.node
.handle_tree_announce(&q_addr, &encoded[1..])
.await;
assert_eq!(
nodes[p_idx].node.metrics().tree.loop_detected.get(),
loop_before + 1
);
// No alternative parent remains, so P falls back to self-root.
assert!(nodes[p_idx].node.tree_state().is_root());
assert_eq!(*nodes[p_idx].node.tree_state().root(), p_addr);
cleanup_nodes(&mut nodes).await;
}
/// When our parent keeps the same root and depth but swaps a mid-chain
/// ancestor, we keep the parent yet must recompute our coordinates and
/// re-announce (the `old_addrs != new_addrs` gate). P is forced into
/// [P, Q, mid, root]; Q re-announces [Q, new_mid, root], leaving root and depth
/// unchanged while replacing the interior ancestor.
#[tokio::test]
async fn test_tree_announce_same_parent_ancestry_update() {
let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
let (p_idx, q_idx) = (0, 1);
let q_addr = *nodes[q_idx].node.node_addr();
let root = make_node_addr(0);
let mid = make_node_addr(1);
let new_mid = make_node_addr(2);
// Force P into a child of Q rooted at `root` via `mid`: [P, Q, mid, root].
{
let ts = nodes[p_idx].node.tree_state_mut();
ts.remove_peer(&q_addr);
ts.update_peer(
ParentDeclaration::new(q_addr, mid, 1, 1000),
TreeCoordinate::from_addrs(vec![q_addr, mid, root]).unwrap(),
);
ts.set_parent(q_addr, 1, 1000, 1000);
ts.recompute_coords();
}
{
let identity = nodes[p_idx].node.identity().clone();
let decl_mut = nodes[p_idx].node.tree_state_mut().my_declaration_mut();
sign_declaration(decl_mut, &identity).unwrap();
}
let depth_before = nodes[p_idx].node.tree_state().my_coords().depth();
assert_eq!(
nodes[p_idx].node.tree_state().my_declaration().parent_id(),
&q_addr
);
// Q keeps root and depth but swaps its mid-chain ancestor mid -> new_mid.
let mut decl = ParentDeclaration::new(q_addr, new_mid, 5, 2000);
sign_declaration(&mut decl, nodes[q_idx].node.identity()).unwrap();
let announce = TreeAnnounce::new(
decl,
TreeCoordinate::from_addrs(vec![q_addr, new_mid, root]).unwrap(),
);
let encoded = announce.encode().unwrap();
let ancestry_before = nodes[p_idx].node.metrics().tree.ancestry_changed.get();
nodes[p_idx]
.node
.handle_tree_announce(&q_addr, &encoded[1..])
.await;
assert_eq!(
nodes[p_idx].node.metrics().tree.ancestry_changed.get(),
ancestry_before + 1
);
// Same parent, same depth, but the recomputed path now runs through new_mid.
assert_eq!(
nodes[p_idx].node.tree_state().my_declaration().parent_id(),
&q_addr
);
assert_eq!(
nodes[p_idx].node.tree_state().my_coords().depth(),
depth_before
);
let path: Vec<NodeAddr> = nodes[p_idx]
.node
.tree_state()
.my_coords()
.node_addrs()
.copied()
.collect();
assert!(
path.contains(&new_mid),
"recomputed path should include the swapped-in ancestor"
);
assert!(
!path.contains(&mid),
"old mid-chain ancestor should be gone from the recomputed path"
);
cleanup_nodes(&mut nodes).await;
}
src/node/tree.rs
+395 -252
View File
@@ -3,15 +3,63 @@
//! Handles building, sending, and receiving TreeAnnounce messages,
//! including periodic root refresh and rate-limited propagation.
use std::collections::HashMap;
use std::collections::{BTreeMap, BTreeSet};
use crate::NodeAddr;
use crate::protocol::TreeAnnounce;
use secp256k1::XOnlyPublicKey;
use secp256k1::schnorr::Signature;
use crate::proto::stp::{ParentDeclaration, Stp, TreeAnnounce, TreeDecision, TreeError};
use crate::{Identity, NodeAddr};
use super::reject::TreeReject;
use super::{Node, NodeError};
use tracing::{debug, info, trace, warn};
/// Sign a node's own tree declaration, writing the 64-byte signature back into
/// it. The key-crypto boundary (§6): `proto::stp` owns the declaration data and
/// the pure `signing_bytes()` serialization; the shell owns the `secp256k1`
/// sign (`Identity::sign` hashes with SHA-256 internally). Mirrors discovery's
/// shell-side proof signing.
pub(super) fn sign_declaration(
decl: &mut ParentDeclaration,
identity: &Identity,
) -> Result<(), TreeError> {
if identity.node_addr() != decl.node_addr() {
return Err(TreeError::InvalidSignature(*decl.node_addr()));
}
let signature = identity.sign(&decl.signing_bytes());
decl.set_signature(signature.to_byte_array());
Ok(())
}
/// Verify a peer's tree declaration signature against their pubkey. The shell
/// side of the key-crypto boundary (§6): runs the `sha2` hash + `secp256k1`
/// schnorr verification over the in-core declaration's `signing_bytes()`.
pub(super) fn verify_declaration(
decl: &ParentDeclaration,
pubkey: &XOnlyPublicKey,
) -> Result<(), TreeError> {
let sig_bytes = decl
.signature()
.ok_or(TreeError::InvalidSignature(*decl.node_addr()))?;
let signature = Signature::from_slice(sig_bytes)
.map_err(|_| TreeError::InvalidSignature(*decl.node_addr()))?;
let secp = secp256k1::Secp256k1::verification_only();
let hash = signing_hash(decl);
secp.verify_schnorr(&signature, &hash, pubkey)
.map_err(|_| TreeError::InvalidSignature(*decl.node_addr()))
}
/// Compute the SHA-256 hash of a declaration's signing bytes (shell side, §6).
fn signing_hash(decl: &ParentDeclaration) -> [u8; 32] {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(decl.signing_bytes());
hasher.finalize().into()
}
impl Node {
/// Build a TreeAnnounce from our current tree state.
fn build_tree_announce(&self) -> Result<TreeAnnounce, NodeError> {
@@ -163,7 +211,7 @@ impl Node {
return;
}
if let Err(e) = announce.declaration.verify(&pubkey) {
if let Err(e) = verify_declaration(&announce.declaration, &pubkey) {
self.metrics().tree.sig_failed.inc();
warn!(
from = %self.peer_display_name(from),
@@ -245,7 +293,7 @@ impl Node {
// receive path and is naturally bounded by the per-peer 500 ms
// tree-announce rate limiter, so it does not storm during normal
// convergence (it stops as soon as the peer adopts our root).
if *announce.ancestry.root_id() > *self.tree_state.root()
if Stp::should_echo(announce.ancestry.root_id(), self.tree_state.root())
&& let Err(e) = self.send_tree_announce_to_peer(from).await
{
debug!(
@@ -267,109 +315,130 @@ impl Node {
// Re-evaluate parent selection with current link costs.
// Exclude peers without MMP RTT data — they are not yet eligible
// as parent candidates (prevents oscillation from optimistic defaults).
let peer_costs: HashMap<NodeAddr, f64> = self
let peer_costs: BTreeMap<NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, peer)| peer.has_srtt())
.map(|(addr, peer)| (*addr, peer.link_cost()))
.collect();
if let Some(new_parent) = self.tree_state.evaluate_parent(&peer_costs) {
let new_seq = self.tree_state.my_declaration().sequence() + 1;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// No peers are excluded from parent candidacy on this branch; the
// non-full/leaf skip is a next-only shell refinement.
let skip: BTreeSet<NodeAddr> = BTreeSet::new();
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp);
// recompute_coords may demote to self_root if the new path would be
// invalid; sign AFTER recompute so the signature covers the final
// declaration.
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
warn!(error = %e, "Failed to sign declaration after parent switch");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
// Monotonic ms for the flap-dampening / hold-down timers (distinct from
// the wall-clock `now_ms` above used for the peer's tree position). Read
// once and threaded into classify + the state mutators.
let mono_now_ms = crate::mmp::mono_ms();
match Stp::classify_announce(&self.tree_state, *from, &peer_costs, &skip, mono_now_ms) {
TreeDecision::Switch {
new_parent,
new_seq,
} => {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
let flap_dampened =
self.tree_state
.set_parent(new_parent, new_seq, timestamp, mono_now_ms);
// recompute_coords may demote to self_root if the new path would be
// invalid; sign AFTER recompute so the signature covers the final
// declaration.
self.tree_state.recompute_coords();
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after parent switch");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_parent = %self.peer_display_name(&new_parent),
new_seq = new_seq,
new_root = %self.tree_state.root(),
depth = self.tree_state.my_coords().depth(),
"Parent switched, invalidated downstream coord cache entries, announcing to all peers"
);
if flap_dampened {
self.metrics().tree.flap_dampened.inc();
warn!("Flap dampening engaged: excessive parent switches detected");
}
self.send_tree_announce_to_all().await;
// Tree structure changed — trigger bloom filter exchange with all peers
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_parent = %self.peer_display_name(&new_parent),
new_seq = new_seq,
new_root = %self.tree_state.root(),
depth = self.tree_state.my_coords().depth(),
"Parent switched, invalidated downstream coord cache entries, announcing to all peers"
);
if flap_dampened {
self.metrics().tree.flap_dampened.inc();
warn!("Flap dampening engaged: excessive parent switches detected");
TreeDecision::SelfRoot => {
// Self is the smallest visible NodeAddr — promote to root rather
// than continuing to advertise a stale ancestry rooted elsewhere.
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state.become_root(timestamp);
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign self-root declaration");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
self.coord_cache
.invalidate_other_roots(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_root = %self.tree_state.root(),
"Self-promoted to root: smallest visible NodeAddr"
);
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
}
self.send_tree_announce_to_all().await;
// Tree structure changed — trigger bloom filter exchange with all peers
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
} else if !self.tree_state.is_root() && self.tree_state.should_be_root() {
// Self is the smallest visible NodeAddr — promote to root rather
// than continuing to advertise a stale ancestry rooted elsewhere.
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state.become_root();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
warn!(error = %e, "Failed to sign self-root declaration");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
self.coord_cache
.invalidate_other_roots(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_root = %self.tree_state.root(),
"Self-promoted to root: smallest visible NodeAddr"
);
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
} else if !self.tree_state.is_root()
&& *self.tree_state.my_declaration().parent_id() == *from
{
// Check for loop: if parent's ancestry now contains us, drop parent
if let Some(parent_coords) = self.tree_state.peer_coords(from)
&& parent_coords.contains(self.identity().node_addr())
{
TreeDecision::LoopDrop => {
self.metrics().tree.loop_detected.inc();
warn!(
parent = %self.peer_display_name(from),
"Parent ancestry contains us — loop detected, dropping parent"
);
let peer_costs: HashMap<NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, peer)| peer.has_srtt())
.map(|(addr, peer)| (*addr, peer.link_cost()))
.collect();
if self.tree_state.handle_parent_lost(&peer_costs) {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
if self
.tree_state
.handle_parent_lost(&peer_costs, timestamp, mono_now_ms)
{
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after loop detection");
self.metrics()
.tree
@@ -385,74 +454,82 @@ impl Node {
self.reset_discovery_backoff();
self.send_tree_announce_to_all().await;
}
return;
}
TreeDecision::AncestryUpdate { parent, new_seq } => {
// Our parent's ancestry changed but we're keeping the same parent.
// Recompute our own coordinates (which derive from parent's ancestry)
// and re-announce so downstream nodes stay current.
//
// Compare the full address path (not just root + depth) so that a
// mid-chain ancestor swap also triggers re-announce. A reroute that
// replaces an interior ancestor without changing the root or the
// path length leaves both `root` and `depth` unchanged but still
// alters our coords; downstream peers must learn the new path or
// they will route into a phantom intermediate that no longer
// exists on our parent's tree.
let old_root = *self.tree_state.root();
let old_depth = self.tree_state.my_coords().depth();
let old_addrs: Vec<NodeAddr> =
self.tree_state.my_coords().node_addrs().copied().collect();
// Our parent's ancestry changed but we're keeping the same parent.
// Recompute our own coordinates (which derive from parent's ancestry)
// and re-announce so downstream nodes stay current.
//
// Compare the full address path (not just root + depth) so that a
// mid-chain ancestor swap also triggers re-announce. A reroute that
// replaces an interior ancestor without changing the root or the
// path length leaves both `root` and `depth` unchanged but still
// alters our coords; downstream peers must learn the new path or
// they will route into a phantom intermediate that no longer
// exists on our parent's tree.
let old_root = *self.tree_state.root();
let old_depth = self.tree_state.my_coords().depth();
let old_addrs: Vec<NodeAddr> =
self.tree_state.my_coords().node_addrs().copied().collect();
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let new_seq = self.tree_state.my_declaration().sequence() + 1;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state
.set_parent(parent, new_seq, timestamp, mono_now_ms);
self.tree_state.recompute_coords();
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after parent update");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state.set_parent(*from, new_seq, timestamp);
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
warn!(error = %e, "Failed to sign declaration after parent update");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
let new_addrs: Vec<NodeAddr> =
self.tree_state.my_coords().node_addrs().copied().collect();
if old_addrs != new_addrs {
self.metrics().tree.ancestry_changed.inc();
info!(
parent = %self.peer_display_name(from),
old_root = %old_root,
new_root = %self.tree_state.root(),
old_depth = old_depth,
new_depth = self.tree_state.my_coords().depth(),
"Parent ancestry changed, re-announcing"
);
self.send_tree_announce_to_all().await;
// Bloom contents do not depend on path structure, only on
// identity sets. Our parent_id is unchanged in this branch,
// so our tree-peer set is unchanged and our outgoing filter
// content is unchanged. Use mark_changed_peers, which
// checks for actual content delta against last_sent_filters,
// instead of mark_all_updates_needed, which marks
// unconditionally regardless of whether content 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);
}
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
let new_addrs: Vec<NodeAddr> =
self.tree_state.my_coords().node_addrs().copied().collect();
if old_addrs != new_addrs {
self.metrics().tree.ancestry_changed.inc();
info!(
parent = %self.peer_display_name(from),
old_root = %old_root,
new_root = %self.tree_state.root(),
old_depth = old_depth,
new_depth = self.tree_state.my_coords().depth(),
"Parent ancestry changed, re-announcing"
);
self.send_tree_announce_to_all().await;
// Bloom contents do not depend on path structure, only on
// identity sets. Our parent_id is unchanged in this branch,
// so our tree-peer set is unchanged and our outgoing filter
// content is unchanged. Use mark_changed_peers, which
// checks for actual content delta against last_sent_filters,
// instead of mark_all_updates_needed, which marks
// unconditionally regardless of whether content 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);
TreeDecision::NoChange => {}
// classify_announce never yields PeriodicRebroadcast (the periodic
// path's no-change tail) nor ParentLost (the removal drive's outcome).
TreeDecision::PeriodicRebroadcast | TreeDecision::ParentLost => {
unreachable!("classify_announce yields neither PeriodicRebroadcast nor ParentLost")
}
}
}
@@ -493,99 +570,134 @@ impl Node {
self.last_parent_reeval = Some(now);
let peer_costs: HashMap<NodeAddr, f64> = self
let peer_costs: BTreeMap<NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, peer)| peer.has_srtt())
.map(|(addr, peer)| (*addr, peer.link_cost()))
.collect();
// No peers are excluded from parent candidacy on this branch; the
// non-full/leaf skip is a next-only shell refinement.
let skip: BTreeSet<NodeAddr> = BTreeSet::new();
if let Some(new_parent) = self.tree_state.evaluate_parent(&peer_costs) {
let new_seq = self.tree_state.my_declaration().sequence() + 1;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Monotonic ms for the flap-dampening / hold-down timers, read once and
// threaded into classify + the state mutators.
let mono_now_ms = crate::mmp::mono_ms();
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp);
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
warn!(error = %e, "Failed to sign declaration after periodic parent re-eval");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
match Stp::classify_periodic(&self.tree_state, &peer_costs, &skip, mono_now_ms) {
TreeDecision::Switch {
new_parent,
new_seq,
} => {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
let flap_dampened =
self.tree_state
.set_parent(new_parent, new_seq, timestamp, mono_now_ms);
self.tree_state.recompute_coords();
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after periodic parent re-eval");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_parent = %self.peer_display_name(&new_parent),
new_seq = new_seq,
new_root = %self.tree_state.root(),
depth = self.tree_state.my_coords().depth(),
trigger = "periodic",
"Parent switched via periodic cost re-evaluation"
);
if flap_dampened {
self.metrics().tree.flap_dampened.inc();
warn!("Flap dampening engaged: excessive parent switches detected");
}
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_parent = %self.peer_display_name(&new_parent),
new_seq = new_seq,
new_root = %self.tree_state.root(),
depth = self.tree_state.my_coords().depth(),
trigger = "periodic",
"Parent switched via periodic cost re-evaluation"
);
if flap_dampened {
self.metrics().tree.flap_dampened.inc();
warn!("Flap dampening engaged: excessive parent switches detected");
TreeDecision::SelfRoot => {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state.become_root(timestamp);
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign self-root declaration in periodic reeval");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
self.coord_cache
.invalidate_other_roots(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_root = %self.tree_state.root(),
trigger = "periodic",
"Self-promoted to root in periodic reeval: smallest visible NodeAddr"
);
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
}
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
} else if !self.tree_state.is_root() && self.tree_state.should_be_root() {
// Clone identity up front to avoid a split borrow against the
// &mut self.tree_state / &mut self.coord_cache calls below (cold path).
let our_identity = self.identity().clone();
self.tree_state.become_root();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
warn!(error = %e, "Failed to sign self-root declaration in periodic reeval");
self.metrics()
.tree
.record_reject(TreeReject::OutboundSignFailed);
return;
TreeDecision::PeriodicRebroadcast => {
// Periodic re-broadcast on no-change: makes TreeAnnounce
// distribution eventually-consistent. Receivers coalesce
// by sequence via ParentDeclaration::is_fresher_than and
// short-circuit at the `if !updated` gate in
// handle_tree_announce; the per-peer 500 ms rate-limiter
// never blocks at this 60 s cadence. Closes the cross-init
// in-flight loss recovery gap where the swap window can
// strand one side's announce on a session-index the other
// side cannot decrypt.
trace!(
seq = self.tree_state.my_declaration().sequence(),
root = %self.tree_state.root(),
"Periodic TreeAnnounce re-broadcast (no state change)"
);
self.send_tree_announce_to_all().await;
}
// classify_periodic never yields these: a periodic tick has no
// announcing peer, so the same-parent loop-drop / ancestry-update
// arms cannot arise, the no-change tail is PeriodicRebroadcast, and
// ParentLost is the removal drive's outcome.
TreeDecision::LoopDrop
| TreeDecision::AncestryUpdate { .. }
| TreeDecision::ParentLost
| TreeDecision::NoChange => {
unreachable!(
"classify_periodic yields only Switch / SelfRoot / PeriodicRebroadcast"
)
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
self.coord_cache
.invalidate_other_roots(our_identity.node_addr());
self.reset_discovery_backoff();
self.metrics().tree.parent_switched.inc();
self.metrics().tree.parent_switches.inc();
info!(
new_root = %self.tree_state.root(),
trigger = "periodic",
"Self-promoted to root in periodic reeval: smallest visible NodeAddr"
);
self.send_tree_announce_to_all().await;
let all_peers: Vec<NodeAddr> = self.peers.keys().copied().collect();
self.bloom_state.mark_all_updates_needed(all_peers);
} else {
// Periodic re-broadcast on no-change: makes TreeAnnounce
// distribution eventually-consistent. Receivers coalesce
// by sequence via ParentDeclaration::is_fresher_than and
// short-circuit at the `if !updated` gate in
// handle_tree_announce; the per-peer 500 ms rate-limiter
// never blocks at this 60 s cadence. Closes the cross-init
// in-flight loss recovery gap where the swap window can
// strand one side's announce on a session-index the other
// side cannot decrypt.
trace!(
seq = self.tree_state.my_declaration().sequence(),
root = %self.tree_state.root(),
"Periodic TreeAnnounce re-broadcast (no state change)"
);
self.send_tree_announce_to_all().await;
}
}
@@ -601,20 +713,50 @@ impl Node {
self.tree_state.remove_peer(node_addr);
if was_parent {
self.metrics().tree.parent_losses.inc();
let peer_costs: HashMap<NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, peer)| peer.has_srtt())
.map(|(addr, peer)| (*addr, peer.link_cost()))
.collect();
let changed = self.tree_state.handle_parent_lost(&peer_costs);
if changed {
if !was_parent {
return false;
}
// The removed peer was our parent. `parent_losses` counts the loss
// itself (independent of whether we recover), so it is stamped here —
// before the recovery mutation — exactly as before.
self.metrics().tree.parent_losses.inc();
let peer_costs: BTreeMap<NodeAddr, f64> = self
.peers
.iter()
.filter(|(_, peer)| peer.has_srtt())
.map(|(addr, peer)| (*addr, peer.link_cost()))
.collect();
// Wall-clock seconds stamped onto the new declaration; monotonic ms for
// the parent re-evaluation's flap timers.
let now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mono_now_ms = crate::mmp::mono_ms();
// Removal is not a pure classify: `handle_parent_lost` is a &mut mutator
// whose returned `changed` bool IS the decision. Drive it and map the
// outcome onto the TreeDecision vocabulary.
let decision = if self
.tree_state
.handle_parent_lost(&peer_costs, now_secs, mono_now_ms)
{
TreeDecision::ParentLost
} else {
TreeDecision::NoChange
};
match decision {
TreeDecision::ParentLost => {
// Re-sign the new declaration. Clone identity to avoid a split
// borrow against the &mut self.tree_state receiver (cold path).
let our_identity = self.identity().clone();
if let Err(e) = self.tree_state.sign_declaration(&our_identity) {
if let Err(e) =
sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
{
warn!(error = %e, "Failed to sign declaration after parent loss");
self.metrics()
.tree
@@ -635,10 +777,11 @@ impl Node {
is_root = self.tree_state.is_root(),
"Tree state updated after parent loss"
);
true
}
changed
} else {
false
TreeDecision::NoChange => false,
// The removal drive constructs only ParentLost / NoChange above.
_ => unreachable!("removal drive yields only ParentLost / NoChange"),
}
}
}