From 009101ee4ac8faa4e8791c5964781e934a71b44f Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 12 Feb 2026 13:39:34 +0000 Subject: [PATCH] SessionDatagram forwarding handler with coordinate cache warming Add handle_session_datagram handler replacing the 0x40 dispatch stub. Transit nodes now decode the datagram envelope, enforce hop limits, warm coordinate caches from SessionSetup/SessionAck/DataPacket payloads, route via find_next_hop, and generate CoordsRequired/PathBroken error signals on routing failure. 14 new tests covering decode errors, hop limit enforcement, local delivery, cache warming for all session message types, single-hop and multi-hop forwarding through live node chains, error signal generation, and cache warming enabling subsequent routing. 375 tests pass. --- src/node/handlers/dispatch.rs | 2 +- src/node/handlers/forwarding.rs | 237 ++++++++++++++ src/node/handlers/mod.rs | 1 + src/node/tests/forwarding.rs | 559 ++++++++++++++++++++++++++++++++ src/node/tests/mod.rs | 1 + 5 files changed, 799 insertions(+), 1 deletion(-) create mode 100644 src/node/handlers/forwarding.rs create mode 100644 src/node/tests/forwarding.rs diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 0197c39..b79ee05 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -36,7 +36,7 @@ impl Node { } 0x40 => { // SessionDatagram - debug!("Received SessionDatagram (not yet implemented)"); + self.handle_session_datagram(from, payload).await; } 0x50 => { // Disconnect diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs new file mode 100644 index 0000000..8ac13cd --- /dev/null +++ b/src/node/handlers/forwarding.rs @@ -0,0 +1,237 @@ +//! SessionDatagram forwarding handler. +//! +//! Handles incoming SessionDatagram (0x40) link messages: decodes the +//! envelope, enforces hop limits, performs coordinate cache warming from +//! 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::protocol::{ + CoordsRequired, DataPacket, PathBroken, SessionAck, SessionDatagram, SessionMessageType, + SessionSetup, +}; +use crate::NodeAddr; +use tracing::debug; + +impl Node { + /// Handle an incoming SessionDatagram from a peer. + /// + /// Called by `dispatch_link_message` for msg_type 0x40. The payload + /// has already had its msg_type byte stripped by dispatch. + pub(in crate::node) async fn handle_session_datagram(&mut self, _from: &NodeAddr, payload: &[u8]) { + let mut datagram = match SessionDatagram::decode(payload) { + Ok(dg) => dg, + Err(e) => { + debug!(error = %e, "Malformed SessionDatagram"); + return; + } + }; + + // Hop limit enforcement: decrement and drop if exhausted + if !datagram.decrement_hop_limit() { + debug!( + src = %datagram.src_addr, + dest = %datagram.dest_addr, + "SessionDatagram hop limit exhausted, dropping" + ); + return; + } + + // Coordinate cache warming from plaintext session-layer headers + self.try_warm_coord_cache(&datagram); + + // Local delivery check + if datagram.dest_addr == *self.node_addr() { + debug!( + src = %datagram.src_addr, + payload_len = datagram.payload.len(), + "SessionDatagram delivered locally" + ); + // TODO: deliver to session layer + return; + } + + // Find next hop toward destination + let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { + Some(peer) => *peer.node_addr(), + None => { + self.send_routing_error(&datagram).await; + return; + } + }; + + // Forward: re-encode (includes 0x40 type byte) and send + let encoded = datagram.encode(); + if let Err(e) = self + .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" + ); + } + } + + /// Attempt to warm the coordinate cache from session-layer payload headers. + /// + /// Transit routers can read the session message type byte and, for + /// SessionSetup and SessionAck, extract plaintext coordinate fields. + /// DataPacket with COORDS_PRESENT has plaintext coords before the + /// encrypted payload. Other types are ignored. + /// + /// Decode failures are logged and silently ignored — they don't block + /// forwarding. + fn try_warm_coord_cache(&mut self, datagram: &SessionDatagram) { + if datagram.payload.is_empty() { + return; + } + + let msg_type = datagram.payload[0]; + let inner = &datagram.payload[1..]; + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + match SessionMessageType::from_byte(msg_type) { + Some(SessionMessageType::SessionSetup) => { + match SessionSetup::decode(inner) { + Ok(setup) => { + self.coord_cache_mut().insert( + datagram.src_addr, + setup.src_coords, + now_ms, + ); + self.coord_cache_mut().insert( + datagram.dest_addr, + setup.dest_coords, + now_ms, + ); + debug!( + src = %datagram.src_addr, + dest = %datagram.dest_addr, + "Cached coords from SessionSetup" + ); + } + Err(e) => { + debug!(error = %e, "Failed to decode SessionSetup for cache warming"); + } + } + } + Some(SessionMessageType::SessionAck) => { + match SessionAck::decode(inner) { + Ok(ack) => { + self.coord_cache_mut().insert( + datagram.src_addr, + ack.src_coords, + now_ms, + ); + debug!( + src = %datagram.src_addr, + "Cached coords from SessionAck" + ); + } + Err(e) => { + debug!(error = %e, "Failed to decode SessionAck for cache warming"); + } + } + } + Some(SessionMessageType::DataPacket) => { + match DataPacket::decode(inner) { + Ok(data) => { + if data.flags.coords_present { + if let Some(src_coords) = data.src_coords { + self.coord_cache_mut().insert( + datagram.src_addr, + src_coords, + now_ms, + ); + } + if let Some(dest_coords) = data.dest_coords { + self.coord_cache_mut().insert( + datagram.dest_addr, + dest_coords, + now_ms, + ); + } + debug!( + src = %datagram.src_addr, + dest = %datagram.dest_addr, + "Cached coords from DataPacket" + ); + } + } + Err(e) => { + debug!(error = %e, "Failed to decode DataPacket for cache warming"); + } + } + } + _ => { + // CoordsRequired, PathBroken, unknown: no coords to cache + } + } + } + + /// Generate and send a routing error signal back to the datagram's source. + /// + /// If we have cached coords for the destination, send PathBroken (we know + /// where it is but can't reach it). Otherwise send CoordsRequired (we + /// don't know where it is). + /// + /// If we can't route the error back to the source either, drop silently. + /// No cascading errors. + async fn send_routing_error(&mut self, original: &SessionDatagram) { + let my_addr = *self.node_addr(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + let error_payload = + if let Some(coords) = self.coord_cache().get(&original.dest_addr, now_ms) { + let coords = coords.clone(); + PathBroken::new(original.dest_addr, my_addr) + .with_last_coords(coords) + .encode() + } else { + CoordsRequired::new(original.dest_addr, my_addr).encode() + }; + + let error_dg = SessionDatagram::new(my_addr, original.src_addr, error_payload); + + 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 error 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 routing error signal" + ); + } else { + debug!( + original_dest = %original.dest_addr, + error_dest = %original.src_addr, + "Sent routing error signal" + ); + } + } +} diff --git a/src/node/handlers/mod.rs b/src/node/handlers/mod.rs index 8213a67..c03d61d 100644 --- a/src/node/handlers/mod.rs +++ b/src/node/handlers/mod.rs @@ -2,6 +2,7 @@ mod dispatch; mod encrypted; +mod forwarding; mod handshake; mod rx_loop; mod timeout; diff --git a/src/node/tests/forwarding.rs b/src/node/tests/forwarding.rs new file mode 100644 index 0000000..371d26e --- /dev/null +++ b/src/node/tests/forwarding.rs @@ -0,0 +1,559 @@ +//! SessionDatagram forwarding tests. +//! +//! Tests for the handle_session_datagram handler including decode errors, +//! hop limit enforcement, local delivery, coordinate cache warming, and +//! multi-hop forwarding through live node topologies. + +use super::*; +use crate::protocol::{DataPacket, SessionAck, SessionDatagram, SessionSetup}; +use crate::tree::TreeCoordinate; +use spanning_tree::{ + cleanup_nodes, process_available_packets, run_tree_test, verify_tree_convergence, + TestNode, +}; + +// ============================================================================ +// Unit Tests +// ============================================================================ + +// --- Decode errors --- + +#[tokio::test] +async fn test_forwarding_decode_error() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + // Too-short payload: should log error and return without panic + node.handle_session_datagram(&from, &[0x00; 5]).await; +} + +// --- Hop limit --- + +#[tokio::test] +async fn test_forwarding_hop_limit_exhausted() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + let src = make_node_addr(0x01); + let dest = make_node_addr(0x02); + let dg = SessionDatagram::new(src, dest, vec![0x10, 0x00, 0x00, 0x00]) + .with_hop_limit(0); + let encoded = dg.encode(); + // Dispatch with payload after msg_type byte + node.handle_session_datagram(&from, &encoded[1..]).await; + // No panic, no send (node has no peers) +} + +#[tokio::test] +async fn test_forwarding_hop_limit_one_drops_at_transit() { + // hop_limit=1 means after decrement it becomes 0 — the datagram can + // still be delivered this hop but would be dropped at the next. + // decrement_hop_limit returns true (1 > 0), so the handler proceeds. + let mut node = make_node(); + let from = make_node_addr(0xAA); + let my_addr = *node.node_addr(); + let src = make_node_addr(0x01); + let dg = SessionDatagram::new(src, my_addr, vec![0x10, 0x00, 0x00, 0x00]) + .with_hop_limit(1); + let encoded = dg.encode(); + // Should succeed — hop_limit=1 decrements to 0 but packet is still processed + node.handle_session_datagram(&from, &encoded[1..]).await; +} + +// --- Local delivery --- + +#[tokio::test] +async fn test_forwarding_local_delivery() { + let mut node = make_node(); + let my_addr = *node.node_addr(); + let from = make_node_addr(0xAA); + let dg = SessionDatagram::new(from, my_addr, vec![0x10, 0x00, 0x00, 0x00]); + let encoded = dg.encode(); + // Should detect local delivery and return without forwarding + node.handle_session_datagram(&from, &encoded[1..]).await; +} + +// --- Direct peer forwarding --- + +#[tokio::test] +async fn test_forwarding_direct_peer() { + // Set up a node with one peer. Send a datagram destined for that peer. + // The handler should forward it directly. + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + + // Build a datagram from some external source destined for node 1 + let external_src = make_node_addr(0xEE); + let dg = SessionDatagram::new(external_src, node1_addr, vec![0x10, 0x00, 0x00, 0x00]); + let encoded = dg.encode(); + + // Handle on node 0: should forward to node 1 (direct peer) + nodes[0] + .node + .handle_session_datagram(&node0_addr, &encoded[1..]) + .await; + + // Process packets — node 1 should receive the forwarded datagram + tokio::time::sleep(Duration::from_millis(50)).await; + let count = process_available_packets(&mut nodes).await; + assert!(count > 0, "Expected forwarded packet to arrive at node 1"); + + cleanup_nodes(&mut nodes).await; +} + +// ============================================================================ +// Coordinate Cache Warming Tests +// ============================================================================ + +#[tokio::test] +async fn test_coord_cache_warming_session_setup() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + let src_addr = make_node_addr(0x01); + let dest_addr = make_node_addr(0x02); + let root_addr = make_node_addr(0xF0); + + let src_coords = TreeCoordinate::from_addrs(vec![src_addr, root_addr]).unwrap(); + let dest_coords = TreeCoordinate::from_addrs(vec![dest_addr, root_addr]).unwrap(); + + let setup = SessionSetup::new(src_coords.clone(), dest_coords.clone()); + let setup_payload = setup.encode(); + + let dg = SessionDatagram::new(src_addr, dest_addr, setup_payload); + let encoded = dg.encode(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + // Before: cache is empty + assert!(node.coord_cache().get(&src_addr, now_ms).is_none()); + assert!(node.coord_cache().get(&dest_addr, now_ms).is_none()); + + // Handle the datagram (will be local delivery or no-route, but cache warming + // happens before routing decision) + node.handle_session_datagram(&from, &encoded[1..]).await; + + // After: both src and dest coords should be cached + let cached_src = node.coord_cache().get(&src_addr, now_ms); + let cached_dest = node.coord_cache().get(&dest_addr, now_ms); + assert!(cached_src.is_some(), "src_addr coords not cached"); + assert!(cached_dest.is_some(), "dest_addr coords not cached"); + + // Verify the cached coords have the right root + let cached_src = cached_src.unwrap(); + let cached_dest = cached_dest.unwrap(); + assert_eq!(cached_src.root_id(), &root_addr); + assert_eq!(cached_dest.root_id(), &root_addr); +} + +#[tokio::test] +async fn test_coord_cache_warming_session_ack() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + let src_addr = make_node_addr(0x01); + let dest_addr = make_node_addr(0x02); + let root_addr = make_node_addr(0xF0); + + let src_coords = TreeCoordinate::from_addrs(vec![src_addr, root_addr]).unwrap(); + + let ack = SessionAck::new(src_coords.clone()); + let ack_payload = ack.encode(); + + let dg = SessionDatagram::new(src_addr, dest_addr, ack_payload); + let encoded = dg.encode(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + assert!(node.coord_cache().get(&src_addr, now_ms).is_none()); + + node.handle_session_datagram(&from, &encoded[1..]).await; + + // SessionAck only caches src_coords (the acknowledger's coords) + let cached_src = node.coord_cache().get(&src_addr, now_ms); + assert!(cached_src.is_some(), "src_addr coords not cached from SessionAck"); + assert_eq!(cached_src.unwrap().root_id(), &root_addr); + + // dest_addr should NOT be cached (SessionAck doesn't carry dest coords) + assert!(node.coord_cache().get(&dest_addr, now_ms).is_none()); +} + +#[tokio::test] +async fn test_coord_cache_warming_data_packet_with_coords() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + let src_addr = make_node_addr(0x01); + let dest_addr = make_node_addr(0x02); + let root_addr = make_node_addr(0xF0); + + let src_coords = TreeCoordinate::from_addrs(vec![src_addr, root_addr]).unwrap(); + let dest_coords = TreeCoordinate::from_addrs(vec![dest_addr, root_addr]).unwrap(); + + let data = DataPacket::new(vec![1, 2, 3, 4]) + .with_coords(src_coords.clone(), dest_coords.clone()); + let data_payload = data.encode(); + + let dg = SessionDatagram::new(src_addr, dest_addr, data_payload); + let encoded = dg.encode(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + assert!(node.coord_cache().get(&src_addr, now_ms).is_none()); + assert!(node.coord_cache().get(&dest_addr, now_ms).is_none()); + + node.handle_session_datagram(&from, &encoded[1..]).await; + + assert!( + node.coord_cache().get(&src_addr, now_ms).is_some(), + "src coords not cached from DataPacket" + ); + assert!( + node.coord_cache().get(&dest_addr, now_ms).is_some(), + "dest coords not cached from DataPacket" + ); +} + +#[tokio::test] +async fn test_coord_cache_warming_opaque_data_packet() { + let mut node = make_node(); + let from = make_node_addr(0xAA); + let src_addr = make_node_addr(0x01); + let dest_addr = make_node_addr(0x02); + + // DataPacket without COORDS_PRESENT — no coords to cache + let data = DataPacket::new(vec![1, 2, 3, 4]); + let data_payload = data.encode(); + + let dg = SessionDatagram::new(src_addr, dest_addr, data_payload); + let encoded = dg.encode(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + node.handle_session_datagram(&from, &encoded[1..]).await; + + assert!( + node.coord_cache().get(&src_addr, now_ms).is_none(), + "Should not cache coords from opaque DataPacket" + ); + assert!( + node.coord_cache().get(&dest_addr, now_ms).is_none(), + "Should not cache coords from opaque DataPacket" + ); +} + +// ============================================================================ +// Integration Tests +// ============================================================================ + +/// Helper: populate all coordinate caches across a set of test nodes. +fn populate_all_coord_caches(nodes: &mut [TestNode]) { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + // Collect all coords first to avoid borrow conflicts + let all_coords: Vec<(NodeAddr, TreeCoordinate)> = nodes + .iter() + .map(|tn| { + ( + *tn.node.node_addr(), + tn.node.tree_state().my_coords().clone(), + ) + }) + .collect(); + + for tn in nodes.iter_mut() { + for (addr, coords) in &all_coords { + if addr != tn.node.node_addr() { + tn.node + .coord_cache_mut() + .insert(*addr, coords.clone(), now_ms); + } + } + } +} + +#[tokio::test] +async fn test_forwarding_single_hop() { + // 3-node chain: 0 -- 1 -- 2 + // Send datagram from node 0 destined for node 2. + // Node 1 should forward it. + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test(3, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node2_addr = *nodes[2].node.node_addr(); + + // Build a SessionDatagram from node 0 to node 2 + let dg = SessionDatagram::new( + node0_addr, + node2_addr, + vec![0x10, 0x00, 0x04, 0x00, 1, 2, 3, 4], + ); + let encoded = dg.encode(); + + // Send from node 0 to node 1 (the first hop) + nodes[0] + .node + .send_encrypted_link_message(&node1_addr, &encoded) + .await + .unwrap(); + + // Process: node 1 receives, decrypts, dispatches to handler, forwards to node 2 + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + + // Give time for the forwarded packet to arrive at node 2 + tokio::time::sleep(Duration::from_millis(50)).await; + let count = process_available_packets(&mut nodes).await; + + // Node 2 should have received the forwarded datagram + // (it sees dest_addr == self, treats as local delivery) + // We verify the chain completed by checking packets were processed. + assert!(count > 0, "Expected forwarded packet at node 2"); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_forwarding_multi_hop() { + // 5-node chain: 0 -- 1 -- 2 -- 3 -- 4 + // Send datagram from node 0 destined for node 4. + let edges = vec![(0, 1), (1, 2), (2, 3), (3, 4)]; + let mut nodes = run_tree_test(5, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node4_addr = *nodes[4].node.node_addr(); + + // Build a SessionDatagram with enough hop_limit for 4 hops + let dg = SessionDatagram::new( + node0_addr, + node4_addr, + vec![0x10, 0x00, 0x04, 0x00, 1, 2, 3, 4], + ); + let encoded = dg.encode(); + + // Inject at node 0 → node 1 + nodes[0] + .node + .send_encrypted_link_message(&node1_addr, &encoded) + .await + .unwrap(); + + // Process multiple rounds to let the datagram traverse the chain + for _ in 0..5 { + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + } + + // Verify no crashes — the datagram should have traversed 1→2→3→4 + // and been delivered locally at node 4. + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_forwarding_hop_limit_prevents_infinite_loops() { + // 3-node chain: 0 -- 1 -- 2 + // Send a datagram with hop_limit=1. It should be forwarded by node 1 + // (decrement to 0) and delivered at node 2 (local delivery). If node 2 + // tried to forward further, the 0 hop_limit would prevent it. + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test(3, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node2_addr = *nodes[2].node.node_addr(); + + let dg = SessionDatagram::new( + node0_addr, + node2_addr, + vec![0x10, 0x00, 0x04, 0x00, 1, 2, 3, 4], + ) + .with_hop_limit(2); // Enough for 0→1 (decrement to 1) and 1→2 (decrement to 0, local delivery) + + let encoded = dg.encode(); + + nodes[0] + .node + .send_encrypted_link_message(&node1_addr, &encoded) + .await + .unwrap(); + + for _ in 0..3 { + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + } + + // No panic, no infinite loop + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_forwarding_no_route_generates_error() { + // 2-node network: 0 -- 1 + // Node 0 receives a datagram from node 1 destined for unknown node. + // Node 0 should generate CoordsRequired back to node 1. + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + verify_tree_convergence(&nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let unknown_dest = make_node_addr(0xFF); + + // Node 1 sends a datagram to unknown dest via node 0 + let dg = SessionDatagram::new(node1_addr, unknown_dest, vec![0x10, 0x00, 0x00, 0x00]); + let encoded = dg.encode(); + + // Inject at node 1 → node 0 + nodes[1] + .node + .send_encrypted_link_message(&node0_addr, &encoded) + .await + .unwrap(); + + // Process: node 0 receives, can't route to unknown_dest, sends error back to node 1 + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + + // Process the error signal arriving at node 1 + tokio::time::sleep(Duration::from_millis(50)).await; + let count = process_available_packets(&mut nodes).await; + assert!(count > 0, "Expected error signal to arrive at node 1"); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_forwarding_with_cache_warming_enables_routing() { + // 4-node chain: 0 -- 1 -- 2 -- 3 + // Initially, only populate coord caches at node 0. + // Send a SessionSetup from node 0 to node 3. + // As it traverses 1 and 2, those nodes should cache coordinates from the + // SessionSetup. Then verify the caches were warmed. + let edges = vec![(0, 1), (1, 2), (2, 3)]; + let mut nodes = run_tree_test(4, &edges, false).await; + verify_tree_convergence(&nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let _node2_addr = *nodes[2].node.node_addr(); + let node3_addr = *nodes[3].node.node_addr(); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + + // Only populate node 0's cache with all coords (the source knows where to send) + let all_coords: Vec<(NodeAddr, TreeCoordinate)> = nodes + .iter() + .map(|tn| { + ( + *tn.node.node_addr(), + tn.node.tree_state().my_coords().clone(), + ) + }) + .collect(); + + // Node 0 gets full cache + for (addr, coords) in &all_coords { + if addr != nodes[0].node.node_addr() { + nodes[0] + .node + .coord_cache_mut() + .insert(*addr, coords.clone(), now_ms); + } + } + + // Nodes 1 and 2 only get their direct peers' coords (from tree state) + // but NOT node 0 or node 3's coords (the endpoints) + // Actually, they need bloom filter hits to route, so let's also ensure + // bloom filters are converged (which they should be from run_tree_test). + + // But nodes 1 and 2 need cached coords to make loop-free forwarding + // decisions. Without coords, find_next_hop returns None. + // This is exactly what the SessionSetup cache warming solves! + // Populate enough so nodes can route to their adjacent peers, + // but NOT the distant endpoint coords. + for i in 0..4 { + for j in 0..4 { + if i != j { + // Give each node coords for its direct peers only + let j_addr = *nodes[j].node.node_addr(); + if nodes[i].node.get_peer(&j_addr).is_some() { + let coords = all_coords.iter().find(|(a, _)| a == &j_addr).unwrap().1.clone(); + nodes[i].node.coord_cache_mut().insert(j_addr, coords, now_ms); + } + } + } + } + + // Build SessionSetup with real coordinates + let src_coords = nodes[0].node.tree_state().my_coords().clone(); + let dest_coords = nodes[3].node.tree_state().my_coords().clone(); + let setup = SessionSetup::new(src_coords, dest_coords); + let setup_payload = setup.encode(); + + let dg = SessionDatagram::new(node0_addr, node3_addr, setup_payload); + let encoded = dg.encode(); + + // Inject: node 0 → node 1 + nodes[0] + .node + .send_encrypted_link_message(&node1_addr, &encoded) + .await + .unwrap(); + + // Process multiple rounds for the datagram to traverse 1→2→3 + for _ in 0..5 { + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + } + + // Verify cache warming: nodes 1 and 2 should now have cached coords + // for both node 0 and node 3 (from the SessionSetup) + let cached_0_at_1 = nodes[1].node.coord_cache().get(&node0_addr, now_ms); + let cached_3_at_1 = nodes[1].node.coord_cache().get(&node3_addr, now_ms); + assert!( + cached_0_at_1.is_some(), + "Node 1 should have cached node 0's coords from SessionSetup" + ); + assert!( + cached_3_at_1.is_some(), + "Node 1 should have cached node 3's coords from SessionSetup" + ); + + let cached_0_at_2 = nodes[2].node.coord_cache().get(&node0_addr, now_ms); + let cached_3_at_2 = nodes[2].node.coord_cache().get(&node3_addr, now_ms); + assert!( + cached_0_at_2.is_some(), + "Node 2 should have cached node 0's coords from SessionSetup" + ); + assert!( + cached_3_at_2.is_some(), + "Node 2 should have cached node 3's coords from SessionSetup" + ); + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 2f893ca..cdf64ac 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -6,6 +6,7 @@ use std::time::Duration; mod bloom; mod disconnect; +mod forwarding; mod handshake; mod routing; mod spanning_tree;