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

@@ -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<NodeAddr> = 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

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

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

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<NodeAddr, VecDeque<Vec<u8>>>,
 
+    // === Pending Discovery Lookups ===
+    /// Tracks in-flight discovery lookups. Maps target NodeAddr to the
+    /// initiation timestamp (Unix ms). Prevents duplicate flood queries.
+    pending_lookups: HashMap<NodeAddr, u64>,
+
     // === 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);

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]