impl Node {
pub(in crate::node) async fn initiate_session(
&mut self,
dest_addr: NodeAddr,
dest_pubkey: PublicKey,
) -> Result<(), NodeError> {
if self.sessions.should_skip_session_initiation(&dest_addr) {
return Ok(());
}
let our_keypair = self.identity.keypair();
let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey);
handshake.set_local_epoch(self.startup_epoch);
let msg1 = handshake
.write_xk_message_1()
.map_err(|e| NodeError::SendFailed {
node_addr: dest_addr,
reason: format!("Noise XK msg1 generation failed: {}", e),
})?;
let our_coords = self.tree_state.my_coords().clone();
let dest_coords = self.get_dest_coords(&dest_addr);
let setup = SessionSetup::new(our_coords, dest_coords).with_handshake(msg1);
let setup_payload = setup.encode();
let my_addr = *self.node_addr();
let mut datagram = SessionDatagram::new(my_addr, dest_addr, setup_payload.clone())
.with_ttl(self.config.node.session.default_ttl);
self.send_session_datagram(&mut datagram).await?;
self.register_identity(dest_addr, dest_pubkey);
let now_ms = Self::now_ms();
let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms;
self.sessions.install_initiating_session(
dest_addr,
dest_pubkey,
handshake,
setup_payload,
now_ms,
resend_interval,
);
debug!(dest = %self.peer_display_name(&dest_addr), "Session initiation started");
Ok(())
}
pub(in crate::node) async fn send_session_data(
&mut self,
dest_addr: &NodeAddr,
src_port: u16,
dst_port: u16,
payload: &[u8],
) -> Result<(), NodeError> {
let now_ms = Self::now_ms();
let send_context = self
.sessions
.session_fsp_send_context(dest_addr, now_ms)
.map_err(|error| error.into_node_error(*dest_addr))?;
let wants_coords = send_context.wants_coords();
let timestamp = send_context.timestamp;
let mut port_payload = Vec::with_capacity(FSP_PORT_HEADER_SIZE + payload.len());
port_payload.extend_from_slice(&src_port.to_le_bytes());
port_payload.extend_from_slice(&dst_port.to_le_bytes());
port_payload.extend_from_slice(payload);
let msg_type = SessionMessageType::DataPacket.to_byte(); let inner_flags = send_context.inner_flags_byte();
let inner_plaintext =
fsp_prepend_inner_header(timestamp, msg_type, inner_flags, &port_payload);
let (include_coords, my_coords, dest_coords) = if wants_coords {
let src = self.tree_state.my_coords().clone();
let dst = self.get_dest_coords(dest_addr);
let coords_size = coords_wire_size(&src) + coords_wire_size(&dst);
let total_wire =
FIPS_OVERHEAD as usize + FSP_PORT_HEADER_SIZE + coords_size + payload.len();
if total_wire <= self.transport_mtu() as usize {
(true, Some(src), Some(dst))
} else {
if let Err(e) = self.send_coords_warmup(dest_addr).await {
debug!(dest = %self.peer_display_name(dest_addr), error = %e,
"Failed to send standalone CoordsWarmup before data packet");
}
(false, None, None)
}
} else {
(false, None, None)
};
if wants_coords {
self.sessions.consume_coords_warmup_packet(dest_addr);
}
let flags = send_context.fsp_flags(include_coords);
let coords = my_coords.as_ref().zip(dest_coords.as_ref());
self.send_session_fsp_plan(SessionFspSendPlan::new(
*dest_addr,
timestamp,
flags,
&inner_plaintext,
coords,
SessionFspSendBookkeeping::Data {
payload_len: payload.len(),
now_ms,
},
))
.await
}
async fn send_session_fsp_plan(
&mut self,
plan: SessionFspSendPlan<'_>,
) -> Result<(), NodeError> {
let dest_addr = plan.dest_addr();
let sealed = self.sessions.seal_session_fsp_send(plan)?;
let (mut datagram, bookkeeping) =
sealed.into_datagram(*self.node_addr(), self.config.node.session.default_ttl);
self.send_session_datagram(&mut datagram).await?;
let _ = self
.sessions
.record_fsp_send_bookkeeping(&dest_addr, bookkeeping);
Ok(())
}
pub(in crate::node) async fn send_ipv6_packet(
&mut self,
dest_addr: &NodeAddr,
ipv6_packet: &[u8],
) -> Result<(), NodeError> {
let compressed = crate::upper::ipv6_shim::compress_ipv6(ipv6_packet).ok_or_else(|| {
NodeError::SendFailed {
node_addr: *dest_addr,
reason: "IPv6 header compression failed".into(),
}
})?;
self.send_session_data(
dest_addr,
FSP_PORT_IPV6_SHIM,
FSP_PORT_IPV6_SHIM,
&compressed,
)
.await
}
pub(in crate::node) async fn handle_endpoint_data_command(
&mut self,
command: NodeEndpointCommand,
) {
match command {
NodeEndpointCommand::Send {
command,
response_tx,
} => {
let result = self.handle_endpoint_send_command(command).await;
let _ = response_tx.send(result);
}
NodeEndpointCommand::SendOneway { command } => {
let _ = self.handle_endpoint_send_command(command).await;
}
NodeEndpointCommand::SendBatchOneway { command, .. } => {
self.handle_endpoint_send_batch_commands(vec![command]).await;
}
NodeEndpointCommand::UpdatePeers { peers, response_tx } => {
let result = self.update_peers(peers).await;
let _ = response_tx.send(result);
}
NodeEndpointCommand::RefreshPeerPaths { npubs, response_tx } => {
let result = self.refresh_peer_paths(npubs).await;
let _ = response_tx.send(result);
}
NodeEndpointCommand::PeerSnapshot { response_tx } => {
let snapshot_now = Instant::now();
let nostr_failure_state: std::collections::HashMap<String, _> = self
.nostr_discovery_handle()
.map(|discovery| {
discovery
.failure_state_snapshot()
.into_iter()
.map(|state| (state.npub.clone(), state))
.collect()
})
.unwrap_or_default();
let mut peers = self
.peers()
.map(|peer| {
let link_id = peer.link_id();
let retry_state = self.retry_pending.get(peer.node_addr());
let npub = peer.npub();
let nostr_state = nostr_failure_state.get(&npub);
let nostr_traversal_cooldown_until_ms =
nostr_state.and_then(|state| state.cooldown_until_ms);
let transport_type = self.get_link(&link_id).and_then(|link| {
self.get_transport(&link.transport_id())
.map(|handle| handle.transport_type().name.to_string())
});
let stats = peer.link_stats();
let direct_probe_pending = retry_state.is_some();
let srtt = peer.mmp().and_then(|mmp| {
mmp.metrics.srtt_ms().map(|value| {
(value.round() as u64, mmp.metrics.srtt_age_ms(snapshot_now))
})
});
NodeEndpointPeer {
npub,
node_addr: *peer.node_addr(),
connected: true,
transport_addr: peer.current_addr().map(|addr| addr.to_string()),
transport_type,
link_id: link_id.as_u64(),
srtt_ms: srtt.map(|(value, _)| value),
srtt_age_ms: srtt.and_then(|(_, age)| age),
packets_sent: stats.packets_sent,
packets_recv: stats.packets_recv,
bytes_sent: stats.bytes_sent,
bytes_recv: stats.bytes_recv,
rekey_in_progress: peer.rekey_in_progress(),
rekey_draining: peer.is_draining(),
current_k_bit: Some(peer.current_k_bit()),
direct_probe_pending,
direct_probe_after_ms: retry_state.map(|state| state.retry_after_ms),
direct_probe_retry_count: retry_state
.map_or(0, |state| state.retry_count),
direct_probe_auto_reconnect: retry_state
.is_some_and(|state| state.reconnect),
direct_probe_expires_at_ms: retry_state
.and_then(|state| state.expires_at_ms),
nostr_traversal_consecutive_failures: nostr_state
.map_or(0, |state| state.consecutive_failures),
nostr_traversal_in_cooldown: nostr_traversal_cooldown_until_ms
.is_some(),
nostr_traversal_cooldown_until_ms,
nostr_traversal_last_observed_skew_ms: nostr_state
.and_then(|state| state.last_observed_skew_ms),
}
})
.collect::<Vec<_>>();
for (node_addr, retry_state) in self.retry_pending.iter() {
if self.peers.contains_key(node_addr)
|| !self
.config
.peers
.iter()
.any(|peer| peer.npub == retry_state.peer_config.npub)
{
continue;
}
let npub = retry_state.peer_config.npub.clone();
let nostr_state = nostr_failure_state.get(&npub);
let nostr_traversal_cooldown_until_ms =
nostr_state.and_then(|state| state.cooldown_until_ms);
peers.push(NodeEndpointPeer {
npub,
node_addr: *node_addr,
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,
direct_probe_pending: true,
direct_probe_after_ms: Some(retry_state.retry_after_ms),
direct_probe_retry_count: retry_state.retry_count,
direct_probe_auto_reconnect: retry_state.reconnect,
direct_probe_expires_at_ms: retry_state.expires_at_ms,
nostr_traversal_consecutive_failures: nostr_state
.map_or(0, |state| state.consecutive_failures),
nostr_traversal_in_cooldown: nostr_traversal_cooldown_until_ms.is_some(),
nostr_traversal_cooldown_until_ms,
nostr_traversal_last_observed_skew_ms: nostr_state
.and_then(|state| state.last_observed_skew_ms),
});
}
let _ = response_tx.send(peers);
}
NodeEndpointCommand::RelaySnapshot { response_tx } => {
let relays = if let Some(discovery) = self.nostr_discovery_handle() {
discovery
.relay_statuses()
.await
.into_iter()
.map(|relay| NodeEndpointRelayStatus {
url: relay.url,
status: relay.status,
})
.collect()
} else {
Vec::new()
};
let _ = response_tx.send(relays);
}
NodeEndpointCommand::UpdateRelays {
advert_relays,
dm_relays,
response_tx,
} => {
let result = if let Some(discovery) = self.nostr_discovery_handle() {
discovery
.update_relays(advert_relays, dm_relays)
.await
.map_err(|error| NodeError::Discovery(error.to_string()))
} else {
Err(NodeError::Discovery(
"Nostr discovery is not running".to_string(),
))
};
let _ = response_tx.send(result);
}
}
}
async fn handle_endpoint_send_command(
&mut self,
command: EndpointSendCommand,
) -> Result<(), NodeError> {
let lane = command.lane();
let (send, queued_at) = command.into_parts();
record_endpoint_command_wait(queued_at, lane, 1);
let _t = crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointSend);
self.send_endpoint_data_send(send).await
}
pub(in crate::node) async fn handle_endpoint_send_batch_commands(
&mut self,
commands: Vec<EndpointSendBatchCommand>,
) {
let Some(first) = commands.first() else {
return;
};
let lane = first.lane();
let remote = first.remote();
let mut payload_count = 0usize;
let mut payloads = Vec::new();
for command in commands {
debug_assert_eq!(command.lane(), lane);
debug_assert_eq!(command.remote(), remote);
let count = command.len();
let (command_remote, command_payloads, queued_at) = command.into_parts();
debug_assert_eq!(command_remote, remote);
record_endpoint_command_wait(queued_at, lane, count as u64);
payload_count = payload_count.saturating_add(count);
payloads.extend(command_payloads);
}
if payload_count == 0 {
return;
}
let dest_addr = *remote.node_addr();
let dest_pubkey = remote.pubkey_full();
self.register_identity(dest_addr, dest_pubkey);
#[cfg(unix)]
if self.encrypt_workers.is_some()
&& self
.sessions
.get(&dest_addr)
.is_some_and(|entry| entry.is_established())
{
self.handle_established_endpoint_send_batch(dest_addr, dest_pubkey, payloads)
.await;
return;
}
self.handle_endpoint_send_batch_slow_path(dest_addr, dest_pubkey, payloads)
.await;
}
}