From 20cf6932cd344916a901189569ade46f0227cb7d Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 22 Feb 2026 21:07:14 +0000 Subject: [PATCH] Implement reactive MtuExceeded error signal (0x22) Add a new session-layer error signal that transit routers send back to the source when a forwarded packet exceeds the next-hop transport MTU. This complements the existing proactive path MTU discovery (min'd at each hop) by providing immediate feedback when oversized packets are dropped, closing the transient window before the proactive mechanism converges. Wire format: 36-byte payload (msg_type + flags + dest_addr + reporter + mtu) with FSP phase=0x0 and U flag set, matching the existing CoordsRequired/PathBroken pattern. Changes: - Add SessionMessageType::MtuExceeded (0x22) and MtuExceeded struct with encode/decode methods to protocol/session.rs - Add NodeError::MtuExceeded variant to propagate structured MTU info from TransportError through send_encrypted_link_message() - Catch MtuExceeded in the forwarding path and send error signal back to the datagram source via send_mtu_exceeded_error(), rate-limited by the existing routing_error_rate_limiter - Handle incoming MtuExceeded at the source by calling PathMtuState::apply_notification() for immediate MTU decrease - Add unit tests for encode/decode roundtrip, boundary MTU values, and too-short payload rejection --- src/node/handlers/forwarding.rs | 81 ++++++++++++++++-- src/node/handlers/session.rs | 48 ++++++++++- src/node/mod.rs | 18 +++- src/protocol/mod.rs | 8 +- src/protocol/session.rs | 144 ++++++++++++++++++++++++++++++++ 5 files changed, 281 insertions(+), 18 deletions(-) diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index dd84087..e182380 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -5,13 +5,13 @@ //! plaintext session-layer headers, routes to the next hop or delivers //! locally, and generates error signals on routing failure. -use crate::node::Node; +use crate::node::{Node, NodeError}; use crate::node::session_wire::{ parse_encrypted_coords, FspCommonPrefix, FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2, }; use crate::protocol::{ - CoordsRequired, PathBroken, SessionAck, SessionDatagram, SessionSetup, + CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup, }; use crate::NodeAddr; use tracing::debug; @@ -77,12 +77,19 @@ impl Node { .send_encrypted_link_message(&next_hop_addr, &encoded) .await { - debug!( - next_hop = %next_hop_addr, - dest = %datagram.dest_addr, - error = %e, - "Failed to forward SessionDatagram" - ); + match e { + NodeError::MtuExceeded { mtu, .. } => { + self.send_mtu_exceeded_error(&datagram, mtu).await; + } + _ => { + debug!( + next_hop = %next_hop_addr, + dest = %datagram.dest_addr, + error = %e, + "Failed to forward SessionDatagram" + ); + } + } } } @@ -261,4 +268,62 @@ impl Node { ); } } + + /// Generate and send an MtuExceeded error signal back to the datagram's source. + /// + /// 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; + } + + let my_addr = *self.node_addr(); + + let error_payload = MtuExceeded::new( + original.dest_addr, + my_addr, + bottleneck_mtu, + ).encode(); + + 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) { + Some(peer) => *peer.node_addr(), + None => { + debug!( + src = %original.src_addr, + dest = %original.dest_addr, + "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) + .await + { + debug!( + next_hop = %next_hop_addr, + error = %e, + "Failed to send MtuExceeded error signal" + ); + } else { + debug!( + original_dest = %original.dest_addr, + error_dest = %original.src_addr, + bottleneck_mtu, + "Sent MtuExceeded error signal" + ); + } + } } diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index c872714..5414cf9 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -18,8 +18,8 @@ use crate::noise::{HandshakeState, HANDSHAKE_MSG1_SIZE, HANDSHAKE_MSG2_SIZE}; use crate::mmp::report::ReceiverReport; use crate::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS}; use crate::protocol::{ - CoordsRequired, FspInnerFlags, PathBroken, PathMtuNotification, SessionAck, SessionDatagram, - SessionMessageType, SessionReceiverReport, SessionSenderReport, SessionSetup, + CoordsRequired, FspInnerFlags, MtuExceeded, PathBroken, PathMtuNotification, SessionAck, + SessionDatagram, SessionMessageType, SessionReceiverReport, SessionSenderReport, SessionSetup, }; use crate::NodeAddr; use secp256k1::PublicKey; @@ -73,6 +73,9 @@ impl Node { Some(SessionMessageType::PathBroken) => { self.handle_path_broken(error_body).await; } + Some(SessionMessageType::MtuExceeded) => { + self.handle_mtu_exceeded(error_body).await; + } _ => { debug!(error_type, "Unknown plaintext error signal type"); } @@ -703,6 +706,47 @@ impl Node { } } + /// Handle an MtuExceeded error signal from a transit router. + /// + /// A transit router couldn't forward our packet because it exceeded the + /// next-hop transport MTU. Apply the reported bottleneck MTU to our + /// PathMtuState for the affected session, causing an immediate decrease. + async fn handle_mtu_exceeded(&mut self, inner: &[u8]) { + let msg = match MtuExceeded::decode(inner) { + Ok(m) => m, + Err(e) => { + debug!(error = %e, "Malformed MtuExceeded"); + return; + } + }; + + let peer_name = self.peer_display_name(&msg.dest_addr); + debug!( + dest = %peer_name, + reporter = %msg.reporter, + bottleneck_mtu = msg.mtu, + "MtuExceeded: transit router reports oversized packet" + ); + + // Apply to PathMtuState: immediate decrease via apply_notification() + if let Some(entry) = self.sessions.get_mut(&msg.dest_addr) + && let Some(mmp) = entry.mmp_mut() + { + let old_mtu = mmp.path_mtu.current_mtu(); + let now = std::time::Instant::now(); + if mmp.path_mtu.apply_notification(msg.mtu, now) { + let new_mtu = mmp.path_mtu.current_mtu(); + info!( + dest = %peer_name, + old_mtu, + new_mtu, + reporter = %msg.reporter, + "Path MTU decreased via reactive MtuExceeded signal" + ); + } + } + } + // === Session Initiation (Send Path) === /// Initiate an end-to-end session with a remote node. diff --git a/src/node/mod.rs b/src/node/mod.rs index 24e65ca..75fa5a1 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -25,7 +25,7 @@ use crate::peer::{ActivePeer, PeerConnection}; use self::rate_limit::HandshakeRateLimiter; use self::routing_error_rate_limit::RoutingErrorRateLimiter; use crate::transport::{ - Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportHandle, TransportId, + Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, TransportHandle, TransportId, }; use crate::transport::udp::UdpTransport; use crate::tree::TreeState; @@ -96,6 +96,9 @@ pub enum NodeError { #[error("send failed to {node_addr}: {reason}")] SendFailed { node_addr: NodeAddr, reason: String }, + #[error("mtu exceeded forwarding to {node_addr}: packet {packet_size} > mtu {mtu}")] + MtuExceeded { node_addr: NodeAddr, packet_size: usize, mtu: u16 }, + #[error("config error: {0}")] Config(#[from] ConfigError), @@ -1156,9 +1159,16 @@ impl Node { .ok_or(NodeError::TransportNotFound(transport_id))?; let bytes_sent = transport.send(&remote_addr, &wire_packet).await - .map_err(|e| NodeError::SendFailed { - node_addr: *node_addr, - reason: format!("transport send: {}", e), + .map_err(|e| match e { + TransportError::MtuExceeded { packet_size, mtu } => NodeError::MtuExceeded { + node_addr: *node_addr, + packet_size, + mtu, + }, + other => NodeError::SendFailed { + node_addr: *node_addr, + reason: format!("transport send: {}", other), + }, })?; // Update send statistics diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 62cbf73..3600fc0 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -37,10 +37,10 @@ pub use tree::TreeAnnounce; pub use filter::FilterAnnounce; pub use discovery::{LookupRequest, LookupResponse}; pub use session::{ - CoordsRequired, FspFlags, FspInnerFlags, PathBroken, PathMtuNotification, SessionAck, - SessionFlags, SessionMessageType, SessionReceiverReport, SessionSenderReport, SessionSetup, - COORDS_REQUIRED_SIZE, PATH_MTU_NOTIFICATION_SIZE, SESSION_RECEIVER_REPORT_SIZE, - SESSION_SENDER_REPORT_SIZE, + CoordsRequired, FspFlags, FspInnerFlags, MtuExceeded, PathBroken, PathMtuNotification, + SessionAck, SessionFlags, SessionMessageType, SessionReceiverReport, SessionSenderReport, + SessionSetup, COORDS_REQUIRED_SIZE, MTU_EXCEEDED_SIZE, PATH_MTU_NOTIFICATION_SIZE, + SESSION_RECEIVER_REPORT_SIZE, SESSION_SENDER_REPORT_SIZE, }; pub(crate) use session::{coords_wire_size, decode_optional_coords, encode_coords}; diff --git a/src/protocol/session.rs b/src/protocol/session.rs index ef0ff78..71cbf2d 100644 --- a/src/protocol/session.rs +++ b/src/protocol/session.rs @@ -42,6 +42,8 @@ pub enum SessionMessageType { CoordsRequired = 0x20, /// Routing failure — local minimum or unreachable (link-layer error signal). PathBroken = 0x21, + /// MTU exceeded — forwarded packet too large for next-hop transport (link-layer error signal). + MtuExceeded = 0x22, } impl SessionMessageType { @@ -57,6 +59,7 @@ impl SessionMessageType { 0x14 => Some(SessionMessageType::CoordsWarmup), 0x20 => Some(SessionMessageType::CoordsRequired), 0x21 => Some(SessionMessageType::PathBroken), + 0x22 => Some(SessionMessageType::MtuExceeded), _ => None, } } @@ -79,6 +82,7 @@ impl fmt::Display for SessionMessageType { SessionMessageType::CoordsWarmup => "CoordsWarmup", SessionMessageType::CoordsRequired => "CoordsRequired", SessionMessageType::PathBroken => "PathBroken", + SessionMessageType::MtuExceeded => "MtuExceeded", }; write!(f, "{}", name) } @@ -952,6 +956,86 @@ impl PathBroken { } } +/// Error indicating a forwarded packet exceeded the next-hop transport MTU. +/// +/// Generated by a transit router when `send_encrypted_link_message()` +/// fails with `TransportError::MtuExceeded`. The reporter includes the +/// bottleneck MTU so the source can immediately reduce its sending MTU. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|-----------|----------|------------------------------------| +/// | 0 | msg_type | 1 byte | 0x22 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr | 16 bytes | The destination we were forwarding | +/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | +/// | 34 | mtu | 2 bytes | Bottleneck MTU (u16 LE) | +/// +/// Payload: 36 bytes +#[derive(Clone, Debug)] +pub struct MtuExceeded { + /// Destination that the oversized packet was heading to. + pub dest_addr: NodeAddr, + /// Router that detected the MTU violation. + pub reporter: NodeAddr, + /// Transport MTU at the bottleneck hop. + pub mtu: u16, +} + +/// Wire size of MtuExceeded payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16) + mtu(2). +pub const MTU_EXCEEDED_SIZE: usize = 36; + +impl MtuExceeded { + /// Create a new MtuExceeded error. + pub fn new(dest_addr: NodeAddr, reporter: NodeAddr, mtu: u16) -> Self { + Self { dest_addr, reporter, mtu } + } + + /// Encode as wire format (4-byte FSP prefix + msg_type + body). + /// + /// Error signals use phase=0x0 with U flag set. + pub fn encode(&self) -> Vec { + let body_len = MTU_EXCEEDED_SIZE; // 36 bytes + let mut buf = Vec::with_capacity(4 + body_len); + // FSP prefix: version 0, phase 0x0, U flag set + buf.push(0x00); // version 0, phase 0x0 + buf.push(0x04); // U flag + let payload_len = body_len as u16; + buf.extend_from_slice(&payload_len.to_le_bytes()); + // msg_type byte + buf.push(SessionMessageType::MtuExceeded.to_byte()); + buf.push(0x00); // reserved flags + buf.extend_from_slice(self.dest_addr.as_bytes()); + buf.extend_from_slice(self.reporter.as_bytes()); + buf.extend_from_slice(&self.mtu.to_le_bytes()); + buf + } + + /// Decode from wire format (after FSP prefix and msg_type byte consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) + mtu(2) = 35 + if payload.len() < 35 { + return Err(ProtocolError::MessageTooShort { + expected: 35, + got: payload.len(), + }); + } + // payload[0] is flags (reserved, ignored) + let mut dest_bytes = [0u8; 16]; + dest_bytes.copy_from_slice(&payload[1..17]); + let mut reporter_bytes = [0u8; 16]; + reporter_bytes.copy_from_slice(&payload[17..33]); + let mtu = u16::from_le_bytes([payload[33], payload[34]]); + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + mtu, + }) + } +} + #[cfg(test)] mod tests { use super::*; @@ -979,6 +1063,7 @@ mod tests { SessionMessageType::PathMtuNotification, SessionMessageType::CoordsRequired, SessionMessageType::PathBroken, + SessionMessageType::MtuExceeded, ]; for ty in types { @@ -1405,4 +1490,63 @@ mod tests { assert_eq!(decoded.path_mtu, mtu); } } + + // ===== 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); + assert_eq!( + SessionMessageType::from_byte(0x22), + Some(SessionMessageType::MtuExceeded) + ); + } + + #[test] + fn test_mtu_exceeded_display() { + assert_eq!(format!("{}", SessionMessageType::MtuExceeded), "MtuExceeded"); + } }