From 801a5ab22297a9bc80063482d660e0165ebdcab2 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 14 Feb 2026 16:59:40 +0000 Subject: [PATCH] Integrate discovery protocol into data plane for multi-hop routing Wire the discovery protocol into the data plane path so that find_next_hop() consults route_cache as a fallback when coord_cache has no entry. When session initiation fails due to missing routes, trigger discovery (initiate_lookup) instead of immediately sending ICMPv6 Destination Unreachable. On discovery completion, retry session initiation for any pending TUN packets. Changes: - find_next_hop() falls back to route_cache when coord_cache misses - handle_tun_outbound() triggers discovery on session failure - handle_lookup_response() retries session after discovery completes - handle_coords_required() triggers discovery for missing coordinates - Add pending_lookups deduplication with 10-second timeout - Periodic cleanup of stale lookups in tick handler - New test: route_cache fallback verification in find_next_hop --- src/node/handlers/discovery.rs | 54 +++++++++++++++++++++++++++++-- src/node/handlers/rx_loop.rs | 1 + src/node/handlers/session.rs | 57 ++++++++++++++++++++++++++------ src/node/mod.rs | 21 ++++++++++-- src/node/tests/routing.rs | 59 ++++++++++++++++++++++++++++++++++ 5 files changed, 177 insertions(+), 15 deletions(-) diff --git a/src/node/handlers/discovery.rs b/src/node/handlers/discovery.rs index 74742de..b6bc67c 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -4,7 +4,7 @@ //! visited filter for loop prevention, and reverse-path forwarding for //! responses. -use crate::node::{Node, RecentRequest}; +use crate::node::{Node, RecentRequest, DISCOVERY_TTL, LOOKUP_TIMEOUT_MS}; use crate::protocol::{LookupRequest, LookupResponse}; use crate::NodeAddr; use tracing::{debug, trace}; @@ -138,11 +138,22 @@ impl Node { "Received LookupResponse, caching route" ); + let target = response.target; self.route_cache.insert( - response.target, + target, response.target_coords, now_ms, ); + + // Clean up pending lookup tracking + self.pending_lookups.remove(&target); + + // If we have pending TUN packets for this target, retry session + // initiation. The route_cache now has coords, so find_next_hop() + // should succeed. + if self.pending_tun_packets.contains_key(&target) { + self.retry_session_after_discovery(target).await; + } } } @@ -251,7 +262,6 @@ impl Node { /// does NOT record the request_id in recent_requests, so when the /// response arrives, it's recognized as "our request" and the /// target's coordinates are cached in route_cache. - #[allow(dead_code)] // Called from integration tests; will be used from event loop pub(in crate::node) async fn initiate_lookup(&mut self, target: &NodeAddr, ttl: u8) { let origin = *self.node_addr(); let origin_coords = self.tree_state().my_coords().clone(); @@ -283,6 +293,44 @@ impl Node { } } + /// Initiate a discovery lookup if one is not already pending for this target. + /// + /// Deduplicates lookups using `pending_lookups` with a timeout. If a + /// lookup was recently initiated and hasn't timed out, this is a no-op. + pub(in crate::node) async fn maybe_initiate_lookup(&mut self, dest: &NodeAddr) { + let now_ms = Self::now_ms(); + if let Some(&initiated_at) = self.pending_lookups.get(dest) { + if now_ms.saturating_sub(initiated_at) < LOOKUP_TIMEOUT_MS { + return; + } + } + self.pending_lookups.insert(*dest, now_ms); + self.initiate_lookup(dest, DISCOVERY_TTL).await; + } + + /// Remove timed-out pending lookups and drain their queued packets. + /// + /// Called periodically from the tick handler. For each timed-out lookup, + /// sends ICMPv6 Destination Unreachable for any queued TUN packets and + /// removes them from the pending queue. + pub(in crate::node) fn purge_stale_lookups(&mut self, now_ms: u64) { + let timed_out: Vec = self + .pending_lookups + .iter() + .filter(|&(_, &ts)| now_ms.saturating_sub(ts) >= LOOKUP_TIMEOUT_MS) + .map(|(addr, _)| *addr) + .collect(); + + for addr in timed_out { + self.pending_lookups.remove(&addr); + if let Some(packets) = self.pending_tun_packets.remove(&addr) { + for pkt in &packets { + self.send_icmpv6_dest_unreachable(pkt); + } + } + } + } + /// Remove expired entries from the recent_requests cache. fn purge_expired_requests(&mut self, current_time_ms: u64) { self.recent_requests diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 0f46e33..d9001fa 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -82,6 +82,7 @@ impl Node { self.process_pending_retries(now_ms).await; self.check_tree_state().await; self.check_bloom_state().await; + self.purge_stale_lookups(now_ms); } } } diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index f0e6888..7a58cfa 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -45,7 +45,7 @@ impl Node { self.handle_data_packet(src_addr, inner).await; } Some(SessionMessageType::CoordsRequired) => { - self.handle_coords_required(inner); + self.handle_coords_required(inner).await; } Some(SessionMessageType::PathBroken) => { self.handle_path_broken(inner); @@ -322,9 +322,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. Future packets should include - /// coordinates (set COORDS_PRESENT flag). - fn handle_coords_required(&mut self, inner: &[u8]) { + /// coordinates for the destination. Trigger discovery to populate + /// the route cache so subsequent routing attempts succeed. + async fn handle_coords_required(&mut self, inner: &[u8]) { let msg = match CoordsRequired::decode(inner) { Ok(m) => m, Err(e) => { @@ -336,8 +336,10 @@ impl Node { debug!( dest = %msg.dest_addr, reporter = %msg.reporter, - "CoordsRequired: transit router needs coordinates" + "CoordsRequired: transit router needs coordinates, initiating discovery" ); + + self.maybe_initiate_lookup(&msg.dest_addr).await; } /// Handle a PathBroken error signal from a transit router. @@ -555,17 +557,20 @@ impl Node { return; } - // No session: initiate one and queue the packet + // No session: initiate one and queue the packet. + // If session initiation fails (no route), trigger discovery and + // queue the packet for retry when discovery completes. if let Err(e) = self.initiate_session(dest_addr, dest_pubkey).await { - debug!(dest = %dest_addr, error = %e, "Failed to initiate session for TUN packet"); - self.send_icmpv6_dest_unreachable(&ipv6_packet); + debug!(dest = %dest_addr, error = %e, "Failed to initiate session, trying discovery"); + self.maybe_initiate_lookup(&dest_addr).await; + self.queue_pending_packet(dest_addr, ipv6_packet); return; } self.queue_pending_packet(dest_addr, ipv6_packet); } /// Send ICMPv6 Destination Unreachable back through TUN. - fn send_icmpv6_dest_unreachable(&self, original_packet: &[u8]) { + pub(in crate::node) fn send_icmpv6_dest_unreachable(&self, original_packet: &[u8]) { use crate::icmp::{build_dest_unreachable, should_send_icmp_error, DestUnreachableCode}; use crate::FipsAddress; @@ -615,4 +620,38 @@ impl Node { } } } + + /// Retry session initiation after discovery provided coordinates. + /// + /// Called when a LookupResponse arrives and we have pending TUN packets + /// for the discovered target. The route_cache now has coords, so + /// `find_next_hop()` should succeed and the SessionSetup can be sent. + pub(in crate::node) async fn retry_session_after_discovery(&mut self, dest_addr: NodeAddr) { + // Look up the destination's public key from the identity cache + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&dest_addr.as_bytes()[0..15]); + let dest_pubkey = match self.lookup_by_fips_prefix(&prefix) { + Some(&(_, pk)) => pk, + None => { + debug!(dest = %dest_addr, "Discovery complete but no identity for session retry"); + return; + } + }; + + // Skip if a session already exists + if let Some(existing) = self.sessions.get(&dest_addr) { + if existing.state().is_established() || existing.state().is_initiating() { + return; + } + } + + match self.initiate_session(dest_addr, dest_pubkey).await { + Ok(()) => { + debug!(dest = %dest_addr, "Session initiated after discovery"); + } + Err(e) => { + debug!(dest = %dest_addr, error = %e, "Session retry after discovery failed"); + } + } + } } diff --git a/src/node/mod.rs b/src/node/mod.rs index 62e0a6b..2bc5f1b 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -186,6 +186,11 @@ type AddrKey = (TransportId, TransportAddr); /// /// The `addr_to_link` map enables dispatching incoming packets to the right /// connection before authentication completes. +/// +/// Discovery lookup constants used across handler modules. +const LOOKUP_TIMEOUT_MS: u64 = 10_000; +const DISCOVERY_TTL: u8 = 64; + pub struct Node { // === Identity === /// This node's cryptographic identity. @@ -258,6 +263,11 @@ pub struct Node { /// Keyed by destination NodeAddr, bounded per-dest and total. pending_tun_packets: HashMap>>, + // === Pending Discovery Lookups === + /// Tracks in-flight discovery lookups. Maps target NodeAddr to the + /// initiation timestamp (Unix ms). Prevents duplicate flood queries. + pending_lookups: HashMap, + // === Resource Limits === /// Maximum connections (0 = unlimited). max_connections: usize, @@ -363,6 +373,7 @@ impl Node { sessions: HashMap::new(), identity_cache: HashMap::new(), pending_tun_packets: HashMap::new(), + pending_lookups: HashMap::new(), max_connections: 256, max_peers: 128, max_links: 256, @@ -420,6 +431,7 @@ impl Node { sessions: HashMap::new(), identity_cache: HashMap::new(), pending_tun_packets: HashMap::new(), + pending_lookups: HashMap::new(), max_connections: 256, max_peers: 128, max_links: 256, @@ -849,7 +861,8 @@ impl Node { /// 5. No route → `None` /// /// Both the bloom filter and tree routing paths require cached destination - /// coordinates. Without coordinates, the node cannot make loop-free + /// coordinates (checked in `coord_cache` first, then `route_cache` as + /// fallback). Without coordinates, the node cannot make loop-free /// forwarding decisions. The caller should signal `CoordsRequired` back /// to the source when `None` is returned for a non-local destination. pub fn find_next_hop(&self, dest_node_addr: &NodeAddr) -> Option<&ActivePeer> { @@ -865,12 +878,14 @@ impl Node { } } - // Look up destination coords (required by both bloom and tree paths) + // Look up destination coords (required by both bloom and tree paths). + // Try coord_cache first (session-based), then route_cache (discovery-based). 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_coords = self.coord_cache.get(dest_node_addr, now_ms)?; + let dest_coords = self.coord_cache.get(dest_node_addr, now_ms) + .or_else(|| self.route_cache.get(dest_node_addr).map(|c| c.coords()))?; // 3. Bloom filter candidates — requires dest_coords for loop-free selection let candidates: Vec<&ActivePeer> = self.destination_in_filters(dest_node_addr); diff --git a/src/node/tests/routing.rs b/src/node/tests/routing.rs index b613e0e..6a57ca3 100644 --- a/src/node/tests/routing.rs +++ b/src/node/tests/routing.rs @@ -258,6 +258,65 @@ fn test_routing_bloom_hit_without_coords_returns_none() { assert!(node.find_next_hop(&dest).is_none()); } +// === Route cache fallback === + +#[test] +fn test_routing_route_cache_fallback() { + // Verify that find_next_hop() falls back to route_cache when + // coord_cache has no entry. This is the key change that enables + // discovery-based routing: initiate_lookup() populates route_cache, + // and find_next_hop() now consults it. + let mut node = make_node(); + let transport_id = TransportId::new(1); + let my_addr = *node.node_addr(); + + // Create a peer + let link_id = LinkId::new(1); + let (conn, id) = make_completed_connection(&mut node, link_id, transport_id, 1000); + let peer_addr = *id.node_addr(); + node.add_connection(conn).unwrap(); + node.promote_connection(link_id, id, 2000).unwrap(); + + // Set up tree: we are root, peer is our child + let peer_coords = TreeCoordinate::from_addrs(vec![peer_addr, my_addr]).unwrap(); + node.tree_state_mut().update_peer( + ParentDeclaration::new(peer_addr, my_addr, 1, 1000), + peer_coords, + ); + + // Create a destination "behind" the peer in the tree + let dest = make_node_addr(99); + let dest_coords = TreeCoordinate::from_addrs(vec![dest, peer_addr, my_addr]).unwrap(); + + // Put dest in peer's bloom filter so there's a candidate + let peer = node.get_peer_mut(&peer_addr).unwrap(); + let mut filter = BloomFilter::new(); + filter.insert(&dest); + peer.update_filter(filter, 1, 3000); + + // Verify: coord_cache has nothing for dest + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + assert!(node.coord_cache().get(&dest, now_ms).is_none()); + + // Without route_cache entry, should return None (same as before) + assert!(node.find_next_hop(&dest).is_none()); + + // Now populate route_cache (as discovery would do) + node.route_cache_mut().insert(dest, dest_coords, now_ms); + + // find_next_hop should succeed via route_cache fallback + let result = node.find_next_hop(&dest); + assert!(result.is_some(), "Should route via route_cache fallback"); + assert_eq!( + result.unwrap().node_addr(), + &peer_addr, + "Should pick peer with bloom filter hit" + ); +} + // === Integration: converged network === #[tokio::test]