diff --git a/src/lib.rs b/src/lib.rs index 5e5285b..9b21602 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,12 @@ //! A distributed, decentralized network routing protocol for mesh nodes //! connecting over arbitrary transports. +// Name the `alloc` crate directly so the sans-IO protocol cores can spell their +// heap-type imports in `no_std`-forward form (`alloc::sync::Arc`, +// `alloc::collections::BTreeMap`). The crate remains `std`; this only reduces the +// distance to extracting the pure cores into a `no_std` crate later. +extern crate alloc; + pub mod bloom; pub mod cache; pub mod config; @@ -16,7 +22,10 @@ pub mod node; pub mod noise; pub mod peer; pub mod perf_profile; +pub(crate) mod proto; pub mod protocol; +#[cfg(test)] +pub(crate) mod testutil; pub mod transport; pub mod tree; pub mod upper; @@ -52,11 +61,14 @@ pub use transport::{ // Re-export protocol types pub use protocol::{ - CoordsRequired, FilterAnnounce, HandshakeMessageType, LinkMessageType, LookupRequest, - LookupResponse, PathBroken, ProtocolError, SessionAck, SessionDatagram, SessionFlags, - SessionMessageType, SessionSetup, TreeAnnounce, + CoordsRequired, FilterAnnounce, HandshakeMessageType, LinkMessageType, PathBroken, + ProtocolError, SessionAck, SessionDatagram, SessionFlags, SessionMessageType, SessionSetup, + TreeAnnounce, }; +// Re-export discovery wire types (relocated from protocol:: to proto::discovery) +pub use proto::discovery::{LookupRequest, LookupResponse}; + // Re-export cache types pub use cache::{CacheEntry, CacheError, CacheStats, CoordCache}; diff --git a/src/node/discovery_rate_limit.rs b/src/node/discovery_rate_limit.rs deleted file mode 100644 index ba1106d..0000000 --- a/src/node/discovery_rate_limit.rs +++ /dev/null @@ -1,376 +0,0 @@ -//! Discovery protocol rate limiting and backoff. -//! -//! Two complementary mechanisms: -//! -//! - **`DiscoveryBackoff`** (originator-side, optional): Exponential -//! suppression of fresh lookups after the per-attempt sequence in -//! `node.discovery.attempt_timeouts_secs` has been exhausted. -//! **Disabled by default** (base/cap = 0); the per-attempt sequence -//! is the only retry pacing in the standard configuration. Reset on -//! topology changes (parent change, new peer, first RTT, reconnection). -//! -//! - **`DiscoveryForwardRateLimiter`** (transit-side): Per-target minimum -//! interval for forwarded requests. Defense-in-depth against misbehaving -//! nodes generating fresh request_ids at high rate. - -use crate::NodeAddr; -use std::collections::HashMap; -use std::time::{Duration, Instant}; - -// ============================================================================ -// Originator-side: Discovery Backoff -// ============================================================================ - -/// Default base backoff after first lookup failure. `0` = disabled. -const DEFAULT_BACKOFF_BASE_SECS: u64 = 0; - -/// Default maximum backoff cap. `0` = disabled. -const DEFAULT_BACKOFF_MAX_SECS: u64 = 0; - -/// Backoff multiplier per consecutive failure. -const BACKOFF_MULTIPLIER: u64 = 2; - -/// Exponential backoff for failed discovery lookups. -/// -/// Tracks targets whose lookups have timed out and suppresses -/// re-initiation with increasing delays. Cleared on topology changes. -pub struct DiscoveryBackoff { - /// Maps target → (suppress_until, consecutive_failures). - entries: HashMap, - /// Base backoff duration (first failure). - base: Duration, - /// Maximum backoff cap. - max: Duration, -} - -struct BackoffEntry { - /// Don't re-initiate until this instant. - suppress_until: Instant, - /// Consecutive failures (drives exponential backoff). - failures: u32, -} - -impl DiscoveryBackoff { - /// Create with default parameters (disabled — base/cap = 0). - pub fn new() -> Self { - Self::with_params(DEFAULT_BACKOFF_BASE_SECS, DEFAULT_BACKOFF_MAX_SECS) - } - - /// Create with custom base and max backoff in seconds. - pub fn with_params(base_secs: u64, max_secs: u64) -> Self { - Self { - entries: HashMap::new(), - base: Duration::from_secs(base_secs), - max: Duration::from_secs(max_secs), - } - } - - /// Check if a lookup for this target is suppressed. - /// - /// Returns true if the target is in backoff and should not be - /// looked up yet. - pub fn is_suppressed(&self, target: &NodeAddr) -> bool { - if let Some(entry) = self.entries.get(target) { - Instant::now() < entry.suppress_until - } else { - false - } - } - - /// Record a lookup failure (timeout) for a target. - /// - /// Increments the failure count and sets the next suppression - /// window using exponential backoff. - pub fn record_failure(&mut self, target: &NodeAddr) { - let now = Instant::now(); - let failures = self.entries.get(target).map_or(0, |e| e.failures) + 1; - - let backoff_secs = self - .base - .as_secs() - .saturating_mul(BACKOFF_MULTIPLIER.saturating_pow(failures.saturating_sub(1))); - let backoff = Duration::from_secs(backoff_secs.min(self.max.as_secs())); - - self.entries.insert( - *target, - BackoffEntry { - suppress_until: now + backoff, - failures, - }, - ); - } - - /// Record a successful lookup — remove backoff for this target. - pub fn record_success(&mut self, target: &NodeAddr) { - self.entries.remove(target); - } - - /// Clear all backoff entries. - /// - /// Called on topology changes that might make previously-unreachable - /// targets reachable (parent change, new peer, first RTT, reconnection). - pub fn reset_all(&mut self) { - self.entries.clear(); - } - - /// Whether any entries exist. - pub fn is_empty(&self) -> bool { - self.entries.is_empty() - } - - /// Current number of entries. - pub fn entry_count(&self) -> usize { - self.entries.len() - } - - /// Get the failure count for a target (for logging). - pub fn failure_count(&self, target: &NodeAddr) -> u32 { - self.entries.get(target).map_or(0, |e| e.failures) - } - - #[cfg(test)] - pub fn len(&self) -> usize { - self.entries.len() - } -} - -impl Default for DiscoveryBackoff { - fn default() -> Self { - Self::new() - } -} - -// ============================================================================ -// Transit-side: Discovery Forward Rate Limiter -// ============================================================================ - -/// Default minimum interval between forwarded lookups for the same target. -const DEFAULT_FORWARD_MIN_INTERVAL: Duration = Duration::from_secs(2); - -/// Maximum age of entries before cleanup. -const FORWARD_MAX_AGE: Duration = Duration::from_secs(60); - -/// Rate limiter for forwarded discovery requests. -/// -/// Tracks the last time a LookupRequest was forwarded for each target -/// and enforces a minimum interval to prevent floods from misbehaving -/// nodes generating fresh request_ids. -pub struct DiscoveryForwardRateLimiter { - last_forwarded: HashMap, - min_interval: Duration, - max_age: Duration, -} - -impl DiscoveryForwardRateLimiter { - /// Create with default parameters (2s interval). - pub fn new() -> Self { - Self { - last_forwarded: HashMap::new(), - min_interval: DEFAULT_FORWARD_MIN_INTERVAL, - max_age: FORWARD_MAX_AGE, - } - } - - /// Create with a custom minimum interval. - pub fn with_interval(min_interval: Duration) -> Self { - Self { - last_forwarded: HashMap::new(), - min_interval, - max_age: FORWARD_MAX_AGE, - } - } - - /// Check if we should forward a lookup for this target. - /// - /// Returns true if enough time has passed since the last forward - /// for this target. Updates internal state when returning true. - pub fn should_forward(&mut self, target: &NodeAddr) -> bool { - let now = Instant::now(); - - if let Some(&last) = self.last_forwarded.get(target) - && now.duration_since(last) < self.min_interval - { - return false; - } - - self.last_forwarded.insert(*target, now); - self.cleanup(now); - true - } - - /// Replace the minimum interval (e.g., set to zero to disable). - #[cfg(test)] - pub fn set_interval(&mut self, interval: Duration) { - self.min_interval = interval; - } - - /// Remove entries older than max_age. - fn cleanup(&mut self, now: Instant) { - self.last_forwarded - .retain(|_, &mut last| now.duration_since(last) < self.max_age); - } - - #[cfg(test)] - pub fn len(&self) -> usize { - self.last_forwarded.len() - } -} - -impl Default for DiscoveryForwardRateLimiter { - fn default() -> Self { - Self::new() - } -} - -// ============================================================================ -// Tests -// ============================================================================ - -#[cfg(test)] -mod tests { - use super::*; - use std::thread; - - fn addr(val: u8) -> NodeAddr { - let mut bytes = [0u8; 16]; - bytes[0] = val; - NodeAddr::from_bytes(bytes) - } - - // --- DiscoveryBackoff tests --- - - #[test] - fn test_backoff_not_suppressed_initially() { - let backoff = DiscoveryBackoff::new(); - assert!(!backoff.is_suppressed(&addr(1))); - } - - #[test] - fn test_backoff_suppressed_after_failure() { - // Backoff is opt-in; exercise the suppression path with explicit params. - let mut backoff = DiscoveryBackoff::with_params(30, 300); - backoff.record_failure(&addr(1)); - assert!(backoff.is_suppressed(&addr(1))); - // Different target not affected - assert!(!backoff.is_suppressed(&addr(2))); - } - - #[test] - fn test_backoff_cleared_on_success() { - let mut backoff = DiscoveryBackoff::with_params(30, 300); - backoff.record_failure(&addr(1)); - assert!(backoff.is_suppressed(&addr(1))); - - backoff.record_success(&addr(1)); - assert!(!backoff.is_suppressed(&addr(1))); - } - - #[test] - fn test_backoff_reset_all() { - let mut backoff = DiscoveryBackoff::new(); - backoff.record_failure(&addr(1)); - backoff.record_failure(&addr(2)); - assert_eq!(backoff.len(), 2); - - backoff.reset_all(); - assert_eq!(backoff.len(), 0); - assert!(!backoff.is_suppressed(&addr(1))); - } - - #[test] - fn test_backoff_exponential() { - let mut backoff = DiscoveryBackoff::with_params(1, 300); - - // First failure: 1s backoff - backoff.record_failure(&addr(1)); - assert_eq!(backoff.failure_count(&addr(1)), 1); - - // Second failure: 2s backoff - backoff.record_failure(&addr(1)); - assert_eq!(backoff.failure_count(&addr(1)), 2); - - // Third failure: 4s backoff - backoff.record_failure(&addr(1)); - assert_eq!(backoff.failure_count(&addr(1)), 3); - } - - #[test] - fn test_backoff_expires() { - let mut backoff = DiscoveryBackoff::with_params(0, 0); - backoff.record_failure(&addr(1)); - // With 0s backoff, should not be suppressed - assert!(!backoff.is_suppressed(&addr(1))); - } - - #[test] - fn test_backoff_capped() { - let mut backoff = DiscoveryBackoff::with_params(1, 10); - - // Record many failures - for _ in 0..20 { - backoff.record_failure(&addr(1)); - } - - // Backoff should be capped at max (10s), not overflow - let entry = backoff.entries.get(&addr(1)).unwrap(); - let remaining = entry.suppress_until.duration_since(Instant::now()); - assert!(remaining <= Duration::from_secs(11)); - } - - // --- DiscoveryForwardRateLimiter tests --- - - #[test] - fn test_forward_first_allowed() { - let mut limiter = DiscoveryForwardRateLimiter::new(); - assert!(limiter.should_forward(&addr(1))); - } - - #[test] - fn test_forward_rapid_rate_limited() { - let mut limiter = DiscoveryForwardRateLimiter::new(); - assert!(limiter.should_forward(&addr(1))); - assert!(!limiter.should_forward(&addr(1))); - assert!(!limiter.should_forward(&addr(1))); - } - - #[test] - fn test_forward_different_targets_independent() { - let mut limiter = DiscoveryForwardRateLimiter::new(); - assert!(limiter.should_forward(&addr(1))); - assert!(limiter.should_forward(&addr(2))); - assert!(!limiter.should_forward(&addr(1))); - assert!(!limiter.should_forward(&addr(2))); - } - - #[test] - fn test_forward_allowed_after_interval() { - let mut limiter = DiscoveryForwardRateLimiter::with_interval(Duration::from_millis(100)); - assert!(limiter.should_forward(&addr(1))); - - thread::sleep(Duration::from_millis(110)); - - assert!(limiter.should_forward(&addr(1))); - } - - #[test] - fn test_forward_cleanup_removes_old() { - let mut limiter = DiscoveryForwardRateLimiter::new(); - assert!(limiter.should_forward(&addr(1))); - assert!(limiter.should_forward(&addr(2))); - assert_eq!(limiter.len(), 2); - - let future = Instant::now() + Duration::from_secs(61); - limiter.cleanup(future); - assert_eq!(limiter.len(), 0); - } - - #[test] - fn test_forward_cleanup_preserves_recent() { - let mut limiter = DiscoveryForwardRateLimiter::new(); - assert!(limiter.should_forward(&addr(1))); - assert_eq!(limiter.len(), 1); - - limiter.cleanup(Instant::now()); - assert_eq!(limiter.len(), 1); - } -} diff --git a/src/node/handlers/discovery.rs b/src/node/handlers/discovery.rs index 76c77cd..46c389c 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -5,15 +5,56 @@ //! bloom filter contains the target. TTL and request_id dedup provide //! safety bounds. +use crate::node::Node; use crate::node::reject::DiscoveryReject; -use crate::node::{Node, RecentRequest}; -use crate::protocol::{LookupRequest, LookupResponse}; +use crate::proto::discovery::{DiscoveryAction, LookupRequest, LookupResponse}; use crate::transport::{TransportAddr, TransportId}; use crate::{NodeAddr, PeerIdentity}; use tracing::{debug, info, trace, warn}; const MAX_RECENT_DISCOVERY_REQUESTS: usize = 4096; +/// Shell adapter exposing the live routing tables to the sans-IO discovery +/// core's `RoutingView` read seam. Lives in `node` so it can read `Node`'s +/// private `peers` map and call the crate-private tree/bloom predicates. +/// +/// Holding `&Node` whole is fine for the forward path because it does not +/// also need `&mut self.discovery` concurrently. A later commit whose core +/// step needs `&mut discovery` while reading routing state should narrow this +/// to borrow only `peers` + `tree_state` instead of the whole node. +struct NodeRoutingView<'a> { + node: &'a Node, +} + +impl crate::proto::discovery::RoutingView for NodeRoutingView<'_> { + fn is_tree_peer(&self, addr: &NodeAddr) -> bool { + self.node.is_tree_peer(addr) + } + fn peers_reaching(&self, target: &NodeAddr) -> Vec { + self.node + .peers + .iter() + .filter(|(_, peer)| peer.may_reach(target)) + .map(|(addr, _)| *addr) + .collect() + } + fn node_is_leaf(&self) -> bool { + self.node.node_profile() == crate::protocol::NodeProfile::Leaf + } + fn peer_is_full(&self, addr: &NodeAddr) -> bool { + self.node + .peers + .get(addr) + .is_some_and(|peer| peer.peer_profile() == crate::protocol::NodeProfile::Full) + } + fn peer_meets_mtu(&self, addr: &NodeAddr, min_mtu: u16) -> bool { + self.node + .peers + .get(addr) + .is_some_and(|peer| self.node.peer_meets_mtu(peer, min_mtu)) + } +} + impl Node { /// Handle an incoming LookupRequest from a peer. /// @@ -39,80 +80,71 @@ impl Node { }; let now_ms = Self::now_ms(); - self.purge_expired_requests(now_ms); - - // Dedup: drop if we've already seen this request_id. - // Also serves as loop protection — tree routing is loop-free, - // but request_id dedup catches edge cases during tree restructuring. - if self.recent_requests.contains_key(&request.request_id) { - self.metrics() - .discovery - .record_reject(DiscoveryReject::ReqDuplicate); - debug!( - request_id = request.request_id, - from = %self.peer_display_name(from), - "Duplicate LookupRequest, dropping" - ); - return; - } - - if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS { - self.metrics() - .discovery - .record_reject(DiscoveryReject::ReqDedupCacheFull); - debug!( - request_id = request.request_id, - from = %self.peer_display_name(from), - recent_requests = self.recent_requests.len(), - max_recent_requests = MAX_RECENT_DISCOVERY_REQUESTS, - "Discovery request dedup cache full, dropping LookupRequest" - ); - return; - } - - // Record for reverse-path forwarding and dedup - self.recent_requests - .insert(request.request_id, RecentRequest::new(*from, now_ms)); - - // Are we the target? - if request.target == *self.node_addr() { - self.metrics().discovery.req_target_is_us.inc(); - debug!( - request_id = request.request_id, - origin = %self.peer_display_name(&request.origin), - "We are the lookup target, generating response" - ); - self.send_lookup_response(&request).await; - return; - } - - // Forward if TTL permits - if request.can_forward() { - // Transit-side rate limit: collapse rapid-fire lookups for the - // same target from misbehaving nodes generating fresh request_ids. - if !self - .discovery_forward_limiter - .should_forward(&request.target) - { + let recent_expiry_ms = self.config().node.discovery.recent_expiry_secs * 1000; + let my_addr = *self.node_addr(); + use crate::proto::discovery::RequestOutcome; + match crate::proto::discovery::classify_request( + &mut self.discovery, + &request, + from, + &my_addr, + now_ms, + recent_expiry_ms, + MAX_RECENT_DISCOVERY_REQUESTS, + ) { + RequestOutcome::Duplicate => { + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqDuplicate); + debug!( + request_id = request.request_id, + from = %self.peer_display_name(from), + "Duplicate LookupRequest, dropping" + ); + } + RequestOutcome::DedupCacheFull { len } => { + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqDedupCacheFull); + debug!( + request_id = request.request_id, + from = %self.peer_display_name(from), + recent_requests = len, + max_recent_requests = MAX_RECENT_DISCOVERY_REQUESTS, + "Discovery request dedup cache full, dropping LookupRequest" + ); + } + RequestOutcome::RespondAsTarget => { + self.metrics().discovery.req_target_is_us.inc(); + debug!( + request_id = request.request_id, + origin = %self.peer_display_name(&request.origin), + "We are the lookup target, generating response" + ); + self.send_lookup_response(&request).await; + } + RequestOutcome::Forward => { + self.metrics().discovery.req_forwarded.inc(); + self.forward_lookup_request(request).await; + } + RequestOutcome::ForwardRateLimited => { self.metrics().discovery.req_forward_rate_limited.inc(); debug!( request_id = request.request_id, target = %self.peer_display_name(&request.target), "Forward rate limited, suppressing LookupRequest" ); - return; } - self.metrics().discovery.req_forwarded.inc(); - self.forward_lookup_request(request).await; - } else { - self.metrics() - .discovery - .record_reject(DiscoveryReject::ReqTtlExhausted); - debug!( - request_id = request.request_id, - target = %self.peer_display_name(&request.target), - "LookupRequest TTL exhausted" - ); + RequestOutcome::TtlExhausted => { + self.metrics() + .discovery + .record_reject(DiscoveryReject::ReqTtlExhausted); + debug!( + request_id = request.request_id, + target = %self.peer_display_name(&request.target), + "LookupRequest TTL exhausted" + ); + } } } @@ -144,150 +176,186 @@ impl Node { let now_ms = Self::now_ms(); // Check if we forwarded this request (transit node) or originated it - if let Some(recent) = self.recent_requests.get_mut(&response.request_id) { - // Already forwarded a response for this request — drop to - // prevent response routing loops. - if recent.response_forwarded { + match crate::proto::discovery::classify_response(&mut self.discovery, response.request_id) { + crate::proto::discovery::ResponseRoute::AlreadyForwarded => { + // Already forwarded a response for this request — drop to + // prevent response routing loops. debug!( request_id = response.request_id, target = %self.peer_display_name(&response.target), "Response already forwarded for this request, dropping" ); - return; } - recent.response_forwarded = true; + crate::proto::discovery::ResponseRoute::Transit { from_peer } => { + // Transit node: reverse-path forward + self.metrics().discovery.resp_forwarded.inc(); - // Transit node: reverse-path forward - let from_peer = recent.from_peer; - self.metrics().discovery.resp_forwarded.inc(); + // Apply path_mtu min() from the outgoing link's transport MTU + self.apply_outgoing_link_mtu_to_response(&mut response, &from_peer); - // Apply path_mtu min() from the outgoing link's transport MTU - self.apply_outgoing_link_mtu_to_response(&mut response, &from_peer); - - debug!( - request_id = response.request_id, - target = %self.peer_display_name(&response.target), - next_hop = %self.peer_display_name(&from_peer), - path_mtu = response.path_mtu, - "Reverse-path forwarding LookupResponse" - ); - - let encoded = response.encode(); - if let Err(e) = self.send_encrypted_link_message(&from_peer, &encoded).await { debug!( + request_id = response.request_id, + target = %self.peer_display_name(&response.target), next_hop = %self.peer_display_name(&from_peer), - error = %e, - "Failed to forward LookupResponse" + path_mtu = response.path_mtu, + "Reverse-path forwarding LookupResponse" ); - } - } else { - // We originated this request — verify proof before caching - let target = response.target; - let path_mtu = response.path_mtu; - // Look up the target's public key from identity_cache - let mut prefix = [0u8; 15]; - prefix.copy_from_slice(&target.as_bytes()[0..15]); - let target_pubkey = match self.lookup_by_fips_prefix(&prefix) { - Some((_addr, pubkey)) => pubkey, - None => { + let encoded = response.encode(); + if let Err(e) = self.send_encrypted_link_message(&from_peer, &encoded).await { + debug!( + next_hop = %self.peer_display_name(&from_peer), + error = %e, + "Failed to forward LookupResponse" + ); + } + } + crate::proto::discovery::ResponseRoute::Originator => { + // We originated this request — verify proof before caching + let target = response.target; + let path_mtu = response.path_mtu; + + // Look up the target's public key from identity_cache + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&target.as_bytes()[0..15]); + let target_pubkey = match self.lookup_by_fips_prefix(&prefix) { + Some((_addr, pubkey)) => pubkey, + None => { + self.metrics() + .discovery + .record_reject(DiscoveryReject::RespIdentityMiss); + warn!( + request_id = response.request_id, + target = %self.peer_display_name(&target), + "identity_cache miss for lookup target, cannot verify proof" + ); + return; + } + }; + + // Verify the proof signature + let (xonly, _parity) = target_pubkey.x_only_public_key(); + let peer_id = PeerIdentity::from_pubkey(xonly); + let proof_data = LookupResponse::proof_bytes( + response.request_id, + &target, + &response.target_coords, + ); + if !peer_id.verify(&proof_data, &response.proof) { self.metrics() .discovery - .record_reject(DiscoveryReject::RespIdentityMiss); + .record_reject(DiscoveryReject::RespProofFailed); warn!( request_id = response.request_id, target = %self.peer_display_name(&target), - "identity_cache miss for lookup target, cannot verify proof" + "LookupResponse proof verification failed, discarding" ); return; } - }; - // Verify the proof signature - let (xonly, _parity) = target_pubkey.x_only_public_key(); - let peer_id = PeerIdentity::from_pubkey(xonly); - let proof_data = - LookupResponse::proof_bytes(response.request_id, &target, &response.target_coords); - if !peer_id.verify(&proof_data, &response.proof) { - self.metrics() - .discovery - .record_reject(DiscoveryReject::RespProofFailed); - warn!( + self.metrics().discovery.resp_accepted.inc(); + + info!( request_id = response.request_id, target = %self.peer_display_name(&target), - "LookupResponse proof verification failed, discarding" + depth = response.target_coords.depth(), + path_mtu = path_mtu, + "Discovery succeeded, proof verified, route cached" ); - return; + + // Apply the accept-side effects: the core clears the success + // state (backoff + pending lookup) and returns the + // cross-subsystem effects for us to drive. + let actions = crate::proto::discovery::on_response_accepted( + &mut self.discovery, + &target, + response.target_coords, + now_ms, + path_mtu, + ); + self.drive_response_actions(actions).await; } + } + } - self.metrics().discovery.resp_accepted.inc(); - - // Clear backoff on success — target is reachable - self.discovery_backoff.record_success(&target); - - info!( - request_id = response.request_id, - target = %self.peer_display_name(&target), - depth = response.target_coords.depth(), - path_mtu = path_mtu, - "Discovery succeeded, proof verified, route cached" - ); - - self.coord_cache - .insert_with_path_mtu(target, response.target_coords, now_ms, path_mtu); - - // Mirror path_mtu into the FipsAddress-keyed read-only lookup - // map used by the TUN reader/writer at TCP MSS clamp time. - let fips_addr = crate::FipsAddress::from_node_addr(&target); - match self.path_mtu_lookup.write() { - Ok(mut map) => { - let prior = map.insert(fips_addr, path_mtu); - debug!( - target = %self.peer_display_name(&target), - fips_addr = %fips_addr, - path_mtu = path_mtu, - prior = ?prior, - map_len = map.len(), - "Wrote path_mtu_lookup from discovery LookupResponse" - ); + /// Drive the cross-subsystem effects returned by the discovery core's + /// accept-side planning. Each arm reproduces the original inline effect + /// exactly (same metrics/logs/writes, same order). + async fn drive_response_actions(&mut self, actions: Vec) { + for action in actions { + match action { + DiscoveryAction::CacheCoords { + target, + coords, + now_ms, + path_mtu, + } => { + self.coord_cache + .insert_with_path_mtu(target, coords, now_ms, path_mtu); } - Err(e) => { - warn!( - target = %self.peer_display_name(&target), - fips_addr = %fips_addr, - path_mtu = path_mtu, - error = %e, - "path_mtu_lookup write lock poisoned; clamp will not see this update" - ); + DiscoveryAction::WritePathMtu { target, path_mtu } => { + // Mirror path_mtu into the FipsAddress-keyed read-only lookup + // map used by the TUN reader/writer at TCP MSS clamp time. + let fips_addr = crate::FipsAddress::from_node_addr(&target); + match self.path_mtu_lookup.write() { + Ok(mut map) => { + let prior = map.insert(fips_addr, path_mtu); + debug!( + target = %self.peer_display_name(&target), + fips_addr = %fips_addr, + path_mtu = path_mtu, + prior = ?prior, + map_len = map.len(), + "Wrote path_mtu_lookup from discovery LookupResponse" + ); + } + Err(e) => { + warn!( + target = %self.peer_display_name(&target), + fips_addr = %fips_addr, + path_mtu = path_mtu, + error = %e, + "path_mtu_lookup write lock poisoned; clamp will not see this update" + ); + } + } + } + DiscoveryAction::ResetWarmupIfEstablished { target } => { + // If an established session exists, reset the warmup counter. + let n = self.config().node.session.coords_warmup_packets; + if let Some(entry) = self.sessions.get_mut(&target) + && entry.is_established() + { + entry.set_coords_warmup_remaining(n); + debug!( + dest = %self.peer_display_name(&target), + warmup_packets = n, + "Reset coords warmup after discovery for existing session" + ); + } + } + DiscoveryAction::RetryQueuedPackets { target } => { + // If we have pending TUN packets for this target, retry session + // initiation. The coord_cache now has coords, so find_next_hop() + // should succeed. + if let Some(packets) = self.pending_tun_packets.get(&target) { + debug!( + dest = %self.peer_display_name(&target), + queued_packets = packets.len(), + "Retrying queued packets after discovery" + ); + self.retry_session_after_discovery(target).await; + } + } + DiscoveryAction::SendLink { peer, bytes } => { + if let Err(e) = self.send_encrypted_link_message(&peer, &bytes).await { + debug!( + peer = %self.peer_display_name(&peer), + error = %e, + "Failed to send discovery link message" + ); + } } - } - - // Clean up pending lookup tracking - self.pending_lookups.remove(&target); - - // If an established session exists, reset the warmup counter. - let n = self.config().node.session.coords_warmup_packets; - if let Some(entry) = self.sessions.get_mut(&target) - && entry.is_established() - { - entry.set_coords_warmup_remaining(n); - debug!( - dest = %self.peer_display_name(&target), - warmup_packets = n, - "Reset coords warmup after discovery for existing session" - ); - } - - // If we have pending TUN packets for this target, retry session - // initiation. The coord_cache now has coords, so find_next_hop() - // should succeed. - if let Some(packets) = self.pending_tun_packets.get(&target) { - debug!( - dest = %self.peer_display_name(&target), - queued_packets = packets.len(), - "Retrying queued packets after discovery" - ); - self.retry_session_after_discovery(target).await; } } } @@ -304,12 +372,16 @@ impl Node { let mut response = LookupResponse::new(request.request_id, request.target, our_coords, proof); - // Route toward origin via reverse path. - let next_hop_addr = if let Some(recent) = self.recent_requests.get(&request.request_id) { - recent.from_peer - } else { - // Fallback: try greedy tree routing toward origin - match self.find_next_hop(&request.origin) { + // Route toward origin. The reverse-path decision (the peer the request + // arrived from, recorded in recent_requests) is the sans-IO core's; the + // greedy tree-route fallback is a &mut coord-cache op kept in the shell. + use crate::proto::discovery::ResponseRouteDecision; + let next_hop_addr = match crate::proto::discovery::plan_response_route( + &self.discovery, + request.request_id, + ) { + ResponseRouteDecision::ReversePath(peer) => peer, + ResponseRouteDecision::NeedsTreeRoute => match self.find_next_hop(&request.origin) { Some(peer) => *peer.node_addr(), None => { debug!( @@ -321,7 +393,7 @@ impl Node { .record_reject(DiscoveryReject::RespNoRoute); return; } - } + }, }; // Fold our outgoing-link MTU into path_mtu so the target-edge link @@ -361,83 +433,58 @@ impl Node { /// bloom contains the target. This recovers from dead ends caused by /// stale bloom filters, tree restructuring, or transit node failures. async fn forward_lookup_request(&mut self, mut request: LookupRequest) { - if !request.forward() { - return; - } - - // Leaf nodes don't forward discovery requests - if self.node_profile() == crate::protocol::NodeProfile::Leaf { - return; - } - - // Collect full tree peers whose bloom filter contains the target - let min_mtu = request.min_mtu; - let forward_to: Vec = self - .peers - .iter() - .filter(|(addr, peer)| { - peer.peer_profile() == crate::protocol::NodeProfile::Full - && self.is_tree_peer(addr) - && peer.may_reach(&request.target) - && self.peer_meets_mtu(peer, min_mtu) - }) - .map(|(addr, _)| *addr) - .collect(); - - // Fallback: if no tree peer matches, try non-tree full bloom-matching peers - let (forward_to, used_fallback) = if forward_to.is_empty() { - let fallback: Vec = self - .peers - .iter() - .filter(|(addr, peer)| { - peer.peer_profile() == crate::protocol::NodeProfile::Full - && !self.is_tree_peer(addr) - && peer.may_reach(&request.target) - && self.peer_meets_mtu(peer, min_mtu) - }) - .map(|(addr, _)| *addr) - .collect(); - if fallback.is_empty() { + // Plan the forward with the sans-IO decision core. The core owns the + // TTL decrement, Leaf suppression, Full+MTU eligibility, tree/fallback + // peer selection, and single-encode fan-out; the shell keeps all + // metrics/logging and drives the sends. + let outcome = { + let rv = NodeRoutingView { node: self }; + crate::proto::discovery::plan_forward(&mut request, &rv) + }; + match outcome { + crate::proto::discovery::ForwardOutcome::TtlExhausted => {} + crate::proto::discovery::ForwardOutcome::LeafNoForward => {} + crate::proto::discovery::ForwardOutcome::NoPeers => { self.metrics().discovery.req_no_tree_peer.inc(); trace!( request_id = request.request_id, "No eligible peers to forward LookupRequest" ); - return; } - (fallback, true) - } else { - (forward_to, false) - }; - - if used_fallback { - self.metrics().discovery.req_fallback_forwarded.inc(); - debug!( - request_id = request.request_id, - target = %self.peer_display_name(&request.target), - ttl = request.ttl, - peer_count = forward_to.len(), - "Forwarding LookupRequest via non-tree fallback" - ); - } else { - debug!( - request_id = request.request_id, - target = %self.peer_display_name(&request.target), - ttl = request.ttl, - peer_count = forward_to.len(), - "Forwarding LookupRequest" - ); - } - - let encoded = request.encode(); - - for peer_addr in forward_to { - if let Err(e) = self.send_encrypted_link_message(&peer_addr, &encoded).await { - debug!( - peer = %self.peer_display_name(&peer_addr), - error = %e, - "Failed to forward LookupRequest to peer" - ); + crate::proto::discovery::ForwardOutcome::Forward { + actions, + used_fallback, + } => { + let peer_count = actions.len(); + if used_fallback { + self.metrics().discovery.req_fallback_forwarded.inc(); + debug!( + request_id = request.request_id, + target = %self.peer_display_name(&request.target), + ttl = request.ttl, + peer_count, + "Forwarding LookupRequest via non-tree fallback" + ); + } else { + debug!( + request_id = request.request_id, + target = %self.peer_display_name(&request.target), + ttl = request.ttl, + peer_count, + "Forwarding LookupRequest" + ); + } + for action in actions { + if let DiscoveryAction::SendLink { peer, bytes } = action + && let Err(e) = self.send_encrypted_link_message(&peer, &bytes).await + { + debug!( + peer = %self.peer_display_name(&peer), + error = %e, + "Failed to forward LookupRequest to peer" + ); + } + } } } } @@ -455,20 +502,16 @@ impl Node { let min_mtu = self.config().tun.mtu(); let request = LookupRequest::generate(*target, origin, ttl, min_mtu); - // Send only to full tree peers whose bloom filter contains the target - let peer_addrs: Vec = self - .peers - .iter() - .filter(|(addr, peer)| { - peer.peer_profile() == crate::protocol::NodeProfile::Full - && self.is_tree_peer(addr) - && peer.may_reach(target) - && self.peer_meets_mtu(peer, request.min_mtu) - }) - .map(|(addr, _)| *addr) - .collect(); + // Tree-peer selection restricted to Full peers meeting min_mtu, plus the + // single encode, live in the sans-IO core. The core keeps the tree-only + // (no non-tree fallback) behavior; the shell drives the sends and keeps + // all metrics/logging. + let actions = { + let rv = NodeRoutingView { node: self }; + crate::proto::discovery::plan_initiate(&request, &rv) + }; - let peer_count = peer_addrs.len(); + let peer_count = actions.len(); debug!( request_id = request.request_id, @@ -479,16 +522,12 @@ impl Node { "Discovery lookup initiated" ); - if peer_count == 0 { - return 0; - } - - let encoded = request.encode(); - - for peer_addr in peer_addrs { - if let Err(e) = self.send_encrypted_link_message(&peer_addr, &encoded).await { + for action in actions { + if let DiscoveryAction::SendLink { peer, bytes } = action + && let Err(e) = self.send_encrypted_link_message(&peer, &bytes).await + { debug!( - peer = %self.peer_display_name(&peer_addr), + peer = %self.peer_display_name(&peer), error = %e, "Failed to send LookupRequest to peer" ); @@ -508,54 +547,49 @@ impl Node { pub(in crate::node) async fn maybe_initiate_lookup(&mut self, dest: &NodeAddr) { let now_ms = Self::now_ms(); - // Dedup: any pending lookup means we are already trying. - if self.pending_lookups.contains_key(dest) { - self.metrics().discovery.req_deduplicated.inc(); - debug!( - target_node = %self.peer_display_name(dest), - "Discovery lookup deduplicated, already pending" - ); - return; - } - - // Optional post-failure suppression. Defaults are 0/0 (inert); - // operators can opt in by setting `node.discovery.backoff_*_secs`. - if self.discovery_backoff.is_suppressed(dest) { - self.metrics().discovery.req_backoff_suppressed.inc(); - debug!( - target_node = %self.peer_display_name(dest), - failures = self.discovery_backoff.failure_count(dest), - "Discovery lookup suppressed by backoff" - ); - return; - } - - // Bloom filter pre-check: if no peer's filter contains the target, - // it's not in the mesh — skip the lookup and record as failure. + // Bloom filter pre-check (view read) BEFORE the core call: if no peer's + // filter contains the target, it's not in the mesh. Reading `self.peers` + // here keeps the `&mut self.discovery` borrow in `initiate_gate` from + // overlapping the immutable peer-table read. let reachable = self.peers.values().any(|peer| peer.may_reach(dest)); - if !reachable { - self.metrics().discovery.req_bloom_miss.inc(); - self.discovery_backoff.record_failure(dest); - debug!( - target_node = %self.peer_display_name(dest), - "Discovery skipped, target not in any peer bloom filter" - ); - return; - } - self.pending_lookups - .insert(*dest, PendingLookup::new(now_ms)); - let ttl = self.config().node.discovery.ttl; - let sent = self.initiate_lookup(dest, ttl).await; + use crate::proto::discovery::InitiateDecision; + match crate::proto::discovery::initiate_gate(&mut self.discovery, dest, now_ms, reachable) { + InitiateDecision::Deduplicated => { + self.metrics().discovery.req_deduplicated.inc(); + debug!( + target_node = %self.peer_display_name(dest), + "Discovery lookup deduplicated, already pending" + ); + } + InitiateDecision::Suppressed { failures } => { + self.metrics().discovery.req_backoff_suppressed.inc(); + debug!( + target_node = %self.peer_display_name(dest), + failures = failures, + "Discovery lookup suppressed by backoff" + ); + } + InitiateDecision::BloomMiss => { + self.metrics().discovery.req_bloom_miss.inc(); + debug!( + target_node = %self.peer_display_name(dest), + "Discovery skipped, target not in any peer bloom filter" + ); + } + InitiateDecision::Proceed => { + let ttl = self.config().node.discovery.ttl; + let sent = self.initiate_lookup(dest, ttl).await; - // If no tree peers had the target, fail immediately - if sent == 0 { - self.pending_lookups.remove(dest); - self.discovery_backoff.record_failure(dest); - debug!( - target_node = %self.peer_display_name(dest), - "Discovery failed, no tree peers with bloom match" - ); + // If no tree peers had the target, fail immediately + if sent == 0 { + crate::proto::discovery::initiate_failed(&mut self.discovery, dest, now_ms); + debug!( + target_node = %self.peer_display_name(dest), + "Discovery failed, no tree peers with bloom match" + ); + } + } } } @@ -570,53 +604,24 @@ impl Node { /// - Otherwise: declare the destination unreachable, drop queued packets, /// and emit ICMPv6 destination-unreachable for each. pub(in crate::node) async fn check_pending_lookups(&mut self, now_ms: u64) { - let timeouts = self.config().node.discovery.attempt_timeouts_secs.clone(); - let max_attempts = timeouts.len() as u8; + let attempt_timeouts = self.config().node.discovery.attempt_timeouts_secs.clone(); + let outcome = + crate::proto::discovery::poll_pending(&mut self.discovery, now_ms, &attempt_timeouts); - // Collect targets needing action - let mut to_retry: Vec = Vec::new(); - let mut to_timeout: Vec = Vec::new(); - - for (&target, entry) in &self.pending_lookups { - let attempt_idx = (entry.attempt as usize).saturating_sub(1); - let attempt_timeout_ms = timeouts.get(attempt_idx).copied().unwrap_or(0) * 1000; - if now_ms.saturating_sub(entry.last_sent_ms) >= attempt_timeout_ms { - if entry.attempt >= max_attempts { - to_timeout.push(target); - } else { - to_retry.push(target); - } + for (target, attempt) in outcome.retries { + let ttl = self.config().node.discovery.ttl; + let sent = self.initiate_lookup(&target, ttl).await; + if sent > 0 { + debug!( + target_node = %self.peer_display_name(&target), + attempt = attempt, + "Discovery retry sent" + ); } } - // Process retries - for target in to_retry { - if let Some(entry) = self.pending_lookups.get_mut(&target) { - entry.attempt += 1; - entry.last_sent_ms = now_ms; - let attempt = entry.attempt; - - let ttl = self.config().node.discovery.ttl; - let sent = self.initiate_lookup(&target, ttl).await; - if sent > 0 { - debug!( - target_node = %self.peer_display_name(&target), - attempt = attempt, - "Discovery retry sent" - ); - } - } - } - - // Process timeouts - for addr in to_timeout { + for (addr, failures) in outcome.timeouts { self.metrics().discovery.resp_timed_out.inc(); - self.pending_lookups.remove(&addr); - - // Record failure for optional backoff - self.discovery_backoff.record_failure(&addr); - let failures = self.discovery_backoff.failure_count(&addr); - let queued = self.pending_tun_packets.remove(&addr); let pkt_count = queued.as_ref().map_or(0, |p| p.len()); info!( @@ -635,12 +640,12 @@ impl Node { /// Reset discovery backoff on topology changes. pub(in crate::node) fn reset_discovery_backoff(&mut self) { - if !self.discovery_backoff.is_empty() { + let cleared = self.discovery.reset_backoff(); + if cleared > 0 { debug!( - entries = self.discovery_backoff.entry_count(), + entries = cleared, "Resetting discovery backoff on topology change" ); - self.discovery_backoff.reset_all(); } } @@ -666,13 +671,6 @@ impl Node { } } - /// Remove expired entries from the recent_requests cache. - fn purge_expired_requests(&mut self, current_time_ms: u64) { - let expiry_ms = self.config().node.discovery.recent_expiry_secs * 1000; - self.recent_requests - .retain(|_, entry| !entry.is_expired(current_time_ms, expiry_ms)); - } - /// Min-fold our outgoing-link MTU into a LookupResponse's `path_mtu`. /// /// Used at both transit-side reverse-path forward and at the target's @@ -759,23 +757,3 @@ impl Node { } } } - -/// Tracks a pending discovery lookup with retry state. -pub struct PendingLookup { - /// When the lookup was first initiated. - pub initiated_ms: u64, - /// When the last attempt was sent. - pub last_sent_ms: u64, - /// Current attempt number (1 = initial, 2 = first retry, ...). - pub attempt: u8, -} - -impl PendingLookup { - pub fn new(now_ms: u64) -> Self { - Self { - initiated_ms: now_ms, - last_sent_ms: now_ms, - attempt: 1, - } - } -} diff --git a/src/node/mod.rs b/src/node/mod.rs index e784e72..7e6911f 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -9,7 +9,6 @@ mod bloom; pub(crate) mod context; #[cfg(unix)] pub(crate) mod decrypt_worker; -mod discovery_rate_limit; #[cfg(unix)] pub(crate) mod encrypt_worker; mod handlers; @@ -29,7 +28,6 @@ mod tests; mod tree; pub(crate) mod wire; -use self::discovery_rate_limit::{DiscoveryBackoff, DiscoveryForwardRateLimiter}; use self::rate_limit::HandshakeRateLimiter; use self::reloadable::Reloadable; use self::routing_error_rate_limit::RoutingErrorRateLimiter; @@ -61,6 +59,7 @@ use crate::bloom::{BloomFilter, BloomState}; use crate::cache::CoordCache; use crate::node::session::SessionEntry; use crate::peer::{ActivePeer, PeerConnection}; +use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter}; use crate::protocol::NodeProfile; #[cfg(unix)] use crate::transport::ethernet::EthernetTransport; @@ -232,39 +231,6 @@ pub struct UpdatePeersOutcome { pub unchanged: usize, } -/// Recent request tracking for dedup and reverse-path forwarding. -/// -/// When a LookupRequest is forwarded through a node, the node stores the -/// request_id and which peer sent it. When the corresponding LookupResponse -/// arrives, it's forwarded back to that peer (reverse-path forwarding). -/// The `response_forwarded` flag prevents response routing loops. -#[derive(Clone, Debug)] -pub(crate) struct RecentRequest { - /// The peer who sent this request to us. - pub(crate) from_peer: NodeAddr, - /// When we received this request (Unix milliseconds). - pub(crate) timestamp_ms: u64, - /// Whether we've already forwarded a response for this request. - /// Prevents response routing loops when convergent request paths - /// create bidirectional entries in recent_requests. - pub(crate) response_forwarded: bool, -} - -impl RecentRequest { - pub(crate) fn new(from_peer: NodeAddr, timestamp_ms: u64) -> Self { - Self { - from_peer, - timestamp_ms, - response_forwarded: false, - } - } - - /// Check if this entry has expired (older than expiry_ms). - pub(crate) fn is_expired(&self, current_time_ms: u64, expiry_ms: u64) -> bool { - current_time_ms.saturating_sub(self.timestamp_ms) > expiry_ms - } -} - /// Key for addr_to_link reverse lookup. type AddrKey = (TransportId, TransportAddr); @@ -335,9 +301,6 @@ pub struct Node { // === Routing === /// Address -> coordinates cache (from session setup and discovery). coord_cache: CoordCache, - /// Recent discovery requests (dedup + reverse-path forwarding). - /// Maps request_id → RecentRequest. - recent_requests: HashMap, /// Per-destination path MTU lookup, keyed by FipsAddress (mirrors /// `coord_cache.entries[*].path_mtu`). Sync read-only access from /// the TUN reader/writer threads at TCP MSS clamp time so the @@ -385,10 +348,11 @@ pub struct Node { /// Packets queued while waiting for session establishment. /// Keyed by destination NodeAddr, bounded per-dest and total. pending_tun_packets: HashMap>>, - // === Pending Discovery Lookups === - /// Tracks in-flight discovery lookups. Maps target NodeAddr to the - /// initiation timestamp (Unix ms). Prevents duplicate flood queries. - pending_lookups: HashMap, + + // === Discovery === + /// Discovery-subsystem state: recent-request dedup cache, in-flight + /// lookups, originator-side backoff, and transit-side forward limiter. + discovery: Discovery, // === Counters === /// Next link ID to allocate. @@ -475,10 +439,6 @@ pub struct Node { routing_error_rate_limiter: RoutingErrorRateLimiter, /// Rate limiter for source-side CoordsRequired/PathBroken responses. coords_response_rate_limiter: RoutingErrorRateLimiter, - /// Backoff for failed discovery lookups (originator-side). - discovery_backoff: DiscoveryBackoff, - /// Rate limiter for forwarded discovery requests (transit-side). - discovery_forward_limiter: DiscoveryForwardRateLimiter, // === Pending Transport Connects === /// Links waiting for transport-level connection establishment before @@ -680,7 +640,6 @@ impl Node { tree_state, bloom_state, coord_cache, - recent_requests: HashMap::new(), transports: HashMap::new(), transport_drops: HashMap::new(), links: HashMap::new(), @@ -692,7 +651,6 @@ impl Node { sessions: HashMap::new(), identity_cache: HashMap::new(), pending_tun_packets: HashMap::new(), - pending_lookups: HashMap::new(), next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), @@ -727,9 +685,9 @@ impl Node { coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval( std::time::Duration::from_millis(coords_response_interval_ms), ), - discovery_backoff: DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs), - discovery_forward_limiter: DiscoveryForwardRateLimiter::with_interval( - std::time::Duration::from_secs(forward_min_interval_secs), + discovery: Discovery::new( + DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs), + DiscoveryForwardRateLimiter::with_interval_ms(forward_min_interval_secs * 1000), ), pending_connects: Vec::new(), retry_pending: HashMap::new(), @@ -843,7 +801,6 @@ impl Node { tree_state, bloom_state, coord_cache, - recent_requests: HashMap::new(), transports: HashMap::new(), transport_drops: HashMap::new(), links: HashMap::new(), @@ -855,7 +812,6 @@ impl Node { sessions: HashMap::new(), identity_cache: HashMap::new(), pending_tun_packets: HashMap::new(), - pending_lookups: HashMap::new(), next_link_id: 1, next_transport_id: 1, stats: stats::NodeStats::new(), @@ -890,8 +846,7 @@ impl Node { coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval( std::time::Duration::from_millis(coords_response_interval_ms), ), - discovery_backoff: DiscoveryBackoff::new(), - discovery_forward_limiter: DiscoveryForwardRateLimiter::new(), + discovery: Discovery::new(DiscoveryBackoff::new(), DiscoveryForwardRateLimiter::new()), pending_connects: Vec::new(), retry_pending: HashMap::new(), nostr_discovery: None, @@ -2485,8 +2440,7 @@ impl Node { /// Disable the discovery forward rate limiter (for tests). #[cfg(test)] pub(crate) fn disable_discovery_forward_rate_limit(&mut self) { - self.discovery_forward_limiter - .set_interval(std::time::Duration::ZERO); + self.discovery.forward_limiter.set_interval_ms(0); } #[cfg(test)] @@ -2598,19 +2552,19 @@ impl Node { /// Number of pending discovery lookups. pub fn pending_lookup_count(&self) -> usize { - self.pending_lookups.len() + self.discovery.pending_lookups.len() } /// Iterate over pending discovery lookups for diagnostics. pub fn pending_lookups_iter( &self, - ) -> impl Iterator { - self.pending_lookups.iter() + ) -> impl Iterator { + self.discovery.pending_lookups.iter() } /// Number of recent discovery requests tracked. pub fn recent_request_count(&self) -> usize { - self.recent_requests.len() + self.discovery.recent_requests.len() } /// Count of destinations with queued TUN packets awaiting session setup. diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index 0d6a56f..dafe6db 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -5,8 +5,7 @@ //! response routing. use super::*; -use crate::node::RecentRequest; -use crate::protocol::{LookupRequest, LookupResponse}; +use crate::proto::discovery::{LookupRequest, LookupResponse, RecentRequest}; use crate::tree::TreeCoordinate; use spanning_tree::{ cleanup_nodes, generate_random_edges, lock_large_network_test, process_available_packets, @@ -23,7 +22,7 @@ async fn test_request_decode_error() { let from = make_node_addr(0xAA); // Too-short payload: should log error and return without panic node.handle_lookup_request(&from, &[0x00; 5]).await; - assert!(node.recent_requests.is_empty()); + assert!(node.discovery.recent_requests.is_empty()); } #[tokio::test] @@ -38,11 +37,11 @@ async fn test_request_dedup() { // First request: accepted node.handle_lookup_request(&from, payload).await; - assert_eq!(node.recent_requests.len(), 1); + assert_eq!(node.discovery.recent_requests.len(), 1); // Duplicate request: dropped node.handle_lookup_request(&from, payload).await; - assert_eq!(node.recent_requests.len(), 1); + assert_eq!(node.discovery.recent_requests.len(), 1); } #[tokio::test] @@ -59,7 +58,7 @@ async fn test_request_target_is_self() { // Should succeed without panic (response send will fail silently // since we have no peers to route toward origin) node.handle_lookup_request(&from, payload).await; - assert!(node.recent_requests.contains_key(&777)); + assert!(node.discovery.recent_requests.contains_key(&777)); } #[tokio::test] @@ -74,7 +73,7 @@ async fn test_request_ttl_zero_not_forwarded() { node.handle_lookup_request(&from, payload).await; // Request recorded, but not forwarded (TTL=0, and no peers anyway) - assert!(node.recent_requests.contains_key(&666)); + assert!(node.discovery.recent_requests.contains_key(&666)); } // ============================================================================ @@ -112,7 +111,7 @@ async fn test_response_originator_caches_route() { let payload = &response.encode()[1..]; // skip msg_type // No entry in recent_requests for 555 → we're the originator - assert!(!node.recent_requests.contains_key(&555)); + assert!(!node.discovery.recent_requests.contains_key(&555)); node.handle_lookup_response(&from, payload).await; @@ -146,7 +145,8 @@ async fn test_response_transit_needs_recent_request() { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_millis() as u64; - node.recent_requests + node.discovery + .recent_requests .insert(444, RecentRequest::new(make_node_addr(0xDD), now_ms)); // Handle response — should try to reverse-path forward to 0xDD @@ -316,14 +316,16 @@ async fn test_recent_request_expiry() { .as_millis() as u64; // Insert an old request (11 seconds ago) - node.recent_requests + node.discovery + .recent_requests .insert(123, RecentRequest::new(make_node_addr(1), now_ms - 11_000)); // Insert a recent request - node.recent_requests + node.discovery + .recent_requests .insert(456, RecentRequest::new(make_node_addr(2), now_ms)); - assert_eq!(node.recent_requests.len(), 2); + assert_eq!(node.discovery.recent_requests.len(), 2); // Trigger purge via a new lookup request let target = make_node_addr(0xBB); @@ -334,9 +336,9 @@ async fn test_recent_request_expiry() { .await; // Old entry (123) should be purged, recent entry (456) and new entry (789) kept - assert!(!node.recent_requests.contains_key(&123)); - assert!(node.recent_requests.contains_key(&456)); - assert!(node.recent_requests.contains_key(&789)); + assert!(!node.discovery.recent_requests.contains_key(&123)); + assert!(node.discovery.recent_requests.contains_key(&456)); + assert!(node.discovery.recent_requests.contains_key(&789)); } // ============================================================================ @@ -373,7 +375,7 @@ async fn test_request_forwarding_two_node() { // Node1 should have recorded the request assert!( - nodes[1].node.recent_requests.contains_key(&42), + nodes[1].node.discovery.recent_requests.contains_key(&42), "Node 1 should have recorded the forwarded request" ); @@ -441,13 +443,13 @@ async fn test_request_three_node_chain() { // Node1 should have been a transit node (has the request_id in recent_requests) assert!( - !nodes[1].node.recent_requests.is_empty(), + !nodes[1].node.discovery.recent_requests.is_empty(), "Node 1 should have recorded the forwarded request" ); // Node2 should have received the request (it's the target) assert!( - !nodes[2].node.recent_requests.is_empty(), + !nodes[2].node.discovery.recent_requests.is_empty(), "Node 2 should have received the request" ); @@ -494,7 +496,7 @@ async fn test_request_dedup_convergent_paths() { // Node2 (the target) must have received the request assert!( - nodes[2].node.recent_requests.contains_key(&300), + nodes[2].node.discovery.recent_requests.contains_key(&300), "Node 2 (target) should have received the request" ); @@ -1159,8 +1161,8 @@ async fn test_open_discovery_sweep_queues_eligible_skips_filtered() { #[tokio::test] async fn test_check_pending_lookups_default_sequence_unreachable() { use crate::bloom::BloomFilter; - use crate::node::handlers::discovery::PendingLookup; use crate::peer::ActivePeer; + use crate::proto::discovery::PendingLookup; use crate::transport::LinkId; use std::sync::mpsc; @@ -1228,7 +1230,8 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { // Inject a PendingLookup directly: attempt=1, last_sent_ms=0. This // mirrors the post-condition of a successful `maybe_initiate_lookup` // at t=0 without depending on wall-clock-derived `Self::now_ms()`. - node.pending_lookups + node.discovery + .pending_lookups .insert(target_addr, PendingLookup::new(0)); let baseline_initiated = node.metrics().discovery.req_initiated.get(); @@ -1238,6 +1241,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { node.check_pending_lookups(1100).await; { let entry = node + .discovery .pending_lookups .get(&target_addr) .expect("still pending"); @@ -1254,6 +1258,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { node.check_pending_lookups(3100).await; { let entry = node + .discovery .pending_lookups .get(&target_addr) .expect("still pending"); @@ -1270,6 +1275,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { node.check_pending_lookups(7100).await; { let entry = node + .discovery .pending_lookups .get(&target_addr) .expect("still pending"); @@ -1285,7 +1291,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { // --- Just-before-final: at t=15099ms the 8s window is not yet reached --- node.check_pending_lookups(15_099).await; assert!( - node.pending_lookups.contains_key(&target_addr), + node.discovery.pending_lookups.contains_key(&target_addr), "8s window not yet expired: pending_lookup must persist" ); assert_eq!( @@ -1308,7 +1314,7 @@ async fn test_check_pending_lookups_default_sequence_unreachable() { // Pending lookup is dropped. assert!( - !node.pending_lookups.contains_key(&target_addr), + !node.discovery.pending_lookups.contains_key(&target_addr), "final timeout must remove the pending_lookups entry" ); // resp_timed_out counter ticked. diff --git a/src/proto/discovery/core.rs b/src/proto/discovery/core.rs new file mode 100644 index 0000000..6f9baf5 --- /dev/null +++ b/src/proto/discovery/core.rs @@ -0,0 +1,404 @@ +//! Sans-IO discovery decision core. +//! +//! Pure, runtime-agnostic decision logic for the discovery protocol. The +//! async I/O adapter in `node::handlers::discovery` decodes wire bytes, +//! calls into this core, and drives the returned actions (the actual +//! encrypted sends). No I/O, no clock, no metrics, no logging here. + +use alloc::sync::Arc; + +use super::state::{Discovery, PendingLookup, RecentRequest}; +use super::wire::LookupRequest; +use crate::NodeAddr; + +/// Read-only view of routing state the discovery core needs. +/// +/// The core defines this interface; the async shell (`node`) implements it +/// over the live peer/tree tables. Keeping it a trait keeps `proto` free of +/// any dependency on `node` and lets the core be unit-tested with a mock. +pub(crate) trait RoutingView { + /// Is `addr` a spanning-tree peer (parent or child)? + fn is_tree_peer(&self, addr: &NodeAddr) -> bool; + /// Peers whose bloom filter may reach `target` (i.e. `may_reach(target)`). + fn peers_reaching(&self, target: &NodeAddr) -> Vec; + /// Is this node a Leaf? Leaves do not transit-forward lookup requests. + fn node_is_leaf(&self) -> bool; + /// Does `addr` advertise the Full profile? Only Full peers are forwarded to. + fn peer_is_full(&self, addr: &NodeAddr) -> bool; + /// Does `addr`'s outgoing link MTU satisfy `min_mtu`? (`min_mtu == 0` means + /// no requirement; a peer with no known transport MTU is not pruned.) + fn peer_meets_mtu(&self, addr: &NodeAddr, min_mtu: u16) -> bool; +} + +/// An I/O action the async shell performs on the core's behalf. +pub(crate) enum DiscoveryAction { + /// Send an encoded discovery PDU to a peer as an encrypted link message. + /// `bytes` is `Arc`-shared so a fan-out encodes once. + SendLink { peer: NodeAddr, bytes: Arc<[u8]> }, + /// Cache the verified destination coordinates + path MTU (coord_cache). + CacheCoords { + target: NodeAddr, + coords: crate::TreeCoordinate, + now_ms: u64, + path_mtu: u16, + }, + /// Mirror path_mtu into the FipsAddress-keyed TUN-shared lookup map. + WritePathMtu { target: NodeAddr, path_mtu: u16 }, + /// Reset the coords-warmup counter if an established session exists. + ResetWarmupIfEstablished { target: NodeAddr }, + /// Retry queued TUN packets for the target if any are pending. + RetryQueuedPackets { target: NodeAddr }, +} + +/// Outcome of planning a LookupRequest forward. +pub(crate) enum ForwardOutcome { + /// TTL was exhausted — nothing to do. + TtlExhausted, + /// This node is a Leaf; leaves do not transit-forward. Silent no-op. + LeafNoForward, + /// No eligible peer had the target in its bloom filter. + NoPeers, + /// Forward: one SendLink per selected peer. `used_fallback` is true when + /// the non-tree bloom-match fallback set was used (no tree peer matched). + Forward { + actions: Vec, + used_fallback: bool, + }, +} + +/// Plan the transit forward of an inbound LookupRequest. +/// +/// Decrements TTL; suppresses forwarding on a Leaf node; restricts candidates to +/// Full peers whose link satisfies the request's `min_mtu`; selects tree peers +/// whose bloom matches the target, else a non-tree bloom-match fallback; encodes +/// the (decremented) request once and emits one SendLink per selected peer. Pure +/// — no I/O, metrics, or logs. +pub(crate) fn plan_forward(request: &mut LookupRequest, rv: &impl RoutingView) -> ForwardOutcome { + if !request.forward() { + return ForwardOutcome::TtlExhausted; + } + // Leaf nodes do not transit-forward discovery requests. + if rv.node_is_leaf() { + return ForwardOutcome::LeafNoForward; + } + let target = request.target; + let min_mtu = request.min_mtu; + // Only Full peers whose outgoing link satisfies min_mtu are eligible. + let eligible: Vec = rv + .peers_reaching(&target) + .into_iter() + .filter(|a| rv.peer_is_full(a) && rv.peer_meets_mtu(a, min_mtu)) + .collect(); + let tree: Vec = eligible + .iter() + .copied() + .filter(|a| rv.is_tree_peer(a)) + .collect(); + let (targets, used_fallback) = if tree.is_empty() { + let fallback: Vec = eligible + .into_iter() + .filter(|a| !rv.is_tree_peer(a)) + .collect(); + if fallback.is_empty() { + return ForwardOutcome::NoPeers; + } + (fallback, true) + } else { + (tree, false) + }; + let bytes: Arc<[u8]> = Arc::from(request.encode()); + let actions = targets + .into_iter() + .map(|peer| DiscoveryAction::SendLink { + peer, + bytes: bytes.clone(), + }) + .collect(); + ForwardOutcome::Forward { + actions, + used_fallback, + } +} + +/// Plan the origination of a freshly-generated LookupRequest. +/// +/// Selects tree peers whose bloom matches the target and emits one SendLink per +/// selected peer, encoding the request once (Arc-shared). Returns an empty Vec +/// when no tree peer matches; the shell treats that as an immediate failure. +/// Pure — no I/O, metrics, or logs; the shell generates and signs the request. +/// +/// NOTE: unlike [`plan_forward`], this does NOT fall back to non-tree +/// (cross-link) bloom-matching peers. That asymmetry is preserved verbatim from +/// the pre-sans-IO `initiate_lookup` to keep this extraction behavior-neutral; +/// it is a known origination gap (ISSUE-2026-0059) whose fix adds the fallback +/// branch as a separate, behavior-changing change. +pub(crate) fn plan_initiate( + request: &LookupRequest, + rv: &impl RoutingView, +) -> Vec { + let min_mtu = request.min_mtu; + let targets: Vec = rv + .peers_reaching(&request.target) + .into_iter() + .filter(|addr| { + rv.is_tree_peer(addr) && rv.peer_is_full(addr) && rv.peer_meets_mtu(addr, min_mtu) + }) + .collect(); + if targets.is_empty() { + return Vec::new(); + } + let bytes: Arc<[u8]> = Arc::from(request.encode()); + targets + .into_iter() + .map(|peer| DiscoveryAction::SendLink { + peer, + bytes: bytes.clone(), + }) + .collect() +} + +/// Classification of an inbound LookupRequest, decided from Discovery state. +pub(crate) enum RequestOutcome { + /// request_id already in the dedup cache — drop. + Duplicate, + /// dedup cache at capacity — drop. `len` is the current cache size (for the log). + DedupCacheFull { len: usize }, + /// We are the lookup target — the shell generates + sends the response. + RespondAsTarget, + /// Forward the request onward (the shell calls the forward planner). + Forward, + /// Transit forward suppressed by the per-target forward rate limiter. + ForwardRateLimited, + /// TTL exhausted, not the target — drop. + TtlExhausted, +} + +/// Classify an inbound LookupRequest against the recent-request dedup cache and +/// the transit forward rate limiter. Purges expired dedup entries, records the +/// request for reverse-path forwarding on the non-drop paths, and decides the +/// route. Pure over Discovery state + node addr + injected clock; no I/O, no view. +pub(crate) fn classify_request( + disc: &mut Discovery, + request: &LookupRequest, + from: &NodeAddr, + my_addr: &NodeAddr, + now_ms: u64, + recent_expiry_ms: u64, + max_recent: usize, +) -> RequestOutcome { + // Purge expired dedup entries (was purge_expired_requests). + disc.recent_requests + .retain(|_, entry| !entry.is_expired(now_ms, recent_expiry_ms)); + + if disc.recent_requests.contains_key(&request.request_id) { + return RequestOutcome::Duplicate; + } + if disc.recent_requests.len() >= max_recent { + return RequestOutcome::DedupCacheFull { + len: disc.recent_requests.len(), + }; + } + disc.recent_requests + .insert(request.request_id, RecentRequest::new(*from, now_ms)); + + if request.target == *my_addr { + return RequestOutcome::RespondAsTarget; + } + if request.can_forward() { + if disc.forward_limiter.should_forward(&request.target, now_ms) { + RequestOutcome::Forward + } else { + RequestOutcome::ForwardRateLimited + } + } else { + RequestOutcome::TtlExhausted + } +} + +/// How an inbound LookupResponse should be routed, decided from the +/// recent-request dedup state. +pub(crate) enum ResponseRoute { + /// A response for a request we forwarded, but we already reverse-forwarded + /// one for this request_id — drop to prevent response routing loops. + AlreadyForwarded, + /// Transit node: reverse-path forward toward `from_peer`. + Transit { from_peer: NodeAddr }, + /// We originated this request — the shell verifies the proof and caches. + Originator, +} + +/// Classify an inbound LookupResponse against the recent-request dedup cache. +/// +/// Pure decision over `Discovery` state: sets `response_forwarded` when this is +/// the first response we transit for the request. No I/O, no view, no metrics. +pub(crate) fn classify_response(disc: &mut Discovery, request_id: u64) -> ResponseRoute { + match disc.recent_requests.get_mut(&request_id) { + Some(recent) => { + if recent.response_forwarded { + ResponseRoute::AlreadyForwarded + } else { + recent.response_forwarded = true; + ResponseRoute::Transit { + from_peer: recent.from_peer, + } + } + } + None => ResponseRoute::Originator, + } +} + +/// Where a LookupResponse we originate as the target should be sent first. +pub(crate) enum ResponseRouteDecision { + /// Send toward the peer the matching request arrived from — the reverse path + /// recorded in `recent_requests` by [`classify_request`]. + ReversePath(NodeAddr), + /// No recorded reverse path: the shell must route greedily toward the origin. + NeedsTreeRoute, +} + +/// Decide the first hop for a LookupResponse we originate as the target, from +/// the recent-request reverse-path record. Pure over `Discovery` state. +/// +/// Only the reverse-path decision is pure. The `NeedsTreeRoute` fallback (greedy +/// tree routing toward the origin) is a `&mut Node` coord-cache operation with a +/// TTL-touch side effect, so it stays in the shell rather than moving here. +pub(crate) fn plan_response_route(disc: &Discovery, request_id: u64) -> ResponseRouteDecision { + match disc.recent_requests.get(&request_id) { + Some(recent) => ResponseRouteDecision::ReversePath(recent.from_peer), + None => ResponseRouteDecision::NeedsTreeRoute, + } +} + +/// Apply the accept-side effects of a verified LookupResponse we originated. +/// +/// Mutates the Discovery success state (clears backoff, drops the pending +/// lookup) and returns the cross-subsystem effects for the shell to drive. +/// Verification is the shell's job — this runs only after the proof checked out. +pub(crate) fn on_response_accepted( + disc: &mut Discovery, + target: &NodeAddr, + coords: crate::TreeCoordinate, + now_ms: u64, + path_mtu: u16, +) -> Vec { + disc.backoff.record_success(target); + disc.pending_lookups.remove(target); + vec![ + DiscoveryAction::CacheCoords { + target: *target, + coords, + now_ms, + path_mtu, + }, + DiscoveryAction::WritePathMtu { + target: *target, + path_mtu, + }, + DiscoveryAction::ResetWarmupIfEstablished { target: *target }, + DiscoveryAction::RetryQueuedPackets { target: *target }, + ] +} + +/// Result of polling the pending-lookup retry ladder at `now_ms`. +/// +/// The core has already applied the state mutations: retried entries have had +/// their attempt bumped and last_sent updated; timed-out entries have been +/// removed and a backoff failure recorded. The shell drives the effects. +pub(crate) struct PollOutcome { + /// (target, new attempt number) — shell re-sends via initiate_lookup. + pub retries: Vec<(NodeAddr, u8)>, + /// (target, failure_count after recording) — shell emits unreachable. + pub timeouts: Vec<(NodeAddr, u32)>, +} + +/// Advance the pending-lookup retry ladder. Pure over `Discovery` state + +/// injected clock: partitions due entries into retries (attempt bumped) and +/// final timeouts (removed + backoff failure recorded). No I/O, no view. +pub(crate) fn poll_pending( + disc: &mut Discovery, + now_ms: u64, + attempt_timeouts_secs: &[u64], +) -> PollOutcome { + let max_attempts = attempt_timeouts_secs.len() as u8; + + // Collect targets needing action (can't mutate while iterating). + let mut retry_targets: Vec = Vec::new(); + let mut timeout_targets: Vec = Vec::new(); + + for (&target, entry) in &disc.pending_lookups { + let idx = (entry.attempt as usize).saturating_sub(1); + let to_ms = attempt_timeouts_secs.get(idx).copied().unwrap_or(0) * 1000; + if now_ms.saturating_sub(entry.last_sent_ms) >= to_ms { + if entry.attempt >= max_attempts { + timeout_targets.push(target); + } else { + retry_targets.push(target); + } + } + } + + let mut retries: Vec<(NodeAddr, u8)> = Vec::new(); + for target in retry_targets { + if let Some(entry) = disc.pending_lookups.get_mut(&target) { + entry.attempt += 1; + entry.last_sent_ms = now_ms; + retries.push((target, entry.attempt)); + } + } + + let mut timeouts: Vec<(NodeAddr, u32)> = Vec::new(); + for target in timeout_targets { + disc.pending_lookups.remove(&target); + disc.backoff.record_failure(&target, now_ms); + let failures = disc.backoff.failure_count(&target); + timeouts.push((target, failures)); + } + + PollOutcome { retries, timeouts } +} + +/// Decision for whether/how to initiate a discovery lookup for a target. +pub(crate) enum InitiateDecision { + /// A lookup is already pending for this target — skip. + Deduplicated, + /// Suppressed by post-failure backoff. `failures` is the current count (for the log). + Suppressed { failures: u32 }, + /// No peer's bloom filter reaches the target — skip (a failure was recorded). + BloomMiss, + /// Proceed: a PendingLookup was inserted; the shell sends the first attempt. + Proceed, +} + +/// Gate a discovery-lookup initiation against pending-dedup, backoff +/// suppression, and bloom reachability (passed in — the shell reads the peer +/// filters). On BloomMiss records a failure; on Proceed inserts the pending +/// lookup. Pure over Discovery state + injected clock; no I/O, no view. +pub(crate) fn initiate_gate( + disc: &mut Discovery, + dest: &NodeAddr, + now_ms: u64, + reachable: bool, +) -> InitiateDecision { + if disc.pending_lookups.contains_key(dest) { + return InitiateDecision::Deduplicated; + } + if disc.backoff.is_suppressed(dest, now_ms) { + return InitiateDecision::Suppressed { + failures: disc.backoff.failure_count(dest), + }; + } + if !reachable { + disc.backoff.record_failure(dest, now_ms); + return InitiateDecision::BloomMiss; + } + disc.pending_lookups + .insert(*dest, PendingLookup::new(now_ms)); + InitiateDecision::Proceed +} + +/// Roll back a lookup whose first attempt reached no tree peers (sent == 0): +/// drop the pending entry and record a backoff failure. +pub(crate) fn initiate_failed(disc: &mut Discovery, dest: &NodeAddr, now_ms: u64) { + disc.pending_lookups.remove(dest); + disc.backoff.record_failure(dest, now_ms); +} diff --git a/src/proto/discovery/limits.rs b/src/proto/discovery/limits.rs new file mode 100644 index 0000000..aed6cf6 --- /dev/null +++ b/src/proto/discovery/limits.rs @@ -0,0 +1,218 @@ +//! Discovery protocol rate limiting and backoff. +//! +//! Two complementary mechanisms: +//! +//! - **`DiscoveryBackoff`** (originator-side, optional): Exponential +//! suppression of fresh lookups after the per-attempt sequence in +//! `node.discovery.attempt_timeouts_secs` has been exhausted. +//! **Disabled by default** (base/cap = 0); the per-attempt sequence +//! is the only retry pacing in the standard configuration. Reset on +//! topology changes (parent change, new peer, first RTT, reconnection). +//! +//! - **`DiscoveryForwardRateLimiter`** (transit-side): Per-target minimum +//! interval for forwarded requests. Defense-in-depth against misbehaving +//! nodes generating fresh request_ids at high rate. + +use crate::NodeAddr; +use alloc::collections::BTreeMap; + +// ============================================================================ +// Originator-side: Discovery Backoff +// ============================================================================ + +/// Default base backoff after first lookup failure. `0` = disabled. +const DEFAULT_BACKOFF_BASE_SECS: u64 = 0; + +/// Default maximum backoff cap. `0` = disabled. +const DEFAULT_BACKOFF_MAX_SECS: u64 = 0; + +/// Backoff multiplier per consecutive failure. +const BACKOFF_MULTIPLIER: u64 = 2; + +/// Exponential backoff for failed discovery lookups. +/// +/// Tracks targets whose lookups have timed out and suppresses +/// re-initiation with increasing delays. Cleared on topology changes. +pub struct DiscoveryBackoff { + /// Maps target → (suppress_until, consecutive_failures). + pub(crate) entries: BTreeMap, + /// Base backoff in milliseconds (first failure). + base_ms: u64, + /// Maximum backoff cap in milliseconds. + max_ms: u64, +} + +pub(crate) struct BackoffEntry { + /// Don't re-initiate until this time (injected `now_ms`). + pub(crate) suppress_until_ms: u64, + /// Consecutive failures (drives exponential backoff). + failures: u32, +} + +impl DiscoveryBackoff { + /// Create with default parameters (disabled — base/cap = 0). + pub fn new() -> Self { + Self::with_params(DEFAULT_BACKOFF_BASE_SECS, DEFAULT_BACKOFF_MAX_SECS) + } + + /// Create with custom base and max backoff in seconds. + pub fn with_params(base_secs: u64, max_secs: u64) -> Self { + Self { + entries: BTreeMap::new(), + base_ms: base_secs * 1000, + max_ms: max_secs * 1000, + } + } + + /// Check if a lookup for this target is suppressed. + /// + /// Returns true if the target is in backoff and should not be + /// looked up yet. + pub fn is_suppressed(&self, target: &NodeAddr, now_ms: u64) -> bool { + if let Some(e) = self.entries.get(target) { + now_ms < e.suppress_until_ms + } else { + false + } + } + + /// Record a lookup failure (timeout) for a target. + /// + /// Increments the failure count and sets the next suppression + /// window using exponential backoff. + pub fn record_failure(&mut self, target: &NodeAddr, now_ms: u64) { + let failures = self.entries.get(target).map_or(0, |e| e.failures) + 1; + + let backoff_ms = self + .base_ms + .saturating_mul(BACKOFF_MULTIPLIER.saturating_pow(failures.saturating_sub(1))) + .min(self.max_ms); + + self.entries.insert( + *target, + BackoffEntry { + suppress_until_ms: now_ms + backoff_ms, + failures, + }, + ); + } + + /// Record a successful lookup — remove backoff for this target. + pub fn record_success(&mut self, target: &NodeAddr) { + self.entries.remove(target); + } + + /// Clear all backoff entries. + /// + /// Called on topology changes that might make previously-unreachable + /// targets reachable (parent change, new peer, first RTT, reconnection). + pub fn reset_all(&mut self) { + self.entries.clear(); + } + + /// Whether any entries exist. + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Current number of entries. + pub fn entry_count(&self) -> usize { + self.entries.len() + } + + /// Get the failure count for a target (for logging). + pub fn failure_count(&self, target: &NodeAddr) -> u32 { + self.entries.get(target).map_or(0, |e| e.failures) + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.entries.len() + } +} + +impl Default for DiscoveryBackoff { + fn default() -> Self { + Self::new() + } +} + +// ============================================================================ +// Transit-side: Discovery Forward Rate Limiter +// ============================================================================ + +/// Default minimum interval between forwarded lookups for the same target. +const DEFAULT_FORWARD_MIN_INTERVAL_MS: u64 = 2_000; + +/// Maximum age of entries before cleanup. +const FORWARD_MAX_AGE_MS: u64 = 60_000; + +/// Rate limiter for forwarded discovery requests. +/// +/// Tracks the last time a LookupRequest was forwarded for each target +/// and enforces a minimum interval to prevent floods from misbehaving +/// nodes generating fresh request_ids. +pub struct DiscoveryForwardRateLimiter { + last_forwarded: BTreeMap, + min_interval_ms: u64, + max_age_ms: u64, +} + +impl DiscoveryForwardRateLimiter { + /// Create with default parameters (2s interval). + pub fn new() -> Self { + Self { + last_forwarded: BTreeMap::new(), + min_interval_ms: DEFAULT_FORWARD_MIN_INTERVAL_MS, + max_age_ms: FORWARD_MAX_AGE_MS, + } + } + + /// Create with a custom minimum interval in milliseconds. + pub fn with_interval_ms(min_interval_ms: u64) -> Self { + Self { + last_forwarded: BTreeMap::new(), + min_interval_ms, + max_age_ms: FORWARD_MAX_AGE_MS, + } + } + + /// Check if we should forward a lookup for this target. + /// + /// Returns true if enough time has passed since the last forward + /// for this target. Updates internal state when returning true. + pub fn should_forward(&mut self, target: &NodeAddr, now_ms: u64) -> bool { + if let Some(&last) = self.last_forwarded.get(target) + && now_ms.saturating_sub(last) < self.min_interval_ms + { + return false; + } + + self.last_forwarded.insert(*target, now_ms); + self.cleanup(now_ms); + true + } + + /// Replace the minimum interval in milliseconds (e.g., set to zero to disable). + #[cfg(test)] + pub fn set_interval_ms(&mut self, interval_ms: u64) { + self.min_interval_ms = interval_ms; + } + + /// Remove entries older than max_age. + pub(crate) fn cleanup(&mut self, now_ms: u64) { + self.last_forwarded + .retain(|_, &mut last| now_ms.saturating_sub(last) < self.max_age_ms); + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.last_forwarded.len() + } +} + +impl Default for DiscoveryForwardRateLimiter { + fn default() -> Self { + Self::new() + } +} diff --git a/src/proto/discovery/mod.rs b/src/proto/discovery/mod.rs new file mode 100644 index 0000000..f3c68cd --- /dev/null +++ b/src/proto/discovery/mod.rs @@ -0,0 +1,33 @@ +//! Sans-IO discovery protocol state. +//! +//! Pure, runtime-agnostic discovery state and rate limiting, migrated out +//! of the async node shell. The async I/O handlers remain in +//! `node::handlers::discovery`. The discovery wire codec now lives here in +//! `wire.rs` (the `LookupRequest` / `LookupResponse` structs), per the +//! wire-migrates-with-subsystem policy. +//! +//! The sans-IO decision core lives in `core.rs`: it defines the `RoutingView` +//! read-seam trait plus the `plan_forward` / `plan_initiate` LookupRequest +//! planners and their `DiscoveryAction` / `ForwardOutcome` types. The async +//! shell decodes wire +//! bytes, calls the planner, and drives the returned actions. + +mod core; +mod limits; +mod state; +mod wire; + +#[cfg(test)] +mod tests; + +pub(crate) use core::{ + DiscoveryAction, ForwardOutcome, InitiateDecision, RequestOutcome, ResponseRoute, + ResponseRouteDecision, RoutingView, classify_request, classify_response, initiate_failed, + initiate_gate, on_response_accepted, plan_forward, plan_initiate, plan_response_route, + poll_pending, +}; +pub(crate) use limits::{DiscoveryBackoff, DiscoveryForwardRateLimiter}; +#[cfg(test)] +pub(crate) use state::RecentRequest; +pub(crate) use state::{Discovery, PendingLookup}; +pub use wire::{LookupRequest, LookupResponse}; diff --git a/src/proto/discovery/state.rs b/src/proto/discovery/state.rs new file mode 100644 index 0000000..aa188f0 --- /dev/null +++ b/src/proto/discovery/state.rs @@ -0,0 +1,110 @@ +//! Discovery-subsystem state owned by [`Node`](crate::node::Node). +//! +//! Groups the four discovery-related state fields (recent-request dedup +//! cache, in-flight lookups, originator-side backoff, transit-side forward +//! rate limiter) behind a single struct so the discovery handlers can +//! evolve toward a sans-IO core without threading four fields through +//! `Node`. + +use alloc::collections::BTreeMap; + +use super::limits::{DiscoveryBackoff, DiscoveryForwardRateLimiter}; +use crate::NodeAddr; + +/// Recent request tracking for dedup and reverse-path forwarding. +/// +/// When a LookupRequest is forwarded through a node, the node stores the +/// request_id and which peer sent it. When the corresponding LookupResponse +/// arrives, it's forwarded back to that peer (reverse-path forwarding). +/// The `response_forwarded` flag prevents response routing loops. +#[derive(Clone, Debug)] +pub(crate) struct RecentRequest { + /// The peer who sent this request to us. + pub(crate) from_peer: NodeAddr, + /// When we received this request (Unix milliseconds). + pub(crate) timestamp_ms: u64, + /// Whether we've already forwarded a response for this request. + /// Prevents response routing loops when convergent request paths + /// create bidirectional entries in recent_requests. + pub(crate) response_forwarded: bool, +} + +impl RecentRequest { + pub(crate) fn new(from_peer: NodeAddr, timestamp_ms: u64) -> Self { + Self { + from_peer, + timestamp_ms, + response_forwarded: false, + } + } + + /// Check if this entry has expired (older than expiry_ms). + pub(crate) fn is_expired(&self, current_time_ms: u64, expiry_ms: u64) -> bool { + current_time_ms.saturating_sub(self.timestamp_ms) > expiry_ms + } +} + +/// Tracks a pending discovery lookup with retry state. +pub struct PendingLookup { + /// When the lookup was first initiated. + pub initiated_ms: u64, + /// When the last attempt was sent. + pub last_sent_ms: u64, + /// Current attempt number (1 = initial, 2 = first retry, ...). + pub attempt: u8, +} + +impl PendingLookup { + pub fn new(now_ms: u64) -> Self { + Self { + initiated_ms: now_ms, + last_sent_ms: now_ms, + attempt: 1, + } + } +} + +/// Discovery-subsystem state. +pub(crate) struct Discovery { + /// Recent discovery requests (dedup + reverse-path forwarding). + /// Maps request_id → RecentRequest. + pub(crate) recent_requests: BTreeMap, + /// Tracks in-flight discovery lookups. Maps target NodeAddr to the + /// initiation timestamp (Unix ms). Prevents duplicate flood queries. + pub(crate) pending_lookups: BTreeMap, + /// Backoff for failed discovery lookups (originator-side). + pub(crate) backoff: DiscoveryBackoff, + /// Rate limiter for forwarded discovery requests (transit-side). + pub(crate) forward_limiter: DiscoveryForwardRateLimiter, +} + +impl Discovery { + /// Create discovery state with the given backoff and forward limiter. + /// + /// The two limiters are constructed by the caller so each `Node` + /// constructor can supply its own configured/default variant, matching + /// the pre-refactor initialization exactly. + pub(crate) fn new( + backoff: DiscoveryBackoff, + forward_limiter: DiscoveryForwardRateLimiter, + ) -> Self { + Self { + recent_requests: BTreeMap::new(), + pending_lookups: BTreeMap::new(), + backoff, + forward_limiter, + } + } + + /// Reset discovery backoff on topology changes. Returns the number of + /// entries cleared (0 if already empty) so the shell can log the reset — + /// observability stays out of the pure core. + pub(crate) fn reset_backoff(&mut self) -> usize { + if self.backoff.is_empty() { + return 0; + } + let cleared = self.backoff.entry_count(); + self.backoff.reset_all(); + cleared + } +} diff --git a/src/proto/discovery/tests/core.rs b/src/proto/discovery/tests/core.rs new file mode 100644 index 0000000..5958ed7 --- /dev/null +++ b/src/proto/discovery/tests/core.rs @@ -0,0 +1,626 @@ +//! Tests for the sans-IO discovery decision core. + +use super::util::{ + MockRoutingView, action_peers, empty_discovery, make_request, make_request_id, + suppressing_discovery, +}; +use crate::TreeCoordinate; +use crate::proto::discovery::*; +use crate::testutil::make_node_addr; + +#[test] +fn picks_only_tree_peers_when_a_tree_peer_matches() { + let tree_peer = make_node_addr(1); + let non_tree_peer = make_node_addr(2); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, true), (non_tree_peer, false, true)], + ..Default::default() + }; + let mut request = make_request(3); + match plan_forward(&mut request, &rv) { + ForwardOutcome::Forward { + actions, + used_fallback, + } => { + assert!(!used_fallback, "tree match must not use fallback"); + assert_eq!(action_peers(&actions), vec![tree_peer]); + } + _ => panic!("expected Forward"), + } +} + +#[test] +fn falls_back_to_non_tree_peers_when_no_tree_peer_matches() { + let non_tree_a = make_node_addr(3); + let non_tree_b = make_node_addr(4); + // A tree peer exists but does not reach the target. + let tree_no_reach = make_node_addr(5); + let rv = MockRoutingView { + peers: vec![ + (tree_no_reach, true, false), + (non_tree_a, false, true), + (non_tree_b, false, true), + ], + ..Default::default() + }; + let mut request = make_request(3); + match plan_forward(&mut request, &rv) { + ForwardOutcome::Forward { + actions, + used_fallback, + } => { + assert!(used_fallback, "no tree match must use fallback"); + assert_eq!(action_peers(&actions), vec![non_tree_a, non_tree_b]); + } + _ => panic!("expected Forward via fallback"), + } +} + +#[test] +fn returns_no_peers_when_nothing_reaches_target() { + let tree_peer = make_node_addr(6); + let non_tree_peer = make_node_addr(7); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, false), (non_tree_peer, false, false)], + ..Default::default() + }; + let mut request = make_request(3); + assert!(matches!( + plan_forward(&mut request, &rv), + ForwardOutcome::NoPeers + )); +} + +#[test] +fn returns_ttl_exhausted_when_ttl_is_zero() { + let tree_peer = make_node_addr(8); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, true)], + ..Default::default() + }; + let mut request = make_request(0); + assert!(matches!( + plan_forward(&mut request, &rv), + ForwardOutcome::TtlExhausted + )); +} + +#[test] +fn initiate_picks_only_tree_peers_and_never_falls_back() { + let tree_peer = make_node_addr(1); + let non_tree_peer = make_node_addr(2); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, true), (non_tree_peer, false, true)], + ..Default::default() + }; + let request = make_request(3); + let actions = plan_initiate(&request, &rv); + assert_eq!(action_peers(&actions), vec![tree_peer]); +} + +#[test] +fn initiate_returns_empty_when_only_non_tree_peers_reach() { + // A tree peer exists but cannot reach the target; only cross-links reach. + // plan_forward would fall back to the cross-links here; plan_initiate does + // NOT — the tree-only origination gap preserved for behavior-neutrality. + let tree_no_reach = make_node_addr(5); + let non_tree_a = make_node_addr(3); + let non_tree_b = make_node_addr(4); + let rv = MockRoutingView { + peers: vec![ + (tree_no_reach, true, false), + (non_tree_a, false, true), + (non_tree_b, false, true), + ], + ..Default::default() + }; + let request = make_request(3); + assert!(plan_initiate(&request, &rv).is_empty()); +} + +#[test] +fn initiate_returns_empty_when_nothing_reaches_target() { + let tree_peer = make_node_addr(6); + let non_tree_peer = make_node_addr(7); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, false), (non_tree_peer, false, false)], + ..Default::default() + }; + let request = make_request(3); + assert!(plan_initiate(&request, &rv).is_empty()); +} + +#[test] +fn leaf_node_does_not_forward() { + let tree_peer = make_node_addr(1); + let rv = MockRoutingView { + peers: vec![(tree_peer, true, true)], + leaf: true, + ..Default::default() + }; + let mut request = make_request(3); + assert!(matches!( + plan_forward(&mut request, &rv), + ForwardOutcome::LeafNoForward + )); +} + +#[test] +fn forward_excludes_non_full_peers() { + let full_tree = make_node_addr(1); + let lite_tree = make_node_addr(2); + let rv = MockRoutingView { + peers: vec![(full_tree, true, true), (lite_tree, true, true)], + not_full: vec![lite_tree], + ..Default::default() + }; + let mut request = make_request(3); + match plan_forward(&mut request, &rv) { + ForwardOutcome::Forward { + actions, + used_fallback, + } => { + assert!(!used_fallback); + assert_eq!(action_peers(&actions), vec![full_tree]); + } + _ => panic!("expected Forward to the Full tree peer only"), + } +} + +#[test] +fn forward_excludes_peers_below_min_mtu() { + let ok_tree = make_node_addr(1); + let small_tree = make_node_addr(2); + let rv = MockRoutingView { + peers: vec![(ok_tree, true, true), (small_tree, true, true)], + mtu_fail: vec![small_tree], + ..Default::default() + }; + let mut request = make_request(3); + match plan_forward(&mut request, &rv) { + ForwardOutcome::Forward { + actions, + used_fallback, + } => { + assert!(!used_fallback); + assert_eq!(action_peers(&actions), vec![ok_tree]); + } + _ => panic!("expected Forward to the MTU-satisfying tree peer only"), + } +} + +#[test] +fn initiate_excludes_non_full_and_mtu_fail_peers() { + let good = make_node_addr(1); + let lite = make_node_addr(2); + let small = make_node_addr(3); + let rv = MockRoutingView { + peers: vec![(good, true, true), (lite, true, true), (small, true, true)], + not_full: vec![lite], + mtu_fail: vec![small], + ..Default::default() + }; + let request = make_request(3); + let actions = plan_initiate(&request, &rv); + assert_eq!(action_peers(&actions), vec![good]); +} + +#[test] +fn response_route_uses_recorded_reverse_path() { + let mut disc = empty_discovery(); + let from = make_node_addr(9); + disc.recent_requests.insert(42, RecentRequest::new(from, 0)); + match plan_response_route(&disc, 42) { + ResponseRouteDecision::ReversePath(peer) => assert_eq!(peer, from), + ResponseRouteDecision::NeedsTreeRoute => panic!("expected ReversePath"), + } +} + +#[test] +fn response_route_needs_tree_route_when_no_record() { + let disc = empty_discovery(); + assert!(matches!( + plan_response_route(&disc, 42), + ResponseRouteDecision::NeedsTreeRoute + )); +} + +#[test] +fn classify_response_transit_on_fresh_forwarded_request() { + let from_peer = make_node_addr(0x11); + let mut disc = empty_discovery(); + disc.recent_requests + .insert(42, RecentRequest::new(from_peer, 1000)); + + match classify_response(&mut disc, 42) { + ResponseRoute::Transit { from_peer: peer } => assert_eq!(peer, from_peer), + _ => panic!("expected Transit"), + } + // The dedup flag must flip after the first transit. + assert!(disc.recent_requests.get(&42).unwrap().response_forwarded); +} + +#[test] +fn classify_response_already_forwarded_on_second_call() { + let from_peer = make_node_addr(0x22); + let mut disc = empty_discovery(); + disc.recent_requests + .insert(7, RecentRequest::new(from_peer, 1000)); + + assert!(matches!( + classify_response(&mut disc, 7), + ResponseRoute::Transit { .. } + )); + assert!(matches!( + classify_response(&mut disc, 7), + ResponseRoute::AlreadyForwarded + )); +} + +#[test] +fn classify_response_originator_when_request_absent() { + let mut disc = empty_discovery(); + assert!(matches!( + classify_response(&mut disc, 999), + ResponseRoute::Originator + )); +} + +#[test] +fn on_response_accepted_clears_state_and_emits_effects() { + let target = make_node_addr(0x5A); + let mut disc = empty_discovery(); + + // Seed a backoff entry and a pending lookup for the target. + disc.backoff.record_failure(&target, 1000); + assert!(!disc.backoff.is_empty(), "precondition: backoff seeded"); + disc.pending_lookups + .insert(target, PendingLookup::new(1000)); + assert!(disc.pending_lookups.contains_key(&target)); + + let coords = TreeCoordinate::root(target); + let now_ms = 12_345u64; + let path_mtu = 1400u16; + let actions = on_response_accepted(&mut disc, &target, coords, now_ms, path_mtu); + + // Success state must be cleared. + assert!( + disc.backoff.is_empty(), + "backoff entry must clear on success" + ); + assert!( + !disc.pending_lookups.contains_key(&target), + "pending lookup must be dropped" + ); + + // Exactly the four effect actions, in order. + assert_eq!(actions.len(), 4, "expected four effect actions"); + match &actions[0] { + DiscoveryAction::CacheCoords { + target: t, + now_ms: n, + path_mtu: p, + .. + } => { + assert_eq!(*t, target); + assert_eq!(*n, now_ms); + assert_eq!(*p, path_mtu); + } + _ => panic!("action[0] must be CacheCoords"), + } + match &actions[1] { + DiscoveryAction::WritePathMtu { + target: t, + path_mtu: p, + } => { + assert_eq!(*t, target); + assert_eq!(*p, path_mtu); + } + _ => panic!("action[1] must be WritePathMtu"), + } + match &actions[2] { + DiscoveryAction::ResetWarmupIfEstablished { target: t } => assert_eq!(*t, target), + _ => panic!("action[2] must be ResetWarmupIfEstablished"), + } + match &actions[3] { + DiscoveryAction::RetryQueuedPackets { target: t } => assert_eq!(*t, target), + _ => panic!("action[3] must be RetryQueuedPackets"), + } +} + +#[test] +fn poll_pending_no_action_before_first_deadline() { + let target = make_node_addr(0x30); + let mut disc = empty_discovery(); + let t0 = 10_000u64; + disc.pending_lookups.insert(target, PendingLookup::new(t0)); + + // Just before the attempt-1 deadline (1s): nothing fires. + let outcome = poll_pending(&mut disc, t0 + 999, &[1, 2, 4, 8]); + assert!(outcome.retries.is_empty(), "no retry before deadline"); + assert!(outcome.timeouts.is_empty(), "no timeout before deadline"); + + // Entry unchanged. + let entry = disc.pending_lookups.get(&target).unwrap(); + assert_eq!(entry.attempt, 1); + assert_eq!(entry.last_sent_ms, t0); +} + +#[test] +fn poll_pending_retries_at_first_deadline() { + let target = make_node_addr(0x31); + let mut disc = empty_discovery(); + let t0 = 10_000u64; + disc.pending_lookups.insert(target, PendingLookup::new(t0)); + + // At the attempt-1 deadline (t0 + 1000): one retry to attempt 2. + let outcome = poll_pending(&mut disc, t0 + 1000, &[1, 2, 4, 8]); + assert_eq!(outcome.retries, vec![(target, 2)]); + assert!(outcome.timeouts.is_empty()); + + // Entry mutated: attempt bumped, last_sent refreshed. + let entry = disc.pending_lookups.get(&target).unwrap(); + assert_eq!(entry.attempt, 2); + assert_eq!(entry.last_sent_ms, t0 + 1000); +} + +#[test] +fn poll_pending_final_timeout_at_max_attempt() { + let target = make_node_addr(0x32); + let mut disc = empty_discovery(); + + // Drive the entry to attempt == max (4) with a known last_sent. + let tn = 50_000u64; + let mut entry = PendingLookup::new(tn); + entry.attempt = 4; + entry.last_sent_ms = tn; + disc.pending_lookups.insert(target, entry); + + // attempt_timeouts_secs[3] == 8 → deadline at tn + 8000. + let outcome = poll_pending(&mut disc, tn + 8000, &[1, 2, 4, 8]); + assert!(outcome.retries.is_empty(), "max attempt cannot retry"); + assert_eq!( + outcome.timeouts, + vec![(target, 1)], + "one timeout, failure #1" + ); + + // Entry removed and a backoff failure recorded. + assert!( + !disc.pending_lookups.contains_key(&target), + "timed-out entry must be removed" + ); + assert_eq!(disc.backoff.failure_count(&target), 1); +} + +// --- classify_request tests --- + +#[test] +fn classify_request_forwards_fresh_and_records_it() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + let request = make_request_id(1, target, 3); + + let outcome = classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096); + assert!(matches!(outcome, RequestOutcome::Forward)); + // Recorded for reverse-path forwarding. + assert!(disc.recent_requests.contains_key(&1)); + assert_eq!(disc.recent_requests.get(&1).unwrap().from_peer, from); +} + +#[test] +fn classify_request_duplicate_on_second_call() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + let request = make_request_id(1, target, 3); + + assert!(matches!( + classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096), + RequestOutcome::Forward + )); + assert!(matches!( + classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096), + RequestOutcome::Duplicate + )); +} + +#[test] +fn classify_request_dedup_cache_full() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + // Fill the cache to max_recent with distinct request_ids. + let max_recent = 3usize; + for id in 100..(100 + max_recent as u64) { + disc.recent_requests + .insert(id, RecentRequest::new(from, 1000)); + } + assert_eq!(disc.recent_requests.len(), max_recent); + + let request = make_request_id(1, target, 3); + match classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, max_recent) { + RequestOutcome::DedupCacheFull { len } => assert_eq!(len, max_recent), + _ => panic!("expected DedupCacheFull"), + } + // The new request must not have been recorded on the drop path. + assert!(!disc.recent_requests.contains_key(&1)); +} + +#[test] +fn classify_request_respond_as_target() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0xAA); + // target == my_addr + let request = make_request_id(1, my_addr, 3); + + assert!(matches!( + classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096), + RequestOutcome::RespondAsTarget + )); + // Recorded before the target decision. + assert!(disc.recent_requests.contains_key(&1)); +} + +#[test] +fn classify_request_ttl_exhausted_for_non_target() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + // ttl 0 → not forwardable, and not the target. + let request = make_request_id(1, target, 0); + + assert!(matches!( + classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096), + RequestOutcome::TtlExhausted + )); +} + +#[test] +fn classify_request_forward_rate_limited() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + // Pre-seed the forward limiter so should_forward(target) returns false + // on the next call within the (default 2s) min interval. + assert!(disc.forward_limiter.should_forward(&target, 1000)); + + let request = make_request_id(1, target, 3); + assert!(matches!( + classify_request(&mut disc, &request, &from, &my_addr, 1000, 5000, 4096), + RequestOutcome::ForwardRateLimited + )); +} + +#[test] +fn classify_request_purges_expired_entries() { + let mut disc = empty_discovery(); + let from = make_node_addr(0x01); + let my_addr = make_node_addr(0x99); + let target = make_node_addr(0xAA); + // Seed an entry that is expired at now_ms with the given expiry window. + // is_expired: now - timestamp > expiry_ms → expired. + disc.recent_requests + .insert(55, RecentRequest::new(from, 1000)); + // now_ms = 10_000, expiry_ms = 5000 → 9000 > 5000 → expired. + let request = make_request_id(1, target, 3); + let outcome = classify_request(&mut disc, &request, &from, &my_addr, 10_000, 5000, 4096); + assert!(matches!(outcome, RequestOutcome::Forward)); + // The expired entry (55) must have been purged. + assert!(!disc.recent_requests.contains_key(&55)); + // The fresh request is recorded. + assert!(disc.recent_requests.contains_key(&1)); +} + +#[test] +fn poll_pending_full_ladder_end_to_end() { + let target = make_node_addr(0x33); + let mut disc = empty_discovery(); + let t0 = 0u64; + disc.pending_lookups.insert(target, PendingLookup::new(t0)); + let ladder = [1u64, 2, 4, 8]; + + // attempt 1 → 2 at deadline 1s + let o = poll_pending(&mut disc, t0 + 1000, &ladder); + assert_eq!(o.retries, vec![(target, 2)]); + // attempt 2 → 3 at deadline 2s after last send + let o = poll_pending(&mut disc, t0 + 1000 + 2000, &ladder); + assert_eq!(o.retries, vec![(target, 3)]); + // attempt 3 → 4 at deadline 4s after last send + let o = poll_pending(&mut disc, t0 + 1000 + 2000 + 4000, &ladder); + assert_eq!(o.retries, vec![(target, 4)]); + // attempt 4 is max → final timeout at deadline 8s after last send + let last = t0 + 1000 + 2000 + 4000; + let o = poll_pending(&mut disc, last + 8000, &ladder); + assert!(o.retries.is_empty()); + assert_eq!(o.timeouts, vec![(target, 1)]); + assert!(!disc.pending_lookups.contains_key(&target)); +} + +// --- initiate_gate / initiate_failed tests --- + +#[test] +fn initiate_gate_deduplicated_when_pending() { + let dest = make_node_addr(0x40); + let mut disc = empty_discovery(); + disc.pending_lookups.insert(dest, PendingLookup::new(500)); + + // reachable=true would otherwise Proceed, but the pending entry wins. + assert!(matches!( + initiate_gate(&mut disc, &dest, 1000, true), + InitiateDecision::Deduplicated + )); +} + +#[test] +fn initiate_gate_suppressed_by_backoff() { + let dest = make_node_addr(0x41); + let mut disc = suppressing_discovery(); + // One failure arms suppression under with_params(30, 300). + disc.backoff.record_failure(&dest, 1000); + assert!( + disc.backoff.is_suppressed(&dest, 1000), + "precondition: suppressed" + ); + + match initiate_gate(&mut disc, &dest, 1000, true) { + InitiateDecision::Suppressed { failures } => assert_eq!(failures, 1), + _ => panic!("expected Suppressed"), + } + // No pending entry was inserted on the suppress path. + assert!(!disc.pending_lookups.contains_key(&dest)); +} + +#[test] +fn initiate_gate_bloom_miss_records_failure() { + let dest = make_node_addr(0x42); + let mut disc = empty_discovery(); + + assert!(matches!( + initiate_gate(&mut disc, &dest, 1000, false), + InitiateDecision::BloomMiss + )); + // A backoff failure was recorded, and no pending entry created. + assert_eq!(disc.backoff.failure_count(&dest), 1); + assert!(!disc.pending_lookups.contains_key(&dest)); +} + +#[test] +fn initiate_gate_proceed_inserts_pending() { + let dest = make_node_addr(0x43); + let mut disc = empty_discovery(); + let now_ms = 7_777u64; + + assert!(matches!( + initiate_gate(&mut disc, &dest, now_ms, true), + InitiateDecision::Proceed + )); + // The pending entry now exists, stamped with now_ms. + let entry = disc + .pending_lookups + .get(&dest) + .expect("Proceed must insert a pending lookup"); + assert_eq!(entry.last_sent_ms, now_ms); + assert_eq!(entry.attempt, 1); +} + +#[test] +fn initiate_failed_drops_pending_and_records_failure() { + let dest = make_node_addr(0x44); + let mut disc = empty_discovery(); + disc.pending_lookups.insert(dest, PendingLookup::new(1000)); + + initiate_failed(&mut disc, &dest, 1000); + assert!( + !disc.pending_lookups.contains_key(&dest), + "pending entry must be dropped" + ); + assert_eq!(disc.backoff.failure_count(&dest), 1); +} diff --git a/src/proto/discovery/tests/limits.rs b/src/proto/discovery/tests/limits.rs new file mode 100644 index 0000000..2422284 --- /dev/null +++ b/src/proto/discovery/tests/limits.rs @@ -0,0 +1,151 @@ +//! Tests for discovery rate limiting and backoff. + +use crate::proto::discovery::{DiscoveryBackoff, DiscoveryForwardRateLimiter}; +use crate::testutil::make_node_addr as addr; + +// --- DiscoveryBackoff tests --- + +#[test] +fn test_backoff_not_suppressed_initially() { + let backoff = DiscoveryBackoff::new(); + assert!(!backoff.is_suppressed(&addr(1), 0)); +} + +#[test] +fn test_backoff_suppressed_after_failure() { + // Backoff is opt-in; exercise the suppression path with explicit params. + let now = 1_000; + let mut backoff = DiscoveryBackoff::with_params(30, 300); + backoff.record_failure(&addr(1), now); + assert!(backoff.is_suppressed(&addr(1), now)); + // Different target not affected + assert!(!backoff.is_suppressed(&addr(2), now)); +} + +#[test] +fn test_backoff_cleared_on_success() { + let now = 1_000; + let mut backoff = DiscoveryBackoff::with_params(30, 300); + backoff.record_failure(&addr(1), now); + assert!(backoff.is_suppressed(&addr(1), now)); + + backoff.record_success(&addr(1)); + assert!(!backoff.is_suppressed(&addr(1), now)); +} + +#[test] +fn test_backoff_reset_all() { + let now = 1_000; + let mut backoff = DiscoveryBackoff::new(); + backoff.record_failure(&addr(1), now); + backoff.record_failure(&addr(2), now); + assert_eq!(backoff.len(), 2); + + backoff.reset_all(); + assert_eq!(backoff.len(), 0); + assert!(!backoff.is_suppressed(&addr(1), now)); +} + +#[test] +fn test_backoff_exponential() { + let now = 1_000; + let mut backoff = DiscoveryBackoff::with_params(1, 300); + + // First failure: 1s backoff + backoff.record_failure(&addr(1), now); + assert_eq!(backoff.failure_count(&addr(1)), 1); + + // Second failure: 2s backoff + backoff.record_failure(&addr(1), now); + assert_eq!(backoff.failure_count(&addr(1)), 2); + + // Third failure: 4s backoff + backoff.record_failure(&addr(1), now); + assert_eq!(backoff.failure_count(&addr(1)), 3); +} + +#[test] +fn test_backoff_expires() { + let now = 1_000; + let mut backoff = DiscoveryBackoff::with_params(0, 0); + backoff.record_failure(&addr(1), now); + // With 0s backoff, should not be suppressed + assert!(!backoff.is_suppressed(&addr(1), now)); +} + +#[test] +fn test_backoff_capped() { + let now = 1_000; + let mut backoff = DiscoveryBackoff::with_params(1, 10); + + // Record many failures + for _ in 0..20 { + backoff.record_failure(&addr(1), now); + } + + // Backoff should be capped at max (10s = 10_000ms), not overflow + let entry = backoff.entries.get(&addr(1)).unwrap(); + let remaining = entry.suppress_until_ms - now; + assert!(remaining <= 11_000); +} + +// --- DiscoveryForwardRateLimiter tests --- + +#[test] +fn test_forward_first_allowed() { + let mut limiter = DiscoveryForwardRateLimiter::new(); + assert!(limiter.should_forward(&addr(1), 0)); +} + +#[test] +fn test_forward_rapid_rate_limited() { + let now = 1_000; + let mut limiter = DiscoveryForwardRateLimiter::new(); + assert!(limiter.should_forward(&addr(1), now)); + assert!(!limiter.should_forward(&addr(1), now)); + assert!(!limiter.should_forward(&addr(1), now)); +} + +#[test] +fn test_forward_different_targets_independent() { + let now = 1_000; + let mut limiter = DiscoveryForwardRateLimiter::new(); + assert!(limiter.should_forward(&addr(1), now)); + assert!(limiter.should_forward(&addr(2), now)); + assert!(!limiter.should_forward(&addr(1), now)); + assert!(!limiter.should_forward(&addr(2), now)); +} + +#[test] +fn test_forward_allowed_after_interval() { + let now = 1_000; + let mut limiter = DiscoveryForwardRateLimiter::with_interval_ms(100); + assert!(limiter.should_forward(&addr(1), now)); + + // Advance past the minimum interval. + assert!(limiter.should_forward(&addr(1), now + 110)); +} + +#[test] +fn test_forward_cleanup_removes_old() { + let now = 1_000; + let mut limiter = DiscoveryForwardRateLimiter::new(); + assert!(limiter.should_forward(&addr(1), now)); + assert!(limiter.should_forward(&addr(2), now)); + assert_eq!(limiter.len(), 2); + + let future = now + 61_000; + limiter.cleanup(future); + assert_eq!(limiter.len(), 0); +} + +#[test] +fn test_forward_cleanup_preserves_recent() { + let now = 1_000; + let mut limiter = DiscoveryForwardRateLimiter::new(); + assert!(limiter.should_forward(&addr(1), now)); + assert_eq!(limiter.len(), 1); + + limiter.cleanup(now); + assert_eq!(limiter.len(), 1); +} diff --git a/src/proto/discovery/tests/mod.rs b/src/proto/discovery/tests/mod.rs new file mode 100644 index 0000000..9e7a06d --- /dev/null +++ b/src/proto/discovery/tests/mod.rs @@ -0,0 +1,7 @@ +//! Discovery subsystem unit tests, extracted from the co-located `#[cfg(test)]` +//! blocks in the sibling source modules. Shared helpers live in `util`. + +mod core; +mod limits; +mod util; +mod wire; diff --git a/src/proto/discovery/tests/util.rs b/src/proto/discovery/tests/util.rs new file mode 100644 index 0000000..2f55648 --- /dev/null +++ b/src/proto/discovery/tests/util.rs @@ -0,0 +1,112 @@ +//! Shared test helpers for the discovery subsystem unit tests. + +use sha2::Digest; + +use crate::proto::discovery::{ + Discovery, DiscoveryAction, DiscoveryBackoff, DiscoveryForwardRateLimiter, LookupRequest, + LookupResponse, RoutingView, +}; +use crate::testutil::make_node_addr; +use crate::{NodeAddr, TreeCoordinate}; + +/// Mock routing view: each `peers` entry is `(addr, is_tree, may_reach)`. The +/// FMP predicates default to eligible — a peer is Full and meets MTU, and the +/// node is not a Leaf — unless named in `not_full` / `mtu_fail` / `leaf`, so +/// tests that only exercise tree/fallback selection need not set them. +#[derive(Default)] +pub(super) struct MockRoutingView { + pub(super) peers: Vec<(NodeAddr, bool, bool)>, + pub(super) leaf: bool, + pub(super) not_full: Vec, + pub(super) mtu_fail: Vec, +} + +impl RoutingView for MockRoutingView { + fn is_tree_peer(&self, addr: &NodeAddr) -> bool { + self.peers + .iter() + .find(|(a, _, _)| a == addr) + .map(|(_, is_tree, _)| *is_tree) + .unwrap_or(false) + } + fn peers_reaching(&self, _target: &NodeAddr) -> Vec { + self.peers + .iter() + .filter(|(_, _, may_reach)| *may_reach) + .map(|(a, _, _)| *a) + .collect() + } + fn node_is_leaf(&self) -> bool { + self.leaf + } + fn peer_is_full(&self, addr: &NodeAddr) -> bool { + !self.not_full.contains(addr) + } + fn peer_meets_mtu(&self, addr: &NodeAddr, _min_mtu: u16) -> bool { + !self.mtu_fail.contains(addr) + } +} + +pub(super) fn make_request(ttl: u8) -> LookupRequest { + let target = make_node_addr(0xAA); + let origin = make_node_addr(0xBB); + LookupRequest::new(1, target, origin, ttl, 0) +} + +/// Build a request with an explicit request_id and target. +pub(super) fn make_request_id(request_id: u64, target: NodeAddr, ttl: u8) -> LookupRequest { + let origin = make_node_addr(0xBB); + LookupRequest::new(request_id, target, origin, ttl, 0) +} + +pub(super) fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::from_addrs(ids.iter().map(|&v| make_node_addr(v)).collect()).unwrap() +} + +pub(super) fn action_peers(actions: &[DiscoveryAction]) -> Vec { + actions + .iter() + .map(|action| match action { + DiscoveryAction::SendLink { peer, .. } => *peer, + _ => panic!("expected SendLink, got a different action variant"), + }) + .collect() +} + +pub(super) fn empty_discovery() -> Discovery { + Discovery::new( + DiscoveryBackoff::default(), + DiscoveryForwardRateLimiter::default(), + ) +} + +/// A Discovery whose backoff is armed (non-zero base/cap) so that a single +/// recorded failure suppresses the target — the default backoff is inert. +pub(super) fn suppressing_discovery() -> Discovery { + Discovery::new( + DiscoveryBackoff::with_params(30, 300), + DiscoveryForwardRateLimiter::default(), + ) +} + +/// Build a `LookupResponse` carrying a valid schnorr proof over its own +/// `proof_bytes`, factoring out the secp/sha256/sign_schnorr setup shared by +/// the wire response roundtrip tests. `path_mtu` is the default `u16::MAX`. +pub(super) fn signed_response( + request_id: u64, + target: &NodeAddr, + coords: &TreeCoordinate, +) -> LookupResponse { + use secp256k1::Secp256k1; + + let secp = Secp256k1::new(); + let mut secret_bytes = [0u8; 32]; + rand::Rng::fill_bytes(&mut rand::rng(), &mut secret_bytes); + let secret_key = secp256k1::SecretKey::from_slice(&secret_bytes) + .expect("32 random bytes is a valid secret key"); + let keypair = secp256k1::Keypair::from_secret_key(&secp, &secret_key); + let proof_data = LookupResponse::proof_bytes(request_id, target, coords); + let digest: [u8; 32] = sha2::Sha256::digest(&proof_data).into(); + let sig = secp.sign_schnorr(&digest, &keypair); + LookupResponse::new(request_id, *target, coords.clone(), sig) +} diff --git a/src/proto/discovery/tests/wire.rs b/src/proto/discovery/tests/wire.rs new file mode 100644 index 0000000..9c82043 --- /dev/null +++ b/src/proto/discovery/tests/wire.rs @@ -0,0 +1,238 @@ +//! Tests for the discovery wire codec (`LookupRequest` / `LookupResponse`). + +use super::util::{make_coords, signed_response}; +use crate::proto::discovery::{LookupRequest, LookupResponse}; +use crate::testutil::make_node_addr; + +#[test] +fn test_lookup_request_forward() { + let target = make_node_addr(1); + let origin = make_node_addr(2); + + let mut request = LookupRequest::new(123, target, origin, 5, 0); + + assert!(request.can_forward()); + assert!(request.forward()); + assert_eq!(request.ttl, 4); +} + +#[test] +fn test_lookup_request_ttl_exhausted() { + let target = make_node_addr(1); + let origin = make_node_addr(2); + + let mut request = LookupRequest::new(123, target, origin, 1, 0); + + assert!(request.forward()); + assert!(!request.can_forward()); + assert!(!request.forward()); +} + +#[test] +fn test_lookup_request_generate() { + let target = make_node_addr(1); + let origin = make_node_addr(2); + + let req1 = LookupRequest::generate(target, origin, 5, 0); + let req2 = LookupRequest::generate(target, origin, 5, 0); + + // Random IDs should differ + assert_ne!(req1.request_id, req2.request_id); +} + +#[test] +fn test_lookup_response_proof_bytes() { + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + let bytes = LookupResponse::proof_bytes(12345, &target, &coords); + + // 8 (request_id) + 16 (target) + 2 (count) + 3*16 (coords) = 74 + assert_eq!(bytes.len(), 74); + assert_eq!(&bytes[0..8], &12345u64.to_le_bytes()); + assert_eq!(&bytes[8..24], target.as_bytes()); + + // Verify coordinate encoding is present + let count = u16::from_le_bytes([bytes[24], bytes[25]]); + assert_eq!(count, 3); // 3 entries in coords +} + +#[test] +fn test_lookup_request_encode_decode_roundtrip() { + let target = make_node_addr(10); + let origin = make_node_addr(20); + + let mut request = LookupRequest::new(12345, target, origin, 8, 1386); + request.forward(); + + let encoded = request.encode(); + assert_eq!(encoded[0], 0x30); + + let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.request_id, 12345); + assert_eq!(decoded.target, target); + assert_eq!(decoded.origin, origin); + assert_eq!(decoded.ttl, 7); // decremented by forward() + assert_eq!(decoded.min_mtu, 1386); +} + +#[test] +fn test_lookup_request_decode_too_short() { + assert!(LookupRequest::decode(&[]).is_err()); + assert!(LookupRequest::decode(&[0u8; 42]).is_err()); +} + +#[test] +fn test_lookup_request_min_mtu_boundary_values() { + let target = make_node_addr(10); + let origin = make_node_addr(20); + + for mtu_val in [0u16, 1386, u16::MAX] { + let request = LookupRequest::new(100, target, origin, 5, mtu_val); + let encoded = request.encode(); + let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.min_mtu, mtu_val); + } +} + +#[test] +fn test_lookup_response_encode_decode_roundtrip() { + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + + let response = signed_response(999, &target, &coords); + + // Default path_mtu should be u16::MAX + assert_eq!(response.path_mtu, u16::MAX); + + let encoded = response.encode(); + assert_eq!(encoded[0], 0x31); + + let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.request_id, 999); + assert_eq!(decoded.target, target); + assert_eq!(decoded.path_mtu, u16::MAX); + assert_eq!(decoded.proof, response.proof); +} + +#[test] +fn test_lookup_response_path_mtu_roundtrip() { + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + + let base = signed_response(999, &target, &coords); + + for mtu_val in [0u16, 1280, 1386, 9000, u16::MAX] { + let mut response = base.clone(); + response.path_mtu = mtu_val; + + let encoded = response.encode(); + let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.path_mtu, mtu_val); + } +} + +#[test] +fn test_lookup_response_path_mtu_not_in_proof_bytes() { + // Verify that proof_bytes does NOT include path_mtu + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + + let bytes = LookupResponse::proof_bytes(12345, &target, &coords); + + // proof_bytes format: request_id(8) + target(16) + coords_encoding(2 + 3*16) = 74 + // No path_mtu(2) in here + assert_eq!(bytes.len(), 74); +} + +#[test] +fn test_lookup_response_decode_too_short() { + assert!(LookupResponse::decode(&[]).is_err()); + assert!(LookupResponse::decode(&[0u8; 50]).is_err()); +} + +#[test] +fn test_lookup_request_with_tlv_roundtrip() { + let target = make_node_addr(10); + let origin = make_node_addr(20); + + let request = LookupRequest::new(555, target, origin, 5, 1280) + .with_tlv(1, vec![0xAA, 0xBB]) + .with_tlv(256, vec![0x01, 0x02, 0x03, 0x04]); + + let encoded = request.encode(); + let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.request_id, 555); + assert_eq!(decoded.min_mtu, 1280); + assert_eq!(decoded.tlv_entries.len(), 2); + assert_eq!(decoded.tlv_entries[0].field_num, 1); + assert_eq!(decoded.tlv_entries[0].value, vec![0xAA, 0xBB]); + assert_eq!(decoded.tlv_entries[1].field_num, 256); + assert_eq!(decoded.tlv_entries[1].value, vec![0x01, 0x02, 0x03, 0x04]); +} + +#[test] +fn test_lookup_request_tlv_forward_compat() { + // Unknown field_nums should be preserved through decode → forward → encode. + let target = make_node_addr(10); + let origin = make_node_addr(20); + + let request = + LookupRequest::new(777, target, origin, 5, 0).with_tlv(9999, vec![0xFF, 0xFE, 0xFD]); + + let encoded = request.encode(); + let mut decoded = LookupRequest::decode(&encoded[1..]).unwrap(); + + // Simulate transit: forward then re-encode. + decoded.forward(); + let re_encoded = decoded.encode(); + let final_decoded = LookupRequest::decode(&re_encoded[1..]).unwrap(); + + assert_eq!(final_decoded.ttl, 4); + assert_eq!(final_decoded.tlv_entries.len(), 1); + assert_eq!(final_decoded.tlv_entries[0].field_num, 9999); + assert_eq!(final_decoded.tlv_entries[0].value, vec![0xFF, 0xFE, 0xFD]); +} + +#[test] +fn test_lookup_response_with_tlv_roundtrip() { + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + + let response = signed_response(999, &target, &coords) + .with_tlv(1, vec![0xAA, 0xBB]) + .with_tlv(500, vec![0x01, 0x02, 0x03]); + + let encoded = response.encode(); + let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.request_id, 999); + assert_eq!(decoded.proof, response.proof); + assert_eq!(decoded.tlv_entries.len(), 2); + assert_eq!(decoded.tlv_entries[0].field_num, 1); + assert_eq!(decoded.tlv_entries[0].value, vec![0xAA, 0xBB]); + assert_eq!(decoded.tlv_entries[1].field_num, 500); + assert_eq!(decoded.tlv_entries[1].value, vec![0x01, 0x02, 0x03]); +} + +#[test] +fn test_lookup_response_tlv_forward_compat() { + // Unknown field_nums preserved through decode → modify path_mtu → encode. + let target = make_node_addr(42); + let coords = make_coords(&[42, 1, 0]); + + let response = signed_response(999, &target, &coords).with_tlv(9999, vec![0xFF, 0xFE, 0xFD]); + + let encoded = response.encode(); + let mut decoded = LookupResponse::decode(&encoded[1..]).unwrap(); + + // Simulate transit: modify path_mtu then re-encode. + decoded.path_mtu = 1280; + let re_encoded = decoded.encode(); + let final_decoded = LookupResponse::decode(&re_encoded[1..]).unwrap(); + + assert_eq!(final_decoded.path_mtu, 1280); + assert_eq!(final_decoded.tlv_entries.len(), 1); + assert_eq!(final_decoded.tlv_entries[0].field_num, 9999); + assert_eq!(final_decoded.tlv_entries[0].value, vec![0xFF, 0xFE, 0xFD]); +} diff --git a/src/protocol/discovery.rs b/src/proto/discovery/wire.rs similarity index 53% rename from src/protocol/discovery.rs rename to src/proto/discovery/wire.rs index 26a9c5e..b732938 100644 --- a/src/protocol/discovery.rs +++ b/src/proto/discovery/wire.rs @@ -1,8 +1,8 @@ //! Discovery messages: LookupRequest and LookupResponse. use crate::NodeAddr; -use crate::protocol::error::ProtocolError; -use crate::protocol::negotiation::TlvEntry; +use crate::protocol::ProtocolError; +use crate::protocol::TlvEntry; use crate::protocol::session::{decode_coords, encode_coords}; use crate::tree::TreeCoordinate; use secp256k1::schnorr::Signature; @@ -332,272 +332,3 @@ impl LookupResponse { }) } } - -#[cfg(test)] -mod tests { - use super::*; - - fn make_node_addr(val: u8) -> NodeAddr { - let mut bytes = [0u8; 16]; - bytes[0] = val; - NodeAddr::from_bytes(bytes) - } - - fn make_coords(ids: &[u8]) -> TreeCoordinate { - TreeCoordinate::from_addrs(ids.iter().map(|&v| make_node_addr(v)).collect()).unwrap() - } - - fn make_test_sig() -> Signature { - use secp256k1::Secp256k1; - let secp = Secp256k1::new(); - let mut secret_bytes = [0u8; 32]; - rand::Rng::fill_bytes(&mut rand::rng(), &mut secret_bytes); - let secret_key = secp256k1::SecretKey::from_slice(&secret_bytes) - .expect("32 random bytes is a valid secret key"); - let keypair = secp256k1::Keypair::from_secret_key(&secp, &secret_key); - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let proof_data = LookupResponse::proof_bytes(999, &target, &coords); - use sha2::Digest; - let digest: [u8; 32] = sha2::Sha256::digest(&proof_data).into(); - secp.sign_schnorr(&digest, &keypair) - } - - #[test] - fn test_lookup_request_forward() { - let target = make_node_addr(1); - let origin = make_node_addr(2); - - let mut request = LookupRequest::new(123, target, origin, 5, 0); - - assert!(request.can_forward()); - assert!(request.forward()); - assert_eq!(request.ttl, 4); - } - - #[test] - fn test_lookup_request_ttl_exhausted() { - let target = make_node_addr(1); - let origin = make_node_addr(2); - - let mut request = LookupRequest::new(123, target, origin, 1, 0); - - assert!(request.forward()); - assert!(!request.can_forward()); - assert!(!request.forward()); - } - - #[test] - fn test_lookup_request_generate() { - let target = make_node_addr(1); - let origin = make_node_addr(2); - - let req1 = LookupRequest::generate(target, origin, 5, 0); - let req2 = LookupRequest::generate(target, origin, 5, 0); - - // Random IDs should differ - assert_ne!(req1.request_id, req2.request_id); - } - - #[test] - fn test_lookup_response_proof_bytes() { - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let bytes = LookupResponse::proof_bytes(12345, &target, &coords); - - // 8 (request_id) + 16 (target) + 2 (count) + 3*16 (coords) = 74 - assert_eq!(bytes.len(), 74); - assert_eq!(&bytes[0..8], &12345u64.to_le_bytes()); - assert_eq!(&bytes[8..24], target.as_bytes()); - - // Verify coordinate encoding is present - let count = u16::from_le_bytes([bytes[24], bytes[25]]); - assert_eq!(count, 3); // 3 entries in coords - } - - #[test] - fn test_lookup_request_encode_decode_roundtrip() { - let target = make_node_addr(10); - let origin = make_node_addr(20); - - let mut request = LookupRequest::new(12345, target, origin, 8, 1386); - request.forward(); - - let encoded = request.encode(); - assert_eq!(encoded[0], 0x30); - - let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.request_id, 12345); - assert_eq!(decoded.target, target); - assert_eq!(decoded.origin, origin); - assert_eq!(decoded.ttl, 7); // decremented by forward() - assert_eq!(decoded.min_mtu, 1386); - assert!(decoded.tlv_entries.is_empty()); - } - - #[test] - fn test_lookup_request_decode_too_short() { - assert!(LookupRequest::decode(&[]).is_err()); - assert!(LookupRequest::decode(&[0u8; 42]).is_err()); - } - - #[test] - fn test_lookup_request_min_mtu_boundary_values() { - let target = make_node_addr(10); - let origin = make_node_addr(20); - - for mtu_val in [0u16, 1386, u16::MAX] { - let request = LookupRequest::new(100, target, origin, 5, mtu_val); - let encoded = request.encode(); - let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.min_mtu, mtu_val); - } - } - - #[test] - fn test_lookup_request_with_tlv_roundtrip() { - let target = make_node_addr(10); - let origin = make_node_addr(20); - - let request = LookupRequest::new(555, target, origin, 5, 1280) - .with_tlv(1, vec![0xAA, 0xBB]) - .with_tlv(256, vec![0x01, 0x02, 0x03, 0x04]); - - let encoded = request.encode(); - let decoded = LookupRequest::decode(&encoded[1..]).unwrap(); - - assert_eq!(decoded.request_id, 555); - assert_eq!(decoded.min_mtu, 1280); - assert_eq!(decoded.tlv_entries.len(), 2); - assert_eq!(decoded.tlv_entries[0].field_num, 1); - assert_eq!(decoded.tlv_entries[0].value, vec![0xAA, 0xBB]); - assert_eq!(decoded.tlv_entries[1].field_num, 256); - assert_eq!(decoded.tlv_entries[1].value, vec![0x01, 0x02, 0x03, 0x04]); - } - - #[test] - fn test_lookup_request_tlv_forward_compat() { - // Unknown field_nums should be preserved through decode→encode - let target = make_node_addr(10); - let origin = make_node_addr(20); - - let request = - LookupRequest::new(777, target, origin, 5, 0).with_tlv(9999, vec![0xFF, 0xFE, 0xFD]); - - let encoded = request.encode(); - let mut decoded = LookupRequest::decode(&encoded[1..]).unwrap(); - - // Simulate transit: forward then re-encode - decoded.forward(); - let re_encoded = decoded.encode(); - let final_decoded = LookupRequest::decode(&re_encoded[1..]).unwrap(); - - assert_eq!(final_decoded.ttl, 4); - assert_eq!(final_decoded.tlv_entries.len(), 1); - assert_eq!(final_decoded.tlv_entries[0].field_num, 9999); - assert_eq!(final_decoded.tlv_entries[0].value, vec![0xFF, 0xFE, 0xFD]); - } - - #[test] - fn test_lookup_response_encode_decode_roundtrip() { - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let sig = make_test_sig(); - - let response = LookupResponse::new(999, target, coords, sig); - - // Default path_mtu should be u16::MAX - assert_eq!(response.path_mtu, u16::MAX); - - let encoded = response.encode(); - assert_eq!(encoded[0], 0x31); - - let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.request_id, 999); - assert_eq!(decoded.target, target); - assert_eq!(decoded.path_mtu, u16::MAX); - assert_eq!(decoded.proof, sig); - assert!(decoded.tlv_entries.is_empty()); - } - - #[test] - fn test_lookup_response_path_mtu_roundtrip() { - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let sig = make_test_sig(); - - for mtu_val in [0u16, 1280, 1386, 9000, u16::MAX] { - let mut response = LookupResponse::new(999, target, coords.clone(), sig); - response.path_mtu = mtu_val; - - let encoded = response.encode(); - let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); - assert_eq!(decoded.path_mtu, mtu_val); - } - } - - #[test] - fn test_lookup_response_path_mtu_not_in_proof_bytes() { - // Verify that proof_bytes does NOT include path_mtu - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - - let bytes = LookupResponse::proof_bytes(12345, &target, &coords); - - // proof_bytes format: request_id(8) + target(16) + coords_encoding(2 + 3*16) = 74 - // No path_mtu(2) in here - assert_eq!(bytes.len(), 74); - } - - #[test] - fn test_lookup_response_decode_too_short() { - assert!(LookupResponse::decode(&[]).is_err()); - assert!(LookupResponse::decode(&[0u8; 50]).is_err()); - } - - #[test] - fn test_lookup_response_with_tlv_roundtrip() { - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let sig = make_test_sig(); - - let response = LookupResponse::new(999, target, coords, sig) - .with_tlv(1, vec![0xAA, 0xBB]) - .with_tlv(500, vec![0x01, 0x02, 0x03]); - - let encoded = response.encode(); - let decoded = LookupResponse::decode(&encoded[1..]).unwrap(); - - assert_eq!(decoded.request_id, 999); - assert_eq!(decoded.proof, sig); - assert_eq!(decoded.tlv_entries.len(), 2); - assert_eq!(decoded.tlv_entries[0].field_num, 1); - assert_eq!(decoded.tlv_entries[0].value, vec![0xAA, 0xBB]); - assert_eq!(decoded.tlv_entries[1].field_num, 500); - assert_eq!(decoded.tlv_entries[1].value, vec![0x01, 0x02, 0x03]); - } - - #[test] - fn test_lookup_response_tlv_forward_compat() { - // Unknown field_nums preserved through decode→modify path_mtu→encode - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - let sig = make_test_sig(); - - let response = - LookupResponse::new(999, target, coords, sig).with_tlv(9999, vec![0xFF, 0xFE, 0xFD]); - - let encoded = response.encode(); - let mut decoded = LookupResponse::decode(&encoded[1..]).unwrap(); - - // Simulate transit: modify path_mtu then re-encode - decoded.path_mtu = 1280; - let re_encoded = decoded.encode(); - let final_decoded = LookupResponse::decode(&re_encoded[1..]).unwrap(); - - assert_eq!(final_decoded.path_mtu, 1280); - assert_eq!(final_decoded.tlv_entries.len(), 1); - assert_eq!(final_decoded.tlv_entries[0].field_num, 9999); - assert_eq!(final_decoded.tlv_entries[0].value, vec![0xFF, 0xFE, 0xFD]); - } -} diff --git a/src/proto/mod.rs b/src/proto/mod.rs new file mode 100644 index 0000000..badef00 --- /dev/null +++ b/src/proto/mod.rs @@ -0,0 +1,6 @@ +//! Sans-IO (runtime-agnostic) protocol state machines. +//! +//! A module here has been migrated out of the async node shell; the async +//! I/O adapters remain in `node::handlers`. + +pub(crate) mod discovery; diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index da5b746..9cab726 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -20,16 +20,14 @@ //! Session-layer datagrams are carried as opaque payloads through the link //! layer, encrypted end-to-end independently of per-hop link encryption. -mod discovery; mod error; mod filter; mod link; mod negotiation; -mod session; +pub(crate) mod session; mod tree; // Re-export all public types at protocol:: level -pub use discovery::{LookupRequest, LookupResponse}; pub use error::ProtocolError; pub use filter::{FilterAnnounce, FilterNack}; pub use link::{ diff --git a/src/testutil.rs b/src/testutil.rs new file mode 100644 index 0000000..4765983 --- /dev/null +++ b/src/testutil.rs @@ -0,0 +1,10 @@ +//! Crate-wide generic test helpers. + +use crate::NodeAddr; + +/// Build a `NodeAddr` from a single discriminating byte in position 0. +pub(crate) fn make_node_addr(val: u8) -> NodeAddr { + let mut bytes = [0u8; 16]; + bytes[0] = val; + NodeAddr::from_bytes(bytes) +}