use super::endpoint_traffic::fmp_plaintext_is_bulk_session_datagram;
use super::*;
use crate::dataplane::{
ActivityTick, DataplaneDirectFspSource, DataplaneEndpointDataRoute, DataplaneFspSendReceipt,
DataplaneFspWrapRoute, DataplaneIngressRoute, DataplaneLiveNodeTurn,
DataplaneLiveOutboundFirsts, DataplaneLiveOwnerRoutes, DataplaneLiveTurnIo,
DataplaneOutputDrop, DataplaneOutputError, DataplaneTunOutboundRoute, OutboundPacket,
OutputTarget, OwnerConfig, OwnerCryptoKeys, OwnerId, PacketClass, TransportPath,
};
use crate::node::session_wire::{FSP_PHASE_MSG2, FSP_PHASE_MSG3, FspCommonPrefix};
use crate::protocol::SessionMessageType;
use std::collections::HashMap;
const DATAPLANE_PENDING_OUTBOUND_FAST_CONTINUATION_TURNS: usize = 2;
const DATAPLANE_PENDING_OUTBOUND_CONTROL_CONTINUATION_TURNS: usize = 8;
const DATAPLANE_PENDING_OUTBOUND_COMPLETION_TIMEOUT: std::time::Duration =
std::time::Duration::from_millis(100);
struct DataplaneFmpOwnerSeed {
owner: OwnerId,
config: OwnerConfig,
keys: OwnerCryptoKeys,
path: TransportPath,
routes: DataplaneLiveOwnerRoutes,
}
struct DataplaneFspOwnerSeed {
owner: OwnerId,
config: OwnerConfig,
keys: OwnerCryptoKeys,
routes: DataplaneLiveOwnerRoutes,
wrap: Option<DataplaneFspWrapRoute>,
path: Option<TransportPath>,
direct_path_mtu: Option<u16>,
}
struct DataplaneFspOwnerSessionSnapshot {
open: ring::aead::LessSafeKey,
seal: ring::aead::LessSafeKey,
counter_authority: crate::noise::SendCounterAuthority,
session_start_ms: u64,
current_k_bit: bool,
previous_draining_k_bit: Option<bool>,
source_peer: PeerIdentity,
is_initiator: bool,
}
struct DataplaneFspOwnerRouteUpdate {
routes: DataplaneLiveOwnerRoutes,
wrap: Option<DataplaneFspWrapRoute>,
path: Option<TransportPath>,
direct_path_mtu: Option<u16>,
next_hop: Option<NodeAddr>,
}
enum DataplanePendingOutboundFailure {
TurnFailed(DataplaneLiveNodeTurn),
Stopped {
turn: DataplaneLiveNodeTurn,
reason: &'static str,
},
Exhausted(DataplaneLiveNodeTurn),
}
#[derive(Clone, Copy)]
struct DataplanePendingOutboundPolicy {
continuation_turns: usize,
await_admitted_output: bool,
}
const DATAPLANE_PENDING_OUTBOUND_FAST_POLICY: DataplanePendingOutboundPolicy =
DataplanePendingOutboundPolicy {
continuation_turns: DATAPLANE_PENDING_OUTBOUND_FAST_CONTINUATION_TURNS,
await_admitted_output: false,
};
const DATAPLANE_PENDING_OUTBOUND_PATIENT_CONTROL_POLICY: DataplanePendingOutboundPolicy =
DataplanePendingOutboundPolicy {
continuation_turns: DATAPLANE_PENDING_OUTBOUND_CONTROL_CONTINUATION_TURNS,
await_admitted_output: true,
};
impl Node {
pub(in crate::node) async fn send_dataplane_fmp_link_plaintext(
&mut self,
node_addr: &NodeAddr,
plaintext: &[u8],
ce_flag: bool,
) -> Result<(), NodeError> {
if !self.dataplane_has_fmp_owner(node_addr) {
return if self.peers.get(node_addr).is_none() {
Err(NodeError::PeerNotFound(*node_addr))
} else {
Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: "dataplane FMP owner not registered".into(),
})
};
}
let Some(send_context) = self.dataplane.fmp_owner_send_context(node_addr) else {
return Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: "dataplane FMP send context unavailable".into(),
});
};
if self.peers.get(node_addr).is_none() {
return Err(NodeError::PeerNotFound(*node_addr));
}
let mut flags = send_context.flags();
if ce_flag {
flags |= FLAG_CE;
}
let outbound = OutboundPacket::fmp(
OwnerId::fmp_node(*node_addr),
send_context.generation(),
dataplane_fmp_link_class(plaintext),
send_context.receiver_idx(),
flags,
crate::transport::PacketBuffer::new(plaintext.to_vec()),
)
.with_activity_tick(ActivityTick::new(Self::now_ms()));
let firsts = DataplaneLiveOutboundFirsts {
initial_outbound: Some(outbound),
collect_transport_sent_receipts: true,
..Default::default()
};
let pending_policy = dataplane_fmp_link_pending_policy(plaintext);
let mut turn = self
.pump_dataplane_pending_outbound_firsts(firsts, 0, 0, 1)
.await;
turn = match self
.drive_dataplane_pending_outbound_turn(
turn,
true,
pending_policy.continuation_turns,
pending_policy.await_admitted_output,
)
.await
{
Ok(turn) => turn,
Err(failure) => {
let failure_turn = match &failure {
DataplanePendingOutboundFailure::TurnFailed(turn)
| DataplanePendingOutboundFailure::Exhausted(turn) => turn,
DataplanePendingOutboundFailure::Stopped { turn, .. } => turn,
};
if let Some(drop) = failure_turn.output_drops().first() {
return Err(self.dataplane_fmp_output_drop_error(*node_addr, drop));
}
return Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: Self::dataplane_pending_outbound_failure_from_stop(
"FMP link send",
&failure,
),
});
}
};
if turn.transport_sent() != 1
|| turn.transport_dropped() != 0
|| turn.summary().outputs_sent() != 1
{
return Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: format!(
"dataplane FMP send unexpected output shape: {:?}",
turn.summary()
),
});
}
let mut sent_receipts = turn.take_transport_sent_receipts();
if sent_receipts.len() != 1 {
return Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: format!(
"dataplane FMP send transport receipt mismatch: {:?}",
turn.summary()
),
});
}
let receipt = sent_receipts.pop().expect("checked one sent receipt");
if receipt.owner != OwnerId::fmp_node(*node_addr) {
return Err(NodeError::SendFailed {
node_addr: *node_addr,
reason: "dataplane FMP send receipt owner mismatch".into(),
});
}
let timestamp_ms = receipt
.fmp_timestamp_ms
.ok_or_else(|| NodeError::SendFailed {
node_addr: *node_addr,
reason: "dataplane FMP timestamp missing".into(),
})?;
let bytes_sent = receipt.payload_len;
self.dataplane.record_fmp_mmp_send_result(
node_addr,
receipt.counter,
timestamp_ms,
bytes_sent,
);
let _ = self
.peers
.record_fmp_send_bookkeeping(node_addr, bytes_sent);
let send_result: Result<usize, TransportError> = Ok(bytes_sent);
self.note_local_send_outcome(node_addr, &send_result);
Ok(())
}
pub(in crate::node) async fn send_dataplane_cached_tun_packet(
&mut self,
dest_addr: &NodeAddr,
packet: Vec<u8>,
) -> Result<(), NodeError> {
if !self.dataplane_has_fsp_owner(dest_addr) {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "dataplane FSP owner not registered for queued TUN packet".into(),
});
}
let turn = self
.pump_dataplane_pending_outbound_firsts(
DataplaneLiveOutboundFirsts {
tun_packet: Some(packet),
..Default::default()
},
0,
1,
1,
)
.await;
if let Some(error) = self.dataplane_cached_tun_drop_error(dest_addr, &turn) {
return Err(error);
}
self.finish_dataplane_pending_outbound_turn(dest_addr, "queued TUN packet", turn, false)
.await
.map(|_| ())
}
fn dataplane_cached_tun_drop_error(
&mut self,
dest_addr: &NodeAddr,
turn: &DataplaneLiveNodeTurn,
) -> Option<NodeError> {
let drop = turn.tun_outbound_drops().first()?;
let packet = drop.packet().to_vec();
let payload_len = drop.payload_len();
match drop.reason() {
crate::dataplane::DataplaneTunOutboundDropReason::MtuExceeded { mtu } => {
self.send_icmpv6_packet_too_big(&packet, mtu);
Some(NodeError::MtuExceeded {
node_addr: *dest_addr,
packet_size: payload_len,
mtu: mtu.min(u32::from(u16::MAX)) as u16,
})
}
crate::dataplane::DataplaneTunOutboundDropReason::NoRoute => {
self.send_icmpv6_dest_unreachable(&packet);
Some(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "dataplane TUN route unavailable".into(),
})
}
crate::dataplane::DataplaneTunOutboundDropReason::InvalidPacket => {
Some(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "dataplane TUN packet invalid".into(),
})
}
}
}
pub(in crate::node) async fn send_dataplane_cached_endpoint_payloads(
&mut self,
dest_addr: &NodeAddr,
payloads: Vec<EndpointDataPayload>,
) -> Result<(), NodeError> {
if payloads.is_empty() {
return Ok(());
}
if !self.dataplane_has_fsp_owner(dest_addr) {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "dataplane FSP owner not registered for queued endpoint data".into(),
});
}
let Some(remote) = self.dataplane_peer_identity(dest_addr) else {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "dataplane endpoint identity unavailable for queued endpoint data".into(),
});
};
let payload_count = payloads.len();
let batch = NodeEndpointDataBatch::from_payloads(remote, payloads, None)
.expect("checked pending endpoint payload batch");
let firsts = DataplaneLiveOutboundFirsts {
endpoint_data_batch: Some(batch),
..Default::default()
};
let turn = self
.pump_dataplane_pending_outbound_firsts(firsts, payload_count, 0, payload_count)
.await;
self.finish_dataplane_pending_outbound_turn(dest_addr, "queued endpoint data", turn, false)
.await
.map(|_| ())
}
pub(in crate::node) async fn send_dataplane_fsp_session_msg(
&mut self,
dest_addr: &NodeAddr,
msg_type: u8,
payload: &[u8],
) -> Result<(), NodeError> {
self.send_dataplane_fsp_control_outbound(
dest_addr,
msg_type,
None,
payload,
None,
"FSP control message",
)
.await
}
pub(in crate::node) async fn send_dataplane_fsp_coords_warmup(
&mut self,
dest_addr: &NodeAddr,
) -> Result<(), NodeError> {
let coords_prefix = self.dataplane_fsp_coords_prefix_for_dest(dest_addr);
self.send_dataplane_fsp_control_outbound(
dest_addr,
SessionMessageType::CoordsWarmup.to_byte(),
Some(crate::node::session_wire::FSP_FLAG_CP),
&[],
Some(coords_prefix),
"FSP coords warmup",
)
.await
}
async fn pump_dataplane_pending_outbound_firsts(
&mut self,
firsts: DataplaneLiveOutboundFirsts,
endpoint_limit: usize,
tun_limit: usize,
crypto_limit: usize,
) -> DataplaneLiveNodeTurn {
let endpoint_tx = self.endpoint_events.sender().unwrap_or_else(|| {
let (tx, rx) = EndpointEventSender::channel(1);
drop(rx);
tx
});
let mut empty_raw_ingress = std::collections::VecDeque::new();
let (_, mut empty_endpoint_data_rx) = endpoint_data_batch_channel(1);
let (_, mut empty_tun_outbound_rx) = crate::upper::tun::tun_outbound_channel(1);
let turn = self
.dataplane
.pump_turn_with_firsts_and_transport_batch(
None,
&mut empty_raw_ingress,
0,
firsts,
DataplaneLiveTurnIo {
endpoint_data_rx: &mut empty_endpoint_data_rx,
endpoint_limit,
tun_outbound_rx: &mut empty_tun_outbound_rx,
tun_limit,
endpoint_tx: &endpoint_tx,
transports: &self.transports,
crypto_limit,
transport_send_batch_packets: self.dataplane_transport_send_batch_packets,
},
)
.await;
Self::observe_dataplane_turn(&turn);
turn
}
async fn send_dataplane_fsp_control_outbound(
&mut self,
dest_addr: &NodeAddr,
msg_type: u8,
fsp_flags_override: Option<u8>,
payload: &[u8],
coords_prefix: Option<Vec<u8>>,
label: &str,
) -> Result<(), NodeError> {
if !self.dataplane_has_fsp_owner(dest_addr) {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("dataplane FSP owner not registered for {label}"),
});
}
let Some(next_hop) = self.dataplane.fsp_owner_next_hop(dest_addr) else {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("dataplane FSP owner route unavailable for {label}"),
});
};
let Some(send_context) = self.dataplane.fsp_owner_send_context(dest_addr) else {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("dataplane FSP owner send context unavailable for {label}"),
});
};
let coords_prefix_len = coords_prefix.as_ref().map_or(0, Vec::len);
let fsp_flags = fsp_flags_override.unwrap_or_else(|| send_context.fsp_flags());
let inner_flags = send_context.inner_flags();
let activity_tick = ActivityTick::new(Self::now_ms());
let mut outbound = OutboundPacket::fsp(
OwnerId::fsp_node(*dest_addr),
send_context.generation(),
dataplane_fsp_control_class(msg_type),
fsp_flags,
crate::transport::PacketBuffer::new(payload.to_vec()),
)
.with_fsp_inner_header(msg_type, inner_flags)
.with_activity_tick(activity_tick);
if let Some(prefix) = coords_prefix {
outbound = outbound.with_fsp_cleartext_prefix(prefix);
} else {
outbound = outbound.without_fsp_auto_coords_warmup();
}
let firsts = DataplaneLiveOutboundFirsts {
initial_outbound: Some(outbound),
collect_transport_sent_receipts: true,
..Default::default()
};
let turn = self
.pump_dataplane_pending_outbound_firsts(firsts, 0, 0, 2)
.await;
let mut turn = match self
.finish_dataplane_pending_outbound_turn(dest_addr, label, turn, true)
.await
{
Ok(turn) => turn,
Err(error) => {
self.record_route_failure(*dest_addr, next_hop);
self.recover_direct_payload_send_failure(*dest_addr, next_hop, &error);
return Err(error);
}
};
if Self::dataplane_sent_fsp_receipt(&mut turn, *dest_addr).is_none() {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("dataplane FSP receipt unavailable for {label}"),
});
}
let frame_bytes = crate::node::session_wire::FSP_INNER_HEADER_SIZE
.saturating_add(payload.len())
.saturating_add(crate::noise::TAG_SIZE);
let datagram_bytes = crate::protocol::SESSION_DATAGRAM_HEADER_SIZE
.saturating_add(crate::node::session_wire::FSP_HEADER_SIZE)
.saturating_add(coords_prefix_len)
.saturating_add(frame_bytes);
self.stats_mut()
.forwarding
.record_originated(datagram_bytes);
Ok(())
}
async fn finish_dataplane_pending_outbound_turn(
&mut self,
dest_addr: &NodeAddr,
label: &str,
turn: DataplaneLiveNodeTurn,
collect_transport_sent_receipts: bool,
) -> Result<DataplaneLiveNodeTurn, NodeError> {
let result = self
.drive_dataplane_pending_outbound_turn(
turn,
collect_transport_sent_receipts,
DATAPLANE_PENDING_OUTBOUND_FAST_POLICY.continuation_turns,
DATAPLANE_PENDING_OUTBOUND_FAST_POLICY.await_admitted_output,
)
.await;
self.process_dataplane_pending_outbound_bookkeeping().await;
match result {
Ok(turn) => Ok(turn),
Err(failure) => Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: Self::dataplane_pending_outbound_failure_from_stop(label, &failure),
}),
}
}
async fn drive_dataplane_pending_outbound_turn(
&mut self,
mut turn: DataplaneLiveNodeTurn,
collect_transport_sent_receipts: bool,
continuation_turns: usize,
await_admitted_output: bool,
) -> Result<DataplaneLiveNodeTurn, DataplanePendingOutboundFailure> {
let mut awaiting_admitted_output = false;
for continuation in 0..=continuation_turns {
let summary = turn.summary();
let sent = Self::dataplane_pending_outbound_sent(&turn);
let deferred =
turn.deferred_endpoint_data_batches_count() > 0 || turn.tun_deferred_packets() > 0;
let failed = turn.has_failures();
let needs_continuation = Self::dataplane_pending_outbound_needs_continuation(&turn);
if failed {
return Err(DataplanePendingOutboundFailure::TurnFailed(turn));
}
if sent {
return Ok(turn);
}
if await_admitted_output && needs_continuation {
awaiting_admitted_output = true;
}
if deferred || (!needs_continuation && !awaiting_admitted_output) {
let reason = if deferred {
"deferred without transport output"
} else {
"made no transport output progress"
};
return Err(DataplanePendingOutboundFailure::Stopped { turn, reason });
}
if continuation == continuation_turns {
return Err(DataplanePendingOutboundFailure::Exhausted(turn));
}
if (needs_continuation || awaiting_admitted_output) && summary.outputs() == 0 {
self.wait_for_dataplane_completion().await;
}
turn = self
.pump_dataplane_pending_outbound_firsts(
DataplaneLiveOutboundFirsts {
collect_transport_sent_receipts,
..Default::default()
},
0,
0,
1,
)
.await;
}
unreachable!("bounded pending outbound continuation loop must return")
}
async fn wait_for_dataplane_completion(&self) {
let notify = self.dataplane.completion_notify();
let _ = tokio::time::timeout(
DATAPLANE_PENDING_OUTBOUND_COMPLETION_TIMEOUT,
notify.notified(),
)
.await;
}
fn dataplane_sent_fsp_receipt(
turn: &mut DataplaneLiveNodeTurn,
dest_addr: NodeAddr,
) -> Option<DataplaneFspSendReceipt> {
let owner = OwnerId::fsp_node(dest_addr);
let mut sent_receipt = None;
for transport_receipt in turn.take_transport_sent_receipts() {
if let Some(receipt) = transport_receipt.fsp_send_receipt
&& receipt.owner == owner
{
sent_receipt = Some(receipt);
}
}
sent_receipt
}
fn dataplane_pending_outbound_sent(turn: &DataplaneLiveNodeTurn) -> bool {
turn.transport_sent() > 0 || turn.summary().outputs_sent() > 0
}
fn dataplane_pending_outbound_needs_continuation(turn: &DataplaneLiveNodeTurn) -> bool {
let summary = turn.summary();
summary.outbound_admitted() > summary.dispatched()
|| (summary.outbound_admitted() > 0 && summary.outputs() == 0)
}
fn dataplane_pending_outbound_failure(label: &str, turn: &DataplaneLiveNodeTurn) -> String {
let summary = turn.summary();
if let Some(drop) = turn.tun_outbound_drops().first() {
return format!(
"dataplane {label} TUN route drop: {:?} ({summary:?})",
drop.reason()
);
}
if let Some(drop) = turn.endpoint_data_drops().first() {
return format!(
"dataplane {label} endpoint route drop: {:?} ({summary:?})",
drop.reason()
);
}
if let Some(drop) = turn.output_drops().first() {
return format!(
"dataplane {label} output drop: {:?} ({summary:?})",
drop.reason()
);
}
if let Some(drop) = turn.drops().first() {
return format!(
"dataplane {label} packet drop: {:?} ({summary:?})",
drop.reason()
);
}
format!("dataplane {label} failed: {summary:?}")
}
fn dataplane_pending_outbound_failure_from_stop(
label: &str,
failure: &DataplanePendingOutboundFailure,
) -> String {
match failure {
DataplanePendingOutboundFailure::TurnFailed(turn) => {
Self::dataplane_pending_outbound_failure(label, turn)
}
DataplanePendingOutboundFailure::Stopped { turn, reason } => {
format!("dataplane {label} {reason}: {:?}", turn.summary())
}
DataplanePendingOutboundFailure::Exhausted(turn) => {
format!(
"dataplane {label} exhausted pending outbound continuation turns: {:?}",
turn.summary()
)
}
}
}
async fn process_dataplane_pending_outbound_bookkeeping(&mut self) -> usize {
let mut processed = 0usize;
for _packet in self.dataplane.take_deferred_tun_packets() {
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
}
pub(in crate::node) fn sync_dataplane_fmp_owner(&mut self, node_addr: &NodeAddr) -> bool {
let Some(seed) = self.dataplane_fmp_owner_seed(node_addr) else {
self.remove_dataplane_fmp_owner(node_addr);
self.refresh_dataplane_fsp_owner_routes_after_fmp_owner_update(node_addr);
return false;
};
self.dataplane
.register_owner_if_missing(seed.owner, seed.config.clone());
let synced = self
.dataplane
.install_owner_fmp_session_routes(
seed.owner,
seed.config,
seed.keys,
seed.path,
seed.routes,
)
.is_ok();
if synced {
self.refresh_dataplane_fsp_owner_routes_after_fmp_owner_update(node_addr);
}
synced
}
pub(in crate::node) fn remove_dataplane_fmp_owner(&mut self, node_addr: &NodeAddr) {
self.dataplane
.unregister_owner(OwnerId::fmp_node(*node_addr));
}
pub(in crate::node) fn dataplane_has_fmp_owner(&self, node_addr: &NodeAddr) -> bool {
self.dataplane.has_owner(OwnerId::fmp_node(*node_addr))
}
pub(in crate::node) fn refresh_dataplane_fsp_owner_routes(
&mut self,
node_addr: &NodeAddr,
) -> bool {
let owner = OwnerId::fsp_node(*node_addr);
let Some(send_context) = self.dataplane.fsp_owner_send_context(node_addr) else {
return false;
};
let update = self.dataplane_fsp_owner_routes(
node_addr,
send_context.generation(),
send_context.fsp_flags(),
send_context.inner_flags(),
);
let route_ready = update.wrap.is_some() || update.path.is_some();
let next_hop_ready = update.path.is_some()
|| update
.next_hop
.is_some_and(|next_hop| self.dataplane_has_fmp_owner(&next_hop));
let direct_path_mtu = update.direct_path_mtu;
let refreshed = self
.dataplane
.replace_owner_fsp_routes(owner, update.routes, update.wrap, update.path)
.is_ok()
&& route_ready
&& next_hop_ready;
if refreshed && let Some(path_mtu) = direct_path_mtu {
let _ = self.dataplane.seed_fsp_path_mtu(*node_addr, path_mtu);
}
refreshed
}
pub(in crate::node) fn refresh_dataplane_fsp_owner_routes_after_fmp_owner_update(
&mut self,
next_hop_addr: &NodeAddr,
) -> usize {
let destinations = self.dataplane.fsp_owner_destinations();
let mut refreshed = 0usize;
for dest in destinations {
let current_uses_next_hop =
self.dataplane.fsp_owner_next_hop(&dest) == Some(*next_hop_addr);
let would_use_next_hop = self
.find_next_hop(&dest)
.is_some_and(|peer| peer.node_addr() == next_hop_addr);
if !(current_uses_next_hop || would_use_next_hop) {
continue;
}
let route_ready = self.refresh_dataplane_fsp_owner_routes(&dest);
if route_ready || current_uses_next_hop {
refreshed = refreshed.saturating_add(1);
}
}
refreshed
}
pub(in crate::node) fn refresh_dataplane_fsp_owner_routes_with_coords_warmup(
&mut self,
node_addr: &NodeAddr,
coords_warmup_remaining: u8,
) -> bool {
let owner = OwnerId::fsp_node(*node_addr);
let coords_prefix = self.dataplane_fsp_coords_prefix(node_addr, coords_warmup_remaining);
let warmup_applied = self
.dataplane
.set_owner_fsp_coords_warmup(owner, coords_warmup_remaining, coords_prefix)
.is_ok();
self.refresh_dataplane_fsp_owner_routes(node_addr) && warmup_applied
}
pub(in crate::node) fn apply_dataplane_fsp_path_mtu_signal(
&mut self,
node_addr: &NodeAddr,
path_mtu: u16,
now: std::time::Instant,
) -> Result<
crate::dataplane::DataplaneFspPathMtuApplyResult,
crate::dataplane::DataplaneFspMmpSkip,
> {
let result = self
.dataplane
.apply_fsp_path_mtu_signal(*node_addr, path_mtu, now)?;
if matches!(
result,
crate::dataplane::DataplaneFspPathMtuApplyResult::Changed(_)
) {
let _ = self.refresh_dataplane_fsp_owner_routes(node_addr);
}
Ok(result)
}
pub(in crate::node) fn set_dataplane_fsp_owner_epoch(
&mut self,
node_addr: &NodeAddr,
current_k_bit: bool,
previous_draining_k_bit: Option<bool>,
) -> bool {
self.dataplane
.set_owner_fsp_epoch(
OwnerId::fsp_node(*node_addr),
current_k_bit,
previous_draining_k_bit,
)
.is_ok()
}
pub(in crate::node) fn install_dataplane_fsp_pending_receive_epoch(
&mut self,
node_addr: &NodeAddr,
pending_k_bit: bool,
open: ring::aead::LessSafeKey,
) -> bool {
self.dataplane
.install_owner_fsp_pending_receive_epoch(
OwnerId::fsp_node(*node_addr),
pending_k_bit,
std::sync::Arc::new(open),
)
.is_ok()
}
pub(in crate::node) fn install_dataplane_fmp_pending_receive_epoch(
&mut self,
node_addr: &NodeAddr,
pending_k_bit: bool,
open: ring::aead::LessSafeKey,
) -> bool {
self.dataplane
.install_owner_fmp_pending_receive_epoch(
OwnerId::fmp_node(*node_addr),
pending_k_bit,
std::sync::Arc::new(open),
)
.is_ok()
}
pub(in crate::node) fn promote_dataplane_authenticated_pending_fmp_epoch(
&mut self,
node_addr: &NodeAddr,
received_k_bit: bool,
) -> bool {
if !self
.dataplane
.fmp_owner_has_pending_receive_epoch(node_addr, received_k_bit)
{
return false;
}
let Some(previous_index) = self
.peers
.get_mut(node_addr)
.and_then(|peer| peer.handle_peer_kbit_flip())
else {
return false;
};
let _ = previous_index;
self.ensure_current_session_index_registered(
node_addr,
"responder authenticated FMP rekey cutover",
);
self.sync_dataplane_fmp_owner(node_addr)
}
pub(in crate::node) fn promote_dataplane_authenticated_pending_fsp_epoch(
&mut self,
node_addr: &NodeAddr,
received_k_bit: bool,
) -> bool {
if !self
.dataplane
.fsp_owner_has_pending_receive_epoch(node_addr, received_k_bit)
{
return false;
}
let now_ms = Self::now_ms();
let promoted = {
let Some(session) = self.sessions.get_mut(node_addr) else {
return false;
};
session.cutover_to_authenticated_pending_epoch(now_ms, received_k_bit)
};
if !promoted {
return false;
}
self.sync_dataplane_fsp_owner_from_current_session(node_addr, 0)
}
pub(in crate::node) fn dataplane_fsp_owner_epoch(
session: &SessionEntry,
) -> (bool, Option<bool>) {
let current_k_bit = session.current_k_bit();
(
current_k_bit,
session.is_draining().then_some(!current_k_bit),
)
}
pub(in crate::node) fn dataplane_has_fsp_owner(&self, node_addr: &NodeAddr) -> bool {
self.dataplane.has_owner(OwnerId::fsp_node(*node_addr))
}
pub(in crate::node) fn dataplane_direct_fsp_sources(
&self,
) -> HashMap<
(
crate::transport::TransportId,
crate::transport::TransportAddr,
),
DataplaneDirectFspSource,
> {
let mut sources = HashMap::new();
for (node_addr, peer) in &self.peers {
let (Some(transport_id), Some(remote_addr)) =
(peer.transport_id(), peer.current_addr().cloned())
else {
continue;
};
let path_mtu = self
.transports
.get(&transport_id)
.map(|transport| transport.link_mtu(&remote_addr))
.unwrap_or_else(|| self.transport_mtu());
sources.insert(
(transport_id, remote_addr),
DataplaneDirectFspSource {
source_addr: *node_addr,
path_mtu,
},
);
}
sources
}
pub(in crate::node) fn sync_dataplane_fsp_owner_from_current_session(
&mut self,
node_addr: &NodeAddr,
coords_warmup_remaining: u8,
) -> bool {
let Some(snapshot) = self
.sessions
.get(node_addr)
.and_then(Self::dataplane_fsp_owner_session_snapshot)
else {
self.remove_dataplane_fsp_owner(node_addr);
return false;
};
self.sync_dataplane_fsp_owner_from_session_snapshot(
node_addr,
snapshot,
coords_warmup_remaining,
)
}
fn sync_dataplane_fsp_owner_from_session_snapshot(
&mut self,
node_addr: &NodeAddr,
snapshot: DataplaneFspOwnerSessionSnapshot,
coords_warmup_remaining: u8,
) -> bool {
let _timer =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DataplaneFspOwnerSync);
crate::perf_profile::record_event(crate::perf_profile::Event::DataplaneFspOwnerSyncCall);
let Some(seed) = self.dataplane_fsp_owner_seed_from_snapshot(
node_addr,
snapshot,
coords_warmup_remaining,
) else {
self.remove_dataplane_fsp_owner(node_addr);
return false;
};
self.apply_dataplane_fsp_owner_seed(seed)
}
fn apply_dataplane_fsp_owner_seed(&mut self, seed: DataplaneFspOwnerSeed) -> bool {
self.dataplane
.register_owner_if_missing(seed.owner, seed.config.clone());
let next_hop_ready = seed
.wrap
.map(DataplaneFspWrapRoute::next_hop_addr)
.is_none_or(|next_hop| self.dataplane_has_fmp_owner(&next_hop));
let synced = self
.dataplane
.install_owner_fsp_session_routes(
seed.owner,
seed.config,
seed.keys,
seed.routes,
seed.wrap,
seed.path,
)
.is_ok()
&& next_hop_ready;
if synced && let Some(path_mtu) = seed.direct_path_mtu {
let _ = self
.dataplane
.seed_fsp_path_mtu(seed.owner.node_addr(), path_mtu);
}
if synced {
crate::perf_profile::record_event(
crate::perf_profile::Event::DataplaneFspOwnerSyncApplied,
);
}
synced
}
pub(in crate::node) fn remove_dataplane_fsp_owner(&mut self, node_addr: &NodeAddr) {
self.dataplane
.unregister_owner(OwnerId::fsp_node(*node_addr));
}
fn dataplane_fmp_owner_seed(&self, node_addr: &NodeAddr) -> Option<DataplaneFmpOwnerSeed> {
let peer = self.peers.get(node_addr)?;
let session = peer.noise_session()?;
let transport_id = peer.transport_id()?;
let remote_addr = peer.current_addr()?.clone();
let receiver_idx = peer.our_index()?.as_u32();
let mut receive_indices = vec![receiver_idx];
for index in [peer.pending_our_index(), peer.previous_our_index()]
.into_iter()
.flatten()
.map(|index| index.as_u32())
{
if !receive_indices.contains(&index) {
receive_indices.push(index);
}
}
let fmp_send_headers = peer.their_index().map(|their_index| {
let mut flags = 0;
if peer.current_k_bit() {
flags |= FLAG_KEY_EPOCH;
}
(their_index.as_u32(), flags)
});
let fmp_mmp_is_initiator = peer.fmp_mmp_is_initiator();
let generation = peer.session_generation();
let session_start_ms = Self::now_ms().wrapping_sub(u64::from(peer.session_elapsed_ms()));
let source_peer = *peer.identity();
let current_k_bit = peer.current_k_bit();
let previous_draining_k_bit = peer.is_draining().then_some(!current_k_bit);
let open = Arc::new(session.recv_cipher_clone()?);
let seal = Arc::new(session.send_cipher_clone()?);
let counter_authority = session.send_counter_authority();
let mut routes = DataplaneLiveOwnerRoutes::new();
for receiver_idx in receive_indices {
routes.push_fmp_ingress(
transport_id,
receiver_idx,
DataplaneIngressRoute::new(
OwnerId::fmp_node(*node_addr),
generation,
OutputTarget::SessionIngress {
local_addr: *self.node_addr(),
},
)
.with_class(PacketClass::Bulk),
);
}
let mut config = self
.dataplane_owner_config(generation)
.with_send_counter_authority(counter_authority)
.with_fmp_session_start_ms(session_start_ms)
.with_fmp_epoch(current_k_bit, previous_draining_k_bit)
.with_source_peer(source_peer);
if let Some((receiver_idx, flags)) = fmp_send_headers {
config = config.with_fmp_send_headers(receiver_idx, flags);
}
config = config.with_fmp_mmp(self.config.node.mmp.clone(), fmp_mmp_is_initiator);
Some(DataplaneFmpOwnerSeed {
owner: OwnerId::fmp_node(*node_addr),
config,
keys: OwnerCryptoKeys::new(open, seal),
path: TransportPath::live(transport_id, remote_addr),
routes,
})
}
fn dataplane_fsp_owner_session_snapshot(
session: &SessionEntry,
) -> Option<DataplaneFspOwnerSessionSnapshot> {
let (open, seal) = session.fsp_crypto_keys()?;
let counter_authority = session.send_counter_authority()?;
let source_peer = session.remote_identity()?;
let current_k_bit = session.current_k_bit();
Some(DataplaneFspOwnerSessionSnapshot {
open,
seal,
counter_authority,
session_start_ms: session.session_start_ms(),
current_k_bit,
previous_draining_k_bit: session.is_draining().then_some(!current_k_bit),
source_peer,
is_initiator: session.is_initiator(),
})
}
fn dataplane_fsp_owner_seed_from_snapshot(
&mut self,
node_addr: &NodeAddr,
snapshot: DataplaneFspOwnerSessionSnapshot,
coords_warmup_remaining: u8,
) -> Option<DataplaneFspOwnerSeed> {
let mut fsp_flags = 0;
if snapshot.current_k_bit {
fsp_flags |= crate::node::session_wire::FSP_FLAG_K;
}
let generation = snapshot.session_start_ms.max(1);
let inner_flags = crate::protocol::FspInnerFlags { spin_bit: false }.to_byte();
let coords_prefix = self.dataplane_fsp_coords_prefix(node_addr, coords_warmup_remaining);
let route_update =
self.dataplane_fsp_owner_routes(node_addr, generation, fsp_flags, inner_flags);
let mut config = self
.dataplane_owner_config(generation)
.with_send_counter_authority(snapshot.counter_authority)
.with_fsp_session_start_ms(snapshot.session_start_ms)
.with_fsp_send_headers(fsp_flags, inner_flags)
.with_fsp_epoch(snapshot.current_k_bit, snapshot.previous_draining_k_bit)
.with_source_peer(snapshot.source_peer);
config = config.with_fsp_mmp(self.config.node.session_mmp.clone(), snapshot.is_initiator);
if coords_warmup_remaining > 0 {
config = config.with_fsp_coords_warmup(coords_warmup_remaining, coords_prefix);
}
Some(DataplaneFspOwnerSeed {
owner: OwnerId::fsp_node(*node_addr),
config,
keys: OwnerCryptoKeys::new(Arc::new(snapshot.open), Arc::new(snapshot.seal)),
routes: route_update.routes,
wrap: route_update.wrap,
path: route_update.path,
direct_path_mtu: route_update.direct_path_mtu,
})
}
fn dataplane_fsp_coords_prefix(
&self,
node_addr: &NodeAddr,
coords_warmup_remaining: u8,
) -> Vec<u8> {
if coords_warmup_remaining == 0 {
return Vec::new();
}
self.dataplane_fsp_coords_prefix_for_dest(node_addr)
}
fn dataplane_fsp_coords_prefix_for_dest(&self, node_addr: &NodeAddr) -> Vec<u8> {
let src = self.tree_state.my_coords().clone();
let dst = self.get_dest_coords(node_addr);
let mut prefix = Vec::with_capacity(
crate::protocol::coords_wire_size(&src) + crate::protocol::coords_wire_size(&dst),
);
crate::protocol::encode_coords(&src, &mut prefix);
crate::protocol::encode_coords(&dst, &mut prefix);
prefix
}
fn dataplane_fsp_owner_routes(
&mut self,
node_addr: &NodeAddr,
generation: u64,
fsp_flags: u8,
inner_flags: u8,
) -> DataplaneFspOwnerRouteUpdate {
let owner = OwnerId::fsp_node(*node_addr);
let Some(next_hop) = self.find_next_hop(node_addr).map(|peer| *peer.node_addr()) else {
return DataplaneFspOwnerRouteUpdate {
routes: DataplaneLiveOwnerRoutes::new(),
wrap: None,
path: None,
direct_path_mtu: None,
next_hop: None,
};
};
let mut direct_path_mtu = None;
let (wrap, path) = if next_hop == *node_addr {
match self.dataplane_direct_fsp_path(node_addr) {
Some((path, path_mtu)) => {
direct_path_mtu = Some(path_mtu);
(None, Some(path))
}
None => (self.dataplane_fsp_wrap_route_to(node_addr, next_hop), None),
}
} else {
(self.dataplane_fsp_wrap_route_to(node_addr, next_hop), None)
};
if wrap.is_none() && path.is_none() {
return DataplaneFspOwnerRouteUpdate {
routes: DataplaneLiveOwnerRoutes::new(),
wrap: None,
path: None,
direct_path_mtu: None,
next_hop: Some(next_hop),
};
};
let mut routes = DataplaneLiveOwnerRoutes::new();
routes.push_fsp_ingress(
*node_addr,
DataplaneIngressRoute::new(
owner,
generation,
OutputTarget::SessionPayload {
local_addr: *self.node_addr(),
},
)
.with_class(PacketClass::Bulk),
);
let tun = DataplaneTunOutboundRoute::fsp_ipv6_shim(
owner,
generation,
PacketClass::Bulk,
fsp_flags,
inner_flags,
)
.with_max_packet_len(self.dataplane_tun_max_packet_len(node_addr));
routes.push_tun_destination(*node_addr, tun);
let mut endpoint =
DataplaneEndpointDataRoute::fsp(owner, generation, fsp_flags, inner_flags);
if direct_path_mtu.is_some() {
endpoint = endpoint.with_direct_transport();
}
routes.push_endpoint_destination(*node_addr, endpoint);
DataplaneFspOwnerRouteUpdate {
routes,
wrap,
path,
direct_path_mtu,
next_hop: Some(next_hop),
}
}
fn dataplane_direct_fsp_path(&self, dest_addr: &NodeAddr) -> Option<(TransportPath, u16)> {
let peer = self.peers.get(dest_addr)?;
let transport_id = peer.transport_id()?;
let remote_addr = peer.current_addr()?.clone();
let path_mtu = self
.transports
.get(&transport_id)
.map(|transport| transport.link_mtu(&remote_addr))
.unwrap_or_else(|| self.transport_mtu());
Some((TransportPath::live(transport_id, remote_addr), path_mtu))
}
fn dataplane_fsp_wrap_route_to(
&mut self,
dest_addr: &NodeAddr,
next_hop: NodeAddr,
) -> Option<DataplaneFspWrapRoute> {
let send_context = self.dataplane.fmp_owner_send_context(&next_hop)?;
let active_path = self
.dataplane
.owner_active_path(OwnerId::fmp_node(next_hop))
.ok()??;
let transport_id = active_path.transport_id;
let remote_addr = active_path.remote_addr.clone();
let fmp_flags = send_context.flags();
let path_mtu = self
.transports
.get(&transport_id)
.map(|transport| transport.link_mtu(&remote_addr))
.unwrap_or_else(|| self.transport_mtu());
let wrap = DataplaneFspWrapRoute::new(
OwnerId::fmp_node(next_hop),
send_context.generation(),
send_context.receiver_idx(),
*self.node_addr(),
*dest_addr,
)
.with_fmp_flags(fmp_flags)
.with_ttl(self.config.node.session.default_ttl)
.with_path_mtu(path_mtu);
Some(wrap)
}
fn dataplane_tun_max_packet_len(&self, dest_addr: &NodeAddr) -> usize {
let effective_mtu = self.effective_ipv6_mtu() as usize;
self.dataplane
.fsp_owner_activity(dest_addr)
.and_then(|activity| activity.current_path_mtu())
.map(crate::upper::icmp::effective_ipv6_mtu)
.map(usize::from)
.filter(|path_ipv6_mtu| *path_ipv6_mtu < effective_mtu)
.unwrap_or(effective_mtu)
}
fn dataplane_owner_config(&self, generation: u64) -> OwnerConfig {
let in_flight_limit = self.config.node.limits.max_pending_inbound.max(1);
OwnerConfig::new(generation, in_flight_limit)
}
fn dataplane_fmp_output_drop_error(
&self,
node_addr: NodeAddr,
drop: &DataplaneOutputDrop,
) -> NodeError {
match drop.reason() {
DataplaneOutputError::MtuExceeded { mtu } => NodeError::MtuExceeded {
node_addr,
packet_size: drop.payload_len(),
mtu,
},
DataplaneOutputError::NoRoute => {
NodeError::LocalRouteUnavailable("dataplane transport route unavailable".into())
}
reason => NodeError::SendFailed {
node_addr,
reason: format!("dataplane transport output failed: {:?}", reason),
},
}
}
}
fn dataplane_fmp_link_class(plaintext: &[u8]) -> PacketClass {
match plaintext
.first()
.and_then(|msg_type| LinkMessageType::from_byte(*msg_type))
{
Some(LinkMessageType::Heartbeat) => PacketClass::Liveness,
Some(LinkMessageType::SenderReport | LinkMessageType::ReceiverReport) => PacketClass::Mmp,
Some(LinkMessageType::SessionDatagram)
if fmp_plaintext_is_bulk_session_datagram(plaintext) =>
{
PacketClass::Bulk
}
_ => PacketClass::Control,
}
}
fn dataplane_fmp_link_pending_policy(plaintext: &[u8]) -> DataplanePendingOutboundPolicy {
if fmp_plaintext_is_fsp_handshake_response_datagram(plaintext) {
DATAPLANE_PENDING_OUTBOUND_PATIENT_CONTROL_POLICY
} else {
DATAPLANE_PENDING_OUTBOUND_FAST_POLICY
}
}
fn fmp_plaintext_is_fsp_handshake_response_datagram(plaintext: &[u8]) -> bool {
if plaintext
.first()
.is_none_or(|ty| *ty != LinkMessageType::SessionDatagram.to_byte())
{
return false;
}
let Some(fsp_payload) = plaintext.get(crate::protocol::SESSION_DATAGRAM_HEADER_SIZE..) else {
return false;
};
FspCommonPrefix::parse(fsp_payload)
.is_some_and(|prefix| matches!(prefix.phase, FSP_PHASE_MSG2 | FSP_PHASE_MSG3))
}
fn dataplane_fsp_control_class(msg_type: u8) -> PacketClass {
match SessionMessageType::from_byte(msg_type) {
Some(
SessionMessageType::SenderReport
| SessionMessageType::ReceiverReport
| SessionMessageType::PathMtuNotification,
) => PacketClass::Mmp,
_ => PacketClass::Control,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dataplane::{
DataplaneLiveOutboundFirsts, DataplaneRawIngress, DataplaneRawIngressDropReason,
PacketProtocol,
};
use crate::peer::{ActivePeer, ActivePeerSession};
use crate::transport::{LinkStats, ReceivedPacket};
use crate::utils::index::SessionIndex;
use std::collections::VecDeque;
fn test_fmp_session(
local: &Identity,
peer: &Identity,
local_epoch: [u8; 8],
peer_epoch: [u8; 8],
) -> crate::noise::NoiseSession {
let mut initiator =
crate::noise::HandshakeState::new_initiator(local.keypair(), peer.pubkey_full());
let mut responder = crate::noise::HandshakeState::new_responder(peer.keypair());
initiator.set_local_epoch(local_epoch);
responder.set_local_epoch(peer_epoch);
let msg1 = initiator.write_message_1().unwrap();
responder.read_message_1(&msg1).unwrap();
let msg2 = responder.write_message_2().unwrap();
initiator.read_message_2(&msg2).unwrap();
initiator.into_session().unwrap()
}
fn invalid_fmp_frame(receiver_idx: SessionIndex, counter: u64, flags: u8) -> Vec<u8> {
let mut frame =
Vec::with_capacity(crate::node::wire::ESTABLISHED_HEADER_SIZE + crate::noise::TAG_SIZE);
frame.push(0);
frame.push(flags);
frame.extend_from_slice(&0u16.to_le_bytes());
frame.extend_from_slice(&receiver_idx.to_le_bytes());
frame.extend_from_slice(&counter.to_le_bytes());
frame.extend_from_slice(&[0; crate::noise::TAG_SIZE]);
frame
}
fn session_datagram_plaintext_with_fsp_prefix(prefix: [u8; 4]) -> Vec<u8> {
let mut plaintext = Vec::with_capacity(crate::protocol::SESSION_DATAGRAM_HEADER_SIZE + 4);
plaintext.push(LinkMessageType::SessionDatagram.to_byte());
plaintext.push(64);
plaintext.extend_from_slice(&1280u16.to_le_bytes());
plaintext.extend_from_slice(&[0; 16]);
plaintext.extend_from_slice(&[1; 16]);
plaintext.extend_from_slice(&prefix);
plaintext
}
#[test]
fn fmp_pending_policy_is_patient_only_for_fsp_handshake_responses() {
let msg1 = session_datagram_plaintext_with_fsp_prefix(
crate::node::session_wire::build_fsp_handshake_prefix(
crate::node::session_wire::FSP_PHASE_MSG1,
0,
),
);
let msg2 = session_datagram_plaintext_with_fsp_prefix(
crate::node::session_wire::build_fsp_handshake_prefix(
crate::node::session_wire::FSP_PHASE_MSG2,
0,
),
);
let msg3 = session_datagram_plaintext_with_fsp_prefix(
crate::node::session_wire::build_fsp_handshake_prefix(
crate::node::session_wire::FSP_PHASE_MSG3,
0,
),
);
let established = session_datagram_plaintext_with_fsp_prefix([
crate::node::session_wire::FSP_VERSION << 4,
0,
0,
0,
]);
assert!(!dataplane_fmp_link_pending_policy(&msg1).await_admitted_output);
assert!(dataplane_fmp_link_pending_policy(&msg2).await_admitted_output);
assert!(dataplane_fmp_link_pending_policy(&msg3).await_admitted_output);
assert!(!dataplane_fmp_link_pending_policy(&established).await_admitted_output);
}
#[tokio::test]
async fn fmp_owner_sync_routes_current_pending_and_previous_receive_indices() {
let mut node = Node::new(Config::new()).unwrap();
let peer_identity_full = Identity::generate();
let peer_identity = PeerIdentity::from_pubkey_full(peer_identity_full.pubkey_full());
let peer_addr = *peer_identity.node_addr();
let transport_id = TransportId::new(77);
let remote_addr = TransportAddr::from_string("127.0.0.1:7777");
let previous_index = SessionIndex::new(10);
let current_index = SessionIndex::new(11);
let pending_index = SessionIndex::new(12);
let current_session =
test_fmp_session(&node.identity, &peer_identity_full, [0x01; 8], [0x02; 8]);
let mut peer = ActivePeer::with_session(
peer_identity,
LinkId::new(77),
1_000,
ActivePeerSession {
session: current_session,
our_index: previous_index,
their_index: SessionIndex::new(20),
transport_id,
current_addr: remote_addr.clone(),
link_stats: LinkStats::new(),
is_initiator: true,
remote_epoch: Some([0x02; 8]),
},
);
let first_pending =
test_fmp_session(&node.identity, &peer_identity_full, [0x03; 8], [0x04; 8]);
peer.set_pending_session(first_pending, current_index, SessionIndex::new(21), false);
assert_eq!(peer.handle_peer_kbit_flip(), Some(previous_index));
let second_pending =
test_fmp_session(&node.identity, &peer_identity_full, [0x05; 8], [0x06; 8]);
peer.set_pending_session(second_pending, pending_index, SessionIndex::new(22), false);
assert_eq!(peer.our_index(), Some(current_index));
assert_eq!(peer.previous_our_index(), Some(previous_index));
assert_eq!(peer.pending_our_index(), Some(pending_index));
node.peers.insert(peer_addr, peer);
assert!(node.sync_dataplane_fmp_owner(&peer_addr));
let mut raw = VecDeque::from([
DataplaneRawIngress::from_received(
PacketProtocol::Fmp,
TransportPath::live(transport_id, remote_addr.clone()),
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
crate::transport::PacketBuffer::new(invalid_fmp_frame(
current_index,
1,
crate::node::wire::FLAG_KEY_EPOCH,
)),
1,
),
),
DataplaneRawIngress::from_received(
PacketProtocol::Fmp,
TransportPath::live(transport_id, remote_addr.clone()),
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
crate::transport::PacketBuffer::new(invalid_fmp_frame(previous_index, 2, 0)),
2,
),
),
DataplaneRawIngress::from_received(
PacketProtocol::Fmp,
TransportPath::live(transport_id, remote_addr.clone()),
ReceivedPacket::with_timestamp(
transport_id,
remote_addr,
crate::transport::PacketBuffer::new(invalid_fmp_frame(pending_index, 3, 0)),
3,
),
),
]);
let (endpoint_tx, endpoint_rx) = EndpointEventSender::channel(1);
drop(endpoint_rx);
let (_, mut endpoint_data_rx) = endpoint_data_batch_channel(1);
let (_, mut tun_outbound_rx) = crate::upper::tun::tun_outbound_channel(1);
let turn = node
.dataplane
.pump_turn_with_firsts_and_transport_batch(
None,
&mut raw,
3,
DataplaneLiveOutboundFirsts::default(),
DataplaneLiveTurnIo {
endpoint_data_rx: &mut endpoint_data_rx,
endpoint_limit: 0,
tun_outbound_rx: &mut tun_outbound_rx,
tun_limit: 0,
endpoint_tx: &endpoint_tx,
transports: &node.transports,
crypto_limit: 3,
transport_send_batch_packets: node.dataplane_transport_send_batch_packets,
},
)
.await;
assert_eq!(turn.summary().inbound_admitted(), 3);
assert!(turn.raw_ingress_drops().is_empty());
assert!(
!turn
.raw_ingress_drops()
.iter()
.any(|drop| drop.reason() == DataplaneRawIngressDropReason::Unrouted)
);
}
}