diff --git a/src/node/tests/disconnect.rs b/src/node/tests/disconnect.rs new file mode 100644 index 0000000..a7a92bc --- /dev/null +++ b/src/node/tests/disconnect.rs @@ -0,0 +1,271 @@ +//! Disconnect and peer removal integration tests. +//! +//! Tests that graceful disconnect messages propagate correctly through +//! multi-node networks and trigger proper cascading cleanup: peer removal, +//! tree reconvergence, and bloom filter recomputation. + +use super::spanning_tree::*; +use super::*; +use crate::protocol::{Disconnect, DisconnectReason}; + +/// 3-node chain: middle node disconnects one peer. +/// +/// Chain: 0 -- 1 -- 2. Node 1 sends Disconnect to node 0. +/// Verifies: +/// - Node 0 removes node 1 from its peer table +/// - Node 0's tree reconverges (becomes its own root since isolated) +/// - Node 1 still has node 2 as a peer +#[tokio::test] +async fn test_disconnect_chain_peer_removal() { + // Build 3-node chain: 0 -- 1 -- 2 + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test(3, &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(); + + // Verify initial state: node 0 has 1 peer (node 1) + assert_eq!(nodes[0].node.peer_count(), 1); + assert!(nodes[0].node.get_peer(&node1_addr).is_some()); + + // Node 1 sends Disconnect(Shutdown) to node 0 + let disconnect = Disconnect::new(DisconnectReason::Shutdown); + let plaintext = disconnect.encode(); + nodes[1] + .node + .send_encrypted_link_message(&node0_addr, &plaintext) + .await + .expect("Failed to send disconnect"); + + // Process the disconnect at node 0 + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + + // Node 0 should have removed node 1 + assert_eq!( + nodes[0].node.peer_count(), + 0, + "Node 0 should have no peers after disconnect" + ); + assert!( + nodes[0].node.get_peer(&node1_addr).is_none(), + "Node 0 should not have node 1 as a peer" + ); + + // Node 0 becomes its own root (isolated) + assert!( + nodes[0].node.tree_state().is_root(), + "Isolated node 0 should be root" + ); + + // Node 1 still has node 2 as a peer (disconnect was only to node 0) + assert!( + nodes[1].node.get_peer(&node2_addr).is_some(), + "Node 1 should still have node 2" + ); + + cleanup_nodes(&mut nodes).await; +} + +/// 4-node star: hub disconnects, spokes reconverge. +/// +/// Star: 0 is hub, connected to 1, 2, 3. Hub sends Disconnect to all. +/// Verifies: +/// - All spokes remove hub from their peer tables +/// - Each spoke becomes its own root (since there are no spoke-spoke links) +#[tokio::test] +async fn test_disconnect_star_hub_departs() { + let edges = vec![(0, 1), (0, 2), (0, 3)]; + let mut nodes = run_tree_test(4, &edges, false).await; + verify_tree_convergence(&nodes); + + let hub_addr = *nodes[0].node.node_addr(); + + // Hub sends Disconnect(Shutdown) to all spokes + let disconnect = Disconnect::new(DisconnectReason::Shutdown); + let plaintext = disconnect.encode(); + for spoke_idx in 1..4 { + let spoke_addr = *nodes[spoke_idx].node.node_addr(); + nodes[0] + .node + .send_encrypted_link_message(&spoke_addr, &plaintext) + .await + .expect("Failed to send disconnect"); + } + + // Process disconnects at all nodes + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + + // All spokes should have removed the hub + for spoke_idx in 1..4 { + assert!( + nodes[spoke_idx].node.get_peer(&hub_addr).is_none(), + "Spoke {} should have removed hub", + spoke_idx + ); + assert_eq!( + nodes[spoke_idx].node.peer_count(), + 0, + "Spoke {} should have no peers (no spoke-spoke links)", + spoke_idx + ); + assert!( + nodes[spoke_idx].node.tree_state().is_root(), + "Isolated spoke {} should become root", + spoke_idx + ); + } + + cleanup_nodes(&mut nodes).await; +} + +/// 5-node chain: interior node departs, network splits into two components. +/// +/// Chain: 0 -- 1 -- 2 -- 3 -- 4. Node 2 sends Disconnect to nodes 1 and 3. +/// Verifies: +/// - Peers removed correctly on both sides +/// - Bloom filters update so routing no longer bridges the partition +/// +/// Note: Tree root reconvergence after partition is not tested here because +/// the tree protocol detects parent loss but not root unreachability. Nodes +/// whose parent is still connected may retain a stale root belief until the +/// root refresh timer fires. This is a known limitation of the current tree +/// protocol — bloom filter routing is the primary mechanism and it updates +/// immediately on peer removal. +#[tokio::test] +async fn test_disconnect_chain_partition() { + 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); + + let node2_addr = *nodes[2].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node3_addr = *nodes[3].node.node_addr(); + + // Node 2 sends Disconnect to nodes 1 and 3 + let disconnect = Disconnect::new(DisconnectReason::Shutdown); + let plaintext = disconnect.encode(); + nodes[2] + .node + .send_encrypted_link_message(&node1_addr, &plaintext) + .await + .expect("Failed to send disconnect to node 1"); + nodes[2] + .node + .send_encrypted_link_message(&node3_addr, &plaintext) + .await + .expect("Failed to send disconnect to node 3"); + + // Process disconnects and let filters reconverge + drain_all_packets(&mut nodes, false).await; + + // Nodes 1 and 3 should have removed node 2 + assert!( + nodes[1].node.get_peer(&node2_addr).is_none(), + "Node 1 should not have node 2 as peer" + ); + assert!( + nodes[3].node.get_peer(&node2_addr).is_none(), + "Node 3 should not have node 2 as peer" + ); + + // Within each component, peers are still connected + let node0_addr = *nodes[0].node.node_addr(); + let node4_addr = *nodes[4].node.node_addr(); + assert!( + nodes[0].node.get_peer(&node1_addr).is_some(), + "Node 0 should still have node 1 as peer" + ); + assert!( + nodes[3].node.get_peer(&node4_addr).is_some(), + "Node 3 should still have node 4 as peer" + ); + + // Bloom filter check: node 0 should NOT see node 4 as reachable + // (bloom filters update immediately on peer removal via split-horizon recomputation) + let node0_reaches_node4 = nodes[0] + .node + .peers() + .any(|peer| peer.may_reach(&node4_addr)); + assert!( + !node0_reaches_node4, + "Node 0 should not see node 4 as reachable after partition" + ); + + // And vice versa + let node4_reaches_node0 = nodes[4] + .node + .peers() + .any(|peer| peer.may_reach(&node0_addr)); + assert!( + !node4_reaches_node0, + "Node 4 should not see node 0 as reachable after partition" + ); + + // Nodes within the same component should still see each other + let node0_reaches_node1 = nodes[0] + .node + .peers() + .any(|peer| peer.may_reach(&node1_addr)); + assert!( + node0_reaches_node1, + "Node 0 should still see node 1 as reachable" + ); + + let node4_reaches_node3 = nodes[4] + .node + .peers() + .any(|peer| peer.may_reach(&node3_addr)); + assert!( + node4_reaches_node3, + "Node 4 should still see node 3 as reachable" + ); + + cleanup_nodes(&mut nodes).await; +} + +/// Verify that different disconnect reasons are handled correctly. +/// +/// Sends each reason code and verifies the peer is removed regardless. +#[tokio::test] +async fn test_disconnect_all_reason_codes() { + let reasons = vec![ + DisconnectReason::Shutdown, + DisconnectReason::Restart, + DisconnectReason::ProtocolError, + DisconnectReason::TransportFailure, + DisconnectReason::ResourceExhaustion, + ]; + + for reason in reasons { + 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(); + + // Node 0 sends disconnect with this reason + let disconnect = Disconnect::new(reason); + let plaintext = disconnect.encode(); + nodes[0] + .node + .send_encrypted_link_message(&node1_addr, &plaintext) + .await + .expect("Failed to send disconnect"); + + tokio::time::sleep(Duration::from_millis(50)).await; + process_available_packets(&mut nodes).await; + + assert!( + nodes[1].node.get_peer(&node0_addr).is_none(), + "Node 1 should remove peer for reason {:?}", + reason + ); + + cleanup_nodes(&mut nodes).await; + } +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 516b8d1..2f893ca 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -5,6 +5,7 @@ use crate::PeerIdentity; use std::time::Duration; mod bloom; +mod disconnect; mod handshake; mod routing; mod spanning_tree; diff --git a/src/node/tests/routing.rs b/src/node/tests/routing.rs index a1f2277..b613e0e 100644 --- a/src/node/tests/routing.rs +++ b/src/node/tests/routing.rs @@ -605,3 +605,292 @@ async fn test_routing_reachability_100_nodes() { cleanup_nodes(&mut nodes).await; } +// === Peer removal stops routing through removed peer === + +/// After removing a peer from a converged chain, routing to destinations +/// previously reachable through that peer should fail. +/// +/// Chain: 0 -- 1 -- 2 -- 3. Remove node 2 from node 1's perspective. +/// Node 0 should no longer be able to route to node 3. +#[tokio::test] +async fn test_routing_stops_after_peer_removal() { + use crate::protocol::{Disconnect, DisconnectReason}; + + 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(); + + // Inject coordinates so routing works before removal + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + let all_coords: Vec<(NodeAddr, crate::tree::TreeCoordinate)> = nodes + .iter() + .map(|tn| (*tn.node.node_addr(), tn.node.tree_state().my_coords().clone())) + .collect(); + + for node in &mut nodes { + for &(ref addr, ref coords) in &all_coords { + if addr != node.node.node_addr() { + node.node.coord_cache_mut().insert(*addr, coords.clone(), now_ms); + } + } + } + + // Verify routing works before removal: node 0 → node 3 + let addr_index = build_addr_index(&nodes); + match simulate_forwarding(&nodes, &addr_index, 0, 3) { + ForwardResult::Delivered(_) => {} + other => panic!("Expected delivery before removal, got {:?}", other), + } + + // Node 2 sends Disconnect to node 1 + let disconnect = Disconnect::new(DisconnectReason::Shutdown); + let plaintext = disconnect.encode(); + nodes[2] + .node + .send_encrypted_link_message(&node1_addr, &plaintext) + .await + .expect("Failed to send disconnect"); + + // Process disconnect and let bloom filters reconverge + drain_all_packets(&mut nodes, false).await; + + // Verify node 1 removed node 2 + assert!( + nodes[1].node.get_peer(&node2_addr).is_none(), + "Node 1 should have removed node 2" + ); + + // Bloom filter check: node 0's peer (node 1) should no longer + // advertise node 3 as reachable + let node0_reaches_node3 = nodes[0] + .node + .peers() + .any(|peer| peer.may_reach(&node3_addr)); + assert!( + !node0_reaches_node3, + "Node 0 should not see node 3 as reachable after partition" + ); + + // Routing from node 0 to node 3 should now fail: no bloom filter hit. + // Greedy tree routing may still have stale coords cached, but without + // bloom filter hits, routing should stop at node 1 (which lost its + // peer to the other side). If stale coords exist, greedy routing could + // still attempt forwarding — but the self-distance check prevents loops. + // Either NoRoute or Loop-with-stale-coords is acceptable here; what + // matters is that delivery does NOT succeed. + match simulate_forwarding(&nodes, &addr_index, 0, 3) { + ForwardResult::NoRoute { .. } => {} // Expected: can't reach node 3 + ForwardResult::Loop { .. } => {} // Also acceptable: stale coords cause loop detection + ForwardResult::Delivered(hops) => { + panic!("Should NOT deliver after partition, but got delivery in {} hops", hops); + } + } + + // But routing within the same component still works: node 2 → node 3 + match simulate_forwarding(&nodes, &addr_index, 2, 3) { + ForwardResult::Delivered(_) => {} + other => panic!("Expected delivery within component, got {:?}", other), + } + + cleanup_nodes(&mut nodes).await; +} + +// === Bloom-filter-only transit routing (no globally injected coords) === + +/// Verify that transit routers can forward using bloom filters alone. +/// +/// In a converged network, only the SOURCE has the destination's coords +/// in its cache (simulating a real first-contact scenario where only the +/// source ran discovery). Transit routers have no cached coords for the +/// destination. Routing should still work because transit routers use +/// bloom filter hits to select next hops. +/// +/// Chain: 0 -- 1 -- 2 -- 3. Only node 0 has node 3's coords cached. +/// Nodes 1 and 2 route using bloom filters only. +#[tokio::test] +async fn test_routing_bloom_only_transit() { + let edges = vec![(0, 1), (1, 2), (2, 3)]; + let mut nodes = run_tree_test(4, &edges, false).await; + verify_tree_convergence(&nodes); + + let node3_addr = *nodes[3].node.node_addr(); + let node3_coords = nodes[3].node.tree_state().my_coords().clone(); + + // Only inject node 3's coords at node 0 (the source). + // Transit nodes (1, 2) have NO coords for node 3 in their caches. + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + nodes[0] + .node + .coord_cache_mut() + .insert(node3_addr, node3_coords, now_ms); + + // Node 0 should find a next hop (bloom filter hit at peer node 1, + // with coords available for tie-breaking at the source) + let hop = nodes[0].node.find_next_hop(&node3_addr); + assert!(hop.is_some(), "Node 0 should route to node 3 (has coords)"); + + // Node 1 should also find a next hop using bloom filter alone. + // But wait — find_next_hop requires dest_coords to be cached when + // bloom filter hits exist (loop prevention). Node 1 has no coords + // for node 3, so it should return None. + let hop_at_1 = nodes[1].node.find_next_hop(&node3_addr); + + // This is the key insight: bloom-filter-only transit routing does NOT + // work in the current implementation because find_next_hop gates bloom + // filter candidate selection on having cached dest_coords. Transit + // routers without coords return None, which is the correct behavior + // (prevents loops) but means the SessionSetup must carry coords to + // warm transit router caches before data packets can flow. + assert!( + hop_at_1.is_none(), + "Node 1 should NOT route without cached coords (loop prevention)" + ); + + // However, node 1 IS a direct peer of node 2, and node 2 IS a direct + // peer of node 3. The "direct peer" priority (step 2 in find_next_hop) + // would handle adjacency. Let's verify node 2 can route to its direct + // peer node 3. + let hop_at_2 = nodes[2].node.find_next_hop(&node3_addr); + assert!( + hop_at_2.is_some(), + "Node 2 should route to node 3 (direct peer)" + ); + assert_eq!( + hop_at_2.unwrap().node_addr(), + &node3_addr, + "Node 2's next hop to node 3 should be node 3 itself" + ); + + cleanup_nodes(&mut nodes).await; +} + +/// 100-node routing: verify that with coords cached ONLY at the source, +/// multi-hop forwarding still works because each transit node either has +/// the destination as a direct peer OR needs coords to break bloom filter +/// ties. +/// +/// This test reveals the boundary: in a converged network, bloom filter +/// routing needs dest_coords at each hop for loop-free forwarding through +/// non-adjacent nodes. Direct peer adjacency handles the last hop. +#[tokio::test] +async fn test_routing_source_only_coords_100_nodes() { + const NUM_NODES: usize = 100; + const TARGET_EDGES: usize = 250; + const SEED: u64 = 42; + + let edges = generate_random_edges(NUM_NODES, TARGET_EDGES, SEED); + let mut nodes = run_tree_test(NUM_NODES, &edges, false).await; + verify_tree_convergence(&nodes); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + // Collect all coords for injection + let all_coords: Vec<(NodeAddr, crate::tree::TreeCoordinate)> = nodes + .iter() + .map(|tn| (*tn.node.node_addr(), tn.node.tree_state().my_coords().clone())) + .collect(); + + let addr_index = build_addr_index(&nodes); + + // Test: for each pair, inject dest coords ONLY at the source. + // Count how many pairs can be delivered vs fail. + let mut source_only_delivered = 0usize; + let mut source_only_failed = 0usize; + let mut total_pairs = 0usize; + + // Test a sample of pairs (all pairs would be expensive) + let sample_pairs: Vec<(usize, usize)> = (0..NUM_NODES) + .step_by(10) + .flat_map(|s| { + (0..NUM_NODES) + .step_by(10) + .filter(move |&d| d != s) + .map(move |d| (s, d)) + }) + .collect(); + + for &(src, dst) in &sample_pairs { + total_pairs += 1; + + // Clear ALL coord caches + for node in &mut nodes { + node.node.coord_cache_mut().clear(); + } + + // Inject dest coords ONLY at the source + let (dest_addr, dest_coords) = &all_coords[dst]; + nodes[src] + .node + .coord_cache_mut() + .insert(*dest_addr, dest_coords.clone(), now_ms); + + match simulate_forwarding(&nodes, &addr_index, src, dst) { + ForwardResult::Delivered(_) => source_only_delivered += 1, + ForwardResult::NoRoute { .. } => source_only_failed += 1, + ForwardResult::Loop { .. } => { + panic!("Routing loop detected with source-only coords: {} -> {}", src, dst); + } + } + } + + eprintln!( + "\n === Source-Only Coords Routing ({} nodes) ===", + NUM_NODES + ); + eprintln!( + " Pairs: {} | Delivered: {} | Failed: {} | Delivery rate: {:.1}%", + total_pairs, + source_only_delivered, + source_only_failed, + source_only_delivered as f64 / total_pairs as f64 * 100.0 + ); + + // With source-only coords, only single-hop (direct peer) destinations + // are guaranteed to be delivered. Multi-hop destinations fail at the + // first transit node that doesn't have dest_coords cached. This + // confirms the protocol's design: SessionSetup MUST carry coords + // to warm transit router caches for multi-hop delivery. + assert!( + source_only_delivered > 0, + "At least some direct-peer pairs should be delivered" + ); + + // Now compare: inject coords at ALL nodes (full cache) and verify 100% + for node in &mut nodes { + for &(ref addr, ref coords) in &all_coords { + if addr != node.node.node_addr() { + node.node.coord_cache_mut().insert(*addr, coords.clone(), now_ms); + } + } + } + + let mut full_cache_failures = 0usize; + for &(src, dst) in &sample_pairs { + match simulate_forwarding(&nodes, &addr_index, src, dst) { + ForwardResult::Delivered(_) => {} + _ => full_cache_failures += 1, + } + } + assert_eq!( + full_cache_failures, 0, + "With full coord caches, all pairs should be delivered" + ); + + cleanup_nodes(&mut nodes).await; +} + diff --git a/src/protocol/link.rs b/src/protocol/link.rs index f02c76e..01f8875 100644 --- a/src/protocol/link.rs +++ b/src/protocol/link.rs @@ -278,6 +278,9 @@ pub struct SessionDatagram { pub payload: Vec, } +/// SessionDatagram fixed header size: msg_type(1) + src_addr(16) + dest_addr(16) + hop_limit(1). +pub const SESSION_DATAGRAM_HEADER_SIZE: usize = 34; + impl SessionDatagram { /// Create a new session datagram. pub fn new(src_addr: NodeAddr, dest_addr: NodeAddr, payload: Vec) -> Self { @@ -309,6 +312,41 @@ impl SessionDatagram { pub fn can_forward(&self) -> bool { self.hop_limit > 0 } + + /// Encode as link-layer message (msg_type + src_addr + dest_addr + hop_limit + payload). + pub fn encode(&self) -> Vec { + let mut buf = Vec::with_capacity(SESSION_DATAGRAM_HEADER_SIZE + self.payload.len()); + buf.push(LinkMessageType::SessionDatagram.to_byte()); + buf.extend_from_slice(self.src_addr.as_bytes()); + buf.extend_from_slice(self.dest_addr.as_bytes()); + buf.push(self.hop_limit); + buf.extend_from_slice(&self.payload); + buf + } + + /// Decode from link-layer payload (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + // src_addr(16) + dest_addr(16) + hop_limit(1) = 33 + if payload.len() < 33 { + return Err(ProtocolError::MessageTooShort { + expected: 33, + got: payload.len(), + }); + } + let mut src_bytes = [0u8; 16]; + src_bytes.copy_from_slice(&payload[0..16]); + let mut dest_bytes = [0u8; 16]; + dest_bytes.copy_from_slice(&payload[16..32]); + let hop_limit = payload[32]; + let inner_payload = payload[33..].to_vec(); + + Ok(Self { + src_addr: NodeAddr::from_bytes(src_bytes), + dest_addr: NodeAddr::from_bytes(dest_bytes), + hop_limit, + payload: inner_payload, + }) + } } // Legacy type alias for compatibility during transition @@ -450,4 +488,61 @@ mod tests { let decoded = Disconnect::decode(&[0x80]).unwrap(); assert_eq!(decoded.reason, DisconnectReason::Other); } + + // ===== SessionDatagram Tests ===== + + fn make_node_addr(val: u8) -> NodeAddr { + let mut bytes = [0u8; 16]; + bytes[0] = val; + NodeAddr::from_bytes(bytes) + } + + #[test] + fn test_session_datagram_encode_decode() { + let src = make_node_addr(0xAA); + let dest = make_node_addr(0xBB); + let payload = vec![0x10, 0x00, 0x05, 0x00, 1, 2, 3, 4, 5]; // DataPacket payload + let dg = SessionDatagram::new(src, dest, payload.clone()) + .with_hop_limit(32); + + let encoded = dg.encode(); + assert_eq!(encoded[0], 0x40); // msg_type + assert_eq!(encoded.len(), SESSION_DATAGRAM_HEADER_SIZE + payload.len()); + + // Decode (after msg_type) + let decoded = SessionDatagram::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.src_addr, src); + assert_eq!(decoded.dest_addr, dest); + assert_eq!(decoded.hop_limit, 32); + assert_eq!(decoded.payload, payload); + } + + #[test] + fn test_session_datagram_empty_payload() { + let dg = SessionDatagram::new(make_node_addr(1), make_node_addr(2), Vec::new()); + + let encoded = dg.encode(); + assert_eq!(encoded.len(), SESSION_DATAGRAM_HEADER_SIZE); + + let decoded = SessionDatagram::decode(&encoded[1..]).unwrap(); + assert!(decoded.payload.is_empty()); + } + + #[test] + fn test_session_datagram_decode_too_short() { + assert!(SessionDatagram::decode(&[]).is_err()); + assert!(SessionDatagram::decode(&[0x00; 20]).is_err()); + } + + #[test] + fn test_session_datagram_hop_limit_roundtrip() { + for hop in [0u8, 1, 64, 128, 255] { + let dg = SessionDatagram::new(make_node_addr(1), make_node_addr(2), vec![0x42]) + .with_hop_limit(hop); + + let encoded = dg.encode(); + let decoded = SessionDatagram::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.hop_limit, hop); + } + } } diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 4315013..9f560fb 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -31,13 +31,15 @@ mod tree; pub use error::ProtocolError; pub use link::{ Disconnect, DisconnectReason, HandshakeMessageType, LinkMessageType, SessionDatagram, + SESSION_DATAGRAM_HEADER_SIZE, }; pub use tree::TreeAnnounce; pub use filter::FilterAnnounce; pub use discovery::{LookupRequest, LookupResponse}; pub use session::{ CoordsRequired, DataFlags, DataPacket, PathBroken, SessionAck, SessionFlags, - SessionMessageType, SessionSetup, DATA_FLAG_COORDS_PRESENT, DATA_HEADER_SIZE, + SessionMessageType, SessionSetup, COORDS_REQUIRED_SIZE, DATA_FLAG_COORDS_PRESENT, + DATA_HEADER_SIZE, }; /// Protocol version for message compatibility. diff --git a/src/protocol/session.rs b/src/protocol/session.rs index 79a1498..423e986 100644 --- a/src/protocol/session.rs +++ b/src/protocol/session.rs @@ -1,5 +1,6 @@ //! Session-layer message types: setup, ack, data, and error messages. +use super::ProtocolError; use crate::tree::TreeCoordinate; use crate::NodeAddr; use std::fmt; @@ -67,6 +68,96 @@ impl fmt::Display for SessionMessageType { } } +// ============================================================================ +// Coordinate Wire Format Helpers +// ============================================================================ + +/// Encode a TreeCoordinate as address-only wire format: count(u16 LE) + addrs(16 × n). +/// +/// Session-layer messages serialize coordinates as NodeAddr arrays (16 bytes each), +/// without the sequence/timestamp metadata used by the tree gossip protocol. +fn encode_coords(coords: &TreeCoordinate, buf: &mut Vec) { + let addrs: Vec<&NodeAddr> = coords.node_addrs().collect(); + let count = addrs.len() as u16; + buf.extend_from_slice(&count.to_le_bytes()); + for addr in addrs { + buf.extend_from_slice(addr.as_bytes()); + } +} + +/// Decode a TreeCoordinate from address-only wire format. +/// +/// Returns the decoded coordinate and the number of bytes consumed. +fn decode_coords(data: &[u8]) -> Result<(TreeCoordinate, usize), ProtocolError> { + if data.len() < 2 { + return Err(ProtocolError::MessageTooShort { + expected: 2, + got: data.len(), + }); + } + let count = u16::from_le_bytes([data[0], data[1]]) as usize; + let needed = 2 + count * 16; + if data.len() < needed { + return Err(ProtocolError::MessageTooShort { + expected: needed, + got: data.len(), + }); + } + if count == 0 { + return Err(ProtocolError::Malformed( + "coordinate with zero entries".into(), + )); + } + let mut addrs = Vec::with_capacity(count); + for i in 0..count { + let offset = 2 + i * 16; + let mut bytes = [0u8; 16]; + bytes.copy_from_slice(&data[offset..offset + 16]); + addrs.push(NodeAddr::from_bytes(bytes)); + } + let coord = TreeCoordinate::from_addrs(addrs) + .map_err(|e| ProtocolError::Malformed(e.to_string()))?; + Ok((coord, needed)) +} + +/// Decode an optional coordinate field (count may be 0). +/// +/// Returns None if count is 0, Some(coord) otherwise, plus bytes consumed. +fn decode_optional_coords(data: &[u8]) -> Result<(Option, usize), ProtocolError> { + if data.len() < 2 { + return Err(ProtocolError::MessageTooShort { + expected: 2, + got: data.len(), + }); + } + let count = u16::from_le_bytes([data[0], data[1]]) as usize; + let needed = 2 + count * 16; + if data.len() < needed { + return Err(ProtocolError::MessageTooShort { + expected: needed, + got: data.len(), + }); + } + if count == 0 { + return Ok((None, 2)); + } + let mut addrs = Vec::with_capacity(count); + for i in 0..count { + let offset = 2 + i * 16; + let mut bytes = [0u8; 16]; + bytes.copy_from_slice(&data[offset..offset + 16]); + addrs.push(NodeAddr::from_bytes(bytes)); + } + let coord = TreeCoordinate::from_addrs(addrs) + .map_err(|e| ProtocolError::Malformed(e.to_string()))?; + Ok((Some(coord), needed)) +} + +/// Encode a count of zero (for empty/absent coordinate fields). +fn encode_empty_coords(buf: &mut Vec) { + buf.extend_from_slice(&0u16.to_le_bytes()); +} + // ============================================================================ // Session Flags // ============================================================================ @@ -126,9 +217,21 @@ impl SessionFlags { /// Session setup to establish cached coordinate state. /// /// Carried inside a SessionDatagram envelope which provides src_addr and -/// dest_addr. The SessionSetup payload contains only coordinates and the -/// Noise handshake data needed for route cache warming and session -/// establishment. +/// dest_addr. The SessionSetup payload contains coordinates, session flags, +/// and the Noise IK handshake message for session establishment. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|------------------|---------|-------------------------------------| +/// | 0 | msg_type | 1 byte | 0x00 | +/// | 1 | flags | 1 byte | Bit 0: REQUEST_ACK, Bit 1: BIDIR | +/// | 2 | src_coords_count | 2 bytes | u16 LE, number of src coord entries | +/// | 4 | src_coords | 16 × n | NodeAddr array (self → root) | +/// | ... | dest_coords_count| 2 bytes | u16 LE, number of dest coord entries| +/// | ... | dest_coords | 16 × m | NodeAddr array (dest → root) | +/// | ... | handshake_len | 2 bytes | u16 LE, Noise payload length | +/// | ... | handshake_payload| variable| Noise IK msg1 (82 bytes typical) | #[derive(Clone, Debug)] pub struct SessionSetup { /// Source coordinates (for return path caching). @@ -137,6 +240,8 @@ pub struct SessionSetup { pub dest_coords: TreeCoordinate, /// Session options. pub flags: SessionFlags, + /// Noise IK handshake message 1. + pub handshake_payload: Vec, } impl SessionSetup { @@ -146,6 +251,7 @@ impl SessionSetup { src_coords, dest_coords, flags: SessionFlags::new(), + handshake_payload: Vec::new(), } } @@ -154,6 +260,67 @@ impl SessionSetup { self.flags = flags; self } + + /// Set the Noise handshake payload. + pub fn with_handshake(mut self, payload: Vec) -> Self { + self.handshake_payload = payload; + self + } + + /// Encode as wire format (msg_type + flags + coords + handshake). + pub fn encode(&self) -> Vec { + let mut buf = Vec::new(); + buf.push(SessionMessageType::SessionSetup.to_byte()); + buf.push(self.flags.to_byte()); + encode_coords(&self.src_coords, &mut buf); + encode_coords(&self.dest_coords, &mut buf); + let hs_len = self.handshake_payload.len() as u16; + buf.extend_from_slice(&hs_len.to_le_bytes()); + buf.extend_from_slice(&self.handshake_payload); + buf + } + + /// Decode from wire format (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + if payload.is_empty() { + return Err(ProtocolError::MessageTooShort { + expected: 1, + got: 0, + }); + } + let flags = SessionFlags::from_byte(payload[0]); + let mut offset = 1; + + let (src_coords, consumed) = decode_coords(&payload[offset..])?; + offset += consumed; + + let (dest_coords, consumed) = decode_coords(&payload[offset..])?; + offset += consumed; + + if payload.len() < offset + 2 { + return Err(ProtocolError::MessageTooShort { + expected: offset + 2, + got: payload.len(), + }); + } + let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize; + offset += 2; + + if payload.len() < offset + hs_len { + return Err(ProtocolError::MessageTooShort { + expected: offset + hs_len, + got: payload.len(), + }); + } + let handshake_payload = payload[offset..offset + hs_len].to_vec(); + + Ok(Self { + src_coords, + dest_coords, + flags, + handshake_payload, + }) + } } // ============================================================================ @@ -164,17 +331,92 @@ impl SessionSetup { /// /// Carried inside a SessionDatagram envelope which provides src_addr and /// dest_addr. The SessionAck payload contains the acknowledger's coordinates -/// for route cache warming. +/// for route cache warming and the Noise IK handshake response. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|------------------|---------|-------------------------------------| +/// | 0 | msg_type | 1 byte | 0x01 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | src_coords_count | 2 bytes | u16 LE | +/// | 4 | src_coords | 16 × n | Acknowledger's coords (for caching) | +/// | ... | handshake_len | 2 bytes | u16 LE, Noise payload length | +/// | ... | handshake_payload| variable| Noise IK msg2 (33 bytes typical) | #[derive(Clone, Debug)] pub struct SessionAck { /// Acknowledger's coordinates. pub src_coords: TreeCoordinate, + /// Reserved flags byte (for forward compatibility). + pub flags: u8, + /// Noise IK handshake message 2. + pub handshake_payload: Vec, } impl SessionAck { /// Create a new session acknowledgement. pub fn new(src_coords: TreeCoordinate) -> Self { - Self { src_coords } + Self { + src_coords, + flags: 0, + handshake_payload: Vec::new(), + } + } + + /// Set the Noise handshake payload. + pub fn with_handshake(mut self, payload: Vec) -> Self { + self.handshake_payload = payload; + self + } + + /// Encode as wire format (msg_type + flags + coords + handshake). + pub fn encode(&self) -> Vec { + let mut buf = Vec::new(); + buf.push(SessionMessageType::SessionAck.to_byte()); + buf.push(self.flags); + encode_coords(&self.src_coords, &mut buf); + let hs_len = self.handshake_payload.len() as u16; + buf.extend_from_slice(&hs_len.to_le_bytes()); + buf.extend_from_slice(&self.handshake_payload); + buf + } + + /// Decode from wire format (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + if payload.is_empty() { + return Err(ProtocolError::MessageTooShort { + expected: 1, + got: 0, + }); + } + let flags = payload[0]; + let mut offset = 1; + + let (src_coords, consumed) = decode_coords(&payload[offset..])?; + offset += consumed; + + if payload.len() < offset + 2 { + return Err(ProtocolError::MessageTooShort { + expected: offset + 2, + got: payload.len(), + }); + } + let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize; + offset += 2; + + if payload.len() < offset + hs_len { + return Err(ProtocolError::MessageTooShort { + expected: offset + hs_len, + got: payload.len(), + }); + } + let handshake_payload = payload[offset..offset + hs_len].to_vec(); + + Ok(Self { + src_coords, + flags, + handshake_payload, + }) } } @@ -246,20 +488,37 @@ pub const DATA_HEADER_SIZE: usize = 4; /// Encrypted application data carried inside a SessionDatagram. /// -/// The 4-byte header contains: -/// - msg_type (1 byte): 0x10 -/// - flags (1 byte): COORDS_PRESENT, etc. -/// - payload_length (2 bytes) +/// ## Wire Format (minimal, no coordinates) /// -/// Addressing (src_addr, dest_addr) and hop_limit are provided by the -/// enclosing SessionDatagram envelope. The total on-wire overhead for a -/// minimal data packet is 34 (SessionDatagram) + 4 (DataPacket) = 38 bytes. +/// | Offset | Field | Size | Description | +/// |--------|----------------|---------|----------------------------| +/// | 0 | msg_type | 1 byte | 0x10 | +/// | 1 | flags | 1 byte | Bit 0: COORDS_PRESENT | +/// | 2 | payload_length | 2 bytes | u16 LE | +/// | 4 | payload | variable| Encrypted application data | +/// +/// ## Wire Format (with coordinates, when COORDS_PRESENT is set) +/// +/// | Offset | Field | Size | Description | +/// |--------|------------------|---------|----------------------------| +/// | 0 | msg_type | 1 byte | 0x10 | +/// | 1 | flags | 1 byte | 0x01 (COORDS_PRESENT) | +/// | 2 | payload_length | 2 bytes | u16 LE | +/// | 4 | src_coords_count | 2 bytes | u16 LE | +/// | 6 | src_coords | 16 × n | Source coordinates | +/// | ... | dest_coords_count| 2 bytes | u16 LE | +/// | ... | dest_coords | 16 × m | Destination coordinates | +/// | ... | payload | variable| Encrypted application data | #[derive(Clone, Debug)] pub struct DataPacket { /// Packet flags. pub flags: DataFlags, /// Payload data (end-to-end encrypted application data). pub payload: Vec, + /// Source coordinates (present when COORDS_PRESENT flag is set). + pub src_coords: Option, + /// Destination coordinates (present when COORDS_PRESENT flag is set). + pub dest_coords: Option, } impl DataPacket { @@ -268,6 +527,8 @@ impl DataPacket { Self { flags: DataFlags::new(), payload, + src_coords: None, + dest_coords: None, } } @@ -277,20 +538,99 @@ impl DataPacket { self } + /// Set coordinates for route cache warming. + pub fn with_coords(mut self, src: TreeCoordinate, dest: TreeCoordinate) -> Self { + self.src_coords = Some(src); + self.dest_coords = Some(dest); + self.flags.coords_present = true; + self + } + /// Get the payload length. pub fn payload_len(&self) -> usize { self.payload.len() } - /// Total packet size (header + payload). + /// Total packet size (header + optional coords + payload). pub fn total_size(&self) -> usize { - DATA_HEADER_SIZE + self.payload.len() + DATA_HEADER_SIZE + self.coords_wire_size() + self.payload.len() } /// Header size in bytes. pub fn header_size(&self) -> usize { DATA_HEADER_SIZE } + + /// Wire size of the optional coordinate fields. + fn coords_wire_size(&self) -> usize { + if !self.flags.coords_present { + return 0; + } + let src_count = self.src_coords.as_ref().map_or(0, |c| c.depth() + 1); + let dest_count = self.dest_coords.as_ref().map_or(0, |c| c.depth() + 1); + 2 + src_count * 16 + 2 + dest_count * 16 + } + + /// Encode as wire format. + pub fn encode(&self) -> Vec { + let mut buf = Vec::new(); + buf.push(SessionMessageType::DataPacket.to_byte()); + buf.push(self.flags.to_byte()); + let payload_len = self.payload.len() as u16; + buf.extend_from_slice(&payload_len.to_le_bytes()); + if self.flags.coords_present { + if let Some(ref src) = self.src_coords { + encode_coords(src, &mut buf); + } else { + encode_empty_coords(&mut buf); + } + if let Some(ref dest) = self.dest_coords { + encode_coords(dest, &mut buf); + } else { + encode_empty_coords(&mut buf); + } + } + buf.extend_from_slice(&self.payload); + buf + } + + /// Decode from wire format (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + if payload.len() < 3 { + return Err(ProtocolError::MessageTooShort { + expected: 3, + got: payload.len(), + }); + } + let flags = DataFlags::from_byte(payload[0]); + let payload_len = u16::from_le_bytes([payload[1], payload[2]]) as usize; + let mut offset = 3; + + let (src_coords, dest_coords) = if flags.coords_present { + let (src, consumed) = decode_optional_coords(&payload[offset..])?; + offset += consumed; + let (dest, consumed) = decode_optional_coords(&payload[offset..])?; + offset += consumed; + (src, dest) + } else { + (None, None) + }; + + if payload.len() < offset + payload_len { + return Err(ProtocolError::MessageTooShort { + expected: offset + payload_len, + got: payload.len(), + }); + } + let data = payload[offset..offset + payload_len].to_vec(); + + Ok(Self { + flags, + payload: data, + src_coords, + dest_coords, + }) + } } // ============================================================================ @@ -305,6 +645,17 @@ impl DataPacket { /// (src_addr=reporter, dest_addr=original_source). Plaintext — not /// end-to-end encrypted, since the transit router has no session with /// the source. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|----------|---------|------------------------------------| +/// | 0 | msg_type | 1 byte | 0x20 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr| 16 bytes| The node_addr we couldn't route to | +/// | 18 | reporter | 16 bytes| NodeAddr of reporting router | +/// +/// Payload: 34 bytes #[derive(Clone, Debug)] pub struct CoordsRequired { /// Destination that couldn't be routed. @@ -313,11 +664,45 @@ pub struct CoordsRequired { pub reporter: NodeAddr, } +/// Wire size of CoordsRequired payload: msg_type(1) + flags(1) + dest_addr(16) + reporter(16). +pub const COORDS_REQUIRED_SIZE: usize = 34; + impl CoordsRequired { /// Create a new CoordsRequired error. pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self { Self { dest_addr, reporter } } + + /// Encode as wire format. + pub fn encode(&self) -> Vec { + let mut buf = Vec::with_capacity(COORDS_REQUIRED_SIZE); + buf.push(SessionMessageType::CoordsRequired.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 + } + + /// Decode from wire format (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) = 33 + if payload.len() < 33 { + return Err(ProtocolError::MessageTooShort { + expected: 33, + 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]); + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + }) + } } /// Error indicating routing failure (local minimum or unreachable). @@ -326,6 +711,17 @@ impl CoordsRequired { /// The reporting router creates a new SessionDatagram with src_addr=reporter /// and dest_addr=original_source, so the `original_src` field from the old /// design is no longer needed — it's the SessionDatagram's dest_addr. +/// +/// ## Wire Format +/// +/// | Offset | Field | Size | Description | +/// |--------|-------------------|----------|-------------------------------| +/// | 0 | msg_type | 1 byte | 0x21 | +/// | 1 | flags | 1 byte | Reserved | +/// | 2 | dest_addr | 16 bytes | The unreachable node_addr | +/// | 18 | reporter | 16 bytes | NodeAddr of reporting router | +/// | 34 | last_coords_count | 2 bytes | u16 LE | +/// | 36 | last_known_coords | 16 × n | Stale coords that failed | #[derive(Clone, Debug)] pub struct PathBroken { /// Destination that couldn't be reached. @@ -351,6 +747,45 @@ impl PathBroken { self.last_known_coords = Some(coords); self } + + /// Encode as wire format. + pub fn encode(&self) -> Vec { + let mut buf = Vec::new(); + buf.push(SessionMessageType::PathBroken.to_byte()); + buf.push(0x00); // reserved flags + buf.extend_from_slice(self.dest_addr.as_bytes()); + buf.extend_from_slice(self.reporter.as_bytes()); + if let Some(ref coords) = self.last_known_coords { + encode_coords(coords, &mut buf); + } else { + encode_empty_coords(&mut buf); + } + buf + } + + /// Decode from wire format (after msg_type byte has been consumed). + pub fn decode(payload: &[u8]) -> Result { + // flags(1) + dest_addr(16) + reporter(16) + coords_count(2) = 35 minimum + 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 (last_known_coords, _consumed) = decode_optional_coords(&payload[33..])?; + + Ok(Self { + dest_addr: NodeAddr::from_bytes(dest_bytes), + reporter: NodeAddr::from_bytes(reporter_bytes), + last_known_coords, + }) + } } #[cfg(test)] @@ -495,4 +930,183 @@ mod tests { assert_eq!(err.reporter, make_node_addr(3)); assert!(err.last_known_coords.is_some()); } + + // ===== Encode/Decode Roundtrip Tests ===== + + #[test] + fn test_session_setup_encode_decode() { + let handshake = vec![0xAA; 82]; // typical Noise IK msg1 + let setup = SessionSetup::new(make_coords(&[1, 2, 0]), make_coords(&[3, 4, 0])) + .with_flags(SessionFlags::new().with_ack().bidirectional()) + .with_handshake(handshake.clone()); + + let encoded = setup.encode(); + + // Verify msg_type byte + assert_eq!(encoded[0], 0x00); + + // Decode (skip msg_type byte) + let decoded = SessionSetup::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.flags, setup.flags); + assert_eq!(decoded.src_coords, setup.src_coords); + assert_eq!(decoded.dest_coords, setup.dest_coords); + assert_eq!(decoded.handshake_payload, handshake); + } + + #[test] + fn test_session_setup_no_handshake() { + let setup = SessionSetup::new(make_coords(&[5, 0]), make_coords(&[6, 0])); + + let encoded = setup.encode(); + let decoded = SessionSetup::decode(&encoded[1..]).unwrap(); + + assert!(decoded.handshake_payload.is_empty()); + assert_eq!(decoded.src_coords, setup.src_coords); + assert_eq!(decoded.dest_coords, setup.dest_coords); + } + + #[test] + fn test_session_ack_encode_decode() { + let handshake = vec![0xBB; 33]; // typical Noise IK msg2 + let ack = SessionAck::new(make_coords(&[7, 8, 0])) + .with_handshake(handshake.clone()); + + let encoded = ack.encode(); + assert_eq!(encoded[0], 0x01); + + let decoded = SessionAck::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.src_coords, ack.src_coords); + assert_eq!(decoded.handshake_payload, handshake); + } + + #[test] + fn test_data_packet_encode_decode_minimal() { + let data = vec![1, 2, 3, 4, 5]; + let packet = DataPacket::new(data.clone()); + + let encoded = packet.encode(); + assert_eq!(encoded[0], 0x10); // msg_type + assert_eq!(encoded[1], 0x00); // flags (no coords) + + let decoded = DataPacket::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.payload, data); + assert!(!decoded.flags.coords_present); + assert!(decoded.src_coords.is_none()); + assert!(decoded.dest_coords.is_none()); + } + + #[test] + fn test_data_packet_encode_decode_with_coords() { + let data = vec![0xFF; 100]; + let src = make_coords(&[1, 2, 0]); + let dest = make_coords(&[3, 4, 0]); + let packet = DataPacket::new(data.clone()) + .with_coords(src.clone(), dest.clone()); + + let encoded = packet.encode(); + assert_eq!(encoded[0], 0x10); + assert_eq!(encoded[1], 0x01); // COORDS_PRESENT + + let decoded = DataPacket::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.payload, data); + assert!(decoded.flags.coords_present); + assert_eq!(decoded.src_coords.unwrap(), src); + assert_eq!(decoded.dest_coords.unwrap(), dest); + } + + #[test] + fn test_coords_required_encode_decode() { + let err = CoordsRequired::new(make_node_addr(0xAA), make_node_addr(0xBB)); + + let encoded = err.encode(); + assert_eq!(encoded.len(), COORDS_REQUIRED_SIZE); + assert_eq!(encoded[0], 0x20); + + let decoded = CoordsRequired::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + } + + #[test] + fn test_path_broken_encode_decode_no_coords() { + let err = PathBroken::new(make_node_addr(0xCC), make_node_addr(0xDD)); + + let encoded = err.encode(); + assert_eq!(encoded[0], 0x21); + + let decoded = PathBroken::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + assert!(decoded.last_known_coords.is_none()); + } + + #[test] + fn test_path_broken_encode_decode_with_coords() { + let coords = make_coords(&[0xCC, 0xDD, 0xEE]); + let err = PathBroken::new(make_node_addr(0x11), make_node_addr(0x22)) + .with_last_coords(coords.clone()); + + let encoded = err.encode(); + let decoded = PathBroken::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.dest_addr, err.dest_addr); + assert_eq!(decoded.reporter, err.reporter); + assert_eq!(decoded.last_known_coords.unwrap(), coords); + } + + #[test] + fn test_session_setup_decode_too_short() { + assert!(SessionSetup::decode(&[]).is_err()); + } + + #[test] + fn test_session_ack_decode_too_short() { + assert!(SessionAck::decode(&[]).is_err()); + } + + #[test] + fn test_data_packet_decode_too_short() { + assert!(DataPacket::decode(&[]).is_err()); + assert!(DataPacket::decode(&[0x00]).is_err()); + } + + #[test] + fn test_coords_required_decode_too_short() { + assert!(CoordsRequired::decode(&[]).is_err()); + assert!(CoordsRequired::decode(&[0x00; 10]).is_err()); + } + + #[test] + fn test_path_broken_decode_too_short() { + assert!(PathBroken::decode(&[]).is_err()); + assert!(PathBroken::decode(&[0x00; 20]).is_err()); + } + + #[test] + fn test_data_packet_large_payload() { + let data = vec![0x42; 65000]; + let packet = DataPacket::new(data.clone()); + + let encoded = packet.encode(); + let decoded = DataPacket::decode(&encoded[1..]).unwrap(); + assert_eq!(decoded.payload.len(), 65000); + assert_eq!(decoded.payload, data); + } + + #[test] + fn test_session_setup_deep_coords() { + // Depth-10 coordinate (11 entries: self + 10 ancestors) + let addrs: Vec = (0..11).collect(); + let src = make_coords(&addrs); + let dest = make_coords(&[20, 21, 22, 23, 24]); + let setup = SessionSetup::new(src.clone(), dest.clone()) + .with_handshake(vec![0x55; 82]); + + let encoded = setup.encode(); + let decoded = SessionSetup::decode(&encoded[1..]).unwrap(); + + assert_eq!(decoded.src_coords, src); + assert_eq!(decoded.dest_coords, dest); + } }