fn materialize_direct_fsp_segments(
segments: &DataplaneDirectFspTransportSegments,
) -> Vec<PacketOutput> {
let mut outputs = Vec::with_capacity(segments.len());
for index in 0..segments.len() {
let mut payload = Vec::with_capacity(segments.payload_len(index));
let mut slices = [None; crate::transport::udp::UDP_PAYLOAD_MAX_SLICES];
let slice_count = segments.payload_slices(index, &mut slices);
for slice in slices.iter().take(slice_count).flatten() {
payload.extend_from_slice(slice);
}
let mut output = segments.output.clone();
output.payload = PacketBuffer::new(payload);
outputs.push(output);
}
outputs
}
#[test]
fn live_ingress_routes_fmp_by_transport_and_receiver_idx() {
let transport_id = TransportId::new(40);
let other_transport_id = TransportId::new(41);
let remote_addr = TransportAddr::from_string("198.51.100.40:9000");
let source_a = NodeAddr::from_bytes([0x40; 16]);
let source_b = NodeAddr::from_bytes([0x41; 16]);
let owner_a = OwnerId::fmp_node(source_a);
let owner_b = OwnerId::fmp_node(source_b);
let route_a = DataplaneIngressRoute::new(owner_a, 7, OutputTarget::Transport)
.with_class(PacketClass::Liveness);
let route_b = DataplaneIngressRoute::new(owner_b, 8, OutputTarget::Transport)
.with_class(PacketClass::Mmp);
let mut routes = DataplaneLiveRouteTable::default();
routes.register_fmp(transport_id, 404, route_a);
let raw = DataplaneRawIngress::from_live_received(
PacketProtocol::Fmp,
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
PacketBuffer::new(fmp_wire(404, 9, 0)),
9_000,
),
);
let header = DataplaneIngressHeader::Fmp(FmpWireHeader::parse(raw.payload.as_slice()).unwrap());
assert_eq!(raw.path.transport_id, transport_id);
assert_eq!(raw.path.remote_addr, remote_addr);
assert_eq!(raw.activity_tick, Some(ActivityTick::new(9_000)));
assert_eq!(routes.route(&raw, header), Some(route_a));
let wrong_transport = DataplaneRawIngress::from_live_received(
PacketProtocol::Fmp,
ReceivedPacket::with_timestamp(
other_transport_id,
remote_addr.clone(),
PacketBuffer::new(fmp_wire(404, 10, 0)),
9_001,
),
);
let header =
DataplaneIngressHeader::Fmp(
FmpWireHeader::parse(wrong_transport.payload.as_slice()).unwrap(),
);
assert_eq!(routes.route(&wrong_transport, header), None);
routes.register_fmp(other_transport_id, 404, route_b);
let header =
DataplaneIngressHeader::Fmp(FmpWireHeader::parse(raw.payload.as_slice()).unwrap());
assert_eq!(routes.route(&raw, header), Some(route_a));
let header =
DataplaneIngressHeader::Fmp(
FmpWireHeader::parse(wrong_transport.payload.as_slice()).unwrap(),
);
assert_eq!(routes.route(&wrong_transport, header), Some(route_b));
routes.register_fmp(transport_id, 404, route_b);
let header =
DataplaneIngressHeader::Fmp(FmpWireHeader::parse(raw.payload.as_slice()).unwrap());
assert_eq!(routes.route(&raw, header), Some(route_b));
assert_eq!(routes.unregister_owner(owner_b), 1);
let header =
DataplaneIngressHeader::Fmp(FmpWireHeader::parse(raw.payload.as_slice()).unwrap());
assert_eq!(routes.route(&raw, header), None);
}
#[test]
fn live_ingress_routes_fsp_require_source_context_and_refresh_cleanly() {
let source = NodeAddr::from_bytes([0x42; 16]);
let owner = OwnerId::fsp_node(source);
let mut routes = DataplaneLiveRouteTable::default();
let old_route = DataplaneIngressRoute::new(owner, 3, OutputTarget::Transport)
.with_class(PacketClass::Bulk);
let new_route = DataplaneIngressRoute::new(owner, 4, OutputTarget::Transport)
.with_class(PacketClass::Mmp);
routes.register_fsp(source, old_route);
let bare_raw = DataplaneRawIngress::from_live_received(
PacketProtocol::Fsp,
ReceivedPacket::with_timestamp(
TransportId::new(42),
TransportAddr::from_string("198.51.100.42:9000"),
PacketBuffer::new(fsp_wire(77, 0)),
1,
),
);
let header =
DataplaneIngressHeader::Fsp(FspWireHeader::parse(bare_raw.payload.as_slice()).unwrap());
assert_eq!(bare_raw.fsp_source, None);
assert_eq!(routes.route(&bare_raw, header), None);
let sourced_raw = bare_raw.clone().with_fsp_source(source);
let header =
DataplaneIngressHeader::Fsp(
FspWireHeader::parse(sourced_raw.payload.as_slice()).unwrap(),
);
assert_eq!(sourced_raw.fsp_source, Some(source));
let routed = routes.route(&sourced_raw, header).expect("sourced FSP route");
assert_eq!(routed.owner, old_route.owner);
assert_eq!(routed.generation, old_route.generation);
assert_eq!(routed.output, old_route.output);
assert_eq!(routed.class, PacketClass::Bulk);
routes.register_fsp(source, new_route);
let header =
DataplaneIngressHeader::Fsp(
FspWireHeader::parse(sourced_raw.payload.as_slice()).unwrap(),
);
assert_eq!(routes.route(&sourced_raw, header), Some(new_route));
assert_eq!(routes.unregister_owner(owner), 1);
let header =
DataplaneIngressHeader::Fsp(
FspWireHeader::parse(sourced_raw.payload.as_slice()).unwrap(),
);
assert_eq!(routes.route(&sourced_raw, header), None);
}
#[test]
fn packet_rx_source_classifies_flagged_direct_fsp_with_source_context() {
let source = NodeAddr::from_bytes([0x44; 16]);
let transport_id = TransportId::new(44);
let remote_addr = TransportAddr::from_string("198.51.100.44:9000");
let (_tx, mut rx) = crate::transport::packet_channel(1);
let mut direct_sources = std::collections::HashMap::new();
direct_sources.insert(
(transport_id, remote_addr.clone()),
DataplaneDirectFspSource {
source_addr: source,
path_mtu: 1400,
},
);
let first = ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
PacketBuffer::new(fsp_wire(
88,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
)),
44_000,
);
let mut source_rx =
DataplaneFmpPacketRxSource::with_first_direct_fsp_sources_and_reassembler(
&mut rx,
Some(first),
std::sync::Arc::new(direct_sources),
None,
);
let mut packets = Vec::new();
assert_eq!(
source_rx.drain_raw_ingress(1, |packet| packets.push(packet)),
1
);
assert!(source_rx.take_control_ingress().is_empty());
assert_eq!(packets.len(), 1);
let packet = &packets[0];
assert_eq!(packet.protocol, PacketProtocol::Fsp);
assert_eq!(packet.fsp_source, Some(source));
assert_eq!(packet.previous_hop, Some(source));
assert_eq!(packet.path_mtu, 1400);
assert_eq!(packet.path.transport_id, transport_id);
assert_eq!(packet.path.remote_addr, remote_addr);
assert_eq!(packet.activity_tick, Some(ActivityTick::new(44_000)));
}
#[test]
fn direct_fsp_transport_segments_reassemble_before_classification() {
let source = NodeAddr::from_bytes([0x45; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(45);
let remote_addr = TransportAddr::from_string("198.51.100.45:9000");
let mut direct_sources = std::collections::HashMap::new();
direct_sources.insert(
(transport_id, remote_addr.clone()),
DataplaneDirectFspSource {
source_addr: source,
path_mtu: 220,
},
);
let direct_sources = std::sync::Arc::new(direct_sources);
let mut wire = fsp_wire(
4242,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
);
wire.extend((0..700).map(|idx| (idx % 251) as u8));
let mut output =
transport_output(owner, 4242, 9, transport_id, remote_addr.clone(), wire.clone());
output.path_mtu = 220;
let segments = match dataplane_direct_fsp_transport_output(output) {
DataplaneDirectFspTransportOutput::Segments(segments) => {
materialize_direct_fsp_segments(&segments)
}
DataplaneDirectFspTransportOutput::Whole(_) => panic!("expected segmented output"),
DataplaneDirectFspTransportOutput::MtuExceeded(_) => panic!("expected segmented output"),
};
assert!(segments.len() > 1);
assert!(segments.iter().all(|segment| segment.payload_len() <= 220));
assert!(segments.iter().all(|segment| {
dataplane_direct_fsp_transport_fragment_is_fragment(segment.payload())
}));
let (_tx, mut rx) = crate::transport::packet_channel(1);
let mut reassembler = DataplaneDirectFspReassembler::default();
let mut packets = Vec::new();
for (idx, segment) in segments.into_iter().rev().enumerate() {
let received = ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload,
45_000 + idx as u64,
);
let mut source_rx =
DataplaneFmpPacketRxSource::with_first_direct_fsp_sources_and_reassembler(
&mut rx,
Some(received),
direct_sources.clone(),
Some(&mut reassembler),
);
assert_eq!(
source_rx.drain_raw_ingress(1, |packet| packets.push(packet)),
1
);
assert!(source_rx.take_control_ingress().is_empty());
}
assert_eq!(packets.len(), 1);
let packet = &packets[0];
assert_eq!(packet.protocol, PacketProtocol::Fsp);
assert_eq!(packet.fsp_source, Some(source));
assert_eq!(packet.path_mtu, 220);
assert_eq!(packet.payload.as_slice(), wire.as_slice());
}
#[test]
fn direct_fsp_reassembly_keeps_existing_record_at_capacity() {
let source = NodeAddr::from_bytes([0x49; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(49);
let remote_addr = TransportAddr::from_string("198.51.100.49:9000");
let make_segments = |counter: u64, ingress_seq: u64| {
let mut wire = fsp_wire(
counter,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
);
wire.extend((0..700).map(|idx| (idx % 251) as u8));
let mut output = transport_output(
owner,
counter,
ingress_seq,
transport_id,
remote_addr.clone(),
wire.clone(),
);
output.path_mtu = 220;
let segments = match dataplane_direct_fsp_transport_output(output) {
DataplaneDirectFspTransportOutput::Segments(segments) => {
materialize_direct_fsp_segments(&segments)
}
DataplaneDirectFspTransportOutput::Whole(_) => panic!("expected segmented output"),
DataplaneDirectFspTransportOutput::MtuExceeded(_) => panic!("expected segmented output"),
};
assert!(segments.len() > 1);
(wire, segments)
};
let (target_wire, target_segments) = make_segments(49_000, 0);
let mut reassembler = DataplaneDirectFspReassembler::default();
assert!(matches!(
reassembler.ingest_fragment(ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
target_segments[0].payload.clone(),
49_000,
)),
DataplaneDirectFspReassemblyResult::Pending
));
for idx in 0..DIRECT_FSP_TRANSPORT_MAX_REASSEMBLY_RECORDS - 1 {
let (_wire, segments) = make_segments(50_000 + idx as u64, idx as u64 + 1);
assert!(matches!(
reassembler.ingest_fragment(ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segments[0].payload.clone(),
49_001 + idx as u64,
)),
DataplaneDirectFspReassemblyResult::Pending
));
}
assert_eq!(
reassembler.entries.len(),
DIRECT_FSP_TRANSPORT_MAX_REASSEMBLY_RECORDS
);
let mut complete = None;
for (idx, segment) in target_segments.iter().enumerate().skip(1) {
match reassembler.ingest_fragment(ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload.clone(),
49_700 + idx as u64,
)) {
DataplaneDirectFspReassemblyResult::Pending => {}
DataplaneDirectFspReassemblyResult::Complete(packet) => {
complete = Some(packet);
}
other => panic!("unexpected reassembly result: {other:?}"),
}
}
let packet = complete.expect("capacity should not evict a live matching record");
assert_eq!(packet.data.as_slice(), target_wire.as_slice());
}
#[test]
fn fast_ingress_routes_direct_fsp_segments_before_packet_channel() {
let source = NodeAddr::from_bytes([0x46; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(46);
let remote_addr = TransportAddr::from_string("198.51.100.46:9000");
let route = DataplaneIngressRoute::new(owner, 10, OutputTarget::Transport)
.with_class(PacketClass::Bulk);
let mut routes = DataplaneLiveRouteTable::default();
routes.register_fsp(source, route);
let mut direct_sources = std::collections::HashMap::new();
direct_sources.insert(
(transport_id, remote_addr.clone()),
DataplaneDirectFspSource {
source_addr: source,
path_mtu: 240,
},
);
routes.set_established_fast_ingress_direct_fsp_sources(std::sync::Arc::new(direct_sources));
let mut wire = fsp_wire(
4646,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
);
wire.extend((0..900).map(|idx| (idx % 251) as u8));
let mut output =
transport_output(owner, 4646, 10, transport_id, remote_addr.clone(), wire.clone());
output.path_mtu = 240;
let segments = match dataplane_direct_fsp_transport_output(output) {
DataplaneDirectFspTransportOutput::Segments(segments) => {
materialize_direct_fsp_segments(&segments)
}
DataplaneDirectFspTransportOutput::Whole(_) => panic!("expected segmented output"),
DataplaneDirectFspTransportOutput::MtuExceeded(_) => panic!("expected segmented output"),
};
assert!(segments.len() > 1);
let (sink, mut fast_rx) =
DataplaneEstablishedFastIngressSink::channel(
routes.established_fast_ingress_snapshot(),
64,
);
let mut first_half: Vec<_> = segments[..segments.len() / 2]
.iter()
.enumerate()
.map(|(idx, segment)| {
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload.clone(),
46_000 + idx as u64,
)
})
.collect();
let first_half_len = first_half.len();
assert_eq!(sink.try_ingest_batch(&mut first_half), first_half_len);
assert!(first_half.is_empty());
assert!(fast_rx.try_recv().is_err());
let mut second_half: Vec<_> = segments[segments.len() / 2..]
.iter()
.enumerate()
.map(|(idx, segment)| {
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload.clone(),
46_100 + idx as u64,
)
})
.collect();
assert_eq!(
sink.try_ingest_batch(&mut second_half),
segments.len() - segments.len() / 2
);
assert!(second_half.is_empty());
let batch = fast_rx.try_recv().expect("direct FSP fast batch");
assert_eq!(batch.len(), 1);
let mut runs = batch.into_runs();
assert_eq!(runs.len(), 1);
let (run_owner, run_lane, mut packets) = runs.pop().unwrap().into_parts();
assert_eq!(run_owner, owner);
assert_eq!(run_lane, PacketClass::Bulk.lane());
let packet = packets.pop().expect("direct FSP socket packet");
assert!(packets.is_empty());
assert_eq!(packet.owner, owner);
assert_eq!(packet.generation, 10);
assert_eq!(packet.counter, 4646);
assert_eq!(packet.class, PacketClass::Bulk);
assert_eq!(packet.output, OutputTarget::Transport);
assert_eq!(
packet.source_path,
Some(TransportPath::live(transport_id, remote_addr))
);
assert_eq!(packet.previous_hop, Some(source));
assert_eq!(packet.path_mtu, 240);
assert_eq!(
packet.wire_flags & crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT
);
assert_eq!(packet.payload.as_slice(), wire.as_slice());
}
#[test]
fn fast_ingress_consumes_direct_fsp_fragments_while_output_queue_full() {
let source = NodeAddr::from_bytes([0x48; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(48);
let remote_addr = TransportAddr::from_string("198.51.100.48:9000");
let route = DataplaneIngressRoute::new(owner, 12, OutputTarget::Transport)
.with_class(PacketClass::Bulk);
let mut routes = DataplaneLiveRouteTable::default();
routes.register_fsp(source, route);
let mut direct_sources = std::collections::HashMap::new();
direct_sources.insert(
(transport_id, remote_addr.clone()),
DataplaneDirectFspSource {
source_addr: source,
path_mtu: 240,
},
);
routes.set_established_fast_ingress_direct_fsp_sources(std::sync::Arc::new(direct_sources));
let (sink, fast_rx) =
DataplaneEstablishedFastIngressSink::channel(
routes.established_fast_ingress_snapshot(),
1,
);
let mut fill_queue = vec![ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
PacketBuffer::new(fsp_wire(
4800,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
)),
48_000,
)];
assert_eq!(sink.try_ingest_batch(&mut fill_queue), 1);
assert!(fill_queue.is_empty());
assert_eq!(fast_rx.len(), 1);
let mut wire = fsp_wire(
4848,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
);
wire.extend((0..900).map(|idx| (idx % 251) as u8));
let mut output =
transport_output(owner, 4848, 12, transport_id, remote_addr.clone(), wire.clone());
output.path_mtu = 240;
let segments = match dataplane_direct_fsp_transport_output(output) {
DataplaneDirectFspTransportOutput::Segments(segments) => {
materialize_direct_fsp_segments(&segments)
}
DataplaneDirectFspTransportOutput::Whole(_) => panic!("expected segmented output"),
DataplaneDirectFspTransportOutput::MtuExceeded(_) => panic!("expected segmented output"),
};
assert!(segments.len() > 2);
let split = segments.len() / 2;
let mut first_half: Vec<_> = segments[..split]
.iter()
.enumerate()
.map(|(idx, segment)| {
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload.clone(),
48_100 + idx as u64,
)
})
.collect();
let first_half_len = first_half.len();
assert_eq!(sink.try_ingest_batch(&mut first_half), first_half_len);
assert!(
first_half.is_empty(),
"pending direct-FSP fragments should not fall back to the packet channel"
);
assert_eq!(fast_rx.len(), 1);
let mut second_half: Vec<_> = segments[split..]
.iter()
.enumerate()
.map(|(idx, segment)| {
ReceivedPacket::with_timestamp(
transport_id,
remote_addr.clone(),
segment.payload.clone(),
48_200 + idx as u64,
)
})
.collect();
let second_half_len = second_half.len();
assert_eq!(sink.try_ingest_batch(&mut second_half), second_half_len);
assert_eq!(
second_half.len(),
1,
"full fast queue should fall back with one reassembled direct-FSP record"
);
assert_eq!(second_half[0].data.as_slice(), wire.as_slice());
assert_eq!(fast_rx.len(), 1);
}
#[test]
fn fast_ingress_preserves_direct_fsp_fragment_until_route_exists() {
let source = NodeAddr::from_bytes([0x47; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(47);
let remote_addr = TransportAddr::from_string("198.51.100.47:9000");
let routes = DataplaneLiveRouteTable::default();
let mut direct_sources = std::collections::HashMap::new();
direct_sources.insert(
(transport_id, remote_addr.clone()),
DataplaneDirectFspSource {
source_addr: source,
path_mtu: 240,
},
);
routes.set_established_fast_ingress_direct_fsp_sources(std::sync::Arc::new(direct_sources));
let mut wire = fsp_wire(
4747,
crate::node::session_wire::FSP_FLAG_DIRECT_TRANSPORT,
);
wire.extend((0..900).map(|idx| (idx % 251) as u8));
let mut output =
transport_output(owner, 4747, 10, transport_id, remote_addr.clone(), wire);
output.path_mtu = 240;
let segments = match dataplane_direct_fsp_transport_output(output) {
DataplaneDirectFspTransportOutput::Segments(segments) => {
materialize_direct_fsp_segments(&segments)
}
DataplaneDirectFspTransportOutput::Whole(_) => panic!("expected segmented output"),
DataplaneDirectFspTransportOutput::MtuExceeded(_) => panic!("expected segmented output"),
};
let fragment = segments.into_iter().next().expect("fragment").payload;
let original = fragment.clone();
let (sink, mut fast_rx) =
DataplaneEstablishedFastIngressSink::channel(
routes.established_fast_ingress_snapshot(),
64,
);
let mut packets = vec![ReceivedPacket::with_timestamp(
transport_id,
remote_addr,
fragment,
47_000,
)];
assert_eq!(sink.try_ingest_batch(&mut packets), 0);
assert_eq!(packets.len(), 1);
assert_eq!(packets[0].data.as_slice(), original.as_slice());
assert!(fast_rx.try_recv().is_err());
}
#[test]
fn live_ingress_keeps_encrypted_fsp_bulk_before_decrypt() {
let source = NodeAddr::from_bytes([0x43; 16]);
let local = NodeAddr::from_bytes([0x44; 16]);
let owner = OwnerId::fsp_node(source);
let transport_id = TransportId::new(43);
let remote_addr = TransportAddr::from_string("198.51.100.43:9000");
let mut routes = DataplaneLiveRouteTable::default();
let route = DataplaneIngressRoute::new(
owner,
3,
OutputTarget::SessionPayload { local_addr: local },
)
.with_class(PacketClass::Bulk);
routes.register_fsp(source, route);
let mut large_wire = fsp_wire(79, 0);
large_wire.resize(
FSP_HEADER_SIZE
.saturating_add(FSP_INNER_HEADER_SIZE)
.saturating_add(crate::node::ENDPOINT_EVENT_TEST_PAYLOAD_LEN)
.saturating_add(AEAD_TAG_SIZE)
.saturating_add(1),
0,
);
let large_raw = DataplaneRawIngress::from_live_received(
PacketProtocol::Fsp,
ReceivedPacket::with_timestamp(
transport_id,
remote_addr,
PacketBuffer::new(large_wire),
43_001,
),
)
.with_fsp_source(source);
let header =
DataplaneIngressHeader::Fsp(FspWireHeader::parse(large_raw.payload.as_slice()).unwrap());
assert_eq!(
routes.route(&large_raw, header).expect("large FSP route").class,
PacketClass::Bulk
);
}