use super::*;
#[test]
fn zero_template_capacity_does_not_retain_timestamp_entries() {
crate::tests::ensure_common_test_schema();
let ty = crate::config::register_data_type_with_description(
"ROUTER_ZERO_TEMPLATE_CAPACITY",
"",
crate::MessageElement::Static(1, crate::MessageDataType::UInt8, crate::MessageClass::Data),
&[crate::DataEndpoint::named("RADIO")],
crate::ReliableMode::None,
1,
)
.unwrap();
let tx = Router::new(RouterConfig::default());
let rx = Router::new(RouterConfig::default());
let opts = RouterSideOptions {
max_side_transport_templates: 0,
..RouterSideOptions::default().with_small_packet_transport(0)
};
let a = tx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
let b = rx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
for index in 0..32 {
let packet = Packet::new(
ty,
&[crate::DataEndpoint::named("RADIO")],
&format!("PEER{index}"),
index,
Arc::<[u8]>::from([0u8]),
)
.unwrap();
let raw = wire_format::pack_packet(&packet);
let frames = tx
.encode_side_transport_frames(
a,
opts.clone(),
crate::shared_bytes::convert(raw.clone()),
)
.unwrap()
.collect::<Vec<_>>();
assert_eq!(
rx.decode_side_transport_frame(b, &frames[0]).unwrap(),
Some(crate::shared_bytes::convert(raw))
);
}
let st = tx.state.lock();
assert!(st.side_transport[&a].tx_template_ids.is_empty());
assert!(
st.side_transport[&a].tx_last_timestamps.is_empty(),
"disabled compression must not leak one timestamp entry per header"
);
drop(st);
let st = rx.state.lock();
assert_eq!(st.side_transport[&b].rx_template_count(), 0);
assert!(
st.side_transport[&b].rx_last_timestamps.is_empty(),
"zero-capacity RX must not retain timestamps without templates"
);
}
#[test]
fn reliable_control_survives_a_missing_compact_dictionary() {
crate::tests::ensure_common_test_schema();
let status = crate::config::register_data_type_with_description(
"ROUTER_RELIABLE_HEADER_RECOVERY",
"ordered status header recovery",
crate::MessageElement::Static(2, crate::MessageDataType::UInt8, crate::MessageClass::Data),
&[crate::DataEndpoint::named("RADIO")],
crate::ReliableMode::Ordered,
200,
)
.unwrap();
for ty in [
crate::DataType::ReliableAck,
crate::DataType::ReliablePartialAck,
crate::DataType::ReliablePacketRequest,
status,
] {
let tx = Router::new(RouterConfig::default());
let rx = Router::new(RouterConfig::default());
let opts = RouterSideOptions::default().with_small_packet_transport(0);
let a = tx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
let b = rx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
let packet = Packet::new(
ty,
crate::message_meta(ty).endpoints_ref(),
"GS",
100,
Arc::<[u8]>::from(vec![0u8; crate::get_needed_message_size(ty)]),
)
.unwrap();
let raw = wire_format::pack_packet(&packet);
let _ = tx
.encode_side_transport_frames(
a,
opts.clone(),
crate::shared_bytes::convert(raw.clone()),
)
.unwrap();
let frames = tx
.encode_side_transport_frames(a, opts, crate::shared_bytes::convert(raw.clone()))
.unwrap()
.collect::<Vec<_>>();
assert_eq!(
rx.decode_side_transport_frame(b, &frames[0]).unwrap(),
Some(crate::shared_bytes::convert(raw)),
"reliability control cannot depend on a possibly lost template"
);
assert!(
tx.state.lock().side_transport[&a]
.tx_template_ids
.is_empty(),
"self-describing reliable frames must not retain unused TX templates"
);
assert_eq!(
rx.state.lock().side_transport[&b].rx_template_count(),
0,
"self-describing reliable frames must not consume the RX dictionary"
);
}
}
#[test]
fn restarting_peer_discovery_request_refreshes_next_application_header() {
for schema_request in [false, true] {
check_restarting_peer_header(schema_request);
}
}
fn check_restarting_peer_header(schema_request: bool) {
crate::tests::ensure_common_test_schema();
let ty = crate::DataType::try_named("ROUTER_RESTART_BEST_EFFORT").unwrap_or_else(|| {
crate::config::register_data_type_with_description(
"ROUTER_RESTART_BEST_EFFORT",
"",
crate::MessageElement::Static(
12,
crate::MessageDataType::UInt8,
crate::MessageClass::Data,
),
&[crate::DataEndpoint::named("RADIO")],
crate::ReliableMode::None,
1,
)
.unwrap()
});
let tx = Router::new_with_clock(RouterConfig::new([]).with_sender("GB"), Box::new(|| 100u64));
let opts = RouterSideOptions::default().with_small_packet_transport(0);
let side = tx.add_side_packed_with_options("uart", |_| Ok(()), opts.clone());
let unrelated = tx.add_side_packed_with_options("radio", |_| Ok(()), opts.clone());
let endpoint = DataEndpoint::named("RADIO");
tx.rx_from_side(
&discovery::build_discovery_announce("GS", 1, &[endpoint]).unwrap(),
side,
)
.unwrap();
let packet = Packet::new(ty, &[endpoint], "AB", 50, Arc::<[u8]>::from([0u8; 12])).unwrap();
let initial = tx
.encode_side_transport_frames(
side,
opts.clone(),
crate::shared_bytes::convert(wire_format::pack_packet(&packet)),
)
.unwrap()
.collect::<Vec<_>>();
tx.decode_side_transport_frame(side, &initial[0])
.unwrap()
.unwrap();
let _ = tx
.encode_side_transport_frames(
unrelated,
opts.clone(),
crate::shared_bytes::convert(wire_format::pack_packet(&packet)),
)
.unwrap();
let restarted = Router::new(RouterConfig::default());
let rx_side = restarted.add_side_packed_with_options("uart", |_| Ok(()), opts.clone());
let request = if schema_request {
discovery::build_discovery_schema_request("GS", 2)
} else {
discovery::build_discovery_topology_request("GS", 2)
}
.unwrap();
tx.rx_from_side(&request, side).unwrap();
let frames = tx
.encode_side_transport_frames(
side,
opts.clone(),
crate::shared_bytes::convert(wire_format::pack_packet(&packet)),
)
.unwrap()
.collect::<Vec<_>>();
assert!(
restarted
.decode_side_transport_frame(rx_side, &frames[0])
.unwrap()
.is_some(),
"a peer requesting discovery must receive a self-describing application header immediately"
);
assert_eq!(
tx.state.lock().side_transport[&side].rx_template_count(),
1,
"incoming dictionaries must survive a transmit refresh"
);
let unrelated_frames = tx
.encode_side_transport_frames(
unrelated,
opts.clone(),
crate::shared_bytes::convert(wire_format::pack_packet(&packet)),
)
.unwrap()
.collect::<Vec<_>>();
let empty = Router::new(RouterConfig::default());
let empty_side = empty.add_side_packed_with_options("wire", |_| Ok(()), opts);
assert!(
empty
.decode_side_transport_frame(empty_side, &unrelated_frames[0])
.unwrap()
.is_none(),
"unrelated links must keep their existing compact dictionary"
);
}
#[test]
fn reliable_missing_route_is_an_error_and_same_packet_can_retry_after_discovery() {
crate::tests::ensure_common_test_schema();
let destination = DataEndpoint::named("RADIO");
let ty = crate::config::register_data_type_with_description(
"NO_ROUTE_RETRY_TEST",
"reliable route recovery",
MessageElement::Static(1, crate::MessageDataType::UInt8, crate::MessageClass::Data),
&[destination],
crate::ReliableMode::Ordered,
200,
)
.unwrap();
let received = Arc::new(core::sync::atomic::AtomicU32::new(0));
let received_cb = received.clone();
let router = Router::new(RouterConfig::default());
let side = router.add_side_packet("can", move |_| {
received_cb.fetch_add(1, Ordering::Relaxed);
Ok(())
});
router
.rx_from_side(
&discovery::build_discovery_announce("PEER", 1, &[DataEndpoint::named("SD_CARD")])
.unwrap(),
side,
)
.unwrap();
let packet = Packet::new(ty, &[destination], "VB", 42, Arc::from([1u8].as_slice())).unwrap();
assert!(
router.tx(packet.clone()).is_err(),
"a missing remote route must not masquerade as successful delivery"
);
assert_eq!(received.load(Ordering::Relaxed), 0);
router
.rx_from_side(
&discovery::build_discovery_announce("PEER", 2, &[destination]).unwrap(),
side,
)
.unwrap();
router.tx(packet).unwrap();
assert_eq!(
received.load(Ordering::Relaxed),
1,
"failed submission must not poison duplicate tracking"
);
}
#[test]
fn discovery_refresh_survives_a_missed_transport_template() {
crate::tests::ensure_common_test_schema();
let tx = Router::new(RouterConfig::default());
let rx = Router::new(RouterConfig::default());
let opts = RouterSideOptions::default().with_small_packet_transport(512);
let tx_side = tx.add_side_packed_with_options("can", |_| Ok(()), opts.clone());
let rx_side = rx.add_side_packed_with_options("can", |_| Ok(()), opts.clone());
for timestamp in [1000, 6000] {
let packet = discovery::build_discovery_announce("GB", timestamp, &[]).unwrap();
let frames = tx
.encode_side_transport_frames(
tx_side,
opts.clone(),
crate::shared_bytes::convert(wire_format::pack_packet(&packet)),
)
.unwrap()
.collect::<Vec<_>>();
if timestamp == 1000 {
continue;
}
let decoded = rx.decode_side_transport_frame(rx_side, &frames[0]).unwrap();
assert!(
decoded.is_some(),
"discovery must bootstrap without a dictionary"
);
}
}
#[test]
fn compact_loss_must_not_change_packet_identity() {
crate::tests::ensure_common_test_schema();
let tx = Router::new_with_clock(RouterConfig::new([]).with_sender("VB"), Box::new(|| 0u64));
let rx = Router::new_with_clock(RouterConfig::new([]).with_sender("GS"), Box::new(|| 0u64));
let opts = RouterSideOptions::default().with_small_packet_transport(0);
let a = tx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
let b = rx.add_side_packed_with_options("wire", |_| Ok(()), opts.clone());
for (index, timestamp) in [10000, 10010, 10020].into_iter().enumerate() {
let pkt = Packet::new(
DataType::named("GPS_DATA"),
&[DataEndpoint::named("RADIO")],
"VB",
timestamp,
Arc::<[u8]>::from([0u8; 12]),
)
.unwrap();
let raw = wire_format::pack_packet(&pkt);
let frames = tx
.encode_side_transport_frames(
a,
opts.clone(),
crate::shared_bytes::convert(raw.clone()),
)
.unwrap()
.collect::<Vec<_>>();
assert_eq!(frames.len(), 1);
if index == 1 {
continue;
} let received = rx
.decode_side_transport_frame(b, &frames[0])
.unwrap()
.unwrap();
assert_eq!(
wire_format::unpack_packet(&received).unwrap().timestamp(),
timestamp
);
assert_eq!(
received, raw,
"loss must not silently rewrite a later packet"
);
}
}
#[test]
fn generated_ack_keeps_original_sender_after_return_cache_eviction() {
check_ack_return_identity("AB");
}
#[test]
fn compact_sender_ack_returns_after_cache_eviction() {
check_ack_return_identity(&format!("@addr:{}", sender_address_u32("AB")));
}
fn check_ack_return_identity(original_sender: &str) {
crate::tests::ensure_common_test_schema();
let router = Router::new_with_clock(
RouterConfig::new([])
.with_sender("GS")
.with_reliable_enabled(true),
Box::new(|| 0u64),
);
let sent = Arc::new(std::sync::Mutex::new(Vec::<Vec<u8>>::new()));
let output = sent.clone();
let side = router.add_side_packed_with_options(
"CAN",
move |bytes| {
output.lock().unwrap().push(bytes.to_vec());
Ok(())
},
RouterSideOptions::default(),
);
let endpoint = DataEndpoint::named("RADIO");
router
.rx_from_side(
&discovery::build_discovery_announce("AB", 0, &[endpoint]).unwrap(),
side,
)
.unwrap();
router.process_all_queues().unwrap();
sent.lock().unwrap().clear();
let packet = Packet::new(
DataType::named("GPS_DATA"),
&[endpoint],
original_sender,
1,
Arc::<[u8]>::from([0u8; 12]),
)
.unwrap();
router.note_reliable_return_route(side, packet.packet_id());
router.queue_end_to_end_reliable_ack(&packet, true).unwrap();
for index in 0..runtime_reliable_max_return_routes().max(1) {
router.note_reliable_return_route(side, packet.packet_id().wrapping_add(index as u64 + 1));
}
assert!(
!router
.state
.lock()
.reliable_return_routes
.contains_key(&packet.packet_id())
);
router.process_all_queues().unwrap();
let frames = sent.lock().unwrap();
let acks: Vec<_> = frames
.iter()
.filter_map(|bytes| wire_format::unpack_packet(bytes).ok())
.filter(Router::is_end_to_end_ack_packet)
.collect();
assert_eq!(
acks.len(),
1,
"ACK must return to AB even after its per-packet return route is evicted"
);
assert_eq!(
Router::end_to_end_ack_sender_hash(&acks[0]),
Some(Router::sender_hash("GS"))
);
assert_eq!(
acks[0].wire_target_senders().get(1),
Some(&Router::sender_hash("AB"))
);
}
#[test]
fn wrapped_acks_use_reserved_headroom_during_memory_pressure() {
extern "C" fn only_control(additional: usize, largest: usize) -> bool {
additional <= 512 && largest <= 160
}
struct Reset;
impl Drop for Reset {
fn drop(&mut self) {
crate::memory_admission::set_probe(None);
}
}
let rx = Router::new(RouterConfig::new([]));
let opts = RouterSideOptions::default().with_small_packet_transport(0);
let b = rx.add_side_packed_with_options("wire", |_| Ok(()), opts);
let packet = Packet::new(
crate::DataType::ReliableAck,
&[crate::DataEndpoint::Discovery],
"GS",
100,
Arc::<[u8]>::from([0; 8]),
)
.unwrap();
let raw = wire_format::pack_packet(&packet);
let mut body = vec![1];
body.extend_from_slice(&raw);
let full = wrap_side_transport_frame(SIDE_TRANSPORT_KIND_FULL, &body);
let (_, _, flags, size, timestamp, nonce, reliable, payload) =
extract_side_header_template(&raw).unwrap();
let mut body = vec![flags, 1];
write_uleb128_local(size, &mut body);
write_uleb128_local(timestamp, &mut body);
if flags & SIDE_TRANSPORT_FLAG_PACKET_NONCE != 0 {
write_uleb128_local(u64::from(nonce), &mut body);
}
if let Some((seq, ack)) = reliable {
write_uleb128_local(u64::from(seq), &mut body);
write_uleb128_local(u64::from(ack), &mut body);
}
body.extend_from_slice(payload);
let compact = wrap_side_transport_frame(SIDE_TRANSPORT_KIND_COMPACT, &body);
let ordinary = Packet::new(
crate::DataType::DiscoverySchema,
&[crate::DataEndpoint::Discovery],
"GS",
101,
Arc::<[u8]>::from([1; 32]),
)
.unwrap();
let ordinary = wire_format::pack_packet(&ordinary);
let _reset = Reset;
crate::memory_admission::set_probe(Some(only_control));
assert!(
rx.decode_side_transport_frame(b, &ordinary).is_err(),
"ordinary ingress must remain refused"
);
for wire in [raw.as_ref(), full.as_ref(), compact.as_ref()] {
let decoded = rx.decode_side_transport_frame(b, wire).unwrap().unwrap();
assert_eq!(decoded, raw);
let packet = wire_format::unpack_packet(&decoded).unwrap();
assert_eq!(packet.data_type(), crate::DataType::ReliableAck);
}
}
#[test]
fn legacy_ack_recovers_after_return_cache_churn_on_two_side_bridge() {
crate::tests::ensure_common_test_schema();
for packed in [false, true] {
for extra_side in [false, true] {
let router =
Router::new_with_clock(RouterConfig::new([]).with_sender("GB"), Box::new(|| 0));
let forwarded = Arc::new(std::sync::Mutex::new(Vec::<Vec<u8>>::new()));
let output = forwarded.clone();
let uart = router.add_side_packed("uart", move |bytes| {
output.lock().unwrap().push(bytes.to_vec());
Ok(())
});
let can = router.add_side_packed("can", |_| Ok(()));
if extra_side {
router.add_side_packed("other", |_| Ok(()));
}
router
.rx_from_side(
&discovery::build_discovery_announce("GS", 0, &[DataEndpoint::named("RADIO")])
.unwrap(),
uart,
)
.unwrap();
router.process_all_queues().unwrap();
forwarded.lock().unwrap().clear();
let packet_id = 1234u64;
router.note_reliable_return_route(uart, packet_id);
for index in 0..runtime_reliable_max_return_routes().max(1) {
router.note_reliable_return_route(can, 10000 + index as u64);
}
assert!(
!router
.state
.lock()
.reliable_return_routes
.contains_key(&packet_id)
);
let ack = Packet::new(
DataType::ReliableAck,
&[DataEndpoint::Discovery],
"E2EACK:VB",
1,
Arc::<[u8]>::from(packet_id.to_le_bytes()),
)
.unwrap();
let raw = wire_format::pack_packet_with_wire_contract(
&ack,
None,
None,
&[Router::sender_hash("VB")],
)
.unwrap();
if packed {
router.rx_packed_from_side(&raw, can).unwrap();
} else {
router
.rx_from_side(&wire_format::unpack_packet(&raw).unwrap(), can)
.unwrap();
}
router.process_all_queues().unwrap();
let count = forwarded
.lock()
.unwrap()
.iter()
.filter_map(|bytes| wire_format::unpack_packet(bytes).ok())
.filter(|p| p.data_type() == DataType::ReliableAck)
.count();
assert_eq!(
count,
usize::from(!extra_side),
"recover only an unambiguous bridge path"
);
}
}
}
#[test]
fn reliable_data_retry_survives_return_route_cache_churn() {
crate::tests::ensure_common_test_schema();
for packed in [false, true] {
let now = Arc::new(core::sync::atomic::AtomicU64::new(10_000));
let clock = now.clone();
let router = Router::new_with_clock(
RouterConfig::new([]).with_sender("GB"),
Box::new(move || clock.load(core::sync::atomic::Ordering::Relaxed)),
);
let forwarded = Arc::new(std::sync::Mutex::new(Vec::<Vec<u8>>::new()));
let output = forwarded.clone();
let uart = router.add_side_packed("uart", move |bytes| {
output.lock().unwrap().push(bytes.to_vec());
Ok(())
});
let can = router.add_side_packed("can", |_| Ok(()));
let ep = DataEndpoint::named("RADIO");
router
.rx_from_side(
&discovery::build_discovery_announce("GS", 0, &[ep]).unwrap(),
uart,
)
.unwrap();
router.process_all_queues().unwrap();
let ty = crate::config::register_data_type_with_description(
"RETURN_CACHE_RETRY_DATA",
"",
crate::MessageElement::Static(
1,
crate::MessageDataType::UInt8,
crate::MessageClass::Data,
),
&[ep],
crate::ReliableMode::Ordered,
200,
)
.unwrap();
let packet = Packet::new(ty, &[ep], "VB", 1, Arc::<[u8]>::from([1])).unwrap();
let raw = wire_format::pack_packet_with_wire_contract(
&packet,
None,
None,
&[Router::sender_hash("GS")],
)
.unwrap();
let deliver = || {
if packed {
router.rx_packed_from_side(&raw, can).unwrap();
} else {
router
.rx_from_side(&wire_format::unpack_packet(&raw).unwrap(), can)
.unwrap();
}
router.process_all_queues().unwrap();
};
let count = || {
forwarded
.lock()
.unwrap()
.iter()
.filter_map(|b| wire_format::unpack_packet(b).ok())
.filter(|p| p.data_type() == ty)
.count()
};
deliver();
assert_eq!(count(), 1);
now.fetch_add(
runtime_reliable_retransmit_ms(),
core::sync::atomic::Ordering::Relaxed,
);
for index in 0..runtime_reliable_max_return_routes().max(1) {
router.note_reliable_return_route(can, 10000 + index as u64);
}
deliver();
assert_eq!(count(), 2, "cache churn must not restart the retry delay");
deliver();
assert_eq!(count(), 2, "immediate echoes remain suppressed");
}
}