#[test]
fn session_runtime_receive_owns_fsp_open_bookkeeping_and_dispatch_metadata() {
let local = Identity::generate();
let peer = Identity::generate();
let (local_session, mut peer_sender) = make_xk_session_pair(&local, &peer);
let mut entry = SessionEntry::new(
*peer.node_addr(),
peer.pubkey_full(),
EndToEndState::Established(local_session),
1_000,
true,
);
entry.mark_established(1_000);
entry.touch_inbound_frame(1_000);
entry.record_decrypt_failure();
let endpoint_payload = b"endpoint runtime receive".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let counter = peer_sender.current_send_counter();
let header = build_fsp_header(counter, 0, plaintext.len() as u16);
let ciphertext = peer_sender
.encrypt_with_aad(&plaintext, &header)
.expect("test frame should encrypt");
let mut wire = header.to_vec();
wire.extend_from_slice(&ciphertext);
let parsed = FspEncryptedHeader::parse(&wire).expect("test frame should parse");
let outcome = SessionRuntimeReceive::new(
&mut entry,
&parsed,
&wire[FSP_HEADER_SIZE..],
1_280,
true,
2_000,
)
.open_established();
match outcome {
FspFrameOutcome::Authentic(message) => {
assert_eq!(message.source_peer().node_addr(), peer.node_addr());
assert_eq!(message.plaintext(), plaintext);
assert_eq!(
message.msg_type(),
SessionMessageType::EndpointData.to_byte()
);
assert_eq!(message.inner_flags_byte(), 0);
assert_eq!(message.timestamp(), 0x0102_0304);
assert_eq!(message.body(), endpoint_payload);
assert!(message.is_application_data());
}
other => panic!("expected authentic FSP frame, got {other:?}"),
}
assert_eq!(entry.consecutive_decrypt_failures(), 0);
assert_eq!(entry.last_inbound_frame_ms(), 2_000);
}
#[test]
fn direct_session_receive_clears_direct_probe_retry_and_degradation() {
use crate::PeerIdentity;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::retry::RetryState;
use crate::peer::ActivePeer;
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
let local = Identity::generate();
let peer = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let peer_addr = *peer_identity.node_addr();
let peer_config = PeerConfig {
npub: peer.npub(),
alias: None,
addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
discovery_fallback_transit: true,
};
let mut config = crate::config::Config::new();
config.peers.push(peer_config.clone());
let mut node = Node::with_identity(local, config).expect("node");
node.config.node.heartbeat_interval_secs = 10;
let mut active_peer = ActivePeer::with_session(
peer_identity,
LinkId::new(9),
1_000,
make_xk_session(&node.identity, &peer),
SessionIndex::new(0x1010),
SessionIndex::new(0x2020),
TransportId::new(0x55),
TransportAddr::from_string("198.51.100.20:61062"),
LinkStats::new(),
true,
&node.config.node.mmp,
None,
);
active_peer.touch(Node::now_ms().saturating_sub(11_000));
node.peers
.insert_with_current_session_index(peer_addr, active_peer);
let mut session = SessionEntry::new(
peer_addr,
peer.pubkey_full(),
EndToEndState::Established(make_xk_session(&node.identity, &peer)),
1_000,
true,
);
session.record_outbound_next_hop(peer_addr);
node.sessions.insert(peer_addr, session);
let mut retry = RetryState::new(peer_config);
retry.reconnect = true;
node.retry_pending.insert(peer_addr, retry);
node.mark_session_direct_path_degraded(peer_addr, Node::now_ms());
assert!(
node.session_direct_path_blocks_direct_payload(&peer_addr, Node::now_ms()),
"fixture should start with the direct path withheld from payload routing"
);
SessionDispatchCommit {
source_addr: peer_addr,
receive_completion: Some(SessionReceiveCompletion {
source_addr: peer_addr,
previous_hop_addr: peer_addr,
body_len: 512,
direct_path: true,
}),
}
.finish_receive(&mut node);
assert!(
!node.retry_pending.contains_key(&peer_addr),
"fresh authenticated payload return on the direct peer path should stop direct-probe churn"
);
assert!(
!node.session_direct_path_blocks_direct_payload(&peer_addr, Node::now_ms()),
"fresh authenticated endpoint data from the direct peer should restore payload routing"
);
}
#[test]
fn direct_session_receive_clears_direct_probe_retry_without_route_marker() {
use crate::PeerIdentity;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::retry::RetryState;
use crate::peer::ActivePeer;
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
let local = Identity::generate();
let peer = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let peer_addr = *peer_identity.node_addr();
let peer_config = PeerConfig {
npub: peer.npub(),
alias: None,
addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
discovery_fallback_transit: true,
};
let mut config = crate::config::Config::new();
config.peers.push(peer_config.clone());
let mut node = Node::with_identity(local, config).expect("node");
node.config.node.heartbeat_interval_secs = 10;
let mut active_peer = ActivePeer::with_session(
peer_identity,
LinkId::new(9),
1_000,
make_xk_session(&node.identity, &peer),
SessionIndex::new(0x1010),
SessionIndex::new(0x2020),
TransportId::new(0x55),
TransportAddr::from_string("198.51.100.20:61062"),
LinkStats::new(),
true,
&node.config.node.mmp,
None,
);
active_peer.touch(Node::now_ms().saturating_sub(11_000));
node.peers
.insert_with_current_session_index(peer_addr, active_peer);
let session = SessionEntry::new(
peer_addr,
peer.pubkey_full(),
EndToEndState::Established(make_xk_session(&node.identity, &peer)),
1_000,
true,
);
node.sessions.insert(peer_addr, session);
let mut retry = RetryState::new(peer_config);
retry.reconnect = true;
node.retry_pending.insert(peer_addr, retry);
SessionDispatchCommit {
source_addr: peer_addr,
receive_completion: Some(SessionReceiveCompletion {
source_addr: peer_addr,
previous_hop_addr: peer_addr,
body_len: 512,
direct_path: true,
}),
}
.finish_receive(&mut node);
assert!(
!node.retry_pending.contains_key(&peer_addr),
"authenticated endpoint data from the direct peer itself should refresh the direct path even if outbound next-hop bookkeeping was absent"
);
}
#[test]
fn application_receive_refreshes_previous_hop_peer_without_direct_source_trust() {
use crate::PeerIdentity;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::retry::RetryState;
use crate::peer::ActivePeer;
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
let local = Identity::generate();
let source = Identity::generate();
let previous_hop = Identity::generate();
let previous_hop_identity = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
let previous_hop_addr = *previous_hop_identity.node_addr();
let source_addr = *source.node_addr();
let previous_hop_config = PeerConfig {
npub: previous_hop.npub(),
alias: None,
addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
discovery_fallback_transit: true,
};
let mut config = crate::config::Config::new();
config.peers.push(previous_hop_config.clone());
let mut node = Node::with_identity(local, config).expect("node");
node.config.node.heartbeat_interval_secs = 10;
let stale_seen_ms = Node::now_ms().saturating_sub(11_000);
let mut active_peer = ActivePeer::with_session(
previous_hop_identity,
LinkId::new(9),
1_000,
make_xk_session(&node.identity, &previous_hop),
SessionIndex::new(0x1010),
SessionIndex::new(0x2020),
TransportId::new(0x55),
TransportAddr::from_string("203.0.113.9:2121"),
LinkStats::new(),
true,
&node.config.node.mmp,
None,
);
active_peer.touch(stale_seen_ms);
node.peers
.insert_with_current_session_index(previous_hop_addr, active_peer);
let session = SessionEntry::new(
source_addr,
source.pubkey_full(),
EndToEndState::Established(make_xk_session(&node.identity, &source)),
1_000,
true,
);
node.sessions.insert(source_addr, session);
let mut retry = RetryState::new(previous_hop_config);
retry.reconnect = true;
node.retry_pending.insert(previous_hop_addr, retry);
SessionDispatchCommit {
source_addr,
receive_completion: Some(SessionReceiveCompletion {
source_addr,
previous_hop_addr,
body_len: 512,
direct_path: false,
}),
}
.finish_receive(&mut node);
let previous_hop_peer = node
.peers
.get(&previous_hop_addr)
.expect("previous hop should remain active");
assert!(
previous_hop_peer.idle_time(Node::now_ms()) <= 1_000,
"accepted application data should refresh the direct previous-hop link"
);
assert!(
!node.retry_pending.contains_key(&previous_hop_addr),
"fresh authenticated data from the direct previous hop should stop link refresh churn"
);
let entry = node
.sessions
.get(&source_addr)
.expect("source session should remain");
assert_eq!(
entry.last_inbound_data_frame_ms(),
1_000,
"previous-hop liveness must not become direct-source payload trust"
);
}
#[test]
fn authenticated_fmp_receive_clears_direct_probe_retry_on_direct_path() {
use crate::PeerIdentity;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::decrypt_worker::DecryptFmpBookkeeping;
use crate::node::retry::RetryState;
use crate::peer::ActivePeer;
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
let local = Identity::generate();
let peer = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let peer_addr = *peer_identity.node_addr();
let peer_config = PeerConfig {
npub: peer.npub(),
alias: None,
addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
discovery_fallback_transit: true,
};
let transport_id = TransportId::new(0x56);
let transport_addr = TransportAddr::from_string("198.51.100.20:61062");
let mut config = crate::config::Config::new();
config.peers.push(peer_config.clone());
let mut node = Node::with_identity(local, config).expect("node");
node.config.node.heartbeat_interval_secs = 10;
let mut active_peer = ActivePeer::with_session(
peer_identity,
LinkId::new(9),
1_000,
make_xk_session(&node.identity, &peer),
SessionIndex::new(0x1011),
SessionIndex::new(0x2021),
transport_id,
transport_addr.clone(),
LinkStats::new(),
true,
&node.config.node.mmp,
None,
);
active_peer.touch(Node::now_ms().saturating_sub(11_000));
node.peers
.insert_with_current_session_index(peer_addr, active_peer);
let mut retry = RetryState::new(peer_config);
retry.reconnect = true;
node.retry_pending.insert(peer_addr, retry);
node.record_worker_authenticated_fmp_receive(
&DecryptFmpBookkeeping {
source_peer: peer_identity,
transport_id,
remote_addr: transport_addr,
packet_timestamp_ms: Node::now_ms(),
packet_len: 256,
fmp_counter: 11,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
Some(&peer_addr),
);
assert!(
!node.retry_pending.contains_key(&peer_addr),
"fresh authenticated FMP return on the direct peer path should stop direct-probe churn"
);
}
#[test]
fn authenticated_fmp_receive_keeps_direct_probe_retry_for_forwarded_path() {
use crate::PeerIdentity;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::decrypt_worker::DecryptFmpBookkeeping;
use crate::node::retry::RetryState;
use crate::peer::ActivePeer;
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
let local = Identity::generate();
let peer = Identity::generate();
let relay = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let peer_addr = *peer_identity.node_addr();
let relay_addr = *relay.node_addr();
let peer_config = PeerConfig {
npub: peer.npub(),
alias: None,
addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
connect_policy: ConnectPolicy::AutoConnect,
auto_reconnect: true,
discovery_fallback_transit: true,
};
let transport_id = TransportId::new(0x57);
let transport_addr = TransportAddr::from_string("198.51.100.20:61062");
let mut config = crate::config::Config::new();
config.peers.push(peer_config.clone());
let mut node = Node::with_identity(local, config).expect("node");
node.config.node.heartbeat_interval_secs = 10;
let mut active_peer = ActivePeer::with_session(
peer_identity,
LinkId::new(9),
1_000,
make_xk_session(&node.identity, &peer),
SessionIndex::new(0x1012),
SessionIndex::new(0x2022),
transport_id,
transport_addr.clone(),
LinkStats::new(),
true,
&node.config.node.mmp,
None,
);
active_peer.touch(Node::now_ms().saturating_sub(11_000));
node.peers
.insert_with_current_session_index(peer_addr, active_peer);
let mut retry = RetryState::new(peer_config);
retry.reconnect = true;
node.retry_pending.insert(peer_addr, retry);
node.record_worker_authenticated_fmp_receive(
&DecryptFmpBookkeeping {
source_peer: peer_identity,
transport_id,
remote_addr: transport_addr,
packet_timestamp_ms: Node::now_ms(),
packet_len: 256,
fmp_counter: 11,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
Some(&relay_addr),
);
assert!(
node.retry_pending.contains_key(&peer_addr),
"forwarded authenticated FMP traffic must not prove the direct peer path is healthy"
);
}
#[test]
fn fsp_receive_sync_requests_worker_refresh_only_on_epoch_promotion() {
let local = Identity::generate();
let peer = Identity::generate();
let mut entry = established_entry(&local, &peer);
entry.mark_established(1_000);
let initial_k_bit = entry.current_k_bit();
entry.set_pending_session(make_xk_session(&local, &peer));
let promoted = entry.apply_fsp_receive_sync_result(
crate::node::session::FspReceiveSync {
counter: 7,
slot: EpochSlot::Pending,
received_k_bit: !initial_k_bit,
timestamp: 0x0102_0304,
plaintext_len: FSP_INNER_HEADER_SIZE + 16,
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
2_000,
Instant::now(),
);
assert!(promoted.is_applied());
assert!(
promoted.refresh_worker_session(),
"pending promotion changes FSP epoch topology and must refresh worker snapshots"
);
assert_eq!(entry.current_k_bit(), !initial_k_bit);
assert!(entry.pending_new_session().is_none());
assert!(entry.previous_highest_counter().is_some());
let current = entry.apply_fsp_receive_sync_result(
crate::node::session::FspReceiveSync {
counter: 8,
slot: EpochSlot::Current,
received_k_bit: entry.current_k_bit(),
timestamp: 0x0102_0305,
plaintext_len: FSP_INNER_HEADER_SIZE + 16,
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
2_100,
Instant::now(),
);
assert!(current.is_applied());
assert!(
!current.refresh_worker_session(),
"ordinary current-epoch replay mirroring must not re-register workers"
);
}
#[test]
fn established_fsp_wire_owns_ciphertext_offset_and_coord_warmup() {
use crate::tree::TreeCoordinate;
let source_addr = node_addr(0x01);
let local_addr = node_addr(0x02);
let root_addr = node_addr(0xf0);
let source_coords = TreeCoordinate::from_addrs(vec![source_addr, root_addr]).unwrap();
let local_coords = TreeCoordinate::from_addrs(vec![local_addr, root_addr]).unwrap();
let header = build_fsp_header(42, FSP_FLAG_CP | FSP_FLAG_K, 20);
let mut wire_bytes = header.to_vec();
encode_coords(&source_coords, &mut wire_bytes);
encode_coords(&local_coords, &mut wire_bytes);
let ciphertext_offset = wire_bytes.len();
wire_bytes.extend_from_slice(&[0xcc; 36]);
let mut wire = EstablishedFspWire::parse(&wire_bytes, source_addr, local_addr)
.expect("wire should parse with coord warmup");
assert_eq!(wire.header.counter, 42);
assert_eq!(wire.header.flags, FSP_FLAG_CP | FSP_FLAG_K);
assert_eq!(wire.ciphertext, &wire_bytes[ciphertext_offset..]);
assert!(wire.has_coord_warmup());
let receive = wire.receive(1_280, true, 2_000);
assert_eq!(receive.header.counter, 42);
assert_eq!(receive.ciphertext, &wire_bytes[ciphertext_offset..]);
assert_eq!(receive.path_mtu, 1_280);
assert!(receive.ce_flag);
assert_eq!(receive.now_ms, 2_000);
let mut coord_cache = crate::cache::CoordCache::new(16, 1_000);
wire.apply_coord_warmup(&mut coord_cache, 1_500);
assert!(!wire.has_coord_warmup());
assert_eq!(
coord_cache
.get(&source_addr, 1_500)
.expect("source coords should be cached")
.root_id(),
&root_addr
);
assert_eq!(
coord_cache
.get(&local_addr, 1_500)
.expect("local coords should be cached")
.root_id(),
&root_addr
);
assert!(matches!(
EstablishedFspWire::parse(&wire_bytes[..FSP_HEADER_SIZE - 1], source_addr, local_addr),
Err(EstablishedFspWireError::BadHeader)
));
let mut truncated_coords = build_fsp_header(43, FSP_FLAG_CP, 20).to_vec();
truncated_coords.extend_from_slice(&2u16.to_le_bytes());
truncated_coords.extend_from_slice(&[0u8; 14]);
assert!(matches!(
EstablishedFspWire::parse(&truncated_coords, source_addr, local_addr),
Err(EstablishedFspWireError::BadCoords(_))
));
}
#[test]
fn session_registry_owns_early_encrypted_handshake_resend_budget() {
let local = Identity::generate();
let peer = Identity::generate();
let handshake = HandshakeState::new_xk_initiator(local.keypair(), peer.pubkey_full());
let mut entry = SessionEntry::new(
*peer.node_addr(),
peer.pubkey_full(),
EndToEndState::Initiating(handshake),
1_000,
true,
);
let payload = vec![0x10, 0x20, 0x30];
entry.set_handshake_payload(payload.clone(), 1_500);
let mut sessions = crate::node::SessionRegistry::default();
sessions.insert(*peer.node_addr(), entry);
match sessions.prepare_handshake_resend_after_early_encrypted_data(peer.node_addr(), 2) {
EarlyEncryptedHandshakeResend::Resend { payload: resend } => {
assert_eq!(resend, payload);
}
other => panic!("expected resend decision, got {other:?}"),
}
assert!(sessions.record_handshake_resend(peer.node_addr(), 2_000));
let entry = sessions
.get(peer.node_addr())
.expect("session should remain");
assert_eq!(entry.resend_count(), 1);
assert_eq!(entry.next_resend_at_ms(), 2_000);
assert_eq!(entry.handshake_payload(), Some(payload.as_slice()));
assert!(matches!(
sessions.prepare_handshake_resend_after_early_encrypted_data(peer.node_addr(), 1),
EarlyEncryptedHandshakeResend::BudgetExhausted
));
let entry = sessions
.get(peer.node_addr())
.expect("session should remain");
assert!(entry.handshake_payload().is_none());
assert_eq!(entry.next_resend_at_ms(), 0);
assert_eq!(
entry.resend_count(),
1,
"clearing an exhausted payload must not rewrite resend history"
);
let missing = node_addr(0x77);
assert!(matches!(
sessions.prepare_handshake_resend_after_early_encrypted_data(&missing, 2),
EarlyEncryptedHandshakeResend::NoPayload
));
assert!(!sessions.record_handshake_resend(&missing, 3_000));
}
#[test]
fn session_registry_owns_established_fsp_open_lookup_and_bookkeeping() {
let local = Identity::generate();
let peer = Identity::generate();
let (local_session, mut peer_sender) = make_xk_session_pair(&local, &peer);
let mut entry = SessionEntry::new(
*peer.node_addr(),
peer.pubkey_full(),
EndToEndState::Established(local_session),
1_000,
true,
);
entry.mark_established(1_000);
entry.record_decrypt_failure();
let endpoint_payload = b"registry runtime receive".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let counter = peer_sender.current_send_counter();
let header = build_fsp_header(counter, 0, plaintext.len() as u16);
let ciphertext = peer_sender
.encrypt_with_aad(&plaintext, &header)
.expect("test frame should encrypt");
let mut wire = header.to_vec();
wire.extend_from_slice(&ciphertext);
let parsed = FspEncryptedHeader::parse(&wire).expect("test frame should parse");
let mut sessions = crate::node::SessionRegistry::default();
sessions.insert(*peer.node_addr(), entry);
let outcome = sessions.open_established_fsp_frame(
peer.node_addr(),
EstablishedFspReceive::new(&parsed, &wire[FSP_HEADER_SIZE..], 1_280, true, 2_000),
);
match outcome {
FspFrameOutcome::Authentic(message) => {
assert_eq!(message.source_peer().node_addr(), peer.node_addr());
assert_eq!(
message.msg_type(),
SessionMessageType::EndpointData.to_byte()
);
assert_eq!(message.body(), endpoint_payload);
}
other => panic!("expected authentic FSP frame, got {other:?}"),
}
let entry = sessions
.get(peer.node_addr())
.expect("session should remain");
assert_eq!(entry.consecutive_decrypt_failures(), 0);
assert_eq!(entry.last_inbound_frame_ms(), 2_000);
let missing = node_addr(0x77);
let outcome = sessions.open_established_fsp_frame(
&missing,
EstablishedFspReceive::new(&parsed, &wire[FSP_HEADER_SIZE..], 1_280, false, 2_001),
);
assert!(matches!(outcome, FspFrameOutcome::UnknownSession));
}
#[test]
fn authenticated_session_message_owns_endpoint_delivery_conversion() {
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let endpoint_payload = b"endpoint delivery".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let message = AuthenticatedSessionMessage::new(
source_peer,
plaintext,
SessionMessageType::EndpointData.to_byte(),
0,
0x0102_0304,
);
assert_eq!(message.body(), endpoint_payload);
let delivery = message.into_endpoint_data_delivery();
assert_eq!(delivery.source_peer, source_peer);
assert_eq!(delivery.payload, endpoint_payload);
}
#[test]
fn authenticated_session_message_can_own_plaintext_inside_wire_buffer() {
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let endpoint_payload = b"buffer endpoint delivery".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let mut buffer = b"outer-fmp-prefix".to_vec();
let plaintext_offset = buffer.len();
buffer.extend_from_slice(&plaintext);
buffer.extend_from_slice(b"outer-fmp-trailer");
let message = AuthenticatedSessionMessage::from_buffer(
source_peer,
buffer,
plaintext_offset,
plaintext.len(),
SessionMessageType::EndpointData.to_byte(),
0,
0x0102_0304,
);
assert_eq!(message.plaintext(), plaintext);
assert_eq!(message.body(), endpoint_payload);
let delivery = message.into_endpoint_data_delivery();
assert_eq!(delivery.source_peer, source_peer);
assert_eq!(delivery.payload, endpoint_payload);
}
#[test]
fn authenticated_session_dispatch_owns_route_ce_and_completion_facts() {
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let previous_hop_addr = node_addr(0x55);
let endpoint_payload = b"endpoint completion".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let dispatch = AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
true,
AuthenticatedSessionMessage::new(
source_peer,
plaintext,
SessionMessageType::EndpointData.to_byte(),
0,
0x0102_0304,
),
);
assert_eq!(dispatch.source_addr(), &source_addr);
assert_eq!(dispatch.previous_hop_addr(), &previous_hop_addr);
assert!(dispatch.ce_flag());
assert_eq!(
dispatch.msg_type(),
SessionMessageType::EndpointData.to_byte()
);
assert_eq!(dispatch.body(), endpoint_payload);
assert_eq!(
dispatch.receive_completion(),
Some(SessionReceiveCompletion {
source_addr,
previous_hop_addr,
body_len: endpoint_payload.len(),
direct_path: false,
})
);
let commit = dispatch.commit();
assert_eq!(commit.source_addr(), &source_addr);
assert_eq!(
commit.receive_completion(),
Some(SessionReceiveCompletion {
source_addr,
previous_hop_addr,
body_len: endpoint_payload.len(),
direct_path: false,
})
);
let local = Identity::generate();
let mut sessions = crate::node::SessionRegistry::default();
sessions.insert(source_addr, established_entry(&local, &peer));
assert!(commit.record_receive(&mut sessions, 0x0bad_cafe));
let entry = sessions.get(&source_addr).expect("session should remain");
assert_eq!(
entry.traffic_counters(),
(0, 1, 0, endpoint_payload.len() as u64)
);
assert_eq!(entry.last_activity(), 0x0bad_cafe);
assert_eq!(
entry.last_inbound_data_frame_ms(),
1000,
"relayed application data must not refresh direct-path trust"
);
let delivery = dispatch.into_endpoint_data_delivery();
assert_eq!(delivery.source_peer, source_peer);
assert_eq!(delivery.payload, endpoint_payload);
let report_plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::SenderReport.to_byte(),
0,
b"report",
);
let report_dispatch = AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
false,
AuthenticatedSessionMessage::new(
source_peer,
report_plaintext,
SessionMessageType::SenderReport.to_byte(),
0,
0x0102_0304,
),
);
assert_eq!(
report_dispatch.receive_completion(),
None,
"MMP reports must not reset session idle/traffic counters"
);
let report_commit = report_dispatch.commit();
assert_eq!(report_commit.source_addr(), &source_addr);
assert_eq!(
report_commit.receive_completion(),
None,
"MMP reports still flush pending packets without recording receive progress"
);
assert!(!report_commit.record_receive(&mut sessions, 0x0bad_f00d));
let entry = sessions.get(&source_addr).expect("session should remain");
assert_eq!(
entry.traffic_counters(),
(0, 1, 0, endpoint_payload.len() as u64)
);
assert_eq!(entry.last_activity(), 0x0bad_cafe);
}
#[test]
fn endpoint_data_fast_dispatch_finishes_receive_without_pending_flush() {
let local = Identity::generate();
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let previous_hop_addr = node_addr(0x55);
let endpoint_payload = b"fast endpoint delivery".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let dispatch = AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
false,
AuthenticatedSessionMessage::new(
source_peer,
plaintext,
SessionMessageType::EndpointData.to_byte(),
0,
0x0102_0304,
),
);
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
let finish = dispatch.dispatch_endpoint_data_fast(&mut node);
assert_eq!(finish.pending_flush_dest(), None);
match endpoint_io.event_rx.try_recv().expect("endpoint event") {
crate::node::NodeEndpointEvent::Data {
source_peer: delivered_source,
payload,
..
} => {
assert_eq!(delivered_source, source_peer);
assert_eq!(payload, endpoint_payload);
}
event => panic!("expected single endpoint data event, got {event:?}"),
}
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(
entry.traffic_counters(),
(0, 1, 0, endpoint_payload.len() as u64)
);
assert_eq!(
entry.last_inbound_data_frame_ms(),
1000,
"fast relayed endpoint data must not refresh direct-path trust"
);
assert!(
!node.pending_session_traffic.has_traffic_for(&source_addr),
"empty pending guard should keep the fast path synchronous"
);
}
#[test]
fn endpoint_data_fast_dispatch_reports_pending_flush_owner() {
let local = Identity::generate();
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let previous_hop_addr = node_addr(0x55);
let endpoint_payload = b"fast endpoint pending".to_vec();
let plaintext = fsp_prepend_inner_header(
0x0102_0304,
SessionMessageType::EndpointData.to_byte(),
0,
&endpoint_payload,
);
let dispatch = AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
false,
AuthenticatedSessionMessage::new(
source_peer,
plaintext,
SessionMessageType::EndpointData.to_byte(),
0,
0x0102_0304,
),
);
let mut node = Node::new(crate::config::Config::new()).expect("node");
let _endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
assert!(
!node
.pending_session_traffic
.push_endpoint_data(source_addr, vec![0xaa], 8, 8)
.destination_dropped()
);
let finish = dispatch.dispatch_endpoint_data_fast(&mut node);
assert_eq!(finish.pending_flush_dest(), Some(source_addr));
assert!(
node.pending_session_traffic.has_traffic_for(&source_addr),
"fast dispatch should report, not synchronously drain, pending traffic"
);
}
#[tokio::test]
async fn worker_direct_session_data_commits_before_endpoint_delivery() {
let local = Identity::generate();
let peer = Identity::generate();
let previous_hop = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let previous_hop_peer = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
let source_addr = *peer.node_addr();
let endpoint_payload = b"worker decoded endpoint".to_vec();
let plaintext_len = FSP_INNER_HEADER_SIZE + endpoint_payload.len();
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
let direct = DecryptDirectSessionData::for_test(
crate::node::decrypt_worker::DecryptFmpBookkeeping {
source_peer: previous_hop_peer,
transport_id: crate::transport::TransportId::new(1),
remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
packet_timestamp_ms: 2_000,
packet_len: 256,
fmp_counter: 11,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
source_addr,
previous_hop_peer,
false,
crate::node::session::FspReceiveSync {
counter: 7,
slot: EpochSlot::Current,
received_k_bit: false,
timestamp: 0x0102_0304,
plaintext_len,
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
endpoint_payload.len(),
DecryptDirectSessionDelivery::EndpointData(EndpointDataDelivery::new(
source_peer,
endpoint_payload.clone(),
)),
);
node.process_direct_session_data_from_worker(direct).await;
match endpoint_io.event_rx.try_recv().expect("endpoint event") {
crate::node::NodeEndpointEvent::Data {
source_peer: delivered_source,
payload,
..
} => {
assert_eq!(delivered_source, source_peer);
assert_eq!(payload, endpoint_payload);
}
event => panic!("expected worker-decoded endpoint data event, got {event:?}"),
}
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(
entry.traffic_counters(),
(0, 1, 0, endpoint_payload.len() as u64)
);
assert_eq!(entry.current_highest_counter(), Some(7));
}
#[tokio::test]
async fn worker_direct_session_commit_updates_metadata_without_payload_bounce() {
let local = Identity::generate();
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let body_len = b"already delivered".len();
let plaintext_len = FSP_INNER_HEADER_SIZE + body_len;
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
let commit = DecryptDirectSessionCommit::for_test(
crate::node::decrypt_worker::DecryptFmpBookkeeping {
source_peer,
transport_id: crate::transport::TransportId::new(1),
remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
packet_timestamp_ms: 2_000,
packet_len: 256,
fmp_counter: 11,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
source_addr,
source_peer,
false,
crate::node::session::FspReceiveSync {
counter: 7,
slot: EpochSlot::Current,
received_k_bit: false,
timestamp: 0x0102_0304,
plaintext_len,
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
body_len,
false,
);
node.process_direct_session_commit_from_worker(commit).await;
assert!(
endpoint_io.event_rx.try_recv().is_err(),
"compact direct commit must not bounce payload bytes through rx_loop"
);
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(entry.traffic_counters(), (0, 1, 0, body_len as u64));
assert_eq!(entry.current_highest_counter(), Some(7));
}
#[tokio::test]
async fn worker_direct_session_commit_batch_flushes_pending_after_ordered_commits() {
let local = Identity::generate();
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let body_len = b"already delivered".len();
let plaintext_len = FSP_INNER_HEADER_SIZE + body_len;
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
assert!(
!node
.pending_session_traffic
.push_endpoint_data(source_addr, vec![0xaa], 8, 8)
.destination_dropped()
);
let commit = |counter| {
DecryptDirectSessionCommit::for_test(
crate::node::decrypt_worker::DecryptFmpBookkeeping {
source_peer,
transport_id: crate::transport::TransportId::new(1),
remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
packet_timestamp_ms: 2_000,
packet_len: 256,
fmp_counter: counter,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
source_addr,
source_peer,
false,
crate::node::session::FspReceiveSync {
counter,
slot: EpochSlot::Current,
received_k_bit: false,
timestamp: 0x0102_0304,
plaintext_len,
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
body_len,
false,
)
};
node.process_direct_session_commit_batch_from_worker(vec![commit(7), commit(8)])
.await;
assert!(
endpoint_io.event_rx.try_recv().is_err(),
"compact direct commit batch must not bounce payload bytes through rx_loop"
);
assert!(
!node.pending_session_traffic.has_traffic_for(&source_addr),
"batch boundary should flush pending session traffic once receive commits are applied"
);
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(entry.current_highest_counter(), Some(8));
}
#[tokio::test]
async fn worker_direct_session_data_batch_delivers_endpoint_data_then_flushes_pending() {
let local = Identity::generate();
let peer = Identity::generate();
let previous_hop = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let previous_hop_peer = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
let source_addr = *peer.node_addr();
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
assert!(
!node
.pending_session_traffic
.push_endpoint_data(source_addr, vec![0xbb], 8, 8)
.destination_dropped()
);
let direct = |counter, payload: &[u8]| {
DecryptDirectSessionData::for_test(
crate::node::decrypt_worker::DecryptFmpBookkeeping {
source_peer: previous_hop_peer,
transport_id: crate::transport::TransportId::new(1),
remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
packet_timestamp_ms: 2_000,
packet_len: 256,
fmp_counter: counter,
inner_timestamp_ms: 22,
fmp_flags: 0,
},
source_addr,
previous_hop_peer,
false,
crate::node::session::FspReceiveSync {
counter,
slot: EpochSlot::Current,
received_k_bit: false,
timestamp: 0x0102_0304,
plaintext_len: FSP_INNER_HEADER_SIZE + payload.len(),
ce_flag: false,
path_mtu: 1_280,
spin_bit: false,
},
payload.len(),
DecryptDirectSessionDelivery::EndpointData(EndpointDataDelivery::new(
source_peer,
payload.to_vec(),
)),
)
};
node.process_direct_session_data_batch_from_worker(vec![
direct(7, b"batch one"),
direct(8, b"batch two"),
])
.await;
for expected in [b"batch one".to_vec(), b"batch two".to_vec()] {
match endpoint_io.event_rx.try_recv().expect("endpoint event") {
crate::node::NodeEndpointEvent::Data {
source_peer: delivered_source,
payload,
..
} => {
assert_eq!(delivered_source, source_peer);
assert_eq!(payload, expected);
}
event => panic!("expected worker-decoded endpoint data event, got {event:?}"),
}
}
assert!(
!node.pending_session_traffic.has_traffic_for(&source_addr),
"batch boundary should flush pending session traffic after direct deliveries"
);
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(entry.current_highest_counter(), Some(8));
}
#[tokio::test]
async fn reserved_link_type_0x03_does_not_deliver_endpoint_data() {
let local = Identity::generate();
let peer = Identity::generate();
let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
let source_addr = *peer.node_addr();
let payload = b"reserved endpoint-shaped payload".to_vec();
let mut plaintext = 0x0102_0304u32.to_le_bytes().to_vec();
plaintext.push(0x03);
plaintext.extend_from_slice(&payload);
let remote_addr = crate::transport::TransportAddr::from_string("127.0.0.1:1234");
let mut node = Node::new(crate::config::Config::new()).expect("node");
let mut endpoint_io = node
.attach_endpoint_data_io(8)
.expect("endpoint I/O should attach");
node.sessions
.insert(source_addr, established_entry(&local, &peer));
node.process_authentic_fmp_plaintext(crate::node::AuthenticatedFmpPlaintext::new(
source_peer,
crate::transport::TransportId::new(1),
&remote_addr,
2_100,
plaintext.len() + crate::node::wire::ESTABLISHED_HEADER_SIZE + crate::noise::TAG_SIZE,
12,
0,
&plaintext,
))
.await;
assert!(
endpoint_io.event_rx.try_recv().is_err(),
"reserved 0x03 link payload must not bypass FSP session data"
);
let entry = node
.sessions
.get(&source_addr)
.expect("session should remain");
assert_eq!(entry.traffic_counters(), (0, 0, 0, 0));
}
#[test]
fn session_runtime_receive_owns_decrypt_failure_recovery_gate() {
let local = Identity::generate();
let peer = Identity::generate();
let mut entry = established_entry(&local, &peer);
entry.mark_established(1_000);
let plaintext_len = FSP_INNER_HEADER_SIZE + 32;
let forged_ciphertext = vec![0u8; plaintext_len + crate::noise::TAG_SIZE];
for attempt in 1..=DECRYPT_FAILURE_RECOVERY_THRESHOLD {
let header = build_fsp_header(attempt as u64, 0, plaintext_len as u16);
let mut wire = header.to_vec();
wire.extend_from_slice(&forged_ciphertext);
let parsed = FspEncryptedHeader::parse(&wire).expect("forged frame should parse");
let outcome = SessionRuntimeReceive::new(
&mut entry,
&parsed,
&wire[FSP_HEADER_SIZE..],
1_280,
false,
1_000 + DECRYPT_FAILURE_RECOVERY_QUIET_MS + attempt as u64,
)
.open_established();
match outcome {
FspFrameOutcome::DecryptFailed {
counter,
consecutive,
recover_session,
..
} => {
assert_eq!(counter, attempt as u64);
assert_eq!(consecutive, attempt);
assert_eq!(
recover_session,
attempt == DECRYPT_FAILURE_RECOVERY_THRESHOLD
);
}
other => panic!("expected decrypt failure, got {other:?}"),
}
}
}