Session 65: RX loop integration test and session layer doc correction

Add test_run_rx_loop_handshake exercising the full packet dispatch
path (UDP → channel → run_rx_loop → process_packet → handler) using
tokio::select! with timeout to release &mut self for assertions.

Correct session layer documentation: sessions are always established
between communicating nodes regardless of adjacency, not only for
non-adjacent traffic. Add adjacent-peer encryption flow example,
fix table description, rename Link Layer subsection to "Hop-by-Hop".

267 tests pass (266 existing + 1 new).

docs/design/fips-intro.md

@@ -196,17 +196,17 @@ traffic.
 
 FIPS uses independent encryption at two layers:
 
-| Layer       | Scope       | Pattern  | Purpose                              |
-|-------------|-------------|----------|--------------------------------------|
-| **Link**    | Hop-by-hop  | Noise IK | Encrypt all traffic on each link    |
-| **Session** | End-to-end  | Noise IK | Encrypt payload across multiple hops |
+| Layer       | Scope       | Pattern  | Purpose                                        |
+|-------------|-------------|----------|------------------------------------------------|
+| **Link**    | Hop-by-hop  | Noise IK | Encrypt all traffic on each link               |
+| **Session** | End-to-end  | Noise IK | Encrypt application payload between endpoints  |
 
-### Link Layer (Peer-to-Peer)
+### Link Layer (Hop-by-Hop)
 
 When two nodes establish a direct connection, they perform a Noise IK handshake.
 This authenticates both parties and establishes symmetric keys for encrypting
 all traffic on that link. Every packet between direct peers is encrypted—gossip
-messages, routing queries, and forwarded traffic alike.
+messages, routing queries, and forwarded session datagrams alike.
 
 The IK pattern is used because outbound connections know the peer's npub from
 configuration, while inbound connections learn the initiator's identity from
@@ -214,15 +214,31 @@ the first handshake message.
 
 ### Session Layer (End-to-End)
 
-For traffic between non-adjacent nodes, FIPS establishes end-to-end encrypted
-sessions using Noise IK. The initiator knows the destination's npub; the
+FIPS establishes end-to-end encrypted sessions between any two communicating
+nodes using Noise IK, regardless of whether they are direct peers or separated
+by intermediate routers. The initiator knows the destination's npub; the
 responder learns the initiator's identity from the handshake—the same asymmetry
 as link-layer connections.
 
+Both layers always apply. For adjacent peers, application traffic is encrypted
+twice: once by the session layer (end-to-end) and once by the link layer
+(hop-by-hop). This uniform model means:
+
+- No special case for "local peer" vs "remote destination"
+- Topology changes (a direct peer becomes reachable only through intermediaries)
+  don't affect sessions
+- The link layer remains purely a transport concern
+
+A packet from A to adjacent peer B:
+
+1. A encrypts payload with A↔B session key
+2. A wraps in SessionDatagram, encrypts with A↔B link key, sends to B
+3. B decrypts link layer, then decrypts session layer to get payload
+
 A packet from A to D through intermediate node B:
 
 1. A encrypts payload with A↔D session key
-2. A encrypts that with A↔B link key, sends to B
+2. A wraps in SessionDatagram, encrypts with A↔B link key, sends to B
 3. B decrypts link layer, sees destination, re-encrypts with B↔D link key
 4. D decrypts link layer, then decrypts session layer to get payload
 

src/node.rs

@@ -2404,4 +2404,194 @@ mod tests {
             t.stop().await.ok();
         }
     }
