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"); + } }