#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
fn closed_tun_to_mesh_queue_stops_reader() {
let (tx, rx) = TunPipelineQueueTx::channel(1);
drop(rx);
assert_eq!(
submit_tun_packet_batch_to_mesh_queue(
&tx,
vec![test_pipeline_packet(test_ipv6_tcp_packet(0x18, 512))],
),
TunQueueSubmit::Closed
);
}
fn mesh_peer_status(
pubkey: impl AsRef<str>,
endpoint_npub: impl AsRef<str>,
transport_addr: Option<&str>,
transport_type: Option<&str>,
connected: bool,
last_seen_at: Option<u64>,
) -> MeshPeerStatus {
MeshPeerStatus {
pubkey: pubkey.as_ref().to_string(),
connected,
endpoint_npub: endpoint_npub.as_ref().to_string(),
transport_addr: transport_addr.map(str::to_string),
transport_type: transport_type.map(str::to_string),
srtt_ms: Some(18),
srtt_age_ms: Some(250),
link_packets_sent: 7,
link_packets_recv: 8,
link_bytes_sent: 900,
link_bytes_recv: 1200,
rekey_in_progress: false,
rekey_draining: false,
current_k_bit: None,
last_outbound_route: None,
direct_probe_pending: false,
direct_probe_after_ms: None,
direct_probe_retry_count: 0,
direct_probe_auto_reconnect: false,
direct_probe_expires_at_ms: None,
nostr_traversal_consecutive_failures: 0,
nostr_traversal_in_cooldown: false,
nostr_traversal_cooldown_until_ms: None,
nostr_traversal_last_observed_skew_ms: None,
last_seen_at,
last_control_seen_at: None,
last_data_seen_at: None,
tx_bytes: 0,
rx_bytes: 0,
error: None,
}
}
#[test]
fn non_roster_endpoint_peers_surface_as_mesh_statuses() {
let keys = Keys::generate();
let pubkey = keys.public_key().to_hex();
let npub = keys.public_key().to_bech32().expect("npub");
let mut peers = HashMap::new();
peers.insert(
pubkey.clone(),
FipsEndpointPeer {
npub: npub.clone(),
node_addr: NodeAddr::from_bytes([7; 16]),
connected: true,
transport_addr: Some("203.0.113.9:9000".to_string()),
transport_type: Some("udp".to_string()),
link_id: 42,
srtt_ms: Some(7),
srtt_age_ms: Some(123),
packets_sent: 11,
packets_recv: 12,
bytes_sent: 1300,
bytes_recv: 1400,
rekey_in_progress: true,
rekey_draining: true,
current_k_bit: Some(true),
last_outbound_route: None,
direct_probe_pending: true,
direct_probe_after_ms: Some(12_345),
direct_probe_retry_count: 3,
direct_probe_auto_reconnect: true,
direct_probe_expires_at_ms: Some(67_890),
nostr_traversal_consecutive_failures: 3,
nostr_traversal_in_cooldown: true,
nostr_traversal_cooldown_until_ms: Some(123_456),
nostr_traversal_last_observed_skew_ms: Some(-250),
},
);
let statuses = other_endpoint_peer_statuses(&peers, 123);
assert_eq!(statuses.len(), 1);
assert_eq!(statuses[0].pubkey, pubkey);
assert_eq!(statuses[0].endpoint_npub, npub);
assert!(statuses[0].connected);
assert_eq!(
statuses[0].transport_addr.as_deref(),
Some("203.0.113.9:9000")
);
assert_eq!(statuses[0].transport_type.as_deref(), Some("udp"));
assert!(statuses[0].direct_probe_pending);
assert_eq!(statuses[0].direct_probe_after_ms, Some(12_345));
assert_eq!(statuses[0].direct_probe_retry_count, 3);
assert!(statuses[0].direct_probe_auto_reconnect);
assert_eq!(statuses[0].direct_probe_expires_at_ms, Some(67_890));
assert_eq!(statuses[0].nostr_traversal_consecutive_failures, 3);
assert!(statuses[0].nostr_traversal_in_cooldown);
assert_eq!(statuses[0].nostr_traversal_cooldown_until_ms, Some(123_456));
assert_eq!(
statuses[0].nostr_traversal_last_observed_skew_ms,
Some(-250)
);
assert_eq!(statuses[0].srtt_ms, Some(7));
assert_eq!(statuses[0].srtt_age_ms, Some(123));
assert_eq!(statuses[0].link_packets_sent, 11);
assert_eq!(statuses[0].link_packets_recv, 12);
assert_eq!(statuses[0].link_bytes_sent, 1300);
assert_eq!(statuses[0].link_bytes_recv, 1400);
assert!(statuses[0].rekey_in_progress);
assert!(statuses[0].rekey_draining);
assert_eq!(statuses[0].current_k_bit, Some(true));
assert_eq!(statuses[0].last_seen_at, Some(123));
assert_eq!(statuses[0].error, None);
}
#[test]
fn stale_participant_refreshes_endpoint_path_when_direct_probe_is_pending() {
let keys = Keys::generate();
let npub = keys.public_key().to_bech32().expect("npub");
let mut peer = FipsEndpointPeer {
npub,
node_addr: NodeAddr::from_bytes([9; 16]),
connected: false,
transport_addr: Some("203.0.113.9:9000".to_string()),
transport_type: Some("udp".to_string()),
link_id: 42,
srtt_ms: Some(70),
srtt_age_ms: Some(120_000),
packets_sent: 100,
packets_recv: 20,
bytes_sent: 8192,
bytes_recv: 1024,
rekey_in_progress: false,
rekey_draining: false,
current_k_bit: None,
last_outbound_route: None,
direct_probe_pending: true,
direct_probe_after_ms: Some(12_345),
direct_probe_retry_count: 3,
direct_probe_auto_reconnect: true,
direct_probe_expires_at_ms: Some(67_890),
nostr_traversal_consecutive_failures: 0,
nostr_traversal_in_cooldown: false,
nostr_traversal_cooldown_until_ms: None,
nostr_traversal_last_observed_skew_ms: None,
};
assert!(
!super::endpoint_path_refresh_due(&peer, Some(120), 123),
"fresh participant traffic must not churn same-path direct probes"
);
assert!(super::endpoint_path_refresh_due(&peer, Some(80), 123));
assert!(!super::endpoint_path_refresh_due(&peer, None, 123));
peer.direct_probe_pending = false;
assert!(!super::endpoint_path_refresh_due(&peer, Some(120), 123));
peer.connected = true;
assert!(
!super::endpoint_path_refresh_due(&peer, Some(80), 123),
"connected endpoint links should not be refreshed from wrapper-level participant staleness alone"
);
}
#[test]
fn retry_only_endpoint_peer_status_keeps_probe_separate_from_link() {
let status = mesh_status_from_endpoint_peer(
"peer".to_string(),
&FipsEndpointPeer {
npub: "npub1peer".to_string(),
node_addr: NodeAddr::from_bytes([8; 16]),
connected: false,
transport_addr: None,
transport_type: None,
link_id: 0,
srtt_ms: None,
srtt_age_ms: None,
packets_sent: 0,
packets_recv: 0,
bytes_sent: 0,
bytes_recv: 0,
rekey_in_progress: false,
rekey_draining: false,
current_k_bit: None,
last_outbound_route: None,
direct_probe_pending: true,
direct_probe_after_ms: Some(12_345),
direct_probe_retry_count: 2,
direct_probe_auto_reconnect: false,
direct_probe_expires_at_ms: Some(67_890),
nostr_traversal_consecutive_failures: 2,
nostr_traversal_in_cooldown: true,
nostr_traversal_cooldown_until_ms: Some(123_456),
nostr_traversal_last_observed_skew_ms: Some(500),
},
123,
);
assert!(!status.connected);
assert!(status.direct_probe_pending);
assert_eq!(status.direct_probe_after_ms, Some(12_345));
assert_eq!(status.direct_probe_retry_count, 2);
assert!(!status.direct_probe_auto_reconnect);
assert_eq!(status.direct_probe_expires_at_ms, Some(67_890));
assert_eq!(status.nostr_traversal_consecutive_failures, 2);
assert!(status.nostr_traversal_in_cooldown);
assert_eq!(status.nostr_traversal_cooldown_until_ms, Some(123_456));
assert_eq!(status.nostr_traversal_last_observed_skew_ms, Some(500));
assert_eq!(status.last_seen_at, None);
assert_eq!(status.transport_addr, None);
assert_eq!(status.error.as_deref(), Some("fips link pending"));
}
#[test]
fn fragmented_control_frames_reassemble_to_original_frame() {
let roster = NetworkRoster {
network_name: "Network 1".to_string(),
devices: (0..12).map(|value| format!("{value:064x}")).collect(),
admins: vec!["f".repeat(64)],
aliases: (0..12)
.map(|value| (format!("{value:064x}"), format!("node-{value}")))
.collect(),
signed_at: 123,
};
let frame = FipsControlFrame::Roster {
network_id: "mesh".to_string(),
roster,
signed_roster: None,
};
let messages = encode_fips_control_messages(&frame).expect("fragment messages");
let mut buffer = ControlFragmentBuffer::default();
let mut reassembled = None;
let source_key = [7u8; 16];
for message in messages {
let decoded = decode_fips_control_frame(&message)
.expect("decode fragment")
.expect("fragment frame");
let FipsControlFrame::Fragment {
id,
index,
total,
data,
} = decoded
else {
panic!("expected fragment");
};
reassembled = buffer
.push(source_key, id, index, total, data, 1)
.expect("push fragment")
.or(reassembled);
}
let decoded = decode_fips_control_frame(&reassembled.expect("reassembled frame"))
.expect("decode reassembled")
.expect("control frame");
assert_eq!(decoded, frame);
}
#[test]
fn private_mesh_mtu_defaults_to_safe_budget() {
let mtu = super::resolve_private_mesh_mtu(None, None, None);
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: nostr_vpn_core::MESH_UNDERLAY_UDP_MTU,
tunnel: nostr_vpn_core::MESH_TUNNEL_MTU,
}
);
}
#[test]
fn private_mesh_mtu_lan_profile_uses_larger_paired_budget() {
let mtu = super::resolve_private_mesh_mtu(Some(" LAN "), None, None);
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: 1420,
tunnel: 1290,
}
);
}
#[test]
fn private_mesh_mtu_app_profile_uses_same_resolver() {
let app = AppConfig {
mesh_mtu_profile: " LAN ".to_string(),
..Default::default()
};
let mtu = super::resolve_private_mesh_mtu_from_sources(Some(&app), None, None, None);
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: 1420,
tunnel: 1290,
}
);
}
#[test]
fn private_mesh_mtu_env_overrides_app_config() {
let app = AppConfig {
mesh_mtu_profile: "lan".to_string(),
mesh_underlay_udp_mtu: 1420,
mesh_tunnel_mtu: 1290,
..Default::default()
};
let mtu = super::resolve_private_mesh_mtu_from_sources(Some(&app), None, Some(1280), None);
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: 1280,
tunnel: 1150,
}
);
}
#[test]
fn private_mesh_mtu_underlay_override_derives_tunnel_budget() {
let mtu = super::resolve_private_mesh_mtu(None, Some(1500), None);
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: 1500,
tunnel: 1370,
}
);
}
#[test]
fn private_mesh_mtu_caps_tunnel_to_underlay_budget() {
let mtu = super::resolve_private_mesh_mtu(None, Some(1280), Some(1420));
assert_eq!(
mtu,
super::MeshMtu {
underlay_udp: 1280,
tunnel: 1150,
}
);
}
#[test]
fn peer_endpoint_hint_addr_accepts_only_udp_socket_addresses() {
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("192.168.50.22:51820")),
Some("192.168.50.22:51820".to_string())
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("peer.example.com:51820")),
Some("peer.example.com:51820".to_string())
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint {
transport: "tcp".to_string(),
addr: "192.168.50.22:51820".to_string(),
}),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("192.168.50.22")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("127.0.0.1:51820")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("100.120.94.10:51820")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("198.51.100.10:51820")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("0.0.0.0:51820")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp("localhost:51820")),
None
);
assert_eq!(
super::peer_endpoint_hint_addr(&PeerEndpointHint::udp(format!(
"{}:51820",
"npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq"
))),
None
);
}
#[test]
fn fips_peer_liveness_prefers_participant_activity_over_link_snapshot() {
assert_eq!(
super::fips_peer_liveness(Some(100), true, None, 120),
(true, None)
);
assert_eq!(
super::fips_peer_liveness(None, true, None, 120),
(true, None)
);
assert_eq!(
super::fips_peer_liveness(Some(10), true, None, 120),
(false, Some("fips participant stale".to_string()))
);
assert_eq!(
super::fips_peer_liveness(None, false, Some("dial failed".to_string()), 120),
(false, Some("dial failed".to_string()))
);
}
#[test]
fn fips_peer_liveness_rejects_far_future_presence() {
assert_eq!(
super::fips_peer_liveness(Some(122), false, None, 120),
(true, None)
);
assert_eq!(
super::fips_peer_liveness(Some(180), false, None, 120),
(false, Some("fips participant stale".to_string()))
);
assert_eq!(
super::fips_peer_liveness(Some(180), true, None, 120),
(false, Some("fips participant stale".to_string()))
);
}
#[test]
fn fips_peer_ping_due_uses_peer_state_intervals() {
assert!(super::fips_peer_ping_due(Some(100), None, true, 120));
assert!(!super::fips_peer_ping_due(Some(100), Some(116), true, 120));
assert!(super::fips_peer_ping_due(Some(100), Some(115), true, 120));
assert!(!super::fips_peer_ping_due(None, Some(116), true, 120));
assert!(super::fips_peer_ping_due(None, Some(115), true, 120));
assert!(!super::fips_peer_ping_due(None, Some(91), false, 120));
assert!(super::fips_peer_ping_due(None, Some(90), false, 120));
}
#[test]
fn fips_peer_ping_due_recovers_from_future_timestamps() {
assert_eq!(
super::fips_peer_ping_interval_secs(Some(122), false, 120),
super::FIPS_PEER_ACTIVE_PING_INTERVAL_SECS
);
assert_eq!(
super::fips_peer_ping_interval_secs(Some(180), false, 120),
super::FIPS_PEER_DISCOVERY_PROBE_INTERVAL_SECS
);
assert!(!super::fips_peer_ping_due(None, Some(122), false, 120));
assert!(super::fips_peer_ping_due(None, Some(180), false, 120));
}
#[test]
fn control_frames_from_rostered_endpoint_resolve_to_participant() {
let keys = Keys::generate();
let participant_pubkey = keys.public_key().to_hex();
let endpoint_npub = keys.public_key().to_bech32().expect("npub");
let source_peer = PeerIdentity::from_npub(&endpoint_npub).expect("source peer");
let mesh = FipsMeshRuntime::new(vec![FipsMeshPeerConfig {
participant_pubkey: participant_pubkey.clone(),
endpoint_npub: endpoint_npub.clone(),
allowed_ips: vec!["10.44.1.2/32".to_string()],
}]);
let frame = FipsControlFrame::Ping {
network_id: "network".to_string(),
sent_at: 42,
};
assert_eq!(
control_frame_source_pubkey(&mesh, source_peer, &frame),
Some(participant_pubkey)
);
}
#[test]
fn control_frames_from_unknown_endpoints_are_limited_to_join_requests() {
let keys = Keys::generate();
let unknown_pubkey = keys.public_key().to_hex();
let unknown_npub = keys.public_key().to_bech32().expect("npub");
let unknown_peer = PeerIdentity::from_npub(&unknown_npub).expect("unknown peer");
let mesh = FipsMeshRuntime::new(Vec::new());
let ping = FipsControlFrame::Ping {
network_id: "network".to_string(),
sent_at: 42,
};
let roster = FipsControlFrame::Roster {
network_id: "network".to_string(),
roster: NetworkRoster {
network_name: "network".to_string(),
devices: Vec::new(),
admins: Vec::new(),
aliases: HashMap::new(),
signed_at: 42,
},
signed_roster: None,
};
let join_request = FipsControlFrame::JoinRequest {
requested_at: 42,
request: MeshJoinRequest {
network_id: "network".to_string(),
invite_secret: String::new(),
requester_node_name: "new-device".to_string(),
},
};
assert!(control_frame_source_pubkey(&mesh, unknown_peer, &ping).is_none());
assert!(control_frame_source_pubkey(&mesh, unknown_peer, &roster).is_none());
assert_eq!(
control_frame_source_pubkey(&mesh, unknown_peer, &join_request),
Some(unknown_pubkey)
);
}
#[test]
fn control_frame_destinations_can_target_pending_join_requester() {
let keys = Keys::generate();
let requester_pubkey = keys.public_key().to_hex();
let requester_npub = keys.public_key().to_bech32().expect("npub");
let mesh = FipsMeshRuntime::new(Vec::new());
let identities = FipsPeerIdentityMap::default();
let destination = control_frame_destination_peer(&mesh, &identities, &requester_pubkey)
.expect("destination identity");
assert_eq!(destination.npub(), requester_npub);
}
#[test]
fn control_frame_destinations_use_configured_endpoint_identity() {
let participant_pubkey = Keys::generate().public_key().to_hex();
let endpoint_npub = Keys::generate().public_key().to_bech32().expect("npub");
let endpoint_identity = PeerIdentity::from_npub(&endpoint_npub).expect("peer identity");
let endpoint_node_addr = *endpoint_identity.node_addr().as_bytes();
let mesh = FipsMeshRuntime::new(vec![FipsMeshPeerConfig {
participant_pubkey: participant_pubkey.clone(),
endpoint_npub: endpoint_npub.clone(),
allowed_ips: Vec::new(),
}]);
let mut identities = FipsPeerIdentityMap::default();
identities
.by_endpoint_node_addr
.insert(endpoint_node_addr, endpoint_identity);
let destination = control_frame_destination_peer(&mesh, &identities, &participant_pubkey)
.expect("destination identity");
assert_eq!(destination.npub(), endpoint_npub);
}
#[tokio::test]
async fn endpoint_data_runtime_sends_and_receives_raw_packets() {
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let packet = ipv4_packet(source, destination);
let sent = runtime
.send_tunnel_packet(&packet)
.await
.expect("send packet");
assert!(sent);
let received = tokio::time::timeout(Duration::from_secs(2), runtime.recv_tunnel_packet())
.await
.expect("packet should arrive")
.expect("receive packet")
.expect("packet should pass admission");
assert_eq!(received, packet);
assert_peer_data_activity(&runtime, &participant_pubkey, packet.len() as u64);
runtime.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn endpoint_data_runtime_sends_and_receives_raw_packet_batch() {
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let first = ipv4_packet(source, destination);
let second = ipv4_packet(source, destination);
let expected_endpoint_data_bytes = (first.len() + second.len()) as u64;
let sent = runtime
.send_tunnel_packet_batch_owned(vec![first.clone(), second.clone()])
.await
.expect("send packet batch");
assert_eq!(sent, 2);
let events = tokio::time::timeout(Duration::from_secs(2), runtime.recv_mesh_event_batch(4))
.await
.expect("packet batch should arrive")
.expect("receive packet batch")
.expect("batch should contain admitted packets");
assert_eq!(events.len(), 2);
let mut packets = events.into_iter().map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
});
assert_eq!(packets.next(), Some(first));
assert_eq!(packets.next(), Some(second));
assert_peer_data_activity(&runtime, &participant_pubkey, expected_endpoint_data_bytes);
runtime.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn endpoint_data_runtime_sends_tun_pipeline_batch_without_repacking() {
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let mut first = ipv4_packet(source, destination);
let mut second = ipv4_packet(source, destination);
first[20] = 1;
second[20] = 2;
let expected_endpoint_data_bytes = (first.len() + second.len()) as u64;
let mut batch = Vec::with_capacity(4);
let batch_capacity = batch.capacity();
batch.push(TunPipelinePacket::new(first.clone()));
batch.push(TunPipelinePacket::new(second.clone()));
let sent = runtime
.send_tun_pipeline_packet_batch(&mut batch)
.await
.expect("send TUN pipeline packet batch");
assert_eq!(sent, 2);
assert!(batch.is_empty());
assert_eq!(batch.capacity(), batch_capacity);
let events = tokio::time::timeout(Duration::from_secs(2), runtime.recv_mesh_event_batch(4))
.await
.expect("packet batch should arrive")
.expect("receive packet batch")
.expect("batch should contain admitted packets");
assert_eq!(events.len(), 2);
let packets: Vec<_> = events
.into_iter()
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![first, second]);
assert_peer_data_activity(&runtime, &participant_pubkey, expected_endpoint_data_bytes);
runtime.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn endpoint_data_runtime_recv_batch_into_reuses_buffers_and_respects_limit() {
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let mut first = ipv4_packet(source, destination);
let mut second = ipv4_packet(source, destination);
let mut third = ipv4_packet(source, destination);
first[20] = 1;
second[20] = 2;
third[20] = 3;
let sent = runtime
.send_tunnel_packet_batch_owned(vec![first.clone(), second.clone(), third.clone()])
.await
.expect("send packet batch");
assert_eq!(sent, 3);
let mut messages = Vec::with_capacity(8);
let mut events = Vec::with_capacity(8);
let message_capacity = messages.capacity();
let event_capacity = events.capacity();
let received = tokio::time::timeout(
Duration::from_secs(2),
runtime.recv_mesh_event_batch_into(&mut messages, &mut events, 2),
)
.await
.expect("packet batch should arrive")
.expect("receive packet batch")
.expect("batch should contain admitted packets");
assert_eq!(received, 2);
assert!(messages.is_empty());
assert_eq!(messages.capacity(), message_capacity);
assert_eq!(events.capacity(), event_capacity);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![first, second]);
let received = tokio::time::timeout(
Duration::from_secs(2),
runtime.recv_mesh_event_batch_into(&mut messages, &mut events, 8),
)
.await
.expect("packet batch should arrive")
.expect("receive packet batch")
.expect("batch should contain admitted packets");
assert_eq!(received, 1);
assert!(messages.is_empty());
assert_eq!(messages.capacity(), message_capacity);
assert_eq!(events.capacity(), event_capacity);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![third]);
runtime.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn endpoint_data_runtime_blocking_recv_batch_into_reuses_buffers_and_respects_limit()
{
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let mut first = ipv4_packet(source, destination);
let mut second = ipv4_packet(source, destination);
let mut third = ipv4_packet(source, destination);
first[20] = 1;
second[20] = 2;
third[20] = 3;
let sent = runtime
.send_tunnel_packet_batch_owned(vec![first.clone(), second.clone(), third.clone()])
.await
.expect("send packet batch");
assert_eq!(sent, 3);
let (runtime, event_capacity) =
tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
let stop = AtomicBool::new(false);
let mut messages = Vec::with_capacity(8);
let mut events = Vec::with_capacity(8);
let message_capacity = messages.capacity();
let event_capacity = events.capacity();
let received = runtime
.recv_mesh_event_batch_blocking_into(&mut messages, &mut events, 2, &stop)?
.expect("batch should contain admitted packets");
assert_eq!(received, 2);
assert!(messages.is_empty());
assert_eq!(messages.capacity(), message_capacity);
assert_eq!(events.capacity(), event_capacity);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![first, second]);
let received = runtime
.recv_mesh_event_batch_blocking_into(&mut messages, &mut events, 8, &stop)?
.expect("batch should contain admitted packets");
assert_eq!(received, 1);
assert!(messages.is_empty());
assert_eq!(messages.capacity(), message_capacity);
assert_eq!(events.capacity(), event_capacity);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![third]);
Ok((runtime, event_capacity))
})
.await
.expect("blocking receiver should join")
.expect("blocking batch receive should succeed");
assert_eq!(event_capacity, 8);
runtime.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn endpoint_data_runtime_blocking_recv_batch_for_each_respects_limit() {
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("nsec");
let participant_pubkey = keys.public_key().to_hex();
let source = Ipv4Addr::new(10, 44, 10, 1);
let destination = Ipv4Addr::new(10, 44, 22, 44);
let peer = FipsMeshPeerConfig::from_participant_pubkey(
&participant_pubkey,
vec![format!("{source}/32"), format!("{destination}/32")],
)
.expect("peer config");
let runtime = FipsPrivateMeshRuntime::bind(nsec, "test-network", vec![peer])
.await
.expect("runtime should bind");
let mut first = ipv4_packet(source, destination);
let mut second = ipv4_packet(source, destination);
let mut third = ipv4_packet(source, destination);
first[20] = 1;
second[20] = 2;
third[20] = 3;
let sent = runtime
.send_tunnel_packet_batch_owned(vec![first.clone(), second.clone(), third.clone()])
.await
.expect("send packet batch");
assert_eq!(sent, 3);
let runtime = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
let stop = AtomicBool::new(false);
let mut events = Vec::with_capacity(8);
let received = runtime
.recv_mesh_event_batch_blocking_for_each(2, &stop, |event| {
events.push(event);
true
})?
.expect("batch should contain admitted packets");
assert_eq!(received, 2);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![first, second]);
let received = runtime
.recv_mesh_event_batch_blocking_for_each(8, &stop, |event| {
events.push(event);
true
})?
.expect("batch should contain admitted packets");
assert_eq!(received, 1);
let packets: Vec<_> = events
.drain(..)
.map(|event| match event {
FipsPrivateMeshEvent::Packet(packet) => packet,
event => panic!("expected packet event, got {event:?}"),
})
.collect();
assert_eq!(packets, vec![third]);
Ok(runtime)
})
.await
.expect("blocking receiver should join")
.expect("blocking callback receive should succeed");
runtime.shutdown().await.expect("shutdown");
}