+
+    /// Integration test: two nodes complete a handshake via run_rx_loop.
+    ///
+    /// Unlike test_two_node_handshake_udp which calls handle_msg1/handle_msg2
+    /// directly, this test exercises the full rx loop dispatch path:
+    /// UDP socket → packet channel → run_rx_loop → process_packet →
+    /// discriminator dispatch → handler.
+    #[tokio::test]
+    async fn test_run_rx_loop_handshake() {
+        use crate::config::UdpConfig;
+        use crate::transport::udp::UdpTransport;
+        use crate::wire::build_msg1;
+        use tokio::time::Duration;
+
+        // === Setup: Two nodes with UDP transports on localhost ===
+
+        let mut node_a = make_node();
+        let mut node_b = make_node();
+
+        let transport_id_a = TransportId::new(1);
+        let transport_id_b = TransportId::new(1);
+
+        let udp_config = UdpConfig {
+            bind_addr: Some("127.0.0.1:0".to_string()),
+            mtu: Some(1280),
+        };
+
+        let (packet_tx_a, packet_rx_a) = packet_channel(64);
+        let (packet_tx_b, packet_rx_b) = packet_channel(64);
+
+        let mut transport_a =
+            UdpTransport::new(transport_id_a, None, udp_config.clone(), packet_tx_a);
+        let mut transport_b =
+            UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b);
+
+        transport_a.start_async().await.unwrap();
+        transport_b.start_async().await.unwrap();
+
+        let addr_b = transport_b.local_addr().unwrap();
+        let remote_addr_b = TransportAddr::from_string(&addr_b.to_string());
+
+        node_a
+            .transports
+            .insert(transport_id_a, TransportHandle::Udp(transport_a));
+        node_b
+            .transports
+            .insert(transport_id_b, TransportHandle::Udp(transport_b));
+
+        // Store packet_rx on nodes for run_rx_loop
+        node_a.packet_rx = Some(packet_rx_a);
+        node_b.packet_rx = Some(packet_rx_b);
+
+        // Set node state to Running (transports need to be operational)
+        node_a.state = NodeState::Running;
+        node_b.state = NodeState::Running;
+
+        // === Phase 1: Node A initiates handshake to Node B ===
+
+        let peer_b_identity =
+            PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
+        let peer_b_node_addr = *peer_b_identity.node_addr();
+
+        let link_id_a = node_a.allocate_link_id();
+        let mut conn_a = PeerConnection::outbound(
+            link_id_a,
+            peer_b_identity.clone(),
+            1000,
+        );
+
+        let our_index_a = node_a.index_allocator.allocate().unwrap();
+        let our_keypair_a = node_a.identity.keypair();
+        let noise_msg1 = conn_a.start_handshake(our_keypair_a, 1000).unwrap();
+        conn_a.set_our_index(our_index_a);
+        conn_a.set_transport_id(transport_id_a);
+        conn_a.set_source_addr(remote_addr_b.clone());
+
+        let wire_msg1 = build_msg1(our_index_a, &noise_msg1);
+
+        let link_a = Link::connectionless(
+            link_id_a,
+            transport_id_a,
+            remote_addr_b.clone(),
+            LinkDirection::Outbound,
+            Duration::from_millis(100),
+        );
+        node_a.links.insert(link_id_a, link_a);
+        node_a.connections.insert(link_id_a, conn_a);
+        node_a.pending_outbound.insert(
+            (transport_id_a, our_index_a.as_u32()),
+            link_id_a,
+        );
+
+        // Send msg1 from A to B over real UDP
+        let transport = node_a.transports.get(&transport_id_a).unwrap();
+        transport
+            .send(&remote_addr_b, &wire_msg1)
+            .await
+            .expect("Failed to send msg1");
+
+        // Small delay to ensure msg1 is received by B's transport
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        // === Phase 2: Run Node B's rx loop (processes msg1, sends msg2) ===
+        //
+        // This is the key difference from test_two_node_handshake_udp:
+        // instead of calling handle_msg1() directly, we run the full rx loop
+        // which dispatches based on the discriminator byte.
+
+        tokio::select! {
+            result = node_b.run_rx_loop() => {
+                panic!("Node B rx loop exited unexpectedly: {:?}", result);
+            }
+            _ = tokio::time::sleep(Duration::from_millis(500)) => {
+                // Timeout: rx loop processed available packets
+            }
+        }
+
+        // Verify Node B promoted the inbound connection via rx loop dispatch
+        let peer_a_node_addr = *PeerIdentity::from_pubkey_full(
+            node_a.identity.pubkey_full(),
+        )
+        .node_addr();
+
+        assert_eq!(node_b.peer_count(), 1, "Node B should have 1 peer after rx loop processed msg1");
+        let peer_a_on_b = node_b
+            .get_peer(&peer_a_node_addr)
+            .expect("Node B should have peer A");
+        assert!(
+            peer_a_on_b.has_session(),
+            "Peer A on B should have NoiseSession"
+        );
+        let our_index_b = peer_a_on_b.our_index().expect("B should have our_index");
+        assert!(
+            peer_a_on_b.their_index().is_some(),
+            "B should have their_index"
+        );
+        assert!(
+            node_b
+                .peers_by_index
+                .contains_key(&(transport_id_b, our_index_b.as_u32())),
+            "Node B peers_by_index should be populated"
+        );
+
+        // === Phase 3: Run Node A's rx loop (processes msg2) ===
+        //
+        // msg2 was sent by Node B during its rx loop processing of msg1.
+        // It arrived at A's UDP transport, which forwarded it to A's packet channel.
+
+        tokio::select! {
+            result = node_a.run_rx_loop() => {
+                panic!("Node A rx loop exited unexpectedly: {:?}", result);
+            }
+            _ = tokio::time::sleep(Duration::from_millis(500)) => {
+                // Timeout: rx loop processed msg2
+            }
+        }
+
+        // Verify Node A promoted the outbound connection via rx loop dispatch
+        assert_eq!(node_a.peer_count(), 1, "Node A should have 1 peer after rx loop processed msg2");
+        let peer_b_on_a = node_a
+            .get_peer(&peer_b_node_addr)
+            .expect("Node A should have peer B");
+        assert!(
+            peer_b_on_a.has_session(),
+            "Peer B on A should have NoiseSession"
+        );
+        assert_eq!(
+            peer_b_on_a.our_index(),
+            Some(our_index_a),
+            "Peer B on A should have our_index matching what we allocated"
+        );
+        assert!(
+            peer_b_on_a.their_index().is_some(),
+            "A should know B's index"
+        );
+        assert!(
+            node_a
+                .peers_by_index
+                .contains_key(&(transport_id_a, our_index_a.as_u32())),
+            "Node A peers_by_index should be populated"
+        );
+
+        // Clean up transports
+        for (_, t) in node_a.transports.iter_mut() {
+            t.stop().await.ok();
+        }
+        for (_, t) in node_b.transports.iter_mut() {
+            t.stop().await.ok();
+        }
+    }
 }