From e03b206f625580b500c03369b22dff8b2da370a3 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 5 Jul 2026 21:59:21 +0000 Subject: [PATCH] proto/discovery: sans-IO state-machine migration + no_std reductions Migrate the FMP discovery decision logic out of the async handlers into synchronous, runtime-agnostic sans-IO state machines owned by the protocol structs, with I/O pushed to the edges. Pulls the full decision surface into a pure core (backoff, rate-limit, planners, response routing), consolidates the tests into a per-module tree with a shared crate testutil, and injects a u64 wall-clock so the core is free of Instant and std time. Also brings the module toward no_std+alloc: the four discovery maps use alloc::collections::BTreeMap (HashMap's RandomState is std-only), Arc is spelled alloc::sync::Arc, the backoff-reset log lives in the shell (the core returns the cleared count so observability stays out of the pure core), and the crate root names alloc directly. The one remaining tether is ProtocolError's std::error::Error coupling in the wire codec. First subsystem of the broader sans-IO refactor; establishes the extraction patterns and conventions carried forward to the remaining protocols. --- src/lib.rs | 18 +- src/node/discovery_rate_limit.rs | 376 --------- src/node/handlers/discovery.rs | 743 +++++++++--------- src/node/mod.rs | 76 +- src/node/tests/discovery.rs | 52 +- src/proto/discovery/core.rs | 380 +++++++++ src/proto/discovery/limits.rs | 218 +++++ src/proto/discovery/mod.rs | 33 + src/proto/discovery/state.rs | 110 +++ src/proto/discovery/tests/core.rs | 544 +++++++++++++ src/proto/discovery/tests/limits.rs | 151 ++++ src/proto/discovery/tests/mod.rs | 7 + src/proto/discovery/tests/util.rs | 98 +++ src/proto/discovery/tests/wire.rs | 156 ++++ .../discovery.rs => proto/discovery/wire.rs} | 193 +---- src/proto/mod.rs | 6 + src/protocol/mod.rs | 4 +- src/testutil.rs | 10 + 18 files changed, 2136 insertions(+), 1039 deletions(-) delete mode 100644 src/node/discovery_rate_limit.rs create mode 100644 src/proto/discovery/core.rs create mode 100644 src/proto/discovery/limits.rs create mode 100644 src/proto/discovery/mod.rs create mode 100644 src/proto/discovery/state.rs create mode 100644 src/proto/discovery/tests/core.rs create mode 100644 src/proto/discovery/tests/limits.rs create mode 100644 src/proto/discovery/tests/mod.rs create mode 100644 src/proto/discovery/tests/util.rs create mode 100644 src/proto/discovery/tests/wire.rs rename src/{protocol/discovery.rs => proto/discovery/wire.rs} (55%) create mode 100644 src/proto/mod.rs create mode 100644 src/testutil.rs 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 e3d8c7b..9a66477 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -5,15 +5,41 @@ //! 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() + } +} + impl Node { /// Handle an incoming LookupRequest from a peer. /// @@ -39,80 +65,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 +161,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 +357,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 +378,7 @@ impl Node { .record_reject(DiscoveryReject::RespNoRoute); return; } - } + }, }; // Fold our outgoing-link MTU into path_mtu so the target-edge link @@ -361,67 +418,56 @@ 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; - } - - // Collect tree peers whose bloom filter contains the target - let forward_to: Vec = self - .peers - .iter() - .filter(|(addr, peer)| self.is_tree_peer(addr) && peer.may_reach(&request.target)) - .map(|(addr, _)| *addr) - .collect(); - - // Fallback: if no tree peer matches, try non-tree bloom-matching peers - let (forward_to, used_fallback) = if forward_to.is_empty() { - let fallback: Vec = self - .peers - .iter() - .filter(|(addr, peer)| !self.is_tree_peer(addr) && peer.may_reach(&request.target)) - .map(|(addr, _)| *addr) - .collect(); - if fallback.is_empty() { + // Plan the forward with the sans-IO decision core. The core owns the + // TTL decrement, 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::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" + ); + } + } } } } @@ -439,15 +485,15 @@ impl Node { let origin_coords = self.tree_state().my_coords().clone(); let request = LookupRequest::generate(*target, origin, origin_coords, ttl, 0); - // Send only to tree peers whose bloom filter contains the target - let peer_addrs: Vec = self - .peers - .iter() - .filter(|(addr, peer)| self.is_tree_peer(addr) && peer.may_reach(target)) - .map(|(addr, _)| *addr) - .collect(); + // Tree-peer bloom-match selection + 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, @@ -458,16 +504,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" ); @@ -487,54 +529,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" + ); + } + } } } @@ -549,53 +586,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!( @@ -614,22 +622,15 @@ 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(); } } - /// 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 @@ -716,23 +717,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 177c36c..4e3c0d7 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; @@ -48,6 +46,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}; #[cfg(unix)] use crate::transport::ethernet::EthernetTransport; use crate::transport::nym::NymTransport; @@ -218,39 +217,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); @@ -319,9 +285,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 @@ -369,10 +332,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. @@ -455,10 +419,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 @@ -658,7 +618,6 @@ impl Node { tree_state, bloom_state, coord_cache, - recent_requests: HashMap::new(), transports: HashMap::new(), transport_drops: HashMap::new(), links: HashMap::new(), @@ -670,7 +629,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(), @@ -704,9 +662,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(), @@ -819,7 +777,6 @@ impl Node { tree_state, bloom_state, coord_cache, - recent_requests: HashMap::new(), transports: HashMap::new(), transport_drops: HashMap::new(), links: HashMap::new(), @@ -831,7 +788,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(), @@ -865,8 +821,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, @@ -2446,8 +2401,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)] @@ -2559,19 +2513,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 671e911..ab3e7c1 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] @@ -39,11 +38,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] @@ -61,7 +60,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] @@ -77,7 +76,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)); } // ============================================================================ @@ -115,7 +114,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; @@ -149,7 +148,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 @@ -319,14 +319,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); @@ -338,9 +340,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)); } // ============================================================================ @@ -379,7 +381,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" ); @@ -447,13 +449,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" ); @@ -502,7 +504,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" ); @@ -1037,8 +1039,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; @@ -1106,7 +1108,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(); @@ -1116,6 +1119,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"); @@ -1132,6 +1136,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"); @@ -1148,6 +1153,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"); @@ -1163,7 +1169,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!( @@ -1186,7 +1192,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..0fe72c4 --- /dev/null +++ b/src/proto/discovery/core.rs @@ -0,0 +1,380 @@ +//! 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; +} + +/// 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, + /// 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; 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; + } + let target = request.target; + let reaching = rv.peers_reaching(&target); + let tree: Vec = reaching + .iter() + .copied() + .filter(|a| rv.is_tree_peer(a)) + .collect(); + let (targets, used_fallback) = if tree.is_empty() { + let fallback: Vec = reaching + .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 targets: Vec = rv + .peers_reaching(&request.target) + .into_iter() + .filter(|addr| rv.is_tree_peer(addr)) + .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..4859e2a --- /dev/null +++ b/src/proto/discovery/tests/core.rs @@ -0,0 +1,544 @@ +//! 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)], + }; + 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), + ], + }; + 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)], + }; + 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)], + }; + 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)], + }; + 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), + ], + }; + 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)], + }; + let request = make_request(3); + assert!(plan_initiate(&request, &rv).is_empty()); +} + +#[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..f81c946 --- /dev/null +++ b/src/proto/discovery/tests/util.rs @@ -0,0 +1,98 @@ +//! 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 entry is `(addr, is_tree, may_reach)`. +pub(super) struct MockRoutingView { + pub(super) peers: Vec<(NodeAddr, bool, bool)>, +} + +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() + } +} + +pub(super) fn make_request(ttl: u8) -> LookupRequest { + let target = make_node_addr(0xAA); + let origin = make_node_addr(0xBB); + let origin_coords = TreeCoordinate::root(origin); + LookupRequest::new(1, target, origin, origin_coords, 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); + let origin_coords = TreeCoordinate::root(origin); + LookupRequest::new(request_id, target, origin, origin_coords, 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..12e0092 --- /dev/null +++ b/src/proto/discovery/tests/wire.rs @@ -0,0 +1,156 @@ +//! 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 coords = make_coords(&[2, 0]); + + let mut request = LookupRequest::new(123, target, origin, coords, 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 coords = make_coords(&[2, 0]); + + let mut request = LookupRequest::new(123, target, origin, coords, 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 coords = make_coords(&[2, 0]); + + let req1 = LookupRequest::generate(target, origin, coords.clone(), 5, 0); + let req2 = LookupRequest::generate(target, origin, coords, 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 coords = make_coords(&[20, 0]); + + let mut request = LookupRequest::new(12345, target, origin, coords, 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); + let coords = make_coords(&[20, 0]); + + for mtu_val in [0u16, 1386, u16::MAX] { + let request = LookupRequest::new(100, target, origin, coords.clone(), 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()); +} diff --git a/src/protocol/discovery.rs b/src/proto/discovery/wire.rs similarity index 55% rename from src/protocol/discovery.rs rename to src/proto/discovery/wire.rs index eb860ac..bc7d318 100644 --- a/src/protocol/discovery.rs +++ b/src/proto/discovery/wire.rs @@ -1,7 +1,7 @@ //! Discovery messages: LookupRequest and LookupResponse. use crate::NodeAddr; -use crate::protocol::error::ProtocolError; +use crate::protocol::ProtocolError; use crate::protocol::session::{decode_coords, encode_coords}; use crate::tree::TreeCoordinate; use secp256k1::schnorr::Signature; @@ -273,194 +273,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() - } - - #[test] - fn test_lookup_request_forward() { - let target = make_node_addr(1); - let origin = make_node_addr(2); - let coords = make_coords(&[2, 0]); - - let mut request = LookupRequest::new(123, target, origin, coords, 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 coords = make_coords(&[2, 0]); - - let mut request = LookupRequest::new(123, target, origin, coords, 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 coords = make_coords(&[2, 0]); - - let req1 = LookupRequest::generate(target, origin, coords.clone(), 5, 0); - let req2 = LookupRequest::generate(target, origin, coords, 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 coords = make_coords(&[20, 0]); - - let mut request = LookupRequest::new(12345, target, origin, coords, 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); - let coords = make_coords(&[20, 0]); - - for mtu_val in [0u16, 1386, u16::MAX] { - let request = LookupRequest::new(100, target, origin, coords.clone(), 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() { - use secp256k1::Secp256k1; - - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - - // Create a dummy signature for testing - 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(999, &target, &coords); - use sha2::Digest; - let digest: [u8; 32] = sha2::Sha256::digest(&proof_data).into(); - let sig = secp.sign_schnorr(&digest, &keypair); - - 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); - } - - #[test] - fn test_lookup_response_path_mtu_roundtrip() { - use secp256k1::Secp256k1; - - let target = make_node_addr(42); - let coords = make_coords(&[42, 1, 0]); - - 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(999, &target, &coords); - use sha2::Digest; - let digest: [u8; 32] = sha2::Sha256::digest(&proof_data).into(); - let sig = secp.sign_schnorr(&digest, &keypair); - - 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()); - } -} 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 99e898c..cfd4000 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -20,15 +20,13 @@ //! 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 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; 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) +}