use crate::dataplane::DataplaneLiveTurnIo;
use crate::discovery::is_punch_packet;
use crate::node::wire::{
COMMON_PREFIX_SIZE, CommonPrefix, FMP_VERSION, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2,
};
use crate::node::{
AuthenticatedFmpReceiveFacts, AuthenticatedLinkMessage, FLAG_CE, LocalSessionPayload, Node,
};
use crate::transport::ReceivedPacket;
use crate::{NodeAddr, PeerIdentity};
use tracing::{debug, trace, warn};
use super::{RxLoopDataplaneIo, RxLoopDataplaneTurnLimits};
impl Node {
pub(in crate::node) async fn drain_dataplane_turn_with_firsts(
&mut self,
io: &mut RxLoopDataplaneIo<'_>,
firsts: crate::dataplane::DataplaneLiveTurnFirsts,
limits: RxLoopDataplaneTurnLimits,
) -> crate::dataplane::DataplaneLiveNodeTurn {
let direct_fsp_sources = std::sync::Arc::new(self.dataplane_direct_fsp_sources());
self.dataplane
.set_established_fast_ingress_direct_fsp_sources(direct_fsp_sources.clone());
let turn = self
.dataplane
.pump_packet_rx_turn_with_firsts_direct_fsp_sources_and_transport_batch(
&mut *io.packet_rx,
firsts,
limits.packet,
direct_fsp_sources,
DataplaneLiveTurnIo {
endpoint_data_rx: &mut *io.endpoint_data_rx,
endpoint_limit: limits.endpoint,
tun_outbound_rx: &mut *io.tun_outbound_rx,
tun_limit: limits.tun,
endpoint_tx: io.endpoint_tx,
transports: &self.transports,
crypto_limit: limits.crypto,
transport_send_batch_packets: self.dataplane_transport_send_batch_packets,
},
)
.await;
Self::observe_dataplane_turn(&turn);
turn
}
pub(in crate::node) async fn drain_dataplane_completion_turn(
&mut self,
io: &mut RxLoopDataplaneIo<'_>,
crypto_limit: usize,
) -> crate::dataplane::DataplaneLiveNodeTurn {
let turn = self
.dataplane
.pump_completion_output_turn_with_transport_batch(DataplaneLiveTurnIo {
endpoint_data_rx: &mut *io.endpoint_data_rx,
endpoint_limit: 0,
tun_outbound_rx: &mut *io.tun_outbound_rx,
tun_limit: 0,
endpoint_tx: io.endpoint_tx,
transports: &self.transports,
crypto_limit,
transport_send_batch_packets: self.dataplane_transport_send_batch_packets,
})
.await;
Self::observe_dataplane_turn(&turn);
turn
}
pub(in crate::node) async fn process_dataplane_control_ingress(
&mut self,
turn: &mut crate::dataplane::DataplaneLiveNodeTurn,
) -> usize {
let mut processed = 0usize;
let fmp_crypto_failures: Vec<_> = turn
.drops()
.iter()
.filter_map(Self::dataplane_fmp_crypto_failure)
.collect();
for (source_addr, counter, authenticated_highest) in fmp_crypto_failures {
if self
.handle_dataplane_fmp_decrypt_failure(&source_addr, counter, authenticated_highest)
.await
{
processed += 1;
}
}
for receipt in turn.take_fmp_ingress_receipts() {
if self.record_dataplane_fmp_ingress_receipt(&receipt) {
processed += 1;
}
}
for ingress in turn.take_fmp_link_ingress() {
if self.process_dataplane_fmp_link_ingress(ingress).await {
processed += 1;
}
}
for warmup in turn.take_fsp_coord_warmups() {
warmup.apply_to(self.coord_cache_mut(), Self::now_ms());
processed += 1;
}
let fsp_crypto_failures: Vec<_> = turn
.drops()
.iter()
.filter_map(Self::dataplane_fsp_crypto_failure)
.collect();
for (source_addr, counter, received_k_bit) in fsp_crypto_failures {
if self
.handle_dataplane_fsp_decrypt_failure(source_addr, counter, received_k_bit)
.await
{
processed += 1;
}
}
for ingress in turn.take_fsp_local_session_ingress() {
if self.process_dataplane_local_session_ingress(ingress).await {
processed += 1;
}
}
processed = processed.saturating_add(
self.process_dataplane_authenticated_ingress(turn.take_fsp_authenticated_ingress())
.await,
);
for control in turn.take_fmp_control_ingress() {
if self.process_dataplane_fmp_control_ingress(control).await {
processed += 1;
}
}
for drop in turn.tun_outbound_drops() {
if self.process_dataplane_tun_outbound_drop(drop) {
processed += 1;
}
}
for packet in self.dataplane.take_deferred_tun_packets() {
self.handle_dataplane_deferred_tun_packet(packet).await;
processed += 1;
}
for batch in self.dataplane.take_deferred_endpoint_data_batches() {
self.handle_endpoint_data_batch_no_established_flush(batch)
.await;
processed += 1;
}
processed
}
fn dataplane_fmp_crypto_failure(
drop: &crate::dataplane::PacketDrop,
) -> Option<(NodeAddr, u64, u64)> {
if drop.owner().protocol() != crate::dataplane::PacketProtocol::Fmp
|| drop.reason() != crate::dataplane::PacketDropReason::CryptoFailed
|| drop.crypto_failure() != Some(crate::dataplane::CryptoFailureKind::Open)
{
return None;
}
Some((
drop.owner().node_addr(),
drop.counter()?,
drop.authenticated_counter_highest().unwrap_or(0),
))
}
fn dataplane_fsp_crypto_failure(
drop: &crate::dataplane::PacketDrop,
) -> Option<(NodeAddr, u64, bool)> {
if drop.owner().protocol() != crate::dataplane::PacketProtocol::Fsp
|| drop.reason() != crate::dataplane::PacketDropReason::CryptoFailed
|| drop.crypto_failure() != Some(crate::dataplane::CryptoFailureKind::Open)
{
return None;
}
let received_k_bit =
drop.wire_flags().unwrap_or(0) & crate::node::session_wire::FSP_FLAG_K != 0;
Some((drop.owner().node_addr(), drop.counter()?, received_k_bit))
}
async fn process_dataplane_local_session_ingress(
&mut self,
ingress: crate::dataplane::DataplaneFspLocalSessionIngress,
) -> bool {
let (source_addr, _previous_hop_addr, _ce_flag, _path_mtu, payload) = ingress.into_parts();
let delivery = LocalSessionPayload::new(source_addr, payload.as_slice());
self.handle_session_payload(delivery).await;
true
}
async fn process_dataplane_fmp_control_ingress(
&mut self,
control: crate::dataplane::DataplaneFmpControlIngress,
) -> bool {
let packet = control.into_packet();
if is_punch_packet(packet.data.as_slice()) {
trace!(
transport_id = %packet.transport_id,
remote_addr = %packet.remote_addr,
bytes = packet.data.len(),
"Dropping stray punch probe/ack from dataplane control ingress"
);
return false;
}
if packet.data.len() < COMMON_PREFIX_SIZE {
return false;
}
let Some(prefix) = CommonPrefix::parse(packet.data.as_slice()) else {
return false;
};
if prefix.version != FMP_VERSION {
self.record_dataplane_fmp_protocol_mismatch(&packet, prefix.version, prefix.phase);
return false;
}
match prefix.phase {
PHASE_MSG1 => {
self.handle_msg1(packet).await;
true
}
PHASE_MSG2 => {
self.handle_msg2(packet).await;
true
}
_ => {
debug!(
phase = prefix.phase,
transport_id = %packet.transport_id,
"Unknown dataplane FMP control phase, dropping"
);
false
}
}
}
fn record_dataplane_fmp_protocol_mismatch(
&mut self,
packet: &ReceivedPacket,
version: u8,
phase: u8,
) {
debug!(
version,
transport_id = %packet.transport_id,
"Unknown dataplane FMP version, dropping"
);
let looks_like_fmp_phase = matches!(phase, PHASE_ESTABLISHED | PHASE_MSG1 | PHASE_MSG2);
if looks_like_fmp_phase
&& self.bootstrap_transports.contains(&packet.transport_id)
&& let Some(npub) = self.bootstrap_transports.peer_npub(&packet.transport_id)
&& let Some(handle) = self.nostr_discovery_handle()
{
let now_ms = Self::now_ms();
let cooldown_secs = handle.protocol_mismatch_cooldown_secs();
if handle.record_protocol_mismatch(npub, now_ms) {
warn!(
peer_npub = %npub,
transport_id = %packet.transport_id,
peer_version = version,
our_version = FMP_VERSION,
cooldown_secs,
"Nostr-discovered peer speaks a different FMP version; suppressing retraversal"
);
}
}
}
fn record_dataplane_fmp_ingress_receipt(
&mut self,
receipt: &crate::dataplane::DataplaneFmpIngressReceipt,
) -> bool {
let fmp = AuthenticatedFmpReceiveFacts::from_dataplane_receipt(receipt);
self.record_authenticated_fmp_receive_facts(fmp, Some(receipt.source_addr()));
true
}
fn process_dataplane_tun_outbound_drop(
&mut self,
drop: &crate::dataplane::DataplaneTunOutboundDrop,
) -> bool {
if drop.packet().is_empty() {
return false;
}
match drop.reason() {
crate::dataplane::DataplaneTunOutboundDropReason::MtuExceeded { mtu } => {
self.send_icmpv6_packet_too_big(drop.packet(), mtu);
true
}
crate::dataplane::DataplaneTunOutboundDropReason::NoRoute => {
self.send_icmpv6_dest_unreachable(drop.packet());
true
}
crate::dataplane::DataplaneTunOutboundDropReason::InvalidPacket => false,
}
}
async fn process_dataplane_fmp_link_ingress(
&mut self,
ingress: crate::dataplane::DataplaneFmpLinkIngress,
) -> bool {
let receipt = ingress.receipt();
let fmp = AuthenticatedFmpReceiveFacts::from_dataplane_receipt(receipt);
debug!(
source = %fmp.source_node_addr(),
counter = fmp.fmp_counter,
flags = fmp.fmp_flags,
inner_timestamp_ms = fmp.inner_timestamp_ms,
packet_len = fmp.packet_len,
msg_type = ?ingress.msg_type(),
payload_len = ingress.payload().len(),
"dataplane FMP link ingress"
);
self.record_authenticated_fmp_receive_facts(fmp, Some(receipt.source_addr()));
let Some(msg_type) = ingress.msg_type() else {
return true;
};
self.dispatch_link_message(AuthenticatedLinkMessage::new(
fmp.source_peer,
msg_type,
ingress.payload(),
fmp.fmp_flags & FLAG_CE != 0,
))
.await;
true
}
pub(in crate::node) fn dataplane_peer_identity(&self, addr: &NodeAddr) -> Option<PeerIdentity> {
if let Some(identity) = self
.sessions
.get(addr)
.and_then(|entry| entry.remote_identity())
{
return Some(identity);
}
if let Some(identity) = self.peers.get(addr).map(|peer| *peer.identity()) {
return Some(identity);
}
self.identity_cache
.iter()
.find_map(|(cached_addr, pubkey, _)| {
(cached_addr == addr).then(|| PeerIdentity::from_pubkey_full(*pubkey))
})
}
pub(super) fn dataplane_packet_activity(
turn: &crate::dataplane::DataplaneLiveNodeTurn,
) -> usize {
let summary = turn.summary();
summary
.raw_ingress_dropped()
.saturating_add(summary.inbound_admitted())
.saturating_add(summary.inbound_dropped())
.saturating_add(turn.fmp_control_ingress().len())
.saturating_add(turn.fmp_link_ingress().len())
.saturating_add(turn.fsp_coord_warmups().len())
.saturating_add(turn.fsp_local_session_ingress().len())
.saturating_add(turn.endpoint_data_packet_count())
.saturating_add(turn.fsp_session_ingress_count())
.saturating_add(turn.deferred_endpoint_data_batches_count())
.saturating_add(turn.tun_deferred_packets())
}
pub(super) fn dataplane_raw_ingress_activity(
turn: &crate::dataplane::DataplaneLiveNodeTurn,
) -> usize {
let summary = turn.summary();
summary
.raw_ingress_dropped()
.saturating_add(summary.inbound_admitted())
.saturating_add(summary.inbound_dropped())
.saturating_add(turn.fmp_control_ingress().len())
}
pub(super) fn dataplane_control_activity(
turn: &crate::dataplane::DataplaneLiveNodeTurn,
) -> usize {
turn.fmp_control_ingress()
.len()
.saturating_add(turn.fmp_link_ingress().len())
.saturating_add(turn.fsp_coord_warmups().len())
.saturating_add(turn.fsp_local_session_ingress().len())
}
pub(in crate::node) fn observe_dataplane_turn(turn: &crate::dataplane::DataplaneLiveNodeTurn) {
if !turn.has_activity() {
return;
}
let summary = turn.summary();
if turn.has_failures() {
debug!(
raw_ingress_dropped = summary.raw_ingress_dropped(),
inbound_dropped = summary.inbound_dropped(),
outbound_dropped = summary.outbound_dropped(),
output_drops = turn.output_drops().len(),
fmp_control_ingress = turn.fmp_control_ingress().len(),
fsp_coord_warmups = turn.fsp_coord_warmups().len(),
fsp_local_session_ingress = turn.fsp_local_session_ingress().len(),
endpoint_data_packets = turn.endpoint_data_packet_count(),
endpoint_data_batches = turn.endpoint_data_batch_count(),
fsp_session_ingress = turn.fsp_session_ingress_count(),
raw_ingress_drops = turn.raw_ingress_drops().len(),
tun_outbound_drops = turn.tun_outbound_drops().len(),
endpoint_data_drops = turn.endpoint_data_drops().len(),
tun_deferred_packets = turn.tun_deferred_packets(),
packet_drops = turn.drops().len(),
transport_dropped = turn.transport_dropped(),
"dataplane turn reported drops"
);
for drop in turn.raw_ingress_drops() {
debug!(
protocol = ?drop.protocol(),
transport_id = ?drop.transport_id(),
remote_addr = ?drop.remote_addr(),
payload_len = drop.payload_len(),
fmp_receiver_idx = ?drop.fmp_receiver_idx(),
reason = ?drop.reason(),
"dataplane raw ingress dropped"
);
}
for drop in turn.endpoint_data_drops() {
debug!(
dest_addr = ?drop.dest_addr(),
payload_len = drop.payload_len(),
reason = ?drop.reason(),
"dataplane endpoint data batch dropped"
);
}
for drop in turn.drops() {
debug!(
owner = ?drop.owner(),
counter = ?drop.counter(),
reason = ?drop.reason(),
crypto_failure = ?drop.crypto_failure(),
wire_flags = ?drop.wire_flags(),
authenticated_counter_highest = ?drop.authenticated_counter_highest(),
"dataplane packet dropped"
);
}
return;
}
trace!(
inbound_admitted = summary.inbound_admitted(),
outbound_admitted = summary.outbound_admitted(),
outputs_sent = summary.outputs_sent(),
transport_sent = turn.transport_sent(),
endpoint_deferred = turn.deferred_endpoint_data_batches_count(),
tun_deferred = turn.tun_deferred_packets(),
fmp_control_ingress = turn.fmp_control_ingress().len(),
fmp_link_ingress = turn.fmp_link_ingress().len(),
fsp_coord_warmups = turn.fsp_coord_warmups().len(),
fsp_local_session_ingress = turn.fsp_local_session_ingress().len(),
endpoint_data_packets = turn.endpoint_data_packet_count(),
endpoint_data_batches = turn.endpoint_data_batch_count(),
fsp_session_ingress = turn.fsp_session_ingress_count(),
"dataplane turn completed"
);
}
}