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

proto/routing: sans-IO transit + hop-selection state machine

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-07 04:12:05 +00:00
parent e03b206f62
commit 9ea57b483a
19 changed files with 1820 additions and 888 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib.rs
+7 -3
View File
@@ -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};
src/node/handlers/forwarding.rs
+206 -147
View File
@@ -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<NextHop> {
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"
);
src/node/handlers/session.rs
+5 -5
View File
@@ -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()
src/node/metrics.rs
+4 -37
View File
@@ -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).
src/node/mod.rs
+99 -141
View File
@@ -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<NodeAddr, ActivePeer>,
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<TreeCoordinate> {
self.coord_cache.get(dest, now_ms).cloned()
}
fn peer_addrs(&self) -> Vec<NodeAddr> {
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<TreeCoordinate> {
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,
src/node/routing_error_rate_limit.rs
-161
View File
@@ -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<NodeAddr, Instant>,
/// 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)));
}
}
src/proto/mod.rs
+1
View File
@@ -4,3 +4,4 @@
//! I/O adapters remain in `node::handlers`.
pub(crate) mod discovery;
pub(crate) mod routing;
src/proto/routing/core.rs
+423
View File
@@ -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<TreeCoordinate>;
/// 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<NodeAddr>;
/// 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<TreeCoordinate>;
}
/// 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<u8>,
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<u8> },
}
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<NextHop>,
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<RouteAction> {
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<RouteAction> {
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<TreeCoordinate>,
}
/// 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<Candidate> {
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<NodeAddr> {
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
}
src/proto/routing/limits.rs
+90
View File
@@ -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<NodeAddr, u64>,
/// 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()
}
}
src/proto/routing/mod.rs
+33
View File
@@ -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};
src/proto/routing/state.rs
+23
View File
@@ -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(),
}
}
}
src/proto/routing/tests/core.rs
+358
View File
@@ -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()
);
}
src/proto/routing/tests/limits.rs
+70
View File
@@ -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));
}
src/proto/routing/tests/mod.rs
+6
View File
@@ -0,0 +1,6 @@
//! Routing subsystem unit tests. Shared helpers live in `util`.
mod core;
mod limits;
mod util;
mod wire;
src/proto/routing/tests/util.rs
+86
View File
@@ -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<NodeAddr>,
pub(super) can_send: bool,
pub(super) link_cost: f64,
pub(super) coords: Option<TreeCoordinate>,
}
/// 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<MockPeer>,
}
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<TreeCoordinate> {
self.coords
.iter()
.find(|(addr, _)| addr == dest)
.map(|(_, coords)| coords.clone())
}
fn peer_addrs(&self) -> Vec<NodeAddr> {
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<TreeCoordinate> {
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()
}
src/proto/routing/tests/wire.rs
+130
View File
@@ -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);
}
}
src/proto/routing/wire.rs
+275
View File
@@ -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<u8> {
// 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<Self, ProtocolError> {
// 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<TreeCoordinate>,
}
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<u8> {
// 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<Self, ProtocolError> {
// 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<u8> {
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<Self, ProtocolError> {
// 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,
})
}
}
src/protocol/mod.rs
+3 -4
View File
@@ -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;
src/protocol/session.rs
+1 -390
View File
@@ -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<u8>) {
pub(crate) fn encode_empty_coords(buf: &mut Vec<u8>) {
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<u8> {
// 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<Self, ProtocolError> {
// 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<TreeCoordinate>,
}
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<u8> {
// 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<Self, ProtocolError> {
// 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<u8> {
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<Self, ProtocolError> {
// 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);