From 9ea57b483a6038a53abf3ecbb7b8af4e3861e478 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Tue, 7 Jul 2026 04:12:05 +0000 Subject: [PATCH] proto/routing: sans-IO transit + hop-selection state machine --- src/lib.rs | 10 +- src/node/handlers/forwarding.rs | 353 ++++++++++++---------- src/node/handlers/session.rs | 10 +- src/node/metrics.rs | 41 +-- src/node/mod.rs | 240 +++++++-------- src/node/routing_error_rate_limit.rs | 161 ---------- src/proto/mod.rs | 1 + src/proto/routing/core.rs | 423 +++++++++++++++++++++++++++ src/proto/routing/limits.rs | 90 ++++++ src/proto/routing/mod.rs | 33 +++ src/proto/routing/state.rs | 23 ++ src/proto/routing/tests/core.rs | 358 +++++++++++++++++++++++ src/proto/routing/tests/limits.rs | 70 +++++ src/proto/routing/tests/mod.rs | 6 + src/proto/routing/tests/util.rs | 86 ++++++ src/proto/routing/tests/wire.rs | 130 ++++++++ src/proto/routing/wire.rs | 275 +++++++++++++++++ src/protocol/mod.rs | 7 +- src/protocol/session.rs | 391 +------------------------ 19 files changed, 1820 insertions(+), 888 deletions(-) delete mode 100644 src/node/routing_error_rate_limit.rs create mode 100644 src/proto/routing/core.rs create mode 100644 src/proto/routing/limits.rs create mode 100644 src/proto/routing/mod.rs create mode 100644 src/proto/routing/state.rs create mode 100644 src/proto/routing/tests/core.rs create mode 100644 src/proto/routing/tests/limits.rs create mode 100644 src/proto/routing/tests/mod.rs create mode 100644 src/proto/routing/tests/util.rs create mode 100644 src/proto/routing/tests/wire.rs create mode 100644 src/proto/routing/wire.rs diff --git a/src/lib.rs b/src/lib.rs index 9b21602..1d23993 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -61,14 +61,18 @@ pub use transport::{ // Re-export protocol types pub use protocol::{ - CoordsRequired, FilterAnnounce, HandshakeMessageType, LinkMessageType, PathBroken, - ProtocolError, SessionAck, SessionDatagram, SessionFlags, SessionMessageType, SessionSetup, - TreeAnnounce, + FilterAnnounce, HandshakeMessageType, LinkMessageType, 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 routing wire types (relocated from protocol:: to proto::routing) +pub use proto::routing::{ + COORDS_REQUIRED_SIZE, CoordsRequired, MTU_EXCEEDED_SIZE, MtuExceeded, PathBroken, +}; + // Re-export cache types pub use cache::{CacheEntry, CacheError, CacheStats, CoordCache}; diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index 8d6f060..7dd4e1c 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -11,11 +11,9 @@ use crate::node::session_wire::{ FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2, FspCommonPrefix, parse_encrypted_coords, }; -use crate::node::{Node, NodeError}; -use crate::protocol::{ - CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionDatagramRef, - SessionSetup, -}; +use crate::node::{Node, NodeError, NodeRoutingView}; +use crate::proto::routing::{DropReason, NextHop, RouteAction, RouteOutcome}; +use crate::protocol::{SessionAck, SessionDatagram, SessionDatagramRef, SessionSetup}; use std::time::{Duration, Instant}; use tracing::{debug, warn}; @@ -43,123 +41,166 @@ impl Node { } }; - // TTL enforcement: decrement for forwarding and drop only if the - // received datagram was already exhausted. - if datagram_ref.ttl == 0 { - self.metrics() - .forwarding - .record_reject_bytes(ForwardingReject::TtlExhausted, payload.len()); - debug!( - src = %datagram_ref.src_addr, - dest = %datagram_ref.dest_addr, - "SessionDatagram TTL exhausted, dropping" - ); - return; - } - let forwarded_ttl = datagram_ref.ttl - 1; + let my_addr = *self.node_addr(); - // Coordinate cache warming from plaintext session-layer headers - self.try_warm_coord_cache_ref(&datagram_ref); - - // Local delivery: dispatch to session layer handlers without - // materializing an owned SessionDatagram payload Vec. - if datagram_ref.dest_addr == *self.node_addr() { - self.metrics().forwarding.record_delivered(payload.len()); - self.handle_session_payload( - &datagram_ref.src_addr, - datagram_ref.payload, - datagram_ref.path_mtu, - incoming_ce, - ) - .await; - return; + // Coordinate cache warming from plaintext session-layer headers. Gated + // on a non-exhausted TTL so a datagram the core will drop as + // TTL-exhausted does not warm the cache, matching the pre-refactor + // ordering (warming ran only after the TTL early-return). + if datagram_ref.ttl != 0 { + self.try_warm_coord_cache_ref(&datagram_ref); } - let mut datagram = datagram_ref.into_owned(); - datagram.ttl = forwarded_ttl; + // Pre-resolve the next hop only for genuine transit packets (TTL > 0 + // and not locally destined) so `find_next_hop`'s coord-cache LRU-touch + // side effect keeps the same scope it had inline. Warming above has + // already run, so the resolution observes freshly cached coords. + let next_hop = if datagram_ref.ttl != 0 && datagram_ref.dest_addr != my_addr { + self.resolve_next_hop(&datagram_ref.dest_addr) + } else { + None + }; - // Find next hop toward destination - let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { - Some(peer) => *peer.node_addr(), - None => { + // Read local congestion once and reuse it for both the CE decision + // (via the view) and the congestion metric/log below, keeping + // `detect_congestion` the single source of truth. + let congested = next_hop + .as_ref() + .map(|nh| self.detect_congestion(&nh.addr)) + .unwrap_or(false); + + // Borrow the routing tables disjointly from `&mut self.routing` for + // the pure decision, then release both before driving the outcome. + let outcome = { + let view = NodeRoutingView { + coord_cache: &self.coord_cache, + peers: &self.peers, + tree_state: &self.tree_state, + congested, + }; + self.routing + .route(&datagram_ref, &my_addr, incoming_ce, next_hop, &view) + }; + + match outcome { + RouteOutcome::Drop { + reason: DropReason::TtlExhausted, + } => { + self.metrics() + .forwarding + .record_reject_bytes(ForwardingReject::TtlExhausted, payload.len()); + debug!( + src = %datagram_ref.src_addr, + dest = %datagram_ref.dest_addr, + "SessionDatagram TTL exhausted, dropping" + ); + } + RouteOutcome::DeliverLocal => { + // Local delivery: dispatch to session layer handlers without + // materializing an owned SessionDatagram payload Vec. + self.metrics().forwarding.record_delivered(payload.len()); + self.handle_session_payload( + &datagram_ref.src_addr, + datagram_ref.payload, + datagram_ref.path_mtu, + incoming_ce, + ) + .await; + } + RouteOutcome::NoRoute => { self.metrics() .forwarding .record_reject_bytes(ForwardingReject::NoRoute, payload.len()); + let original = datagram_ref.into_owned(); debug!( - src = %self.peer_display_name(&datagram.src_addr), - dest = %self.peer_display_name(&datagram.dest_addr), + src = %self.peer_display_name(&original.src_addr), + dest = %self.peer_display_name(&original.dest_addr), bytes = payload.len(), "Dropping transit SessionDatagram: no route to destination" ); - self.send_routing_error(&datagram).await; - return; + self.send_routing_error(&original).await; } - }; + RouteOutcome::Forward { + next_hop, + bytes, + outgoing_ce, + } => { + let dest = datagram_ref.dest_addr; - // Apply path_mtu min() from the outgoing link's transport MTU - if let Some(peer) = self.peers.get(&next_hop_addr) + // ECN CE relay: congestion was detected locally above; emit the + // metric and rate-limited log at the transit chokepoint. + if congested { + self.metrics().congestion.congestion_detected.inc(); + let now = Instant::now(); + let should_log = self + .last_congestion_log + .map(|t| now.duration_since(t) >= Duration::from_secs(5)) + .unwrap_or(true); + if should_log { + self.last_congestion_log = Some(now); + debug!(next_hop = %next_hop, "Congestion detected, CE flag set on forwarded packet"); + } + } + + match self + .send_encrypted_link_message_with_ce(&next_hop, &bytes, outgoing_ce) + .await + { + Err(NodeError::MtuExceeded { mtu, .. }) => { + self.metrics() + .forwarding + .record_reject_bytes(ForwardingReject::MtuExceeded, payload.len()); + self.send_mtu_exceeded_error(dest, datagram_ref.src_addr, mtu) + .await; + } + Err(e) => { + self.metrics() + .forwarding + .record_reject_bytes(ForwardingReject::SendError, payload.len()); + debug!( + next_hop = %next_hop, + dest = %dest, + error = %e, + "Failed to forward SessionDatagram" + ); + } + Ok(()) => { + self.metrics().forwarding.record_forwarded(bytes.len()); + // Classify this transit forward by route class (partition + // of forwarded_packets). Done here, at the data-plane + // chokepoint, so the error-signal routing callers of + // find_next_hop are excluded. + let class = self.classify_forward(&dest, &next_hop); + self.metrics().forwarding.record_route_class(class); + if outgoing_ce { + self.metrics().congestion.ce_forwarded.inc(); + } + } + } + } + } + } + + /// Resolve the next hop toward `dest` into its address plus the outgoing + /// link's transport MTU. Returns `None` when there is no route. + /// + /// The MTU defaults to `u16::MAX` (a no-op min-fold) when the peer's + /// transport is not resolvable, matching the pre-refactor inline behavior + /// where the MTU `if let` chain simply did not fire. + fn resolve_next_hop(&mut self, dest: &NodeAddr) -> Option { + let addr = *self.find_next_hop(dest)?.node_addr(); + let link_mtu = if let Some(peer) = self.peers.get(&addr) && let Some(tid) = peer.transport_id() && let Some(transport) = self.transports.get(&tid) { - if let Some(addr) = peer.current_addr() { - datagram.path_mtu = datagram.path_mtu.min(transport.link_mtu(addr)); - } else { - datagram.path_mtu = datagram.path_mtu.min(transport.mtu()); - } - } - - // ECN CE relay: propagate incoming CE and detect local congestion - let local_congestion = self.detect_congestion(&next_hop_addr); - let outgoing_ce = incoming_ce || local_congestion; - if local_congestion { - self.metrics().congestion.congestion_detected.inc(); - let now = Instant::now(); - let should_log = self - .last_congestion_log - .map(|t| now.duration_since(t) >= Duration::from_secs(5)) - .unwrap_or(true); - if should_log { - self.last_congestion_log = Some(now); - debug!(next_hop = %next_hop_addr, "Congestion detected, CE flag set on forwarded packet"); - } - } - - // Forward: re-encode (includes 0x00 type byte) and send - let encoded = datagram.encode(); - if let Err(e) = self - .send_encrypted_link_message_with_ce(&next_hop_addr, &encoded, outgoing_ce) - .await - { - match e { - NodeError::MtuExceeded { mtu, .. } => { - self.metrics() - .forwarding - .record_reject_bytes(ForwardingReject::MtuExceeded, payload.len()); - self.send_mtu_exceeded_error(&datagram, mtu).await; - } - _ => { - self.metrics() - .forwarding - .record_reject_bytes(ForwardingReject::SendError, payload.len()); - debug!( - next_hop = %next_hop_addr, - dest = %datagram.dest_addr, - error = %e, - "Failed to forward SessionDatagram" - ); - } + match peer.current_addr() { + Some(link_addr) => transport.link_mtu(link_addr), + None => transport.mtu(), } } else { - self.metrics().forwarding.record_forwarded(encoded.len()); - // Classify this transit forward by route class (partition of - // forwarded_packets). Done here, at the data-plane chokepoint, so - // the error-signal routing callers of find_next_hop are excluded. - let class = self.classify_forward(&datagram.dest_addr, &next_hop_addr); - self.metrics().forwarding.record_route_class(class); - if outgoing_ce { - self.metrics().congestion.ce_forwarded.inc(); - } - } + u16::MAX + }; + Some(NextHop { addr, link_mtu }) } /// Attempt to warm the coordinate cache from session-layer payload headers. @@ -260,35 +301,41 @@ impl Node { /// If we can't route the error back to the source either, drop silently. /// No cascading errors. async fn send_routing_error(&mut self, original: &SessionDatagram) { - // Rate limit: one error signal per destination per 100ms - if !self - .routing_error_rate_limiter - .should_send(&original.dest_addr) - { - return; - } - let my_addr = *self.node_addr(); - let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); + let default_ttl = self.config().node.session.default_ttl; - let error_payload = - if let Some(coords) = self.coord_cache().get(&original.dest_addr, now_ms) { - let coords = coords.clone(); - PathBroken::new(original.dest_addr, my_addr) - .with_last_coords(coords) - .encode() - } else { - CoordsRequired::new(original.dest_addr, my_addr).encode() + // Pure decision: rate-limit gate + PathBroken/CoordsRequired choice + + // error-PDU encode. Borrow the routing tables disjointly from + // `&mut self.routing`, then release them before the reverse-hop lookup. + let action = { + let view = NodeRoutingView { + coord_cache: &self.coord_cache, + peers: &self.peers, + tree_state: &self.tree_state, + congested: false, }; + self.routing.synth_routing_error( + &original.dest_addr, + &original.src_addr, + &my_addr, + &view, + now_ms, + default_ttl, + ) + }; + let RouteAction::SendError { toward, bytes } = match action { + Some(action) => action, + // Rate limited: drop silently. No cascading errors. + None => return, + }; - let error_dg = SessionDatagram::new(my_addr, original.src_addr, error_payload) - .with_ttl(self.config().node.session.default_ttl); - - let next_hop_addr = match self.find_next_hop(&original.src_addr) { + // Resolve the reverse link hop only now, after the gate passed, so + // `find_next_hop`'s coord-cache touch keeps its pre-refactor scope. + let next_hop_addr = match self.find_next_hop(&toward) { Some(peer) => *peer.node_addr(), None => { debug!( @@ -300,9 +347,8 @@ impl Node { } }; - let encoded = error_dg.encode(); if let Err(e) = self - .send_encrypted_link_message(&next_hop_addr, &encoded) + .send_encrypted_link_message(&next_hop_addr, &bytes) .await { debug!( @@ -324,37 +370,50 @@ impl Node { /// Called when `send_encrypted_link_message()` fails with /// `NodeError::MtuExceeded` during forwarding. The signal tells the /// source the bottleneck MTU so it can immediately reduce its path MTU. - async fn send_mtu_exceeded_error(&mut self, original: &SessionDatagram, bottleneck_mtu: u16) { - // Rate limit: reuse routing_error_rate_limiter keyed on dest_addr - if !self - .routing_error_rate_limiter - .should_send(&original.dest_addr) - { - return; - } - + /// + /// `dest` is the failed datagram's destination (rate-limit key); `toward` + /// is its source, where the signal is routed back. + async fn send_mtu_exceeded_error( + &mut self, + dest: NodeAddr, + toward: NodeAddr, + bottleneck_mtu: u16, + ) { let my_addr = *self.node_addr(); + let now_ms = Self::now_ms(); + let default_ttl = self.config().node.session.default_ttl; - let error_payload = MtuExceeded::new(original.dest_addr, my_addr, bottleneck_mtu).encode(); + // Pure decision: rate-limit gate + MtuExceeded PDU + encode. + let action = self.routing.synth_mtu_exceeded( + &dest, + &toward, + &my_addr, + bottleneck_mtu, + now_ms, + default_ttl, + ); + let RouteAction::SendError { toward, bytes } = match action { + Some(action) => action, + // Rate limited: drop silently. No cascading errors. + None => return, + }; - let error_dg = SessionDatagram::new(my_addr, original.src_addr, error_payload) - .with_ttl(self.config().node.session.default_ttl); - - let next_hop_addr = match self.find_next_hop(&original.src_addr) { + // Resolve the reverse link hop only now, after the gate passed, so + // `find_next_hop`'s coord-cache touch keeps its pre-refactor scope. + let next_hop_addr = match self.find_next_hop(&toward) { Some(peer) => *peer.node_addr(), None => { debug!( - src = %original.src_addr, - dest = %original.dest_addr, + src = %toward, + dest = %dest, "Cannot route MtuExceeded signal back to source, dropping" ); return; } }; - let encoded = error_dg.encode(); if let Err(e) = self - .send_encrypted_link_message(&next_hop_addr, &encoded) + .send_encrypted_link_message(&next_hop_addr, &bytes) .await { debug!( @@ -364,8 +423,8 @@ impl Node { ); } else { debug!( - original_dest = %original.dest_addr, - error_dest = %original.src_addr, + original_dest = %dest, + error_dest = %toward, bottleneck_mtu, "Sent MtuExceeded error signal" ); diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 3bd6230..d198369 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -24,14 +24,14 @@ use crate::node::{Node, NodeError}; use crate::noise::{ HandshakeState, XK_HANDSHAKE_MSG1_SIZE, XK_HANDSHAKE_MSG2_SIZE, XK_HANDSHAKE_MSG3_SIZE, }; +use crate::proto::routing::{CoordsRequired, MtuExceeded, PathBroken}; #[cfg(unix)] use crate::protocol::LinkMessageType; #[cfg(unix)] use crate::protocol::SESSION_DATAGRAM_HEADER_SIZE; use crate::protocol::{ - CoordsRequired, FspInnerFlags, MtuExceeded, PathBroken, PathMtuNotification, SessionAck, - SessionDatagram, SessionMessageType, SessionMsg3, SessionReceiverReport, SessionSenderReport, - SessionSetup, + FspInnerFlags, PathMtuNotification, SessionAck, SessionDatagram, SessionMessageType, + SessionMsg3, SessionReceiverReport, SessionSenderReport, SessionSetup, }; use crate::protocol::{coords_wire_size, encode_coords}; #[cfg(unix)] @@ -1125,7 +1125,7 @@ impl Node { // Send standalone CoordsWarmup immediately (rate-limited) if self .coords_response_rate_limiter - .should_send(&msg.dest_addr) + .should_send(&msg.dest_addr, Self::now_ms()) { if let Some(entry) = self.sessions.get(&msg.dest_addr) && entry.is_established() @@ -1186,7 +1186,7 @@ impl Node { // Send standalone CoordsWarmup immediately (rate-limited) if self .coords_response_rate_limiter - .should_send(&msg.dest_addr) + .should_send(&msg.dest_addr, Self::now_ms()) { if let Some(entry) = self.sessions.get(&msg.dest_addr) && entry.is_established() diff --git a/src/node/metrics.rs b/src/node/metrics.rs index 84f5ffc..e94d70a 100644 --- a/src/node/metrics.rs +++ b/src/node/metrics.rs @@ -90,43 +90,10 @@ pub struct ForwardingMetrics { } /// Route class of a transit-forwarded packet, classified from tree -/// coordinates at the forwarding decision point. The six variants -/// partition `forwarded_packets` exactly. -/// -/// Two variants are up-and-over forwards (destination not in the chosen -/// peer's subtree); they differ in whether they depend on a child -/// advertising cross-link reach *upward* to its parent: -/// - `TreeDownCross`: the chosen peer is our tree descendant, but the -/// destination is *not* in that child's subtree. The forward only fired -/// because the child advertised cross-link reach upward to us, beyond its -/// own subtree. If children advertised only their subtree upward, this -/// forward would route up instead, so its count measures how much -/// forwarding depends on the upward cross-link advertisement — the -/// dive-to-tree-child cut-through. -/// - `CrosslinkAscend`: the chosen peer is lateral (neither ancestor nor -/// descendant) and the destination is not in its subtree. This is a node -/// using its *own* cross-link, learned via the peer's split-horizon -/// advertisement to its neighbors, so it does not depend on any upward -/// advertisement. Tracked alongside `TreeDownCross` as the lateral -/// up-and-over contrast. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RouteClass { - /// Chosen peer is our ancestor (tree-up). - TreeUp, - /// Chosen peer is our descendant and dest is in its subtree (canonical - /// tree-down). - TreeDown, - /// Chosen peer is our descendant but dest is *not* in its subtree: the - /// dive-to-tree-child cut-through enabled by upward cross-link - /// advertisement. - TreeDownCross, - /// Chosen peer is lateral and dest is in its subtree (subtree entry). - CrosslinkDescend, - /// Chosen peer is lateral and dest is not in its subtree (up-and-over). - CrosslinkAscend, - /// Chosen peer is the destination itself (degenerate direct hop). - DirectPeer, -} +/// coordinates at the forwarding decision point. Defined by the sans-IO +/// routing core and re-exported here for the forwarding-metrics counters +/// ([`ForwardingMetrics::record_route_class`]). +pub(crate) use crate::proto::routing::RouteClass; impl ForwardingMetrics { /// Record a received packet of `bytes` payload (packets and bytes). diff --git a/src/node/mod.rs b/src/node/mod.rs index 4e3c0d7..50d7b44 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -18,7 +18,6 @@ mod rate_limit; pub(crate) mod reject; mod reloadable; mod retry; -mod routing_error_rate_limit; pub(crate) mod session; pub(crate) mod session_wire; pub(crate) mod stats; @@ -30,7 +29,6 @@ pub(crate) mod wire; use self::rate_limit::HandshakeRateLimiter; use self::reloadable::Reloadable; -use self::routing_error_rate_limit::RoutingErrorRateLimiter; /// Half-range of the symmetric jitter applied to the per-session rekey timer. /// Each session draws an offset uniformly from `[-REKEY_JITTER_SECS, @@ -47,6 +45,7 @@ use crate::cache::CoordCache; use crate::node::session::SessionEntry; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter}; +use crate::proto::routing::{self, Router, RoutingErrorRateLimiter}; #[cfg(unix)] use crate::transport::ethernet::EthernetTransport; use crate::transport::nym::NymTransport; @@ -62,7 +61,7 @@ use crate::upper::hosts::HostMap; use crate::upper::icmp_rate_limit::IcmpRateLimiter; use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; use crate::utils::index::IndexAllocator; -use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity}; +use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity, TreeCoordinate}; use rand::Rng; use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt; @@ -415,8 +414,8 @@ pub struct Node { msg1_rate_limiter: HandshakeRateLimiter, /// Rate limiter for ICMP Packet Too Big messages. icmp_rate_limiter: IcmpRateLimiter, - /// Rate limiter for routing error signals (CoordsRequired / PathBroken). - routing_error_rate_limiter: RoutingErrorRateLimiter, + /// Routing-subsystem state (routing error-signal rate limiter). + routing: Router, /// Rate limiter for source-side CoordsRequired/PathBroken responses. coords_response_rate_limiter: RoutingErrorRateLimiter, @@ -658,9 +657,9 @@ impl Node { pending_outbound: HashMap::new(), msg1_rate_limiter, icmp_rate_limiter: IcmpRateLimiter::new(), - routing_error_rate_limiter: RoutingErrorRateLimiter::new(), - coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval( - std::time::Duration::from_millis(coords_response_interval_ms), + routing: Router::new(), + coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms( + coords_response_interval_ms, ), discovery: Discovery::new( DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs), @@ -817,9 +816,9 @@ impl Node { pending_outbound: HashMap::new(), msg1_rate_limiter, icmp_rate_limiter: IcmpRateLimiter::new(), - routing_error_rate_limiter: RoutingErrorRateLimiter::new(), - coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval( - std::time::Duration::from_millis(coords_response_interval_ms), + routing: Router::new(), + coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms( + coords_response_interval_ms, ), discovery: Discovery::new(DiscoveryBackoff::new(), DiscoveryForwardRateLimiter::new()), pending_connects: Vec::new(), @@ -2607,11 +2606,27 @@ impl Node { // 3. Bloom filter candidates — requires dest_coords for loop-free selection. // If no candidate is strictly closer, fall through to tree routing. - let candidates: Vec<&ActivePeer> = self.destination_in_filters(dest_node_addr); + // The sans-IO core assembles the candidate snapshot over the + // `RoutingView` seam (enumerate peers, apply the bloom `may_reach` + // filter, snapshot each), then picks the winner; the shell supplies + // only the raw per-peer reads. + let candidates = { + let view = NodeRoutingView { + coord_cache: &self.coord_cache, + peers: &self.peers, + tree_state: &self.tree_state, + congested: false, + }; + routing::routing_candidates(&view, dest_node_addr) + }; if !candidates.is_empty() - && let Some(peer) = self.select_best_candidate(&candidates, &dest_coords) + && let Some(next_hop) = routing::select_best_candidate( + &candidates, + &dest_coords, + self.tree_state.my_coords(), + ) { - return Some(peer); + return self.peers.get(&next_hop); } // 4. Greedy tree routing fallback @@ -2622,142 +2637,29 @@ impl Node { /// Classify a transit forward by route class from tree coordinates. /// - /// Called at the transit chokepoint after `find_next_hop` returns a peer, - /// so the six classes partition `forwarded_packets` exactly. The branch - /// that `find_next_hop` took (bloom vs greedy-tree) is *not* the route - /// class: a peer can be selected by either, so the cut-through splits - /// (`TreeDownCross`, `CrosslinkAscend`) are decided here from coordinates, - /// not from which branch fired. - /// - /// Inputs: our coords (`tree_state.my_coords`), the chosen peer's coords - /// (`tree_state.peer_coords`), and the destination coords (re-read from the - /// coord cache, which `find_next_hop` just touched). Both the tree-down and - /// cross-link branches split on whether the destination is in the chosen - /// peer's subtree; when the dest coords are unavailable that test defaults - /// to "not in subtree", i.e. the up-and-over variant (`TreeDownCross` for a - /// descendant peer, `CrosslinkAscend` for a lateral one). + /// Thin shell adapter over the pure [`routing::classify_forward`]: it + /// pre-resolves the destination coordinates from the coord cache (the + /// sole impurity — a read-only lookup, no LRU touch) and reads our own + /// and the chosen peer's coordinates from tree state, then defers the + /// six-way classification to the sans-IO routing core. pub(crate) fn classify_forward( &self, dest: &NodeAddr, chosen_peer: &NodeAddr, ) -> metrics::RouteClass { - // Degenerate: the next hop is the destination itself (Branch 2). - if chosen_peer == dest { - return metrics::RouteClass::DirectPeer; - } - - let my_addr = self.node_addr(); - let my_coords = self.tree_state.my_coords(); - - // Tree-up: the chosen peer is our ancestor. - if my_coords.has_ancestor(chosen_peer) { - return metrics::RouteClass::TreeUp; - } - - // Whether the destination is in the chosen peer's subtree. Both the - // tree-down and cross-link splits below turn on this same test, so it - // is computed once. On the live transit path the dest coords are - // always present here: `find_next_hop` looks them up with an early - // return, so a coord-cache miss yields no next hop to classify (the - // caller signals `CoordsRequired` instead of forwarding). The miss - // branch below is therefore defensive — reachable only by direct - // unit-test calls — and defaults the test to "not in subtree", i.e. - // the up-and-over variant of whichever branch fires (TreeDownCross for - // a descendant peer, CrosslinkAscend for a lateral one), matching the - // original cross-link default-to-ascend. let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let dest_in_peer_subtree = self - .coord_cache - .get(dest, now_ms) - .is_some_and(|dest_coords| dest_coords.has_ancestor(chosen_peer)); - - // Tree-down: the chosen peer is our descendant (we are its ancestor). - // Split by subtree membership: a dest genuinely below the child is the - // canonical tree-down; a dest *not* below it means we only forwarded - // down because the child advertised cross-link reach upward, beyond its - // own subtree — the dive-to-tree-child cut-through (TreeDownCross). - if let Some(peer_coords) = self.tree_state.peer_coords(chosen_peer) - && peer_coords.has_ancestor(my_addr) - { - return if dest_in_peer_subtree { - metrics::RouteClass::TreeDown - } else { - metrics::RouteClass::TreeDownCross - }; - } - - // Cross-link (lateral): split by whether the destination is in the - // chosen peer's subtree. Descend = subtree entry; ascend = up-and-over - // via the node's own cross-link (learned from the peer's split-horizon - // advertisement, independent of any upward advertisement). - if dest_in_peer_subtree { - return metrics::RouteClass::CrosslinkDescend; - } - - metrics::RouteClass::CrosslinkAscend - } - - /// Select the best peer from a set of bloom filter candidates. - /// - /// Uses distance from each candidate's tree coordinates to the destination - /// as the primary metric (after link_cost). Only selects peers that are - /// strictly closer to the destination than we are (self-distance check - /// prevents routing loops). - /// - /// Ordering: `(link_cost, distance_to_dest, node_addr)`. - fn select_best_candidate<'a>( - &'a self, - candidates: &[&'a ActivePeer], - dest_coords: &crate::tree::TreeCoordinate, - ) -> Option<&'a ActivePeer> { - let my_distance = self.tree_state.my_coords().distance_to(dest_coords); - - let mut best: Option<(&ActivePeer, f64, usize)> = None; - - for &candidate in candidates { - if !candidate.can_send() { - continue; - } - - let cost = candidate.link_cost(); - - let dist = self - .tree_state - .peer_coords(candidate.node_addr()) - .map(|pc| pc.distance_to(dest_coords)) - .unwrap_or(usize::MAX); - - // Self-distance check: only consider peers strictly closer - // to the destination than we are (prevents routing loops) - if dist >= my_distance { - continue; - } - - let dominated = match &best { - None => true, - Some((_, best_cost, best_dist)) => { - cost < *best_cost - || (cost == *best_cost && dist < *best_dist) - || (cost == *best_cost - && dist == *best_dist - && candidate.node_addr() < best.as_ref().unwrap().0.node_addr()) - } - }; - - if dominated { - best = Some((candidate, cost, dist)); - } - } - - best.map(|(peer, _, _)| peer) - } - - /// Check if a destination is in any peer's bloom filter. - pub fn destination_in_filters(&self, dest: &NodeAddr) -> Vec<&ActivePeer> { - self.peers.values().filter(|p| p.may_reach(dest)).collect() + let dest_coords = self.coord_cache.get(dest, now_ms).cloned(); + routing::classify_forward( + dest, + chosen_peer, + self.node_addr(), + self.tree_state.my_coords(), + dest_coords.as_ref(), + self.tree_state.peer_coords(chosen_peer), + ) } /// Get the TUN packet sender channel. @@ -3002,6 +2904,62 @@ impl Node { } } +/// Shell-side [`routing::RoutingView`] seam over live `Node` state — the sole +/// routing read adapter the shell retains. It hands the sans-IO routing core +/// raw per-peer reads (enumeration plus `may_reach` / `can_send` / `link_cost` +/// / `coords`) so the candidate assembly, selection, and error synthesis all +/// live in `proto::routing::core`; no routing decision or assembly logic +/// remains here. +/// +/// Field-narrowed to `coord_cache` + `peers` + `tree_state` (never `&Node` +/// whole) so it borrows disjointly from `&mut self.routing` on the +/// forward/synth path, where the handler also holds the mutable `Router`. +/// +/// Two call sites: +/// - `find_next_hop` builds it to assemble bloom candidates via the `peer_*` +/// reads; it never queries `is_congested`, so it leaves `congested` false. +/// - `handle_session_datagram` builds it for `Router::route` / `synth_*`, +/// which read `is_congested` (precomputed once for the resolved next hop) +/// and `cached_coords`. +pub(in crate::node) struct NodeRoutingView<'a> { + pub(in crate::node) coord_cache: &'a CoordCache, + pub(in crate::node) peers: &'a HashMap, + pub(in crate::node) tree_state: &'a TreeState, + pub(in crate::node) congested: bool, +} + +impl routing::RoutingView for NodeRoutingView<'_> { + fn is_congested(&self, _next_hop: &NodeAddr) -> bool { + self.congested + } + + fn cached_coords(&self, dest: &NodeAddr, now_ms: u64) -> Option { + self.coord_cache.get(dest, now_ms).cloned() + } + + fn peer_addrs(&self) -> Vec { + self.peers.keys().copied().collect() + } + + fn peer_may_reach(&self, peer: &NodeAddr, dest: &NodeAddr) -> bool { + self.peers.get(peer).is_some_and(|p| p.may_reach(dest)) + } + + fn peer_can_send(&self, peer: &NodeAddr) -> bool { + self.peers.get(peer).is_some_and(|p| p.can_send()) + } + + fn peer_link_cost(&self, peer: &NodeAddr) -> f64 { + self.peers + .get(peer) + .map_or(f64::INFINITY, |p| p.link_cost()) + } + + fn peer_coords(&self, peer: &NodeAddr) -> Option { + self.tree_state.peer_coords(peer).cloned() + } +} + /// Project an MMP metrics block into the snapshot /// [`EntityMmp`](crate::control::snapshot::EntityMmp) shared by `show_peers` /// (link-layer, `path_mtu = None`) and `show_sessions` (session-layer, diff --git a/src/node/routing_error_rate_limit.rs b/src/node/routing_error_rate_limit.rs deleted file mode 100644 index 32cdab4..0000000 --- a/src/node/routing_error_rate_limit.rs +++ /dev/null @@ -1,161 +0,0 @@ -//! Routing error signal rate limiting. -//! -//! Prevents routing error floods (CoordsRequired / PathBroken) by -//! rate-limiting error signals per destination address at transit nodes. - -use crate::NodeAddr; -use std::collections::HashMap; -use std::time::{Duration, Instant}; - -/// Rate limiter for routing error signals (CoordsRequired / PathBroken). -/// -/// Tracks the last time a routing error was sent for each destination -/// address and enforces a minimum interval to prevent floods. -pub struct RoutingErrorRateLimiter { - /// Maps destination NodeAddr to the last time we sent an error about it. - last_sent: HashMap, - /// Minimum interval between error signals for the same destination. - min_interval: Duration, - /// Maximum age of entries before cleanup. - max_age: Duration, -} - -impl RoutingErrorRateLimiter { - /// Create a new rate limiter. - /// - /// Default: max 10 errors/sec per destination (100ms interval). - pub fn new() -> Self { - Self { - last_sent: HashMap::new(), - min_interval: Duration::from_millis(100), - max_age: Duration::from_secs(10), - } - } - - /// Create a rate limiter with a custom minimum interval. - pub fn with_interval(min_interval: Duration) -> Self { - Self { - last_sent: HashMap::new(), - min_interval, - max_age: Duration::from_secs(10), - } - } - - /// Check if we should send a routing error for this destination. - /// - /// Returns true if enough time has passed since the last error for - /// this destination, or if this is the first error. Updates internal - /// state when returning true. - pub fn should_send(&mut self, dest_addr: &NodeAddr) -> bool { - let now = Instant::now(); - - if let Some(&last) = self.last_sent.get(dest_addr) - && now.duration_since(last) < self.min_interval - { - return false; - } - - self.last_sent.insert(*dest_addr, now); - self.cleanup(now); - true - } - - /// Remove entries older than max_age. - fn cleanup(&mut self, now: Instant) { - self.last_sent - .retain(|_, &mut last| now.duration_since(last) < self.max_age); - } - - #[cfg(test)] - pub fn len(&self) -> usize { - self.last_sent.len() - } -} - -impl Default for RoutingErrorRateLimiter { - fn default() -> Self { - Self::new() - } -} - -#[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) - } - - #[test] - fn test_first_send_allowed() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - } - - #[test] - fn test_rapid_sends_rate_limited() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - assert!(!limiter.should_send(&addr(1))); - assert!(!limiter.should_send(&addr(1))); - } - - #[test] - fn test_different_destinations_independent() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - assert!(limiter.should_send(&addr(2))); - assert!(!limiter.should_send(&addr(1))); - assert!(!limiter.should_send(&addr(2))); - } - - #[test] - fn test_send_allowed_after_interval() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - - thread::sleep(Duration::from_millis(110)); - - assert!(limiter.should_send(&addr(1))); - } - - #[test] - fn test_cleanup_removes_old_entries() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - assert!(limiter.should_send(&addr(2))); - assert_eq!(limiter.len(), 2); - - let future = Instant::now() + Duration::from_secs(11); - limiter.cleanup(future); - assert_eq!(limiter.len(), 0); - } - - #[test] - fn test_cleanup_preserves_recent_entries() { - let mut limiter = RoutingErrorRateLimiter::new(); - assert!(limiter.should_send(&addr(1))); - assert_eq!(limiter.len(), 1); - - limiter.cleanup(Instant::now()); - assert_eq!(limiter.len(), 1); - } - - #[test] - fn test_with_interval_custom_rate() { - let mut limiter = RoutingErrorRateLimiter::with_interval(Duration::from_millis(500)); - assert!(limiter.should_send(&addr(1))); - assert!(!limiter.should_send(&addr(1))); - - // Still rate-limited after 200ms (would pass with default 100ms) - thread::sleep(Duration::from_millis(200)); - assert!(!limiter.should_send(&addr(1))); - - // Allowed after 500ms total - thread::sleep(Duration::from_millis(350)); - assert!(limiter.should_send(&addr(1))); - } -} diff --git a/src/proto/mod.rs b/src/proto/mod.rs index badef00..78c2fc8 100644 --- a/src/proto/mod.rs +++ b/src/proto/mod.rs @@ -4,3 +4,4 @@ //! I/O adapters remain in `node::handlers`. pub(crate) mod discovery; +pub(crate) mod routing; diff --git a/src/proto/routing/core.rs b/src/proto/routing/core.rs new file mode 100644 index 0000000..2260169 --- /dev/null +++ b/src/proto/routing/core.rs @@ -0,0 +1,423 @@ +//! Sans-IO routing decision core. +//! +//! Pure, runtime-agnostic transit-forward decision for SessionDatagrams. The +//! async I/O adapter in `node::handlers::forwarding` decodes the wire bytes, +//! pre-resolves the next hop, builds a [`RoutingView`] over live node state, +//! calls [`Router::route`], and drives the returned [`RouteOutcome`] (the +//! actual encrypted sends, metrics, and logging). No I/O, no clock, no +//! metrics, no logging here. +//! +//! This module also holds the pure candidate assembly ([`routing_candidates`]) +//! and the hop-selection / route-classification helpers +//! ([`select_best_candidate`], [`classify_forward`]). The assembly enumerates +//! peers through the [`RoutingView`] seam, applies the bloom `may_reach` +//! filter, and snapshots each survivor into a [`Candidate`]; the shell hands +//! over only raw per-peer reads (enumeration plus `may_reach` / `can_send` / +//! `link_cost` / `coords`). Selection and classification then consume the +//! assembled set. All routing narrowing and decision logic lives here. + +use super::state::Router; +use super::wire::{CoordsRequired, MtuExceeded, PathBroken}; +use crate::protocol::{SessionDatagram, SessionDatagramRef}; +use crate::{NodeAddr, TreeCoordinate}; + +/// Read-only view of routing state the routing core needs. +/// +/// The core defines this interface; the async shell (`node`) implements it +/// over the live peer/coord/congestion state. 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 the outgoing link toward `next_hop` congested (ECN local signal)? + fn is_congested(&self, next_hop: &NodeAddr) -> bool; + /// Cached destination coordinates for `dest`, if any (read-only lookup). + /// + /// Drives the PathBroken-vs-CoordsRequired choice in + /// [`Router::synth_routing_error`]. + fn cached_coords(&self, dest: &NodeAddr, now_ms: u64) -> Option; + + /// Node addresses of every currently-active peer — the raw enumeration the + /// candidate assembly iterates. No filtering or ordering is applied here; + /// [`routing_candidates`] applies the bloom `may_reach` narrowing in core. + fn peer_addrs(&self) -> Vec; + /// Does `peer`'s bloom filter indicate it may reach `dest`? The raw + /// per-peer predicate the core assembly filters candidates on. + fn peer_may_reach(&self, peer: &NodeAddr, dest: &NodeAddr) -> bool; + /// Can `peer`'s session currently carry a forward? + fn peer_can_send(&self, peer: &NodeAddr) -> bool; + /// `peer`'s outgoing link cost (lower is preferred). + fn peer_link_cost(&self, peer: &NodeAddr) -> f64; + /// `peer`'s tree coordinates, if known. + fn peer_coords(&self, peer: &NodeAddr) -> Option; +} + +/// A next hop the shell resolved for a transit forward: the peer address and +/// the outgoing link's transport MTU (already narrowed to the specific link). +pub(crate) struct NextHop { + pub addr: NodeAddr, + pub link_mtu: u16, +} + +/// Why a datagram was dropped without forwarding or delivering. +pub(crate) enum DropReason { + /// Received TTL was already exhausted (0) — cannot decrement further. + TtlExhausted, +} + +/// Outcome of routing an inbound SessionDatagram. +pub(crate) enum RouteOutcome { + /// Drop the datagram; the shell records the reason-specific metric + log. + Drop { reason: DropReason }, + /// Deliver to the local session layer. Carries no bytes — the shell + /// services delivery from the borrowed datagram ref, avoiding a copy. + DeliverLocal, + /// Forward toward `next_hop`. `bytes` is the fully re-encoded datagram + /// (TTL decremented, path MTU min-folded), the single copy the shell would + /// have produced itself. `outgoing_ce` is the CE flag to set on the send. + Forward { + next_hop: NodeAddr, + bytes: Vec, + outgoing_ce: bool, + }, + /// No route to the destination. The shell synthesizes a routing error + /// signal back toward the source. + NoRoute, +} + +/// An I/O action the async shell performs on the core's behalf. +pub(crate) enum RouteAction { + /// Route the encoded error datagram in `bytes` toward `toward` (the failed + /// datagram's source). The shell resolves the outgoing link hop for + /// `toward` and performs the encrypted send. `toward` is the routing + /// target, not a pre-resolved link hop: the reverse hop is resolved + /// shell-side *after* the rate-limit gate so `find_next_hop`'s cache touch + /// keeps the same post-gate scope it had inline. + SendError { toward: NodeAddr, bytes: Vec }, +} + +impl Router { + /// Decide the fate of an inbound SessionDatagram: drop (TTL), local + /// delivery, transit forward, or no-route. Pure over the datagram, the + /// shell-resolved next hop, and the [`RoutingView`] reads. + /// + /// The shell pre-resolves `next_hop` only for genuine transit packets + /// (TTL > 0 and dest not local), so `find_next_hop`'s LRU-touch side + /// effect keeps the same scope it has today. `route` still re-checks TTL + /// and local delivery authoritatively. + pub(crate) fn route( + &mut self, + dg: &SessionDatagramRef<'_>, + my_addr: &NodeAddr, + incoming_ce: bool, + next_hop: Option, + rv: &impl RoutingView, + ) -> RouteOutcome { + if dg.ttl == 0 { + return RouteOutcome::Drop { + reason: DropReason::TtlExhausted, + }; + } + if dg.dest_addr == *my_addr { + return RouteOutcome::DeliverLocal; + } + let nh = match next_hop { + Some(nh) => nh, + None => return RouteOutcome::NoRoute, + }; + + // Re-encode with decremented TTL and the path MTU min-folded against + // the outgoing link. This is the single owned copy + encode the shell + // performed inline today. + let mut datagram = SessionDatagram::new(dg.src_addr, dg.dest_addr, dg.payload.to_vec()); + datagram.ttl = dg.ttl - 1; + datagram.path_mtu = dg.path_mtu.min(nh.link_mtu); + let outgoing_ce = incoming_ce || rv.is_congested(&nh.addr); + let bytes = datagram.encode(); + RouteOutcome::Forward { + next_hop: nh.addr, + bytes, + outgoing_ce, + } + } + + /// Synthesize a routing error signal for an undeliverable transit datagram. + /// + /// Applies the per-destination rate-limit gate, then chooses the error PDU + /// from cached coordinate state: PathBroken (with last-known coords) when + /// `dest` is cached — we know where it is but cannot reach it — otherwise + /// CoordsRequired. The chosen PDU is wrapped in a fresh SessionDatagram + /// addressed back to `toward` (the failed datagram's source) and encoded. + /// + /// Returns `None` when the rate-limit gate suppresses the signal (the shell + /// drops silently). On `Some`, the shell resolves the reverse link hop for + /// `toward` and sends — resolving the hop only after this gate preserves + /// the pre-refactor ordering (rate-limit before `find_next_hop`'s cache + /// touch) and lets the shell distinguish suppression from no-reverse-route + /// for logging. + pub(crate) fn synth_routing_error( + &mut self, + dest: &NodeAddr, + toward: &NodeAddr, + my_addr: &NodeAddr, + rv: &impl RoutingView, + now_ms: u64, + default_ttl: u8, + ) -> Option { + if !self.error_limiter.should_send(dest, now_ms) { + return None; + } + let error_payload = match rv.cached_coords(dest, now_ms) { + Some(coords) => PathBroken::new(*dest, *my_addr) + .with_last_coords(coords) + .encode(), + None => CoordsRequired::new(*dest, *my_addr).encode(), + }; + let error_dg = SessionDatagram::new(*my_addr, *toward, error_payload).with_ttl(default_ttl); + Some(RouteAction::SendError { + toward: *toward, + bytes: error_dg.encode(), + }) + } + + /// Synthesize an MtuExceeded error signal after a forward send failed with + /// a bottleneck MTU. Applies the per-destination rate-limit gate (the same + /// limiter as [`synth_routing_error`]), then builds the MtuExceeded PDU + /// carrying `bottleneck_mtu`, wraps it in a fresh SessionDatagram addressed + /// back to `toward` (the failed datagram's source), and encodes it. + /// + /// Returns `None` when the gate suppresses the signal. On `Some`, the shell + /// resolves the reverse link hop for `toward` and sends — resolving the hop + /// only after this gate preserves the pre-refactor ordering (rate-limit + /// before `find_next_hop`'s cache touch). No coordinate read is involved; + /// unlike routing errors, the PDU is unconditional once the gate passes. + pub(crate) fn synth_mtu_exceeded( + &mut self, + dest: &NodeAddr, + toward: &NodeAddr, + my_addr: &NodeAddr, + bottleneck_mtu: u16, + now_ms: u64, + default_ttl: u8, + ) -> Option { + if !self.error_limiter.should_send(dest, now_ms) { + return None; + } + let error_payload = MtuExceeded::new(*dest, *my_addr, bottleneck_mtu).encode(); + let error_dg = SessionDatagram::new(*my_addr, *toward, error_payload).with_ttl(default_ttl); + Some(RouteAction::SendError { + toward: *toward, + bytes: error_dg.encode(), + }) + } +} + +/// Route class of a transit-forwarded packet, classified from tree +/// coordinates at the forwarding decision point. The six variants +/// partition `forwarded_packets` exactly. +/// +/// Two variants are up-and-over forwards (destination not in the chosen +/// peer's subtree); they differ in whether they depend on a child +/// advertising cross-link reach *upward* to its parent: +/// - `TreeDownCross`: the chosen peer is our tree descendant, but the +/// destination is *not* in that child's subtree. The forward only fired +/// because the child advertised cross-link reach upward to us, beyond its +/// own subtree. If children advertised only their subtree upward, this +/// forward would route up instead, so its count measures how much +/// forwarding depends on the upward cross-link advertisement — the +/// dive-to-tree-child cut-through. +/// - `CrosslinkAscend`: the chosen peer is lateral (neither ancestor nor +/// descendant) and the destination is not in its subtree. This is a node +/// using its *own* cross-link, learned via the peer's split-horizon +/// advertisement to its neighbors, so it does not depend on any upward +/// advertisement. Tracked alongside `TreeDownCross` as the lateral +/// up-and-over contrast. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RouteClass { + /// Chosen peer is our ancestor (tree-up). + TreeUp, + /// Chosen peer is our descendant and dest is in its subtree (canonical + /// tree-down). + TreeDown, + /// Chosen peer is our descendant but dest is *not* in its subtree: the + /// dive-to-tree-child cut-through enabled by upward cross-link + /// advertisement. + TreeDownCross, + /// Chosen peer is lateral and dest is in its subtree (subtree entry). + CrosslinkDescend, + /// Chosen peer is lateral and dest is not in its subtree (up-and-over). + CrosslinkAscend, + /// Chosen peer is the destination itself (degenerate direct hop). + DirectPeer, +} + +/// A bloom-filter routing candidate, snapshotted by [`routing_candidates`] +/// from the per-peer reads the [`RoutingView`] seam exposes. +/// +/// The assembly applies the bloom `may_reach` narrowing before building each +/// snapshot, so [`select_best_candidate`] is a pure consumer of an +/// already-narrowed set and names no shell peer type. +pub(crate) struct Candidate { + /// The candidate peer's node address. + pub addr: NodeAddr, + /// Whether the peer's session can currently carry a forward. + pub can_send: bool, + /// The outgoing link cost (lower is preferred). + pub link_cost: f64, + /// The candidate's tree coordinates, if known. + pub coords: Option, +} + +/// Assemble the bloom-filter routing candidates toward `dest`. +/// +/// Enumerates every peer through the [`RoutingView`] seam, applies the bloom +/// `may_reach` filter, and snapshots each surviving peer's send-eligibility, +/// link cost, and tree coordinates into a [`Candidate`]. Pure over the seam's +/// primitive reads — the shell hands over raw per-peer data only, so all +/// narrowing and snapshotting happens here and [`select_best_candidate`] +/// consumes an already-assembled set. Candidate order follows the seam's +/// enumeration, which the selection ordering renders immaterial (it breaks +/// ties deterministically on `node_addr`). +pub(crate) fn routing_candidates(rv: &impl RoutingView, dest: &NodeAddr) -> Vec { + rv.peer_addrs() + .into_iter() + .filter(|peer| rv.peer_may_reach(peer, dest)) + .map(|peer| Candidate { + can_send: rv.peer_can_send(&peer), + link_cost: rv.peer_link_cost(&peer), + coords: rv.peer_coords(&peer), + addr: peer, + }) + .collect() +} + +/// Select the best next hop from a set of bloom-filter candidates. +/// +/// Uses each candidate's tree-coordinate distance to the destination as the +/// primary metric (after link cost). Only peers strictly closer to the +/// destination than we are (`my_coords`) are eligible — the self-distance +/// check that prevents routing loops. +/// +/// Ordering: `(link_cost, distance_to_dest, node_addr)`. Returns the winning +/// peer's address, or `None` when no candidate is send-ready and strictly +/// closer to the destination than us. +pub(crate) fn select_best_candidate( + candidates: &[Candidate], + dest_coords: &TreeCoordinate, + my_coords: &TreeCoordinate, +) -> Option { + let my_distance = my_coords.distance_to(dest_coords); + + let mut best: Option<(&Candidate, f64, usize)> = None; + + for candidate in candidates { + if !candidate.can_send { + continue; + } + + let cost = candidate.link_cost; + + let dist = candidate + .coords + .as_ref() + .map(|pc| pc.distance_to(dest_coords)) + .unwrap_or(usize::MAX); + + // Self-distance check: only consider peers strictly closer + // to the destination than we are (prevents routing loops) + if dist >= my_distance { + continue; + } + + let dominated = match &best { + None => true, + Some((_, best_cost, best_dist)) => { + cost < *best_cost + || (cost == *best_cost && dist < *best_dist) + || (cost == *best_cost + && dist == *best_dist + && candidate.addr < best.as_ref().unwrap().0.addr) + } + }; + + if dominated { + best = Some((candidate, cost, dist)); + } + } + + best.map(|(candidate, _, _)| candidate.addr) +} + +/// Classify a transit forward by route class from tree coordinates. +/// +/// Pure re-expression of the node-shell classifier. The shell pre-resolves +/// the destination coordinates from its cache (the sole impurity — a +/// read-only cache lookup) and reads `my_coords` / `peer_coords` from tree +/// state, then calls this. Called at the transit chokepoint after +/// `find_next_hop` returns a peer, so the six classes partition +/// `forwarded_packets` exactly. The branch that `find_next_hop` took (bloom +/// vs greedy-tree) is *not* the route class: a peer can be selected by +/// either, so the cut-through splits (`TreeDownCross`, `CrosslinkAscend`) are +/// decided here from coordinates, not from which branch fired. +/// +/// Both the tree-down and cross-link branches split on whether the +/// destination is in the chosen peer's subtree; when `dest_coords` is +/// unavailable that test defaults to "not in subtree", i.e. the up-and-over +/// variant (`TreeDownCross` for a descendant peer, `CrosslinkAscend` for a +/// lateral one). +pub(crate) fn classify_forward( + dest: &NodeAddr, + chosen_peer: &NodeAddr, + my_addr: &NodeAddr, + my_coords: &TreeCoordinate, + dest_coords: Option<&TreeCoordinate>, + peer_coords: Option<&TreeCoordinate>, +) -> RouteClass { + // Degenerate: the next hop is the destination itself (Branch 2). + if chosen_peer == dest { + return RouteClass::DirectPeer; + } + + // Tree-up: the chosen peer is our ancestor. + if my_coords.has_ancestor(chosen_peer) { + return RouteClass::TreeUp; + } + + // Whether the destination is in the chosen peer's subtree. Both the + // tree-down and cross-link splits below turn on this same test, so it + // is computed once. On the live transit path the dest coords are + // always present here: `find_next_hop` looks them up with an early + // return, so a coord-cache miss yields no next hop to classify (the + // caller signals `CoordsRequired` instead of forwarding). The miss + // branch below is therefore defensive — reachable only by direct + // unit-test calls — and defaults the test to "not in subtree", i.e. + // the up-and-over variant of whichever branch fires (TreeDownCross for + // a descendant peer, CrosslinkAscend for a lateral one), matching the + // original cross-link default-to-ascend. + let dest_in_peer_subtree = + dest_coords.is_some_and(|dest_coords| dest_coords.has_ancestor(chosen_peer)); + + // Tree-down: the chosen peer is our descendant (we are its ancestor). + // Split by subtree membership: a dest genuinely below the child is the + // canonical tree-down; a dest *not* below it means we only forwarded + // down because the child advertised cross-link reach upward, beyond its + // own subtree — the dive-to-tree-child cut-through (TreeDownCross). + if let Some(peer_coords) = peer_coords + && peer_coords.has_ancestor(my_addr) + { + return if dest_in_peer_subtree { + RouteClass::TreeDown + } else { + RouteClass::TreeDownCross + }; + } + + // Cross-link (lateral): split by whether the destination is in the + // chosen peer's subtree. Descend = subtree entry; ascend = up-and-over + // via the node's own cross-link (learned from the peer's split-horizon + // advertisement, independent of any upward advertisement). + if dest_in_peer_subtree { + return RouteClass::CrosslinkDescend; + } + + RouteClass::CrosslinkAscend +} diff --git a/src/proto/routing/limits.rs b/src/proto/routing/limits.rs new file mode 100644 index 0000000..d16cdfb --- /dev/null +++ b/src/proto/routing/limits.rs @@ -0,0 +1,90 @@ +//! Routing error signal rate limiting. +//! +//! Prevents routing error floods (CoordsRequired / PathBroken) by +//! rate-limiting error signals per destination address at transit nodes. +//! +//! Runtime-agnostic: the clock is injected as `now_ms` (Unix milliseconds, +//! the `Node::now_ms()` wall-clock basis) rather than read internally, and +//! per-destination state lives in an `alloc` `BTreeMap` for `no_std` +//! portability and deterministic ordering. + +use crate::NodeAddr; +use alloc::collections::BTreeMap; + +/// Default minimum interval between error signals: 100 ms (max 10 errors/sec +/// per destination). +const DEFAULT_MIN_INTERVAL_MS: u64 = 100; + +/// Maximum age of a per-destination entry before cleanup: 10 s. +const MAX_AGE_MS: u64 = 10_000; + +/// Rate limiter for routing error signals (CoordsRequired / PathBroken). +/// +/// Tracks the last time a routing error was sent for each destination +/// address and enforces a minimum interval to prevent floods. +pub struct RoutingErrorRateLimiter { + /// Maps destination NodeAddr to the last time (Unix ms) we sent an error + /// about it. + last_sent: BTreeMap, + /// Minimum interval between error signals for the same destination (ms). + min_interval_ms: u64, + /// Maximum age of entries before cleanup (ms). + max_age_ms: u64, +} + +impl RoutingErrorRateLimiter { + /// Create a new rate limiter. + /// + /// Default: max 10 errors/sec per destination (100ms interval). + pub fn new() -> Self { + Self { + last_sent: BTreeMap::new(), + min_interval_ms: DEFAULT_MIN_INTERVAL_MS, + max_age_ms: MAX_AGE_MS, + } + } + + /// Create a rate limiter with a custom minimum interval in milliseconds. + pub fn with_interval_ms(min_interval_ms: u64) -> Self { + Self { + last_sent: BTreeMap::new(), + min_interval_ms, + max_age_ms: MAX_AGE_MS, + } + } + + /// Check if we should send a routing error for this destination at + /// `now_ms` (Unix milliseconds). + /// + /// Returns true if enough time has passed since the last error for + /// this destination, or if this is the first error. Updates internal + /// state when returning true. + pub fn should_send(&mut self, dest_addr: &NodeAddr, now_ms: u64) -> bool { + if let Some(&last) = self.last_sent.get(dest_addr) + && now_ms.saturating_sub(last) < self.min_interval_ms + { + return false; + } + + self.last_sent.insert(*dest_addr, now_ms); + self.cleanup(now_ms); + true + } + + /// Remove entries older than max_age. + pub(crate) fn cleanup(&mut self, now_ms: u64) { + self.last_sent + .retain(|_, &mut last| now_ms.saturating_sub(last) < self.max_age_ms); + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.last_sent.len() + } +} + +impl Default for RoutingErrorRateLimiter { + fn default() -> Self { + Self::new() + } +} diff --git a/src/proto/routing/mod.rs b/src/proto/routing/mod.rs new file mode 100644 index 0000000..532f3bb --- /dev/null +++ b/src/proto/routing/mod.rs @@ -0,0 +1,33 @@ +//! Sans-IO routing protocol state. +//! +//! Pure, runtime-agnostic routing state and decision core, migrated out of +//! the async node shell. The async I/O handlers remain in +//! `node::handlers::forwarding`. +//! +//! - `core.rs` — the `RoutingView` read-seam trait, the `NextHop` / +//! `RouteOutcome` types, `Router::route`, the pure transit-forward +//! decision (TTL, local-vs-forward, path-MTU min-fold, ECN CE), and the +//! pure candidate assembly + hop-selection / route-classification helpers +//! (`Candidate`, `RouteClass`, `routing_candidates`, `select_best_candidate`, +//! `classify_forward`). The assembly reads raw per-peer data through the +//! `RoutingView` seam; the shell keeps only the seam impl. +//! - `state.rs` — `Router`, the routing-subsystem state owned by `Node`. +//! - `limits.rs` — the routing error-signal rate limiter. +//! - `wire.rs` — the routing error-signal PDUs (`CoordsRequired`, +//! `PathBroken`, `MtuExceeded`), relocated from `protocol::session`. + +mod core; +mod limits; +mod state; +mod wire; + +#[cfg(test)] +mod tests; + +pub(crate) use core::{ + DropReason, NextHop, RouteAction, RouteClass, RouteOutcome, RoutingView, classify_forward, + routing_candidates, select_best_candidate, +}; +pub(crate) use limits::RoutingErrorRateLimiter; +pub(crate) use state::Router; +pub use wire::{COORDS_REQUIRED_SIZE, CoordsRequired, MTU_EXCEEDED_SIZE, MtuExceeded, PathBroken}; diff --git a/src/proto/routing/state.rs b/src/proto/routing/state.rs new file mode 100644 index 0000000..ead7070 --- /dev/null +++ b/src/proto/routing/state.rs @@ -0,0 +1,23 @@ +//! Routing-subsystem state owned by [`Node`](crate::node::Node). +//! +//! Groups the routing error-signal rate limiter behind a single struct so +//! the forwarding handlers can evolve toward a sans-IO core without +//! threading the limiter field through `Node`. + +use super::limits::RoutingErrorRateLimiter; + +/// Routing-subsystem state. +pub(crate) struct Router { + /// Rate limiter for routing error signals (CoordsRequired / PathBroken). + pub(crate) error_limiter: RoutingErrorRateLimiter, +} + +impl Router { + /// Create routing state with a default error-signal rate limiter, + /// matching the pre-refactor initialization exactly. + pub(crate) fn new() -> Self { + Self { + error_limiter: RoutingErrorRateLimiter::new(), + } + } +} diff --git a/src/proto/routing/tests/core.rs b/src/proto/routing/tests/core.rs new file mode 100644 index 0000000..9dded93 --- /dev/null +++ b/src/proto/routing/tests/core.rs @@ -0,0 +1,358 @@ +//! Tests for the sans-IO routing decision core. + +use super::util::{MockPeer, MockRoutingView, make_coords, make_datagram_ref, make_next_hop}; +use crate::TreeCoordinate; +use crate::proto::routing::{ + DropReason, RouteAction, RouteOutcome, Router, RoutingView, routing_candidates, +}; +use crate::protocol::{SessionDatagramRef, SessionMessageType}; +use crate::testutil::make_node_addr; + +/// Decode a forwarded byte buffer (which carries the leading msg_type byte) +/// back into a borrowed view so tests can inspect the re-encoded header. +fn decode_forward(bytes: &[u8]) -> SessionDatagramRef<'_> { + SessionDatagramRef::decode(&bytes[1..]).expect("forwarded datagram re-decodes") +} + +#[test] +fn ttl_zero_drops_as_exhausted() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let dg = make_datagram_ref(0, make_node_addr(0x20)); + let rv = MockRoutingView::new(false); + let out = router.route(&dg, &my_addr, false, None, &rv); + assert!(matches!( + out, + RouteOutcome::Drop { + reason: DropReason::TtlExhausted + } + )); +} + +#[test] +fn destination_is_self_delivers_local() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let dg = make_datagram_ref(5, my_addr); + let rv = MockRoutingView::new(false); + // A next hop is irrelevant for local delivery; the shell would pass None. + let out = router.route(&dg, &my_addr, false, None, &rv); + assert!(matches!(out, RouteOutcome::DeliverLocal)); +} + +#[test] +fn transit_without_next_hop_is_no_route() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let dg = make_datagram_ref(5, make_node_addr(0x20)); + let rv = MockRoutingView::new(false); + let out = router.route(&dg, &my_addr, false, None, &rv); + assert!(matches!(out, RouteOutcome::NoRoute)); +} + +#[test] +fn forward_decrements_ttl_and_folds_link_mtu() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let nh_addr = make_node_addr(0x30); + let dg = make_datagram_ref(5, make_node_addr(0x20)); + let rv = MockRoutingView::new(false); + let out = router.route( + &dg, + &my_addr, + false, + Some(make_next_hop(nh_addr, 1400)), + &rv, + ); + match out { + RouteOutcome::Forward { + next_hop, + bytes, + outgoing_ce, + } => { + assert_eq!(next_hop, nh_addr); + assert!(!outgoing_ce); + let decoded = decode_forward(&bytes); + assert_eq!(decoded.ttl, 4, "TTL decremented once"); + assert_eq!(decoded.path_mtu, 1400, "link MTU is the smaller bound"); + } + _ => panic!("expected Forward"), + } +} + +#[test] +fn forward_keeps_smaller_datagram_path_mtu() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let nh_addr = make_node_addr(0x30); + let mut dg = make_datagram_ref(5, make_node_addr(0x20)); + dg.path_mtu = 900; // datagram already bounded below the link MTU + let rv = MockRoutingView::new(false); + let out = router.route( + &dg, + &my_addr, + false, + Some(make_next_hop(nh_addr, 1400)), + &rv, + ); + match out { + RouteOutcome::Forward { bytes, .. } => { + let decoded = decode_forward(&bytes); + assert_eq!(decoded.path_mtu, 900, "datagram MTU is the smaller bound"); + } + _ => panic!("expected Forward"), + } +} + +#[test] +fn outgoing_ce_set_by_incoming_ce() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let nh_addr = make_node_addr(0x30); + let dg = make_datagram_ref(5, make_node_addr(0x20)); + let rv = MockRoutingView::new(false); // not locally congested + let out = router.route(&dg, &my_addr, true, Some(make_next_hop(nh_addr, 1400)), &rv); + match out { + RouteOutcome::Forward { outgoing_ce, .. } => assert!(outgoing_ce), + _ => panic!("expected Forward"), + } +} + +#[test] +fn outgoing_ce_set_by_local_congestion() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let nh_addr = make_node_addr(0x30); + let dg = make_datagram_ref(5, make_node_addr(0x20)); + let rv = MockRoutingView::new(true); // locally congested + let out = router.route( + &dg, + &my_addr, + false, + Some(make_next_hop(nh_addr, 1400)), + &rv, + ); + match out { + RouteOutcome::Forward { outgoing_ce, .. } => assert!(outgoing_ce), + _ => panic!("expected Forward"), + } +} + +#[test] +fn outgoing_ce_clear_when_neither_signal() { + let mut router = Router::new(); + let my_addr = make_node_addr(0x10); + let nh_addr = make_node_addr(0x30); + let dg = make_datagram_ref(5, make_node_addr(0x20)); + let rv = MockRoutingView::new(false); + let out = router.route( + &dg, + &my_addr, + false, + Some(make_next_hop(nh_addr, 1400)), + &rv, + ); + match out { + RouteOutcome::Forward { outgoing_ce, .. } => assert!(!outgoing_ce), + _ => panic!("expected Forward"), + } +} + +#[test] +fn cached_coords_reads_the_view_table() { + let target = make_node_addr(0x40); + let rv = MockRoutingView { + congested: false, + coords: vec![(target, TreeCoordinate::root(target))], + peers: Vec::new(), + }; + assert!(rv.cached_coords(&target, 0).is_some()); + assert!(rv.cached_coords(&make_node_addr(0x41), 0).is_none()); +} + +#[test] +fn routing_candidates_filters_by_may_reach_and_snapshots() { + let dest = make_node_addr(0x50); + let reacher = make_node_addr(0x60); + let non_reacher = make_node_addr(0x61); + let reacher_coords = make_coords(&[0x01, 0x60]); + let rv = MockRoutingView { + peers: vec![ + MockPeer { + addr: reacher, + reach: vec![dest], + can_send: true, + link_cost: 2.5, + coords: Some(reacher_coords.clone()), + }, + MockPeer { + // Bloom filter does not contain dest — narrowed out in core. + addr: non_reacher, + reach: Vec::new(), + can_send: true, + link_cost: 1.0, + coords: None, + }, + ], + ..MockRoutingView::new(false) + }; + + let candidates = routing_candidates(&rv, &dest); + + assert_eq!( + candidates.len(), + 1, + "only peers whose bloom may_reach the dest survive assembly" + ); + let c = &candidates[0]; + assert_eq!(c.addr, reacher); + assert!(c.can_send); + assert_eq!(c.link_cost, 2.5); + assert_eq!(c.coords, Some(reacher_coords)); +} + +/// Extract the error-PDU msg_type byte from an encoded routing-error action. +/// The action bytes are a SessionDatagram (leading link msg_type byte, then +/// the header); its payload is the error PDU, whose msg_type sits at offset 4 +/// after the 4-byte FSP prefix. +fn error_pdu_type(action: &RouteAction) -> u8 { + let RouteAction::SendError { bytes, .. } = action; + let dg = SessionDatagramRef::decode(&bytes[1..]).expect("error datagram re-decodes"); + dg.payload[4] +} + +#[test] +fn synth_uses_pathbroken_when_coords_cached() { + let mut router = Router::new(); + let dest = make_node_addr(0x20); + let source = make_node_addr(0x21); + let my_addr = make_node_addr(0x10); + let rv = MockRoutingView { + congested: false, + coords: vec![(dest, TreeCoordinate::root(dest))], + peers: Vec::new(), + }; + let action = router + .synth_routing_error(&dest, &source, &my_addr, &rv, 0, 64) + .expect("gate passes on first call"); + let RouteAction::SendError { toward, .. } = &action; + assert_eq!( + *toward, source, + "error routes back toward the failed source" + ); + assert_eq!( + error_pdu_type(&action), + SessionMessageType::PathBroken.to_byte(), + "cached coords select PathBroken", + ); +} + +#[test] +fn synth_uses_coords_required_when_not_cached() { + let mut router = Router::new(); + let dest = make_node_addr(0x20); + let source = make_node_addr(0x21); + let my_addr = make_node_addr(0x10); + let rv = MockRoutingView::new(false); // empty coord table + let action = router + .synth_routing_error(&dest, &source, &my_addr, &rv, 0, 64) + .expect("gate passes on first call"); + assert_eq!( + error_pdu_type(&action), + SessionMessageType::CoordsRequired.to_byte(), + "absent coords select CoordsRequired", + ); +} + +#[test] +fn synth_rate_limit_gate_suppresses_second_call() { + let mut router = Router::new(); + let dest = make_node_addr(0x20); + let source = make_node_addr(0x21); + let my_addr = make_node_addr(0x10); + let rv = MockRoutingView::new(false); + // First call for this destination passes the gate. + assert!( + router + .synth_routing_error(&dest, &source, &my_addr, &rv, 0, 64) + .is_some() + ); + // An immediate second call for the same destination is within the + // rate-limit window and is suppressed (no sleeps needed — the two calls + // are microseconds apart, well under the 100 ms interval). + assert!( + router + .synth_routing_error(&dest, &source, &my_addr, &rv, 0, 64) + .is_none() + ); + // A different destination is independent and still allowed. + let other = make_node_addr(0x22); + assert!( + router + .synth_routing_error(&other, &source, &my_addr, &rv, 0, 64) + .is_some() + ); +} + +/// Extract the bottleneck MTU (trailing u16 LE) from an MtuExceeded action's +/// PDU. Layout after the outer link byte + SessionDatagram header: FSP prefix +/// (4) + msg_type (1) + flags (1) + dest_addr (16) + reporter (16) + mtu (2). +fn mtu_exceeded_bottleneck(action: &RouteAction) -> u16 { + let RouteAction::SendError { bytes, .. } = action; + let dg = SessionDatagramRef::decode(&bytes[1..]).expect("error datagram re-decodes"); + let p = dg.payload; + u16::from_le_bytes([p[38], p[39]]) +} + +#[test] +fn synth_mtu_exceeded_carries_bottleneck_and_targets_source() { + let mut router = Router::new(); + let dest = make_node_addr(0x20); + let source = make_node_addr(0x21); + let my_addr = make_node_addr(0x10); + let action = router + .synth_mtu_exceeded(&dest, &source, &my_addr, 1280, 0, 64) + .expect("gate passes on first call"); + let RouteAction::SendError { toward, .. } = &action; + assert_eq!( + *toward, source, + "signal routes back toward the failed source" + ); + assert_eq!( + error_pdu_type(&action), + SessionMessageType::MtuExceeded.to_byte(), + "PDU is an MtuExceeded signal", + ); + assert_eq!( + mtu_exceeded_bottleneck(&action), + 1280, + "bottleneck MTU is carried verbatim", + ); +} + +#[test] +fn synth_mtu_exceeded_rate_limit_gate_suppresses_second_call() { + let mut router = Router::new(); + let dest = make_node_addr(0x20); + let source = make_node_addr(0x21); + let my_addr = make_node_addr(0x10); + // First call for this destination passes the gate. + assert!( + router + .synth_mtu_exceeded(&dest, &source, &my_addr, 1280, 0, 64) + .is_some() + ); + // Immediate second call for the same destination is suppressed. + assert!( + router + .synth_mtu_exceeded(&dest, &source, &my_addr, 1280, 0, 64) + .is_none() + ); + // A different destination is independent and still allowed. + let other = make_node_addr(0x22); + assert!( + router + .synth_mtu_exceeded(&other, &source, &my_addr, 1280, 0, 64) + .is_some() + ); +} diff --git a/src/proto/routing/tests/limits.rs b/src/proto/routing/tests/limits.rs new file mode 100644 index 0000000..d93f41b --- /dev/null +++ b/src/proto/routing/tests/limits.rs @@ -0,0 +1,70 @@ +//! Tests for routing error-signal rate limiting. + +use crate::proto::routing::RoutingErrorRateLimiter; +use crate::testutil::make_node_addr as addr; + +#[test] +fn test_first_send_allowed() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); +} + +#[test] +fn test_rapid_sends_rate_limited() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); + assert!(!limiter.should_send(&addr(1), 0)); + assert!(!limiter.should_send(&addr(1), 50)); +} + +#[test] +fn test_different_destinations_independent() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); + assert!(limiter.should_send(&addr(2), 0)); + assert!(!limiter.should_send(&addr(1), 0)); + assert!(!limiter.should_send(&addr(2), 0)); +} + +#[test] +fn test_send_allowed_after_interval() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); + // 110 ms later, past the 100 ms window. + assert!(limiter.should_send(&addr(1), 110)); +} + +#[test] +fn test_cleanup_removes_old_entries() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); + assert!(limiter.should_send(&addr(2), 0)); + assert_eq!(limiter.len(), 2); + + // 11 s later, both entries exceed the 10 s max age. + limiter.cleanup(11_000); + assert_eq!(limiter.len(), 0); +} + +#[test] +fn test_cleanup_preserves_recent_entries() { + let mut limiter = RoutingErrorRateLimiter::new(); + assert!(limiter.should_send(&addr(1), 0)); + assert_eq!(limiter.len(), 1); + + limiter.cleanup(0); + assert_eq!(limiter.len(), 1); +} + +#[test] +fn test_with_interval_custom_rate() { + let mut limiter = RoutingErrorRateLimiter::with_interval_ms(500); + assert!(limiter.should_send(&addr(1), 0)); + assert!(!limiter.should_send(&addr(1), 0)); + + // Still rate-limited at 200 ms (would pass with the default 100 ms). + assert!(!limiter.should_send(&addr(1), 200)); + + // Allowed at 500 ms. + assert!(limiter.should_send(&addr(1), 500)); +} diff --git a/src/proto/routing/tests/mod.rs b/src/proto/routing/tests/mod.rs new file mode 100644 index 0000000..5a632a7 --- /dev/null +++ b/src/proto/routing/tests/mod.rs @@ -0,0 +1,6 @@ +//! Routing subsystem unit tests. Shared helpers live in `util`. + +mod core; +mod limits; +mod util; +mod wire; diff --git a/src/proto/routing/tests/util.rs b/src/proto/routing/tests/util.rs new file mode 100644 index 0000000..644a882 --- /dev/null +++ b/src/proto/routing/tests/util.rs @@ -0,0 +1,86 @@ +//! Shared test helpers for the routing subsystem unit tests. + +use crate::proto::routing::{NextHop, RoutingView}; +use crate::protocol::SessionDatagramRef; +use crate::testutil::make_node_addr; +use crate::{NodeAddr, TreeCoordinate}; + +/// A mock peer for the candidate-assembly seam: the set of destinations its +/// bloom filter reaches, its send state, link cost, and tree coordinates. +pub(super) struct MockPeer { + pub(super) addr: NodeAddr, + pub(super) reach: Vec, + pub(super) can_send: bool, + pub(super) link_cost: f64, + pub(super) coords: Option, +} + +/// Mock routing view: a fixed congestion answer, a small coord table, and a +/// set of peers the candidate assembly enumerates through the seam. +pub(super) struct MockRoutingView { + pub(super) congested: bool, + pub(super) coords: Vec<(NodeAddr, TreeCoordinate)>, + pub(super) peers: Vec, +} + +impl MockRoutingView { + pub(super) fn new(congested: bool) -> Self { + Self { + congested, + coords: Vec::new(), + peers: Vec::new(), + } + } + + fn peer(&self, addr: &NodeAddr) -> Option<&MockPeer> { + self.peers.iter().find(|p| p.addr == *addr) + } +} + +impl RoutingView for MockRoutingView { + fn is_congested(&self, _next_hop: &NodeAddr) -> bool { + self.congested + } + fn cached_coords(&self, dest: &NodeAddr, _now_ms: u64) -> Option { + self.coords + .iter() + .find(|(addr, _)| addr == dest) + .map(|(_, coords)| coords.clone()) + } + fn peer_addrs(&self) -> Vec { + self.peers.iter().map(|p| p.addr).collect() + } + fn peer_may_reach(&self, peer: &NodeAddr, dest: &NodeAddr) -> bool { + self.peer(peer).is_some_and(|p| p.reach.contains(dest)) + } + fn peer_can_send(&self, peer: &NodeAddr) -> bool { + self.peer(peer).is_some_and(|p| p.can_send) + } + fn peer_link_cost(&self, peer: &NodeAddr) -> f64 { + self.peer(peer).map_or(f64::INFINITY, |p| p.link_cost) + } + fn peer_coords(&self, peer: &NodeAddr) -> Option { + self.peer(peer).and_then(|p| p.coords.clone()) + } +} + +/// Build a borrowed datagram with the given TTL and destination. The source is +/// a fixed address and the payload is empty (routing decisions never inspect +/// it); `path_mtu` starts at the maximum so tests can observe the min-fold. +pub(super) fn make_datagram_ref(ttl: u8, dest: NodeAddr) -> SessionDatagramRef<'static> { + SessionDatagramRef { + src_addr: make_node_addr(0x01), + dest_addr: dest, + ttl, + path_mtu: u16::MAX, + payload: &[], + } +} + +pub(super) fn make_next_hop(addr: NodeAddr, link_mtu: u16) -> NextHop { + NextHop { addr, link_mtu } +} + +pub(super) fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::from_addrs(ids.iter().map(|&v| make_node_addr(v)).collect()).unwrap() +} diff --git a/src/proto/routing/tests/wire.rs b/src/proto/routing/tests/wire.rs new file mode 100644 index 0000000..c69cd96 --- /dev/null +++ b/src/proto/routing/tests/wire.rs @@ -0,0 +1,130 @@ +//! Tests for the routing error-signal wire PDUs (`CoordsRequired`, +//! `PathBroken`, `MtuExceeded`). + +use super::util::make_coords; +use crate::proto::routing::{ + COORDS_REQUIRED_SIZE, CoordsRequired, MTU_EXCEEDED_SIZE, MtuExceeded, PathBroken, +}; +use crate::testutil::make_node_addr; + +#[test] +fn test_coords_required() { + let err = CoordsRequired::new(make_node_addr(1), make_node_addr(2)); + + assert_eq!(err.dest_addr, make_node_addr(1)); + assert_eq!(err.reporter, make_node_addr(2)); +} + +#[test] +fn test_path_broken() { + let err = PathBroken::new(make_node_addr(2), make_node_addr(3)) + .with_last_coords(make_coords(&[2, 0])); + + assert_eq!(err.dest_addr, make_node_addr(2)); + assert_eq!(err.reporter, make_node_addr(3)); + assert!(err.last_known_coords.is_some()); +} + +#[test] +fn test_coords_required_encode_decode() { + let err = CoordsRequired::new(make_node_addr(0xAA), make_node_addr(0xBB)); + + let encoded = err.encode(); + // 4 prefix + 1 msg_type + 1 flags + 16 dest + 16 reporter = 38 + assert_eq!(encoded.len(), 4 + COORDS_REQUIRED_SIZE); + // Check FSP prefix: phase 0x0, U flag + assert_eq!(encoded[0], 0x00); + assert_eq!(encoded[1], 0x04); // U flag + // msg_type after prefix + assert_eq!(encoded[4], 0x20); + + // decode after prefix + msg_type consumed + let decoded = CoordsRequired::decode(&encoded[5..]).unwrap(); + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); +} + +#[test] +fn test_path_broken_encode_decode_no_coords() { + let err = PathBroken::new(make_node_addr(0xCC), make_node_addr(0xDD)); + + let encoded = err.encode(); + // Check FSP prefix + assert_eq!(encoded[0], 0x00); + assert_eq!(encoded[1], 0x04); // U flag + assert_eq!(encoded[4], 0x21); // msg_type + + let decoded = PathBroken::decode(&encoded[5..]).unwrap(); + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + assert!(decoded.last_known_coords.is_none()); +} + +#[test] +fn test_path_broken_encode_decode_with_coords() { + let coords = make_coords(&[0xCC, 0xDD, 0xEE]); + let err = PathBroken::new(make_node_addr(0x11), make_node_addr(0x22)) + .with_last_coords(coords.clone()); + + let encoded = err.encode(); + let decoded = PathBroken::decode(&encoded[5..]).unwrap(); + + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + assert_eq!(decoded.last_known_coords.unwrap(), coords); +} + +#[test] +fn test_coords_required_decode_too_short() { + assert!(CoordsRequired::decode(&[]).is_err()); + assert!(CoordsRequired::decode(&[0x00; 10]).is_err()); +} + +#[test] +fn test_path_broken_decode_too_short() { + assert!(PathBroken::decode(&[]).is_err()); + assert!(PathBroken::decode(&[0x00; 20]).is_err()); +} + +#[test] +fn test_mtu_exceeded_encode_size() { + let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400); + let encoded = err.encode(); + // 4 prefix + 36 body = 40 + assert_eq!(encoded.len(), 4 + MTU_EXCEEDED_SIZE); +} + +#[test] +fn test_mtu_exceeded_encode_decode() { + let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400); + + let encoded = err.encode(); + // Check FSP prefix: phase 0x0, U flag + assert_eq!(encoded[0], 0x00); + assert_eq!(encoded[1], 0x04); // U flag + // msg_type after prefix + assert_eq!(encoded[4], 0x22); + + // decode after prefix + msg_type consumed + let decoded = MtuExceeded::decode(&encoded[5..]).unwrap(); + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + assert_eq!(decoded.mtu, 1400); +} + +#[test] +fn test_mtu_exceeded_decode_too_short() { + assert!(MtuExceeded::decode(&[]).is_err()); + assert!(MtuExceeded::decode(&[0x00; 20]).is_err()); + assert!(MtuExceeded::decode(&[0x00; 34]).is_err()); // exactly 1 byte short +} + +#[test] +fn test_mtu_exceeded_boundary_mtu_values() { + for mtu in [0u16, 1280, 1500, u16::MAX] { + let err = MtuExceeded::new(make_node_addr(1), make_node_addr(2), mtu); + let encoded = err.encode(); + let decoded = MtuExceeded::decode(&encoded[5..]).unwrap(); + assert_eq!(decoded.mtu, mtu); + } +} diff --git a/src/proto/routing/wire.rs b/src/proto/routing/wire.rs new file mode 100644 index 0000000..864d131 --- /dev/null +++ b/src/proto/routing/wire.rs @@ -0,0 +1,275 @@ +//! Routing error-signal wire PDUs. +//! +//! Plaintext link-layer error signals emitted by a transit router that +//! cannot forward a `SessionDatagram`: `CoordsRequired` (coordinate cache +//! miss), `PathBroken` (routing failure / local minimum), and `MtuExceeded` +//! (next-hop transport MTU too small). Relocated verbatim from +//! `protocol::session`; the shared coordinate helpers and the +//! `SessionMessageType` discriminant stay in `protocol` and are imported +//! downward here (a `proto -> protocol` dependency is allowed). + +use crate::NodeAddr; +use crate::protocol::ProtocolError; +use crate::protocol::session::{ + SessionMessageType, decode_optional_coords, encode_coords, encode_empty_coords, +}; +use crate::tree::TreeCoordinate; + +/// Link-layer error signal indicating router cache miss. +/// +/// Generated by a transit router when it cannot forward a SessionDatagram +/// due to missing cached coordinates for the destination. Carried inside +/// a new SessionDatagram addressed back to the original source +/// (src_addr=reporter, dest_addr=original_source). Plaintext — not +/// end-to-end encrypted, since the transit router has no session with +/// the source. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|----------|---------|------------------------------------| +/// | 0 | msg_type | 1 byte | 0x20 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr| 16 bytes| The node_addr we couldn't route to | +/// | 18 | reporter | 16 bytes| NodeAddr of reporting router | +/// +/// Payload: 34 bytes +#[derive(Clone, Debug)] +pub struct CoordsRequired { + /// Destination that couldn't be routed. + pub dest_addr: NodeAddr, + /// Router reporting the miss. + pub reporter: NodeAddr, +} + +/// Wire size of CoordsRequired payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16). +pub const COORDS_REQUIRED_SIZE: usize = 34; + +impl CoordsRequired { + /// Create a new CoordsRequired error. + pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self { + Self { + dest_addr, + reporter, + } + } + + /// Encode as wire format (4-byte FSP prefix + msg_type + body). + /// + /// Error signals use phase=0x0 with U flag set. + pub fn encode(&self) -> Vec { + // Body: msg_type + flags(reserved) + dest_addr + reporter + let body_len = 1 + 1 + 16 + 16; // 34 bytes + let mut buf = Vec::with_capacity(4 + body_len); + // FSP prefix: version 0, phase 0x0, U flag set + buf.push(0x00); // version 0, phase 0x0 + buf.push(0x04); // U flag + let payload_len = body_len as u16; + buf.extend_from_slice(&payload_len.to_le_bytes()); + // msg_type byte (after prefix, before body) + buf.push(SessionMessageType::CoordsRequired.to_byte()); + buf.push(0x00); // reserved flags + buf.extend_from_slice(self.dest_addr.as_bytes()); + buf.extend_from_slice(self.reporter.as_bytes()); + buf + } + + /// Decode from wire format (after FSP prefix and msg_type byte consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) = 33 + if payload.len() < 33 { + return Err(ProtocolError::MessageTooShort { + expected: 33, + got: payload.len(), + }); + } + // payload[0] is flags (reserved, ignored) + let mut dest_bytes = [0u8; 16]; + dest_bytes.copy_from_slice(&payload[1..17]); + let mut reporter_bytes = [0u8; 16]; + reporter_bytes.copy_from_slice(&payload[17..33]); + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + }) + } +} + +/// Error indicating routing failure (local minimum or unreachable). +/// +/// Carried inside a SessionDatagram addressed back to the original source. +/// The reporting router creates a new SessionDatagram with src_addr=reporter +/// and dest_addr=original_source, so the `original_src` field from the old +/// design is no longer needed — it's the SessionDatagram's dest_addr. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|-------------------|----------|-------------------------------| +/// | 0 | msg_type | 1 byte | 0x21 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr | 16 bytes | The unreachable node_addr | +/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | +/// | 34 | last_coords_count | 2 bytes | u16 LE | +/// | 36 | last_known_coords | 16 × n | Stale coords that failed | +#[derive(Clone, Debug)] +pub struct PathBroken { + /// Destination that couldn't be reached. + pub dest_addr: NodeAddr, + /// Node that detected the failure. + pub reporter: NodeAddr, + /// Optional: last known coordinates of destination. + pub last_known_coords: Option, +} + +impl PathBroken { + /// Create a new PathBroken error. + pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self { + Self { + dest_addr, + reporter, + last_known_coords: None, + } + } + + /// Add last known coordinates. + pub fn with_last_coords(mut self, coords: TreeCoordinate) -> Self { + self.last_known_coords = Some(coords); + self + } + + /// Encode as wire format (4-byte FSP prefix + msg_type + body). + /// + /// Error signals use phase=0x0 with U flag set. + pub fn encode(&self) -> Vec { + // Build body first to compute length + let mut body = Vec::new(); + body.push(SessionMessageType::PathBroken.to_byte()); + body.push(0x00); // reserved flags + body.extend_from_slice(self.dest_addr.as_bytes()); + body.extend_from_slice(self.reporter.as_bytes()); + if let Some(ref coords) = self.last_known_coords { + encode_coords(coords, &mut body); + } else { + encode_empty_coords(&mut body); + } + + // Prepend FSP prefix: version 0, phase 0x0, U flag set + let payload_len = body.len() as u16; + let mut buf = Vec::with_capacity(4 + body.len()); + buf.push(0x00); // version 0, phase 0x0 + buf.push(0x04); // U flag + buf.extend_from_slice(&payload_len.to_le_bytes()); + buf.extend_from_slice(&body); + buf + } + + /// Decode from wire format (after FSP prefix and msg_type byte consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) + coords_count(2) = 35 minimum + if payload.len() < 35 { + return Err(ProtocolError::MessageTooShort { + expected: 35, + got: payload.len(), + }); + } + // payload[0] is flags (reserved, ignored) + let mut dest_bytes = [0u8; 16]; + dest_bytes.copy_from_slice(&payload[1..17]); + let mut reporter_bytes = [0u8; 16]; + reporter_bytes.copy_from_slice(&payload[17..33]); + + let (last_known_coords, _consumed) = decode_optional_coords(&payload[33..])?; + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + last_known_coords, + }) + } +} + +/// Error indicating a forwarded packet exceeded the next-hop transport MTU. +/// +/// Generated by a transit router when `send_encrypted_link_message()` +/// fails with `TransportError::MtuExceeded`. The reporter includes the +/// bottleneck MTU so the source can immediately reduce its sending MTU. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|-----------|----------|------------------------------------| +/// | 0 | msg_type | 1 byte | 0x22 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr | 16 bytes | The destination we were forwarding | +/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | +/// | 34 | mtu | 2 bytes | Bottleneck MTU (u16 LE) | +/// +/// Payload: 36 bytes +#[derive(Clone, Debug)] +pub struct MtuExceeded { + /// Destination that the oversized packet was heading to. + pub dest_addr: NodeAddr, + /// Router that detected the MTU violation. + pub reporter: NodeAddr, + /// Transport MTU at the bottleneck hop. + pub mtu: u16, +} + +/// Wire size of MtuExceeded payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16) + mtu(2). +pub const MTU_EXCEEDED_SIZE: usize = 36; + +impl MtuExceeded { + /// Create a new MtuExceeded error. + pub fn new(dest_addr: NodeAddr, reporter: NodeAddr, mtu: u16) -> Self { + Self { + dest_addr, + reporter, + mtu, + } + } + + /// Encode as wire format (4-byte FSP prefix + msg_type + body). + /// + /// Error signals use phase=0x0 with U flag set. + pub fn encode(&self) -> Vec { + let body_len = MTU_EXCEEDED_SIZE; // 36 bytes + let mut buf = Vec::with_capacity(4 + body_len); + // FSP prefix: version 0, phase 0x0, U flag set + buf.push(0x00); // version 0, phase 0x0 + buf.push(0x04); // U flag + let payload_len = body_len as u16; + buf.extend_from_slice(&payload_len.to_le_bytes()); + // msg_type byte + buf.push(SessionMessageType::MtuExceeded.to_byte()); + buf.push(0x00); // reserved flags + buf.extend_from_slice(self.dest_addr.as_bytes()); + buf.extend_from_slice(self.reporter.as_bytes()); + buf.extend_from_slice(&self.mtu.to_le_bytes()); + buf + } + + /// Decode from wire format (after FSP prefix and msg_type byte consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) + mtu(2) = 35 + if payload.len() < 35 { + return Err(ProtocolError::MessageTooShort { + expected: 35, + got: payload.len(), + }); + } + // payload[0] is flags (reserved, ignored) + let mut dest_bytes = [0u8; 16]; + dest_bytes.copy_from_slice(&payload[1..17]); + let mut reporter_bytes = [0u8; 16]; + reporter_bytes.copy_from_slice(&payload[17..33]); + let mtu = u16::from_le_bytes([payload[33], payload[34]]); + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + mtu, + }) + } +} diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index cfd4000..85e53ad 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -34,10 +34,9 @@ pub use link::{ SESSION_DATAGRAM_HEADER_SIZE, SessionDatagram, SessionDatagramRef, }; pub use session::{ - COORDS_REQUIRED_SIZE, CoordsRequired, FspFlags, FspInnerFlags, MTU_EXCEEDED_SIZE, MtuExceeded, - PATH_MTU_NOTIFICATION_SIZE, PathBroken, PathMtuNotification, SESSION_RECEIVER_REPORT_SIZE, - SESSION_SENDER_REPORT_SIZE, SessionAck, SessionFlags, SessionMessageType, SessionMsg3, - SessionReceiverReport, SessionSenderReport, SessionSetup, + FspFlags, FspInnerFlags, PATH_MTU_NOTIFICATION_SIZE, PathMtuNotification, + SESSION_RECEIVER_REPORT_SIZE, SESSION_SENDER_REPORT_SIZE, SessionAck, SessionFlags, + SessionMessageType, SessionMsg3, SessionReceiverReport, SessionSenderReport, SessionSetup, }; pub(crate) use session::{coords_wire_size, decode_optional_coords, encode_coords}; pub use tree::TreeAnnounce; diff --git a/src/protocol/session.rs b/src/protocol/session.rs index 6fa9465..7e91eda 100644 --- a/src/protocol/session.rs +++ b/src/protocol/session.rs @@ -177,7 +177,7 @@ pub(crate) fn decode_optional_coords( } /// Encode a count of zero (for empty/absent coordinate fields). -fn encode_empty_coords(buf: &mut Vec) { +pub(crate) fn encode_empty_coords(buf: &mut Vec) { buf.extend_from_slice(&0u16.to_le_bytes()); } @@ -885,269 +885,6 @@ impl PathMtuNotification { } } -// ============================================================================ -// Error Messages -// ============================================================================ - -/// Link-layer error signal indicating router cache miss. -/// -/// Generated by a transit router when it cannot forward a SessionDatagram -/// due to missing cached coordinates for the destination. Carried inside -/// a new SessionDatagram addressed back to the original source -/// (src_addr=reporter, dest_addr=original_source). Plaintext — not -/// end-to-end encrypted, since the transit router has no session with -/// the source. -/// -/// ## Wire Format -/// -/// | Offset | Field | Size | Description | -/// |--------|----------|---------|------------------------------------| -/// | 0 | msg_type | 1 byte | 0x20 | -/// | 1 | flags | 1 byte | Reserved | -/// | 2 | dest_addr| 16 bytes| The node_addr we couldn't route to | -/// | 18 | reporter | 16 bytes| NodeAddr of reporting router | -/// -/// Payload: 34 bytes -#[derive(Clone, Debug)] -pub struct CoordsRequired { - /// Destination that couldn't be routed. - pub dest_addr: NodeAddr, - /// Router reporting the miss. - pub reporter: NodeAddr, -} - -/// Wire size of CoordsRequired payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16). -pub const COORDS_REQUIRED_SIZE: usize = 34; - -impl CoordsRequired { - /// Create a new CoordsRequired error. - pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self { - Self { - dest_addr, - reporter, - } - } - - /// Encode as wire format (4-byte FSP prefix + msg_type + body). - /// - /// Error signals use phase=0x0 with U flag set. - pub fn encode(&self) -> Vec { - // Body: msg_type + flags(reserved) + dest_addr + reporter - let body_len = 1 + 1 + 16 + 16; // 34 bytes - let mut buf = Vec::with_capacity(4 + body_len); - // FSP prefix: version 0, phase 0x0, U flag set - buf.push(0x00); // version 0, phase 0x0 - buf.push(0x04); // U flag - let payload_len = body_len as u16; - buf.extend_from_slice(&payload_len.to_le_bytes()); - // msg_type byte (after prefix, before body) - buf.push(SessionMessageType::CoordsRequired.to_byte()); - buf.push(0x00); // reserved flags - buf.extend_from_slice(self.dest_addr.as_bytes()); - buf.extend_from_slice(self.reporter.as_bytes()); - buf - } - - /// Decode from wire format (after FSP prefix and msg_type byte consumed). - pub fn decode(payload: &[u8]) -> Result { - // flags(1) + dest_addr(16) + reporter(16) = 33 - if payload.len() < 33 { - return Err(ProtocolError::MessageTooShort { - expected: 33, - got: payload.len(), - }); - } - // payload[0] is flags (reserved, ignored) - let mut dest_bytes = [0u8; 16]; - dest_bytes.copy_from_slice(&payload[1..17]); - let mut reporter_bytes = [0u8; 16]; - reporter_bytes.copy_from_slice(&payload[17..33]); - - Ok(Self { - dest_addr: NodeAddr::from_bytes(dest_bytes), - reporter: NodeAddr::from_bytes(reporter_bytes), - }) - } -} - -/// Error indicating routing failure (local minimum or unreachable). -/// -/// Carried inside a SessionDatagram addressed back to the original source. -/// The reporting router creates a new SessionDatagram with src_addr=reporter -/// and dest_addr=original_source, so the `original_src` field from the old -/// design is no longer needed — it's the SessionDatagram's dest_addr. -/// -/// ## Wire Format -/// -/// | Offset | Field | Size | Description | -/// |--------|-------------------|----------|-------------------------------| -/// | 0 | msg_type | 1 byte | 0x21 | -/// | 1 | flags | 1 byte | Reserved | -/// | 2 | dest_addr | 16 bytes | The unreachable node_addr | -/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | -/// | 34 | last_coords_count | 2 bytes | u16 LE | -/// | 36 | last_known_coords | 16 × n | Stale coords that failed | -#[derive(Clone, Debug)] -pub struct PathBroken { - /// Destination that couldn't be reached. - pub dest_addr: NodeAddr, - /// Node that detected the failure. - pub reporter: NodeAddr, - /// Optional: last known coordinates of destination. - pub last_known_coords: Option, -} - -impl PathBroken { - /// Create a new PathBroken error. - pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self { - Self { - dest_addr, - reporter, - last_known_coords: None, - } - } - - /// Add last known coordinates. - pub fn with_last_coords(mut self, coords: TreeCoordinate) -> Self { - self.last_known_coords = Some(coords); - self - } - - /// Encode as wire format (4-byte FSP prefix + msg_type + body). - /// - /// Error signals use phase=0x0 with U flag set. - pub fn encode(&self) -> Vec { - // Build body first to compute length - let mut body = Vec::new(); - body.push(SessionMessageType::PathBroken.to_byte()); - body.push(0x00); // reserved flags - body.extend_from_slice(self.dest_addr.as_bytes()); - body.extend_from_slice(self.reporter.as_bytes()); - if let Some(ref coords) = self.last_known_coords { - encode_coords(coords, &mut body); - } else { - encode_empty_coords(&mut body); - } - - // Prepend FSP prefix: version 0, phase 0x0, U flag set - let payload_len = body.len() as u16; - let mut buf = Vec::with_capacity(4 + body.len()); - buf.push(0x00); // version 0, phase 0x0 - buf.push(0x04); // U flag - buf.extend_from_slice(&payload_len.to_le_bytes()); - buf.extend_from_slice(&body); - buf - } - - /// Decode from wire format (after FSP prefix and msg_type byte consumed). - pub fn decode(payload: &[u8]) -> Result { - // flags(1) + dest_addr(16) + reporter(16) + coords_count(2) = 35 minimum - if payload.len() < 35 { - return Err(ProtocolError::MessageTooShort { - expected: 35, - got: payload.len(), - }); - } - // payload[0] is flags (reserved, ignored) - let mut dest_bytes = [0u8; 16]; - dest_bytes.copy_from_slice(&payload[1..17]); - let mut reporter_bytes = [0u8; 16]; - reporter_bytes.copy_from_slice(&payload[17..33]); - - let (last_known_coords, _consumed) = decode_optional_coords(&payload[33..])?; - - Ok(Self { - dest_addr: NodeAddr::from_bytes(dest_bytes), - reporter: NodeAddr::from_bytes(reporter_bytes), - last_known_coords, - }) - } -} - -/// Error indicating a forwarded packet exceeded the next-hop transport MTU. -/// -/// Generated by a transit router when `send_encrypted_link_message()` -/// fails with `TransportError::MtuExceeded`. The reporter includes the -/// bottleneck MTU so the source can immediately reduce its sending MTU. -/// -/// ## Wire Format -/// -/// | Offset | Field | Size | Description | -/// |--------|-----------|----------|------------------------------------| -/// | 0 | msg_type | 1 byte | 0x22 | -/// | 1 | flags | 1 byte | Reserved | -/// | 2 | dest_addr | 16 bytes | The destination we were forwarding | -/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | -/// | 34 | mtu | 2 bytes | Bottleneck MTU (u16 LE) | -/// -/// Payload: 36 bytes -#[derive(Clone, Debug)] -pub struct MtuExceeded { - /// Destination that the oversized packet was heading to. - pub dest_addr: NodeAddr, - /// Router that detected the MTU violation. - pub reporter: NodeAddr, - /// Transport MTU at the bottleneck hop. - pub mtu: u16, -} - -/// Wire size of MtuExceeded payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16) + mtu(2). -pub const MTU_EXCEEDED_SIZE: usize = 36; - -impl MtuExceeded { - /// Create a new MtuExceeded error. - pub fn new(dest_addr: NodeAddr, reporter: NodeAddr, mtu: u16) -> Self { - Self { - dest_addr, - reporter, - mtu, - } - } - - /// Encode as wire format (4-byte FSP prefix + msg_type + body). - /// - /// Error signals use phase=0x0 with U flag set. - pub fn encode(&self) -> Vec { - let body_len = MTU_EXCEEDED_SIZE; // 36 bytes - let mut buf = Vec::with_capacity(4 + body_len); - // FSP prefix: version 0, phase 0x0, U flag set - buf.push(0x00); // version 0, phase 0x0 - buf.push(0x04); // U flag - let payload_len = body_len as u16; - buf.extend_from_slice(&payload_len.to_le_bytes()); - // msg_type byte - buf.push(SessionMessageType::MtuExceeded.to_byte()); - buf.push(0x00); // reserved flags - buf.extend_from_slice(self.dest_addr.as_bytes()); - buf.extend_from_slice(self.reporter.as_bytes()); - buf.extend_from_slice(&self.mtu.to_le_bytes()); - buf - } - - /// Decode from wire format (after FSP prefix and msg_type byte consumed). - pub fn decode(payload: &[u8]) -> Result { - // flags(1) + dest_addr(16) + reporter(16) + mtu(2) = 35 - if payload.len() < 35 { - return Err(ProtocolError::MessageTooShort { - expected: 35, - got: payload.len(), - }); - } - // payload[0] is flags (reserved, ignored) - let mut dest_bytes = [0u8; 16]; - dest_bytes.copy_from_slice(&payload[1..17]); - let mut reporter_bytes = [0u8; 16]; - reporter_bytes.copy_from_slice(&payload[17..33]); - let mtu = u16::from_le_bytes([payload[33], payload[34]]); - - Ok(Self { - dest_addr: NodeAddr::from_bytes(dest_bytes), - reporter: NodeAddr::from_bytes(reporter_bytes), - mtu, - }) - } -} - #[cfg(test)] mod tests { use super::*; @@ -1224,28 +961,6 @@ mod tests { assert!(!setup.flags.bidirectional); } - // ===== CoordsRequired Tests ===== - - #[test] - fn test_coords_required() { - let err = CoordsRequired::new(make_node_addr(1), make_node_addr(2)); - - assert_eq!(err.dest_addr, make_node_addr(1)); - assert_eq!(err.reporter, make_node_addr(2)); - } - - // ===== PathBroken Tests ===== - - #[test] - fn test_path_broken() { - let err = PathBroken::new(make_node_addr(2), make_node_addr(3)) - .with_last_coords(make_coords(&[2, 0])); - - assert_eq!(err.dest_addr, make_node_addr(2)); - assert_eq!(err.reporter, make_node_addr(3)); - assert!(err.last_known_coords.is_some()); - } - // ===== Encode/Decode Roundtrip Tests ===== #[test] @@ -1301,55 +1016,6 @@ mod tests { assert_eq!(decoded.handshake_payload, handshake); } - #[test] - fn test_coords_required_encode_decode() { - let err = CoordsRequired::new(make_node_addr(0xAA), make_node_addr(0xBB)); - - let encoded = err.encode(); - // 4 prefix + 1 msg_type + 1 flags + 16 dest + 16 reporter = 38 - assert_eq!(encoded.len(), 4 + COORDS_REQUIRED_SIZE); - // Check FSP prefix: phase 0x0, U flag - assert_eq!(encoded[0], 0x00); - assert_eq!(encoded[1], 0x04); // U flag - // msg_type after prefix - assert_eq!(encoded[4], 0x20); - - // decode after prefix + msg_type consumed - let decoded = CoordsRequired::decode(&encoded[5..]).unwrap(); - assert_eq!(decoded.dest_addr, err.dest_addr); - assert_eq!(decoded.reporter, err.reporter); - } - - #[test] - fn test_path_broken_encode_decode_no_coords() { - let err = PathBroken::new(make_node_addr(0xCC), make_node_addr(0xDD)); - - let encoded = err.encode(); - // Check FSP prefix - assert_eq!(encoded[0], 0x00); - assert_eq!(encoded[1], 0x04); // U flag - assert_eq!(encoded[4], 0x21); // msg_type - - let decoded = PathBroken::decode(&encoded[5..]).unwrap(); - assert_eq!(decoded.dest_addr, err.dest_addr); - assert_eq!(decoded.reporter, err.reporter); - assert!(decoded.last_known_coords.is_none()); - } - - #[test] - fn test_path_broken_encode_decode_with_coords() { - let coords = make_coords(&[0xCC, 0xDD, 0xEE]); - let err = PathBroken::new(make_node_addr(0x11), make_node_addr(0x22)) - .with_last_coords(coords.clone()); - - let encoded = err.encode(); - let decoded = PathBroken::decode(&encoded[5..]).unwrap(); - - assert_eq!(decoded.dest_addr, err.dest_addr); - assert_eq!(decoded.reporter, err.reporter); - assert_eq!(decoded.last_known_coords.unwrap(), coords); - } - #[test] fn test_session_setup_decode_too_short() { assert!(SessionSetup::decode(&[]).is_err()); @@ -1360,18 +1026,6 @@ mod tests { assert!(SessionAck::decode(&[]).is_err()); } - #[test] - fn test_coords_required_decode_too_short() { - assert!(CoordsRequired::decode(&[]).is_err()); - assert!(CoordsRequired::decode(&[0x00; 10]).is_err()); - } - - #[test] - fn test_path_broken_decode_too_short() { - assert!(PathBroken::decode(&[]).is_err()); - assert!(PathBroken::decode(&[0x00; 20]).is_err()); - } - #[test] fn test_session_setup_deep_coords() { // Depth-10 coordinate (11 entries: self + 10 ancestors) @@ -1612,49 +1266,6 @@ mod tests { // ===== MtuExceeded Tests ===== - #[test] - fn test_mtu_exceeded_encode_size() { - let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400); - let encoded = err.encode(); - // 4 prefix + 36 body = 40 - assert_eq!(encoded.len(), 4 + MTU_EXCEEDED_SIZE); - } - - #[test] - fn test_mtu_exceeded_encode_decode() { - let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400); - - let encoded = err.encode(); - // Check FSP prefix: phase 0x0, U flag - assert_eq!(encoded[0], 0x00); - assert_eq!(encoded[1], 0x04); // U flag - // msg_type after prefix - assert_eq!(encoded[4], 0x22); - - // decode after prefix + msg_type consumed - let decoded = MtuExceeded::decode(&encoded[5..]).unwrap(); - assert_eq!(decoded.dest_addr, err.dest_addr); - assert_eq!(decoded.reporter, err.reporter); - assert_eq!(decoded.mtu, 1400); - } - - #[test] - fn test_mtu_exceeded_decode_too_short() { - assert!(MtuExceeded::decode(&[]).is_err()); - assert!(MtuExceeded::decode(&[0x00; 20]).is_err()); - assert!(MtuExceeded::decode(&[0x00; 34]).is_err()); // exactly 1 byte short - } - - #[test] - fn test_mtu_exceeded_boundary_mtu_values() { - for mtu in [0u16, 1280, 1500, u16::MAX] { - let err = MtuExceeded::new(make_node_addr(1), make_node_addr(2), mtu); - let encoded = err.encode(); - let decoded = MtuExceeded::decode(&encoded[5..]).unwrap(); - assert_eq!(decoded.mtu, mtu); - } - } - #[test] fn test_mtu_exceeded_message_type_value() { assert_eq!(SessionMessageType::MtuExceeded.to_byte(), 0x22);