From f825fa242f0c763aa2453497ec5d9a8f45fde02a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 19 Feb 2026 15:25:30 +0000 Subject: [PATCH] Hybrid coordinate warmup: CoordsWarmup message and proactive fallback Implement hybrid coordinate cache warming strategy: piggyback coords via CP flag when they fit within transport MTU, send standalone CoordsWarmup (0x14) message when they don't. On CoordsRequired or PathBroken receipt, send CoordsWarmup immediately with source-side rate limiting (default 2s per destination, configurable). - Add CoordsWarmup = 0x14 session message type (empty body, CP flag) - Add send_coords_warmup() following send_session_msg() pattern - Restructure send_session_data() to send standalone warmup before data packet when piggybacked coords exceed MTU - Add immediate CoordsWarmup response in handle_coords_required() and handle_path_broken() with per-destination rate limiting - Add coords_response_interval_ms config (node.session) - Add RoutingErrorRateLimiter::with_interval() constructor - Zero transit-path changes: existing try_warm_coord_cache() handles CoordsWarmup identically to CP-flagged data packets - Update design docs (session layer, wire formats, mesh operation, configuration) --- docs/design/fips-configuration.md | 4 +- docs/design/fips-mesh-operation.md | 44 +++++-- docs/design/fips-session-layer.md | 93 ++++++++++---- docs/design/fips-wire-formats.md | 32 ++++- src/config/node.rs | 7 + src/node/handlers/session.rs | 183 ++++++++++++++++++++++----- src/node/mod.rs | 10 ++ src/node/routing_error_rate_limit.rs | 24 ++++ src/protocol/session.rs | 4 + 9 files changed, 324 insertions(+), 77 deletions(-) diff --git a/docs/design/fips-configuration.md b/docs/design/fips-configuration.md index 130df5e..0025f8b 100644 --- a/docs/design/fips-configuration.md +++ b/docs/design/fips-configuration.md @@ -148,7 +148,8 @@ Controls end-to-end session behavior and packet queuing. | `node.session.pending_packets_per_dest` | usize | `16` | Queue depth per destination during session establishment | | `node.session.pending_max_destinations` | usize | `256` | Max destinations with pending packets | | `node.session.idle_timeout_secs` | u64 | `90` | Idle session timeout; established sessions with no application data for this duration are removed. MMP reports (SenderReport, ReceiverReport, PathMtuNotification) do not count as activity | -| `node.session.coords_warmup_packets` | u8 | `5` | Number of initial data packets per session that include the CP flag for transit cache warmup; also the reset count on CoordsRequired receipt | +| `node.session.coords_warmup_packets` | u8 | `5` | Number of initial data packets per session that include the CP flag for transit cache warmup; also the reset count on CoordsRequired/PathBroken receipt | +| `node.session.coords_response_interval_ms` | u64 | `2000` | Minimum interval (ms) between standalone CoordsWarmup responses to CoordsRequired/PathBroken signals per destination | The anti-replay window size (2048 packets) is a compile-time constant and not configurable. @@ -327,6 +328,7 @@ node: pending_max_destinations: 256 idle_timeout_secs: 90 coords_warmup_packets: 5 + coords_response_interval_ms: 2000 mmp: mode: full # full | lightweight | minimal log_interval_secs: 30 diff --git a/docs/design/fips-mesh-operation.md b/docs/design/fips-mesh-operation.md index 345737c..345eebd 100644 --- a/docs/design/fips-mesh-operation.md +++ b/docs/design/fips-mesh-operation.md @@ -338,17 +338,19 @@ After the handshake completes, the entire forward and reverse paths have cached coordinates for both endpoints. Subsequent data packets use minimal headers (no coordinates) and route efficiently through the warmed caches. -## CP (Coords Present) Warmup +## Hybrid Coordinate Warmup (CP + CoordsWarmup) -The CP flag in the FSP common prefix provides a secondary cache-warming -mechanism that complements SessionSetup. See -[fips-session-layer.md](fips-session-layer.md) for the warmup-then-reactive -strategy. +The CP flag in the FSP common prefix and the standalone CoordsWarmup message +(0x14) together provide a hybrid cache-warming mechanism that complements +SessionSetup. See [fips-session-layer.md](fips-session-layer.md) for the +full warmup strategy. Transit nodes parse the CP flag from the FSP header and extract source and destination coordinates from the cleartext section between the header and ciphertext — no decryption needed. This is the same caching operation -performed for SessionSetup coordinates. +performed for SessionSetup coordinates. CoordsWarmup messages use the same +CP-flag format and are handled identically by transit nodes via the existing +`try_warm_coord_cache()` path. ## Error Recovery @@ -361,15 +363,22 @@ corrective action. coordinates for the destination. It cannot make a forwarding decision. **Transit node action**: + 1. Create a new SessionDatagram addressed back to the original source, carrying a CoordsRequired payload identifying the unreachable destination 2. Route the error via `find_next_hop(src_addr)` 3. If the source is also unreachable, drop silently (no cascading errors) **Source recovery**: -1. Initiate discovery (LookupRequest flood) for the destination -2. Reset CP warmup counter — subsequent data packets include coordinates -3. When discovery completes, warmup counter resets again (covers timing gap) + +1. Immediately send a standalone CoordsWarmup (0x14) message to re-warm + transit caches along the path (rate-limited: at most one per destination + per configurable interval, default 2s) +2. Reset CP warmup counter — subsequent data packets piggyback coordinates + when possible, or trigger additional CoordsWarmup messages when + piggybacking would exceed the transport MTU +3. Initiate discovery (LookupRequest flood) for the destination +4. When discovery completes, warmup counter resets again (covers timing gap) The crypto session remains active throughout — only routing state is refreshed. @@ -384,9 +393,12 @@ tree distance metric). The cached coordinates may be stale. source. **Source recovery**: -1. Remove stale coordinates from cache -2. Initiate discovery for the destination -3. Reset CP warmup counter + +1. Immediately send a standalone CoordsWarmup (0x14) message (rate-limited, + same per-destination interval as CoordsRequired response) +2. Remove stale coordinates from cache +3. Initiate discovery for the destination +4. Reset CP warmup counter ### Error Signal Rate Limiting @@ -394,6 +406,12 @@ Both error types are rate-limited at transit nodes: maximum one error per destination per 100ms. This prevents storms during topology changes when many packets to the same destination hit the same routing failure simultaneously. +At the source side, CoordsWarmup responses to CoordsRequired/PathBroken are +independently rate-limited: at most one standalone CoordsWarmup per destination +per `coords_response_interval_ms` (default 2000ms, configurable). This +prevents amplification where a burst of error signals would generate a +corresponding burst of warmup messages. + Error signals (CoordsRequired, PathBroken) are handled asynchronously outside the packet receive path, allowing the RX loop to continue processing without blocking on discovery or session repair. @@ -554,7 +572,7 @@ recovery). | Flush coord cache on parent change | **Implemented** | | LookupRequest/LookupResponse discovery | **Implemented** | | SessionSetup self-bootstrapping | **Implemented** | -| CP warmup-then-reactive | **Implemented** | +| Hybrid coordinate warmup (CP + CoordsWarmup) | **Implemented** | | CoordsRequired recovery | **Implemented** | | PathBroken recovery | **Implemented** | | Error signal rate limiting | **Implemented** | diff --git a/docs/design/fips-session-layer.md b/docs/design/fips-session-layer.md index ce3a7de..cce3ab6 100644 --- a/docs/design/fips-session-layer.md +++ b/docs/design/fips-session-layer.md @@ -83,10 +83,12 @@ may traverse multiple hops, each with independent link encryption. FLP signals routing failures asynchronously: - **CoordsRequired**: A transit node lacks the destination's tree coordinates. - FSP responds by re-initiating discovery and resetting the coordinate warmup - strategy. -- **PathBroken**: Greedy routing reached a dead end. FSP responds by - re-discovering the destination's current coordinates and resetting warmup. + FSP responds by sending a standalone CoordsWarmup (0x14) message + (rate-limited), re-initiating discovery, and resetting the coordinate warmup + counter. +- **PathBroken**: Greedy routing reached a dead end. FSP responds by sending + a standalone CoordsWarmup (rate-limited), re-discovering the destination's + current coordinates, and resetting the warmup counter. Both signals are generated by transit nodes (not the destination) and travel back to the source inside a new SessionDatagram. They are plaintext (not @@ -190,6 +192,7 @@ independent keys and sessions. The full Noise descriptor is `Noise_IK_secp256k1_ChaChaPoly_SHA256`. The IK pattern: + - **msg1** (`→ e, es, s, ss`): Initiator sends ephemeral key, encrypts static key to responder. Four DH operations establish session keys. - **msg2** (`← e, ee, se`): Responder sends ephemeral key. Both parties now @@ -262,7 +265,7 @@ regardless of what packets were lost or reordered. The same `ReplayWindow` and `decrypt_with_replay_check()` implementation is used at both the link and session layers. -## CP (Coords Present) Warmup Strategy +## Hybrid Coordinate Warmup Strategy Session establishment (SessionSetup/SessionAck) warms transit node coordinate caches along the path. But coordinate caches have a finite TTL (default 300s), @@ -270,14 +273,43 @@ and entries may be evicted under memory pressure. When a transit node's cache entry expires, it cannot forward data packets (which carry only addresses, not coordinates) and sends a CoordsRequired error. -FSP uses a warmup-then-reactive strategy to keep transit caches populated: +FSP uses a hybrid warmup strategy combining proactive piggybacking with +reactive standalone messages to keep transit caches populated: -### Warmup Phase +### Proactive Warmup Phase After session establishment, the first N data packets (configurable, default 5) -include both source and destination coordinates via the CP flag in the FSP -common prefix. The coordinates appear in cleartext between the 12-byte header -and the ciphertext, allowing transit nodes to cache them without decryption. +per session attempt to piggyback source and destination coordinates via the CP +flag in the FSP common prefix. The coordinates appear in cleartext between the +12-byte header and the ciphertext, allowing transit nodes to cache them without +decryption. + +If piggybacking coordinates would cause the total packet to exceed the +transport MTU, the source sends a standalone **CoordsWarmup** message (0x14) +first, followed by the data packet without the CP flag. This ensures transit +caches are warmed even when data packets are near the MTU limit. + +### CoordsWarmup Message (0x14) + +CoordsWarmup is an encrypted FSP message (phase 0x0) with the CP flag set and +inner msg_type 0x14. It carries cleartext source and destination coordinates +between the FSP header and the AEAD ciphertext, using the same format as +CP-flagged data packets. The inner body is empty — the message exists solely +to deliver coordinates to transit nodes. + +Transit nodes extract coordinates via the existing `try_warm_coord_cache()` +code path with zero changes to the transit forwarding logic. CoordsWarmup is +indistinguishable from any other CP-flagged message at the transit layer. + +Wire format: + +```text +FSP header (12 bytes, AAD): ver=0, phase=0, flags=CP, counter, payload_len +Cleartext coords: src_coords + dst_coords (same encoding as CP flag) +AEAD ciphertext: inner_header(6) + Poly1305 tag(16) = 22 bytes + +Total FSP payload: 12 + coords + 22 +``` ### Steady State @@ -285,31 +317,32 @@ After the warmup count is reached, FSP clears the CP flag and sends minimal data packets (12-byte header + ciphertext). Transit nodes serve from their coordinate caches. -### Reactive Recovery +### Reactive Re-Warm -When FSP receives a CoordsRequired signal: +When FSP receives a CoordsRequired or PathBroken signal: -1. The warmup counter resets — subsequent data packets include coordinates again -2. A new LookupRequest may be initiated to rediscover the destination's - current coordinates -3. When the LookupResponse arrives for an established session, the warmup +1. A standalone CoordsWarmup message is sent immediately to re-warm transit + caches, rate-limited at one per destination per configurable interval + (default 2000ms, `node.session.coords_response_interval_ms`) +2. The warmup counter resets — subsequent data packets piggyback coordinates + again when possible (or send additional CoordsWarmup messages when the + data packet would exceed the MTU) +3. A new LookupRequest may be initiated to rediscover the destination's + current coordinates (always for PathBroken; optionally for CoordsRequired) +4. When the LookupResponse arrives for an established session, the warmup counter resets again (handling the timing gap where warmup packets might fire before transit caches are repopulated by discovery) -When FSP receives a PathBroken signal: - -1. A LookupRequest is initiated to discover the destination's current - coordinates (which may have changed due to topology change) -2. The warmup counter resets - -Both signals are rate-limited at transit nodes (100ms per destination) to -prevent storms during topology changes. +The source-side rate limiting prevents amplification: at most one standalone +CoordsWarmup response per destination per `coords_response_interval_ms` +(default 2s). This is independent of the transit-node rate limiting on error +signal generation (100ms per destination). ### Warmup State Machine ```text ┌──────────────┐ - │ WARMUP │ ◄── Send first N packets with coords + │ WARMUP │ ◄── Send first N packets with coords (CP or CoordsWarmup) └──────┬───────┘ │ N packets sent without CoordsRequired ▼ @@ -318,8 +351,13 @@ prevent storms during topology changes. └──────┬───────┘ │ CoordsRequired or PathBroken received ▼ + ┌──────────────────────────┐ + │ SEND CoordsWarmup (0x14) │ ◄── Immediate standalone warmup (rate-limited) + └──────────┬───────────────┘ + │ + ▼ ┌──────────────┐ - │ WARMUP │ ◄── Counter reset, send coords again + │ WARMUP │ ◄── Counter reset, piggyback coords again └──────────────┘ ``` @@ -336,6 +374,7 @@ configurable size (default 10K entries). There is no TTL — entries are evicted only when the cache is full and space is needed for a new entry. Cache population mechanisms: + - **DNS lookup**: The primary path. Resolving `npub1xxx...xxx.fips` derives the IPv6 address and populates the identity cache. - **Inbound traffic**: Authenticated sessions from other nodes populate the @@ -454,7 +493,7 @@ MMP session metrics session=npub1tdwa...84le rtt=4.3ms loss=0.6% jitter=0.2ms go | Session establishment (Noise IK) | **Implemented** | | End-to-end encryption (ChaCha20-Poly1305) | **Implemented** | | Explicit counter replay protection | **Implemented** | -| CP warmup-then-reactive | **Implemented** | +| Hybrid coordinate warmup (CP + CoordsWarmup) | **Implemented** | | FSP wire format (prefix, AAD, inner header) | **Implemented** | | Session-layer MMP | **Implemented** | | Identity cache (LRU-only) | **Implemented** | diff --git a/docs/design/fips-wire-formats.md b/docs/design/fips-wire-formats.md index 67f4ae7..27de02e 100644 --- a/docs/design/fips-wire-formats.md +++ b/docs/design/fips-wire-formats.md @@ -526,10 +526,11 @@ body. | 0x11 | SenderReport | MMP sender-side metrics report | | 0x12 | ReceiverReport | MMP receiver-side metrics report | | 0x13 | PathMtuNotification | End-to-end path MTU echo | +| 0x14 | CoordsWarmup | Standalone coordinate cache warming | | 0x20 | CoordsRequired | Error: transit node lacks destination coordinates | | 0x21 | PathBroken | Error: greedy routing reached dead end | -Message types 0x10-0x13 are carried inside the AEAD ciphertext (dispatched +Message types 0x10-0x14 are carried inside the AEAD ciphertext (dispatched by the `msg_type` field in the encrypted inner header). Types 0x20-0x21 are plaintext error signals (U flag set, no encryption). @@ -586,6 +587,34 @@ Sent by the destination to report the observed forward-path MTU. **Total body: 2 bytes** (plus FSP common prefix + encrypted header + AEAD tag). +### CoordsWarmup (0x14) + +Standalone coordinate cache warming message. Sent when piggybacking coordinates +via the CP flag on a data packet would exceed the transport MTU, or as an +immediate response to CoordsRequired/PathBroken signals (rate-limited). + +CoordsWarmup is an encrypted FSP message with the CP flag set and an empty +body. Transit nodes extract coordinates via the existing CP-flag parsing +path — no transit-side changes required. + +**Wire format**: + +```text +FSP header (12 bytes, AAD): ver=0, phase=0, flags=CP, counter, payload_len +Cleartext coords: src_coords + dst_coords (same encoding as CP flag) +AEAD ciphertext: inner_header(6) + Poly1305 tag(16) = 22 bytes + +Total FSP payload: 12 + coords + 22 +``` + +The cleartext coords section uses the same variable-length encoding as any +CP-flagged message: `src_coords_count(2) + src_coords(16×n) + +dest_coords_count(2) + dest_coords(16×m)`. + +**Typical size** (depth-3 tree): 12 + (2+64+2+64) + 22 = **166 bytes** FSP +payload. With SessionDatagram + link overhead: 166 + 36 + 37 = **239 bytes** +on the wire. + ### CoordsRequired (0x20) Plaintext error signal — transit node lacks coordinates for destination. @@ -707,6 +736,7 @@ endpoint session keys). | SenderReport | 12 + 6 + 46 + 16 bytes | MMP metrics | | ReceiverReport | 12 + 6 + 66 + 16 bytes | MMP metrics | | PathMtuNotification | 12 + 6 + 2 + 16 bytes | MTU signal | +| CoordsWarmup | 12 + coords + 6 + 16 bytes | Standalone warmup (empty body) | | CoordsRequired | 38 bytes | Fixed (prefix + msg_type + body) | | PathBroken | 35 + 16n bytes | Includes stale coords | diff --git a/src/config/node.rs b/src/config/node.rs index a3d36a8..fc8fbb4 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -233,6 +233,11 @@ pub struct SessionConfig { /// Also used as the reset count on CoordsRequired receipt. #[serde(default = "SessionConfig::default_coords_warmup_packets")] pub coords_warmup_packets: u8, + /// Minimum interval (ms) between standalone CoordsWarmup responses to + /// CoordsRequired/PathBroken signals, per destination + /// (`node.session.coords_response_interval_ms`). + #[serde(default = "SessionConfig::default_coords_response_interval_ms")] + pub coords_response_interval_ms: u64, } impl Default for SessionConfig { @@ -243,6 +248,7 @@ impl Default for SessionConfig { pending_max_destinations: 256, idle_timeout_secs: 90, coords_warmup_packets: 5, + coords_response_interval_ms: 2000, } } } @@ -253,6 +259,7 @@ impl SessionConfig { fn default_pending_max_destinations() -> usize { 256 } fn default_idle_timeout_secs() -> u64 { 90 } fn default_coords_warmup_packets() -> u8 { 5 } + fn default_coords_response_interval_ms() -> u64 { 2000 } } /// Session-layer Metrics Measurement Protocol (`node.session_mmp.*`). diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 8450572..524ee13 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -251,6 +251,11 @@ impl Node { Some(SessionMessageType::PathMtuNotification) => { self.handle_session_path_mtu_notification(src_addr, rest); } + Some(SessionMessageType::CoordsWarmup) => { + // Standalone coordinate warming — coords already extracted + // from CP flag by transit nodes. No action needed at endpoint. + trace!(src = %self.peer_display_name(src_addr), "CoordsWarmup received"); + } _ => { debug!(src = %self.peer_display_name(src_addr), msg_type, "Unknown session message type, dropping"); } @@ -586,8 +591,9 @@ impl Node { /// Handle a CoordsRequired error signal from a transit router. /// /// The router couldn't route our packet because it lacks cached - /// coordinates for the destination. Trigger discovery to populate - /// the route cache so subsequent routing attempts succeed. + /// coordinates for the destination. Send a standalone CoordsWarmup + /// immediately (rate-limited), trigger discovery, and reset the + /// warmup counter for subsequent data packets. async fn handle_coords_required(&mut self, inner: &[u8]) { let msg = match CoordsRequired::decode(inner) { Ok(m) => m, @@ -600,12 +606,26 @@ impl Node { debug!( dest = %msg.dest_addr, reporter = %msg.reporter, - "CoordsRequired: transit router needs coordinates, initiating discovery" + "CoordsRequired: transit router needs coordinates" ); + // Send standalone CoordsWarmup immediately (rate-limited) + if self.coords_response_rate_limiter.should_send(&msg.dest_addr) { + if let Some(entry) = self.sessions.get(&msg.dest_addr) + && entry.is_established() + && let Err(e) = self.send_coords_warmup(&msg.dest_addr).await + { + debug!(dest = %msg.dest_addr, error = %e, + "Failed to send CoordsWarmup in response to CoordsRequired"); + } + } else { + trace!(dest = %msg.dest_addr, + "CoordsRequired response rate-limited, skipping standalone CoordsWarmup"); + } + self.maybe_initiate_lookup(&msg.dest_addr).await; - // Reset coords warmup counter so the next N packets include + // Reset coords warmup counter so the next N packets also include // COORDS_PRESENT, re-warming transit caches along the path. if let Some(entry) = self.sessions.get_mut(&msg.dest_addr) { let n = self.config.node.session.coords_warmup_packets; @@ -621,8 +641,8 @@ impl Node { /// Handle a PathBroken error signal from a transit router. /// /// The router has coordinates but still can't route to the destination. - /// Invalidate cached coordinates, trigger re-discovery, and reset the - /// COORDS_PRESENT warmup counter so the new path gets warmed. + /// Send a standalone CoordsWarmup immediately (rate-limited), invalidate + /// cached coordinates, trigger re-discovery, and reset the warmup counter. async fn handle_path_broken(&mut self, inner: &[u8]) { let msg = match PathBroken::decode(inner) { Ok(m) => m, @@ -638,6 +658,20 @@ impl Node { "PathBroken: transit router reports routing failure" ); + // Send standalone CoordsWarmup immediately (rate-limited) + if self.coords_response_rate_limiter.should_send(&msg.dest_addr) { + if let Some(entry) = self.sessions.get(&msg.dest_addr) + && entry.is_established() + && let Err(e) = self.send_coords_warmup(&msg.dest_addr).await + { + debug!(dest = %msg.dest_addr, error = %e, + "Failed to send CoordsWarmup in response to PathBroken"); + } + } else { + trace!(dest = %msg.dest_addr, + "PathBroken response rate-limited, skipping standalone CoordsWarmup"); + } + // Invalidate stale cached coordinates self.coord_cache.remove(&msg.dest_addr); @@ -721,37 +755,29 @@ impl Node { plaintext: &[u8], ) -> Result<(), NodeError> { let now_ms = Self::now_ms(); - let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { + + // First borrow: read session metadata (NLL releases before coord decision) + let entry = self.sessions.get(dest_addr).ok_or_else(|| NodeError::SendFailed { node_addr: *dest_addr, reason: "no session".into(), })?; - - // Check warmup counter and get session timestamp let wants_coords = entry.coords_warmup_remaining() > 0; let timestamp = entry.session_timestamp(now_ms); let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit()); + if !entry.is_established() { + return Err(NodeError::SendFailed { + node_addr: *dest_addr, + reason: "session not established".into(), + }); + } - let session = match entry.state_mut() { - EndToEndState::Established(s) => s, - _ => { - return Err(NodeError::SendFailed { - node_addr: *dest_addr, - reason: "session not established".into(), - }); - } - }; - - // Get counter before encrypting (encrypt will increment it) - let counter = session.current_send_counter(); - - // FSP inner header: [timestamp:4 LE][msg_type:1][inner_flags:1] + plaintext + // Build inner plaintext (doesn't depend on counter) let msg_type = SessionMessageType::DataPacket.to_byte(); // 0x10 let inner_flags = FspInnerFlags { spin_bit }.to_byte(); let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, plaintext); // Determine whether coords fit within transport MTU. - // Total wire size: FIPS_OVERHEAD (106) + coords_size + plaintext.len() - // The transport MTU is the UDP payload budget available for the FLP frame. + // If not, send standalone CoordsWarmup before the data packet. let (include_coords, my_coords, dest_coords) = if wants_coords { let src = self.tree_state.my_coords().clone(); let dst = self.get_dest_coords(dest_addr); @@ -760,29 +786,28 @@ impl Node { if total_wire <= self.transport_mtu() as usize { (true, Some(src), Some(dst)) } else { - trace!("CP flag skipped: {} bytes with coords exceeds transport MTU {}", - total_wire, self.transport_mtu()); + // Coords don't fit piggybacked — send standalone CoordsWarmup first + if let Err(e) = self.send_coords_warmup(dest_addr).await { + debug!(dest = %self.peer_display_name(dest_addr), error = %e, + "Failed to send standalone CoordsWarmup before data packet"); + } (false, None, None) } } else { (false, None, None) }; - // Decrement warmup counter only if we actually include coords - if include_coords + // Decrement warmup counter if we sent coords (piggybacked or standalone) + if wants_coords && let Some(entry) = self.sessions.get_mut(dest_addr) { entry.set_coords_warmup_remaining(entry.coords_warmup_remaining() - 1); } - // Build FSP flags (CP flag only if coords will be included) + // Build FSP flags (CP flag only if coords will be piggybacked) let flags = if include_coords { FSP_FLAG_CP } else { 0 }; - // Build 12-byte FSP header (used as AAD for AEAD) - let payload_len = inner_plaintext.len() as u16; - let header = build_fsp_header(counter, flags, payload_len); - - // Re-borrow session for encryption + // Borrow session for counter + encryption (after potential standalone send) let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { node_addr: *dest_addr, reason: "no session".into(), @@ -796,6 +821,11 @@ impl Node { }); } }; + let counter = session.current_send_counter(); + + // Build 12-byte FSP header (used as AAD for AEAD) + let payload_len = inner_plaintext.len() as u16; + let header = build_fsp_header(counter, flags, payload_len); // Encrypt with AAD binding to the FSP header let ciphertext = session.encrypt_with_aad(&inner_plaintext, &header).map_err(|e| { @@ -910,6 +940,89 @@ impl Node { Ok(()) } + /// Send a standalone CoordsWarmup message to warm transit node caches. + /// + /// Constructs an encrypted FSP message with CP flag set and + /// msg_type=CoordsWarmup. Transit nodes extract the cleartext + /// coordinates via `try_warm_coord_cache()` (same as CP-flagged data + /// packets). The encrypted inner payload is the 6-byte inner header + /// with no application data. + async fn send_coords_warmup( + &mut self, + dest_addr: &NodeAddr, + ) -> Result<(), NodeError> { + let now_ms = Self::now_ms(); + + let my_coords = self.tree_state.my_coords().clone(); + let dest_coords = self.get_dest_coords(dest_addr); + + // Read session metadata + let entry = self.sessions.get(dest_addr).ok_or_else(|| NodeError::SendFailed { + node_addr: *dest_addr, + reason: "no session".into(), + })?; + let timestamp = entry.session_timestamp(now_ms); + let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit()); + + // Get mutable access for encryption + let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { + node_addr: *dest_addr, + reason: "no session".into(), + })?; + let session = match entry.state_mut() { + EndToEndState::Established(s) => s, + _ => { + return Err(NodeError::SendFailed { + node_addr: *dest_addr, + reason: "session not established".into(), + }); + } + }; + + let counter = session.current_send_counter(); + + // FSP inner header only, no body payload + let msg_type = SessionMessageType::CoordsWarmup.to_byte(); + let inner_flags = FspInnerFlags { spin_bit }.to_byte(); + let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, &[]); + + // Build FSP header with CP flag + let payload_len = inner_plaintext.len() as u16; + let header = build_fsp_header(counter, FSP_FLAG_CP, payload_len); + + // Encrypt with AAD + let ciphertext = session.encrypt_with_aad(&inner_plaintext, &header).map_err(|e| { + NodeError::SendFailed { + node_addr: *dest_addr, + reason: format!("session encrypt failed: {}", e), + } + })?; + + // Assemble: header(12) + coords + ciphertext + let coords_size = coords_wire_size(&my_coords) + coords_wire_size(&dest_coords); + let mut fsp_payload = Vec::with_capacity(FSP_HEADER_SIZE + coords_size + ciphertext.len()); + fsp_payload.extend_from_slice(&header); + encode_coords(&my_coords, &mut fsp_payload); + encode_coords(&dest_coords, &mut fsp_payload); + fsp_payload.extend_from_slice(&ciphertext); + + let my_addr = *self.node_addr(); + let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload) + .with_ttl(self.config.node.session.default_ttl); + + self.send_session_datagram(&mut datagram).await?; + + // Record in MMP (infrastructure traffic — no idle timer touch) + if let Some(entry) = self.sessions.get_mut(dest_addr) + && let Some(mmp) = entry.mmp_mut() + { + mmp.sender.record_sent(counter, timestamp, ciphertext.len()); + } + + debug!(dest = %self.peer_display_name(dest_addr), "Sent standalone CoordsWarmup"); + Ok(()) + } + /// Route and send a SessionDatagram through the mesh. /// /// Finds the next hop for the destination, seeds path_mtu from the diff --git a/src/node/mod.rs b/src/node/mod.rs index 5dd90f9..a9ac41a 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -319,6 +319,8 @@ pub struct Node { icmp_rate_limiter: IcmpRateLimiter, /// Rate limiter for routing error signals (CoordsRequired / PathBroken). routing_error_rate_limiter: RoutingErrorRateLimiter, + /// Rate limiter for source-side CoordsRequired/PathBroken responses. + coords_response_rate_limiter: RoutingErrorRateLimiter, // === Connection Retry === /// Retry state for peers whose outbound connections have failed. @@ -373,6 +375,7 @@ impl Node { let max_connections = config.node.limits.max_connections; let max_peers = config.node.limits.max_peers; let max_links = config.node.limits.max_links; + let coords_response_interval_ms = config.node.session.coords_response_interval_ms; Ok(Self { identity, @@ -413,6 +416,9 @@ impl Node { 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), + ), retry_pending: HashMap::new(), peer_aliases: HashMap::new(), }) @@ -450,6 +456,7 @@ impl Node { let max_connections = config.node.limits.max_connections; let max_peers = config.node.limits.max_peers; let max_links = config.node.limits.max_links; + let coords_response_interval_ms = config.node.session.coords_response_interval_ms; Self { identity, @@ -490,6 +497,9 @@ impl Node { 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), + ), retry_pending: HashMap::new(), peer_aliases: HashMap::new(), } diff --git a/src/node/routing_error_rate_limit.rs b/src/node/routing_error_rate_limit.rs index 7c617bc..6c42dd8 100644 --- a/src/node/routing_error_rate_limit.rs +++ b/src/node/routing_error_rate_limit.rs @@ -32,6 +32,15 @@ impl RoutingErrorRateLimiter { } } + /// 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 @@ -135,4 +144,19 @@ mod tests { limiter.cleanup(Instant::now()); assert_eq!(limiter.len(), 1); } + + #[test] + fn test_with_interval_custom_rate() { + let mut limiter = RoutingErrorRateLimiter::with_interval(Duration::from_millis(500)); + assert!(limiter.should_send(&addr(1))); + assert!(!limiter.should_send(&addr(1))); + + // Still rate-limited after 200ms (would pass with default 100ms) + thread::sleep(Duration::from_millis(200)); + assert!(!limiter.should_send(&addr(1))); + + // Allowed after 500ms total + thread::sleep(Duration::from_millis(350)); + assert!(limiter.should_send(&addr(1))); + } } diff --git a/src/protocol/session.rs b/src/protocol/session.rs index b143ec0..2ab465d 100644 --- a/src/protocol/session.rs +++ b/src/protocol/session.rs @@ -34,6 +34,8 @@ pub enum SessionMessageType { ReceiverReport = 0x12, /// Path MTU notification (discovered path MTU). PathMtuNotification = 0x13, + /// Standalone coordinate cache warming (empty body, coords in CP flag). + CoordsWarmup = 0x14, // Link-layer error signals (0x20-0x2F) — plaintext, from transit routers /// Router cache miss — needs coordinates (link-layer error signal). @@ -52,6 +54,7 @@ impl SessionMessageType { 0x11 => Some(SessionMessageType::SenderReport), 0x12 => Some(SessionMessageType::ReceiverReport), 0x13 => Some(SessionMessageType::PathMtuNotification), + 0x14 => Some(SessionMessageType::CoordsWarmup), 0x20 => Some(SessionMessageType::CoordsRequired), 0x21 => Some(SessionMessageType::PathBroken), _ => None, @@ -73,6 +76,7 @@ impl fmt::Display for SessionMessageType { SessionMessageType::SenderReport => "SenderReport", SessionMessageType::ReceiverReport => "ReceiverReport", SessionMessageType::PathMtuNotification => "PathMtuNotification", + SessionMessageType::CoordsWarmup => "CoordsWarmup", SessionMessageType::CoordsRequired => "CoordsRequired", SessionMessageType::PathBroken => "PathBroken", };