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

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

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.

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

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};
 

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