#[cfg(all(test, feature = "discovery"))]
#[path = "tests/relay_ack_recovery.rs"]
mod ack_recovery_tests;
#[cfg(test)]
#[allow(clippy::duplicate_mod)]
#[path = "tests/side_template_index.rs"]
mod side_template_index_tests;
use crate::side_transport::{
SIDE_TRANSPORT_FLAG_PACKET_NONCE, SIDE_TRANSPORT_KIND_CHUNK, SIDE_TRANSPORT_KIND_COMPACT,
SIDE_TRANSPORT_KIND_COMPACT_DELTA, SIDE_TRANSPORT_KIND_COMPACT_SAME_TIMESTAMP,
SIDE_TRANSPORT_KIND_FULL, SideChunkAssembly, SideCompactTimestampMode, SideHeaderTemplate,
SideTransportFrames, extract_side_header_template, extract_side_header_template_view,
parse_side_transport_wrapper, read_uleb128_local, reconstruct_side_compact_frame,
wrap_side_transport_frame, write_uleb128_local,
};
#[cfg(all(test, feature = "discovery"))]
#[path = "tests/relay_restart_transport.rs"]
mod restart_transport_tests;
#[cfg(all(test, feature = "discovery"))]
mod compact_summary_tests {
use super::*;
#[test]
fn compact_loss_and_reordering_preserve_packet_identity() {
crate::tests::ensure_common_test_schema();
let tx = Relay::new(Box::new(|| 0));
let rx = Relay::new(Box::new(|| 0));
let opts = RelaySideOptions::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 mut encoded = Vec::new();
for timestamp in [10000, 10010, 10020, 10030] {
let pkt = Packet::new(
crate::DataType::named("GPS_DATA"),
&[crate::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();
assert_eq!(frames.len(), 1);
encoded.push((raw, frames.into_iter().next().unwrap()));
}
for index in [0, 3, 2, 3] {
let (raw, frame) = &encoded[index];
let received = rx.decode_side_transport_frame(b, frame).unwrap().unwrap();
assert_eq!(&received, raw);
}
}
#[test]
fn relay_preserves_compact_routes_without_inventing_endpoint_ownership() {
crate::tests::ensure_common_test_schema();
let endpoint = crate::DataEndpoint::named("RADIO");
let relay = Relay::new(Box::new(|| 0));
let mut sender = DiscoverySenderState {
advertised_reachable: vec![endpoint],
advertised_reachable_timesync_sources: vec!["clock-behind-bridge".into()],
topology_boards: vec![TopologyBoardNode {
sender_id: "bridge".into(),
reachable_endpoints: vec![],
reachable_timesync_sources: vec![],
connections: vec![],
}],
..Default::default()
};
Relay::refresh_sender_topology_state(&mut sender);
let mut route = DiscoverySideState::default();
route.announcers.insert("bridge".into(), sender);
Relay::recompute_discovery_side_state(&mut route);
let mut state = relay.state.lock();
state.discovery_routes.insert(0, route);
assert!(
relay
.advertised_discovery_endpoints_for_link_locked(&state, 0, true, Some(1))
.contains(&endpoint)
);
assert!(
!relay
.advertised_discovery_endpoints_for_link_locked(&state, 0, true, Some(0))
.contains(&endpoint)
);
assert!(
!relay
.advertised_discovery_endpoints_for_link_locked(
&state,
DISCOVERY_ROUTE_TTL_MS + 1,
true,
Some(1)
)
.contains(&endpoint)
);
assert_eq!(
relay.advertised_discovery_timesync_sources_for_link_locked(&state, 0, Some(1)),
vec!["clock-behind-bridge".to_string()]
);
assert!(
relay
.advertised_discovery_timesync_sources_for_link_locked(&state, 0, Some(0))
.is_empty()
);
assert!(
relay
.advertised_discovery_timesync_sources_for_link_locked(
&state,
DISCOVERY_ROUTE_TTL_MS + 1,
Some(1)
)
.is_empty()
);
let topology =
relay.advertised_discovery_topology_for_link_locked(&state, 0, true, Some(1));
assert!(
topology
.iter()
.find(|node| node.sender_id == "bridge")
.unwrap()
.reachable_endpoints
.is_empty()
);
}
}
use crate::config::{
RuntimeMemoryConfig, runtime_reliable_max_end_to_end_ack_cache,
runtime_reliable_max_end_to_end_pending, runtime_reliable_max_pending,
runtime_reliable_max_retries, runtime_reliable_max_return_routes,
runtime_reliable_retransmit_ms,
};
use crate::diagnostics::{
AdaptiveLinkStats, DiscoveryRuntimeStats, QueueRuntimeStats, ReliableRuntimeStats,
RouteModeStats, RouteOverrideStats, RoutePriorityStats, RouteWeightStats, RuntimeSideStats,
RuntimeStatsSnapshot, RuntimeTypeStats, TypedRouteOverrideStats,
};
#[cfg(feature = "discovery")]
use crate::discovery::{
self, ClientStatsSnapshot, DISCOVERY_INCREMENTAL_RETRY_COUNT, DISCOVERY_ROUTE_TTL_MS,
DISCOVERY_SLOW_LINK_PING_INTERVAL_MS, DiscoveryCadenceState,
TIMESYNC_SLOW_LINK_MIN_INTERVAL_MS, TopologyAnnouncerRoute, TopologyBoardNode,
TopologySideRoute, TopologySnapshot,
};
#[cfg(feature = "discovery")]
use crate::packet::sender_address_u32;
use crate::packet::{Packet, hash_bytes_u64};
use crate::queue::{BoundedDeque, ByteCost};
use crate::wire_format;
use crate::{is_reliable_type, message_meta, message_priority, reliable_mode};
use crate::{
router::{Clock, CompactTimestampOmissionPolicy, SideTransportProfile},
{
RouteSelectionMode, TelemetryError, TelemetryResult,
lock::{ReentryGate, ReentryGuard, RouterMutex},
},
};
use alloc::borrow::ToOwned;
use alloc::boxed::Box;
use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
use alloc::string::{String, ToString};
use alloc::{sync::Arc, vec, vec::Vec};
use core::mem::size_of;
const MAX_PRIORITY_BURST: u8 = 8;
pub type RelaySideId = usize;
const CONTROL_SLOW_LINK_CAPACITY_BPS: u64 = 512;
pub const IPV4_LIKE_COMPACT_HEADER_TARGET_BYTES: usize =
crate::router::IPV4_LIKE_COMPACT_HEADER_TARGET_BYTES;
pub const IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES: usize =
crate::router::IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES;
pub const DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT: usize =
crate::router::DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT;
type PacketHandlerFn = dyn Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static;
type PackedHandlerFn = dyn Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static;
#[derive(Clone)]
pub enum RelayTxHandlerFn {
Packed(Arc<PackedHandlerFn>),
Packet(Arc<PacketHandlerFn>),
}
#[derive(Clone, Debug)]
pub struct RelaySideOptions {
pub reliable_enabled: bool,
pub link_local_enabled: bool,
pub ingress_enabled: bool,
pub egress_enabled: bool,
pub header_template_enabled: bool,
pub max_frame_bytes: usize,
pub compact_header_target_bytes: usize,
pub max_side_transport_templates: usize,
pub omit_unchanged_compact_timestamps: bool,
pub compact_timestamp_omission_types: CompactTimestampOmissionPolicy,
pub side_transport_profile: SideTransportProfile,
}
impl Default for RelaySideOptions {
fn default() -> Self {
Self {
reliable_enabled: false,
link_local_enabled: false,
ingress_enabled: true,
egress_enabled: true,
header_template_enabled: false,
max_frame_bytes: 0,
compact_header_target_bytes: 0,
max_side_transport_templates: DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT,
omit_unchanged_compact_timestamps: false,
compact_timestamp_omission_types: CompactTimestampOmissionPolicy::none(),
side_transport_profile: SideTransportProfile::Canonical,
}
}
}
impl RelaySideOptions {
#[inline]
pub fn with_small_packet_transport(mut self, max_frame_bytes: usize) -> Self {
self.header_template_enabled = true;
self.max_frame_bytes = max_frame_bytes;
self.compact_header_target_bytes = IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES;
self.side_transport_profile = SideTransportProfile::Ipv6Like;
self
}
#[inline]
pub fn with_ipv4_like_compact_header_target(mut self) -> Self {
self.header_template_enabled = true;
self.compact_header_target_bytes = IPV4_LIKE_COMPACT_HEADER_TARGET_BYTES;
self.omit_unchanged_compact_timestamps = true;
self.side_transport_profile = SideTransportProfile::Ipv4Like;
self
}
#[inline]
pub fn with_ipv6_like_compact_header_target(mut self) -> Self {
self.header_template_enabled = true;
self.compact_header_target_bytes = IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES;
self.side_transport_profile = SideTransportProfile::Ipv6Like;
self
}
#[inline]
pub fn with_template_transport(mut self) -> Self {
self.header_template_enabled = true;
self.side_transport_profile = SideTransportProfile::Template;
self
}
#[inline]
pub fn with_omitted_unchanged_compact_timestamps(mut self) -> Self {
self.header_template_enabled = true;
self.omit_unchanged_compact_timestamps = true;
self
}
#[inline]
pub fn with_omitted_unchanged_compact_timestamps_for_type(
mut self,
ty: crate::DataType,
) -> Self {
self.header_template_enabled = true;
self.compact_timestamp_omission_types.insert(ty);
self
}
#[inline]
pub fn effective_transport_profile(&self) -> SideTransportProfile {
if !self.header_template_enabled && self.max_frame_bytes == 0 {
SideTransportProfile::Canonical
} else if self.side_transport_profile == SideTransportProfile::Canonical {
SideTransportProfile::Template
} else {
self.side_transport_profile
}
}
#[cfg(feature = "discovery")]
#[inline]
pub fn link_capabilities(&self) -> discovery::LinkCapabilities {
let mut flags = discovery::LINK_CAPABILITY_END_TO_END_RELIABILITY;
if self.header_template_enabled {
flags |= discovery::LINK_CAPABILITY_HEADER_TEMPLATES;
}
if self.max_frame_bytes != 0 {
flags |= discovery::LINK_CAPABILITY_CHUNKING;
}
if self.reliable_enabled {
flags |= discovery::LINK_CAPABILITY_RELIABILITY;
}
if self.omit_unchanged_compact_timestamps
|| !self.compact_timestamp_omission_types.is_empty()
{
flags |= discovery::LINK_CAPABILITY_OMIT_UNCHANGED_TIMESTAMPS;
}
#[cfg(feature = "cryptography")]
{
flags |= discovery::LINK_CAPABILITY_CRYPTO;
}
discovery::LinkCapabilities {
version: 1,
flags,
profile: self.effective_transport_profile().discovery_code(),
max_frame_bytes: self.max_frame_bytes.min(u32::MAX as usize) as u32,
compact_header_target_bytes: self.compact_header_target_bytes.min(u32::MAX as usize)
as u32,
max_side_transport_templates: self.max_side_transport_templates.min(u32::MAX as usize)
as u32,
}
}
}
#[derive(Clone)]
pub struct RelaySide {
pub name: Arc<str>,
pub tx_handler: RelayTxHandlerFn,
pub opts: RelaySideOptions,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RelayItem {
Packed(crate::SharedBytes),
Packet(Arc<Packet>),
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct RelayRxItem {
src: RelaySideId,
data: RelayItem,
priority: u8,
}
impl ByteCost for RelayRxItem {
#[cfg(feature = "compact-packet-store")]
fn park_for_queue(&mut self) -> TelemetryResult<()> {
match &mut self.data {
RelayItem::Packet(packet) => Arc::make_mut(packet).park_for_queue()?,
RelayItem::Packed(bytes) => bytes.park_for_queue()?,
}
Ok(())
}
fn byte_cost(&self) -> usize {
match &self.data {
RelayItem::Packed(bytes) => bytes.len(),
RelayItem::Packet(pkt) => pkt.byte_cost(),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct RelayTxItem {
src: Option<RelaySideId>,
dst: RelaySideId,
data: RelayItem,
priority: u8,
retry_after_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct RelayReplayItem {
dst: RelaySideId,
bytes: crate::SharedBytes,
priority: u8,
retry_after_ms: u64,
}
impl ByteCost for RelayTxItem {
#[cfg(feature = "compact-packet-store")]
fn park_for_queue(&mut self) -> TelemetryResult<()> {
match &mut self.data {
RelayItem::Packet(packet) => Arc::make_mut(packet).park_for_queue()?,
RelayItem::Packed(bytes) => bytes.park_for_queue()?,
}
Ok(())
}
fn byte_cost(&self) -> usize {
(match &self.data {
RelayItem::Packed(bytes) => bytes.len(),
RelayItem::Packet(pkt) => pkt.byte_cost(),
})
.saturating_add(size_of::<u64>())
}
}
impl ByteCost for RelayReplayItem {
#[cfg(feature = "compact-packet-store")]
fn park_for_queue(&mut self) -> TelemetryResult<()> {
self.bytes.park_for_queue()
}
fn byte_cost(&self) -> usize {
(self.bytes.len()).saturating_add(size_of::<u64>())
}
}
#[derive(Debug, Clone)]
struct ReliableTxState {
next_seq: u32,
sent_order: VecDeque<u32>,
sent: BTreeMap<u32, ReliableSent>,
}
#[derive(Debug, Clone)]
struct ReliableSent {
bytes: crate::SharedBytes,
last_send_ms: u64,
retries: u32,
queued: bool,
partial_acked: bool,
}
#[derive(Debug, Clone)]
struct ReliableRxState {
expected_seq: u32,
buffered: BTreeMap<u32, crate::SharedBytes>,
}
#[derive(Debug, Clone)]
struct ReliableReturnRouteState {
side: RelaySideId,
last_replay_ms: u64,
}
#[cfg(feature = "discovery")]
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct DiscoverySenderState {
advertised_address: Option<u32>,
reachable: Vec<crate::DataEndpoint>,
advertised_reachable: Vec<crate::DataEndpoint>,
reachable_timesync_sources: Vec<String>,
advertised_reachable_timesync_sources: Vec<String>,
topology_boards: Vec<TopologyBoardNode>,
has_full_topology: bool,
has_schema: bool,
last_seen_ms: u64,
}
#[inline]
fn is_internal_control_type(ty: crate::DataType) -> bool {
if matches!(
ty,
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
| crate::DataType::P2pMessage
) {
return true;
}
#[cfg(feature = "timesync")]
if matches!(
ty,
crate::DataType::TimeSyncAnnounce
| crate::DataType::TimeSyncRequest
| crate::DataType::TimeSyncResponse
) {
return true;
}
#[cfg(feature = "discovery")]
if discovery::is_discovery_type(ty) {
return true;
}
let _ = ty;
false
}
#[cfg(feature = "discovery")]
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct DiscoverySideState {
reachable: Vec<crate::DataEndpoint>,
reachable_timesync_sources: Vec<String>,
last_seen_ms: u64,
announcers: BTreeMap<String, DiscoverySenderState>,
}
#[derive(Clone, Debug, Default)]
struct SideTransportState {
tx_chunk_progress: Option<Box<crate::side_transport::SideTxProgress>>,
tx_template_ids: BTreeMap<u64, u32>,
tx_templates: BTreeMap<u64, SideHeaderTemplate>,
tx_last_timestamps: BTreeMap<u32, u64>,
tx_compact_uses: BTreeMap<u32, u8>,
rx_templates: BTreeMap<u64, u32>,
rx_templates_by_id: BTreeMap<u32, SideHeaderTemplate>,
rx_last_timestamps: BTreeMap<u32, u64>,
rx_chunks: BTreeMap<u32, SideChunkAssembly>,
next_template_id: u32,
}
impl SideTransportState {
fn refresh_transmit_header_templates(&mut self) {
for uses in self.tx_compact_uses.values_mut() {
*uses = u8::MAX;
}
}
fn tx_template_count(&self) -> usize {
self.tx_template_ids.len()
}
fn rx_template_count(&self) -> usize {
self.rx_templates_by_id.len()
}
fn trim_tx_templates_to(&mut self, limit: usize) -> usize {
let mut removed = 0usize;
while self.tx_template_ids.len() > limit {
let Some(old_hash) = self.tx_template_ids.keys().next().copied() else {
break;
};
if let Some(old_id) = self.tx_template_ids.remove(&old_hash) {
self.tx_last_timestamps.remove(&old_id);
self.tx_compact_uses.remove(&old_id);
}
self.tx_templates.remove(&old_hash);
removed += 1;
}
removed
}
fn insert_tx_template(
&mut self,
template: SideHeaderTemplate,
template_id: u32,
max_templates: usize,
) -> bool {
if max_templates == 0 {
return false;
}
let mut evicted = false;
if self.tx_template_ids.len() >= max_templates
&& !self.tx_template_ids.contains_key(&template.hash)
&& let Some(old_hash) = self.tx_template_ids.keys().next().copied()
{
if let Some(old_id) = self.tx_template_ids.remove(&old_hash) {
self.tx_last_timestamps.remove(&old_id);
self.tx_compact_uses.remove(&old_id);
}
self.tx_templates.remove(&old_hash);
evicted = true;
}
self.tx_template_ids.insert(template.hash, template_id);
self.tx_templates.insert(template.hash, template);
self.tx_compact_uses.insert(template_id, 0);
evicted
}
fn insert_rx_template(
&mut self,
template_id: u32,
template: SideHeaderTemplate,
max_templates: usize,
) -> bool {
if max_templates == 0 {
return false;
}
if let Some(old_id) = self.rx_templates.get(&template.hash).copied()
&& old_id != template_id
{
self.rx_templates_by_id.remove(&old_id);
self.rx_last_timestamps.remove(&old_id);
self.rx_templates.remove(&template.hash);
}
if let Some(previous) = self.rx_templates_by_id.get(&template_id)
&& previous.hash != template.hash
{
self.rx_templates.remove(&previous.hash);
self.rx_last_timestamps.remove(&template_id);
}
let mut evicted = false;
if self.rx_templates_by_id.len() >= max_templates
&& !self.rx_templates_by_id.contains_key(&template_id)
&& let Some((&old_hash, &old_id)) = self.rx_templates.first_key_value()
{
self.rx_templates.remove(&old_hash);
self.rx_templates_by_id.remove(&old_id);
self.rx_last_timestamps.remove(&old_id);
evicted = true;
}
self.rx_templates.insert(template.hash, template_id);
self.rx_templates_by_id.insert(template_id, template);
evicted
}
}
#[derive(Debug, Clone, Default)]
struct AdaptiveRouteStats {
estimated_bandwidth_bps: u64,
peak_bandwidth_bps: u64,
last_observed_ms: u64,
last_slow_observed_ms: u64,
sample_count: u64,
window_started_ms: u64,
window_bytes: u64,
peak_usage_bps: u64,
}
#[cfg(feature = "discovery")]
#[derive(Debug, Clone, Default)]
struct DiscoverySideThrottleState {
next_topology_request_ms: u64,
next_schema_request_ms: u64,
next_ping_ms: u64,
last_liveness_tx_ms: u64,
pending_incremental: bool,
incremental_retries_remaining: u8,
has_sent_full: bool,
last_topology: Vec<TopologyBoardNode>,
}
#[cfg(feature = "discovery")]
impl DiscoverySideThrottleState {
fn liveness_due(&self, now_ms: u64) -> bool {
now_ms >= self.next_ping_ms
&& now_ms.saturating_sub(self.last_liveness_tx_ms)
>= DISCOVERY_SLOW_LINK_PING_INTERVAL_MS
}
}
#[cfg(all(feature = "discovery", feature = "timesync"))]
#[derive(Debug, Clone, Default)]
struct TimeSyncSideThrottleState {
next_allowed_ms: u64,
}
#[cfg(feature = "discovery")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DiscoveryAdvertiseLevel {
MinimalPing,
Incremental,
Full,
}
impl AdaptiveRouteStats {
#[inline]
fn observe(&mut self, bytes: usize, sample_bps: u64, now_ms: u64) {
self.estimated_bandwidth_bps = if self.estimated_bandwidth_bps == 0 {
sample_bps
} else if sample_bps >= self.estimated_bandwidth_bps {
self.estimated_bandwidth_bps
.saturating_mul(3)
.saturating_add(sample_bps.saturating_mul(5))
/ 8
} else {
self.estimated_bandwidth_bps
.saturating_mul(7)
.saturating_add(sample_bps)
/ 8
};
self.peak_bandwidth_bps = self.peak_bandwidth_bps.max(sample_bps);
self.last_observed_ms = now_ms;
if sample_bps > 0 && sample_bps <= CONTROL_SLOW_LINK_CAPACITY_BPS {
self.last_slow_observed_ms = now_ms;
}
self.sample_count = self.sample_count.saturating_add(1);
if self.window_started_ms == 0 || now_ms.saturating_sub(self.window_started_ms) > 1_000 {
self.window_started_ms = now_ms;
self.window_bytes = 0;
}
self.window_bytes = self.window_bytes.saturating_add(bytes as u64);
self.peak_usage_bps = self.peak_usage_bps.max(self.current_usage_bps(now_ms));
}
#[inline]
fn current_usage_bps(&self, now_ms: u64) -> u64 {
if self.window_started_ms == 0 {
return 0;
}
let elapsed_ms = now_ms.saturating_sub(self.window_started_ms).max(1);
(u128::from(self.window_bytes).saturating_mul(1000) / u128::from(elapsed_ms))
.min(u128::from(u64::MAX)) as u64
}
#[inline]
fn available_headroom_bps(&self, now_ms: u64) -> u64 {
let capacity = self
.estimated_bandwidth_bps
.max(self.peak_bandwidth_bps)
.max(1);
capacity.saturating_sub(self.current_usage_bps(now_ms))
}
#[inline]
fn weight(&self, now_ms: u64) -> u64 {
self.available_headroom_bps(now_ms).max(1)
}
#[inline]
fn snapshot(&self, now_ms: u64, auto_balancing_enabled: bool) -> AdaptiveLinkStats {
let current_usage_bps = self.current_usage_bps(now_ms);
let estimated_capacity_bps = self.estimated_bandwidth_bps.max(1);
let peak_capacity_bps = self.peak_bandwidth_bps.max(estimated_capacity_bps);
let available_headroom_bps = peak_capacity_bps.saturating_sub(current_usage_bps);
AdaptiveLinkStats {
auto_balancing_enabled,
estimated_capacity_bps,
peak_capacity_bps,
current_usage_bps,
peak_usage_bps: self.peak_usage_bps.max(current_usage_bps),
available_headroom_bps,
effective_weight: available_headroom_bps.max(1),
last_observed_ms: self.last_observed_ms,
sample_count: self.sample_count,
}
}
}
#[derive(Debug, Clone, Default)]
struct TypeRuntimeStatsInner {
tx_packets: u64,
tx_bytes: u64,
rx_packets: u64,
rx_bytes: u64,
relayed_tx_packets: u64,
relayed_tx_bytes: u64,
relayed_rx_packets: u64,
relayed_rx_bytes: u64,
tx_retries: u64,
handler_failures: u64,
}
#[derive(Debug, Clone, Default)]
struct SideRuntimeStatsInner {
tx_packets: u64,
tx_bytes: u64,
rx_packets: u64,
rx_bytes: u64,
relayed_tx_packets: u64,
relayed_tx_bytes: u64,
relayed_rx_packets: u64,
relayed_rx_bytes: u64,
tx_retries: u64,
tx_handler_failures: u64,
total_handler_retries: u64,
side_transport_full_frames: u64,
side_transport_compact_frames: u64,
side_transport_compact_delta_frames: u64,
side_transport_compact_omitted_timestamp_frames: u64,
side_transport_chunk_frames: u64,
side_transport_raw_bytes: u64,
side_transport_wire_bytes: u64,
side_transport_bytes_saved: u64,
side_transport_min_compact_overhead_bytes: Option<usize>,
side_transport_max_compact_overhead_bytes: Option<usize>,
side_transport_compact_target_misses: u64,
side_transport_template_evictions: u64,
data_types: BTreeMap<u32, TypeRuntimeStatsInner>,
}
impl SideRuntimeStatsInner {
fn type_stats_mut(&mut self, ty: crate::DataType) -> &mut TypeRuntimeStatsInner {
self.data_types.entry(ty.as_u32()).or_default()
}
fn note_tx(&mut self, ty: crate::DataType, bytes: usize, retries: usize) {
self.tx_packets = self.tx_packets.saturating_add(1);
self.tx_bytes = self.tx_bytes.saturating_add(bytes as u64);
self.relayed_tx_packets = self.relayed_tx_packets.saturating_add(1);
self.relayed_tx_bytes = self.relayed_tx_bytes.saturating_add(bytes as u64);
self.tx_retries = self.tx_retries.saturating_add(retries as u64);
self.total_handler_retries = self.total_handler_retries.saturating_add(retries as u64);
let stats = self.type_stats_mut(ty);
stats.tx_packets = stats.tx_packets.saturating_add(1);
stats.tx_bytes = stats.tx_bytes.saturating_add(bytes as u64);
stats.relayed_tx_packets = stats.relayed_tx_packets.saturating_add(1);
stats.relayed_tx_bytes = stats.relayed_tx_bytes.saturating_add(bytes as u64);
stats.tx_retries = stats.tx_retries.saturating_add(retries as u64);
}
fn note_rx(&mut self, ty: crate::DataType, bytes: usize) {
self.rx_packets = self.rx_packets.saturating_add(1);
self.rx_bytes = self.rx_bytes.saturating_add(bytes as u64);
self.relayed_rx_packets = self.relayed_rx_packets.saturating_add(1);
self.relayed_rx_bytes = self.relayed_rx_bytes.saturating_add(bytes as u64);
let stats = self.type_stats_mut(ty);
stats.rx_packets = stats.rx_packets.saturating_add(1);
stats.rx_bytes = stats.rx_bytes.saturating_add(bytes as u64);
stats.relayed_rx_packets = stats.relayed_rx_packets.saturating_add(1);
stats.relayed_rx_bytes = stats.relayed_rx_bytes.saturating_add(bytes as u64);
}
fn note_tx_failure(&mut self, ty: crate::DataType, retries: usize) {
self.tx_handler_failures = self.tx_handler_failures.saturating_add(1);
self.tx_retries = self.tx_retries.saturating_add(retries as u64);
self.total_handler_retries = self.total_handler_retries.saturating_add(retries as u64);
let stats = self.type_stats_mut(ty);
stats.handler_failures = stats.handler_failures.saturating_add(1);
stats.tx_retries = stats.tx_retries.saturating_add(retries as u64);
}
fn note_side_transport_full(&mut self, raw_bytes: usize, wire_bytes: usize) {
self.side_transport_full_frames = self.side_transport_full_frames.saturating_add(1);
self.note_side_transport_bytes(raw_bytes, wire_bytes);
}
fn note_side_transport_compact(
&mut self,
raw_bytes: usize,
wire_bytes: usize,
compact_overhead_bytes: usize,
used_timestamp_delta: bool,
omitted_timestamp: bool,
) {
self.side_transport_compact_frames = self.side_transport_compact_frames.saturating_add(1);
if used_timestamp_delta {
self.side_transport_compact_delta_frames =
self.side_transport_compact_delta_frames.saturating_add(1);
}
if omitted_timestamp {
self.side_transport_compact_omitted_timestamp_frames = self
.side_transport_compact_omitted_timestamp_frames
.saturating_add(1);
}
self.note_side_transport_bytes(raw_bytes, wire_bytes);
self.side_transport_min_compact_overhead_bytes = Some(
self.side_transport_min_compact_overhead_bytes
.map_or(compact_overhead_bytes, |v| v.min(compact_overhead_bytes)),
);
self.side_transport_max_compact_overhead_bytes = Some(
self.side_transport_max_compact_overhead_bytes
.map_or(compact_overhead_bytes, |v| v.max(compact_overhead_bytes)),
);
}
fn note_side_transport_chunks(&mut self, chunks: usize) {
self.side_transport_chunk_frames = self
.side_transport_chunk_frames
.saturating_add(chunks as u64);
}
fn note_side_transport_bytes(&mut self, raw_bytes: usize, wire_bytes: usize) {
self.side_transport_raw_bytes = self
.side_transport_raw_bytes
.saturating_add(raw_bytes as u64);
self.side_transport_wire_bytes = self
.side_transport_wire_bytes
.saturating_add(wire_bytes as u64);
if raw_bytes > wire_bytes {
self.side_transport_bytes_saved = self
.side_transport_bytes_saved
.saturating_add((raw_bytes - wire_bytes) as u64);
}
}
fn note_side_transport_compact_target_miss(&mut self) {
self.side_transport_compact_target_misses =
self.side_transport_compact_target_misses.saturating_add(1);
}
fn note_side_transport_template_eviction(&mut self) {
self.side_transport_template_evictions =
self.side_transport_template_evictions.saturating_add(1);
}
}
#[derive(Clone, Debug)]
pub struct RelayConfig {
sender: Arc<str>,
memory: RuntimeMemoryConfig,
}
impl RelayConfig {
pub fn new() -> Self {
Self::default()
}
pub fn with_sender<S: AsRef<str>>(mut self, sender: S) -> Self {
self.sender = Arc::from(sender.as_ref());
self
}
pub fn with_memory_config(mut self, memory: RuntimeMemoryConfig) -> TelemetryResult<Self> {
memory.validate()?;
self.memory = memory;
Ok(self)
}
fn sender(&self) -> Arc<str> {
self.sender.clone()
}
fn memory_config(&self) -> RuntimeMemoryConfig {
self.memory
}
}
impl Default for RelayConfig {
fn default() -> Self {
Self {
sender: Arc::from("RELAY"),
memory: RuntimeMemoryConfig::default(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RouteSelectionOrigin {
Flood,
Discovered,
}
struct RelayInner {
memory: RuntimeMemoryConfig,
sides: Vec<Option<RelaySide>>,
route_overrides: BTreeMap<(Option<RelaySideId>, RelaySideId), bool>,
typed_route_overrides: BTreeMap<(Option<RelaySideId>, u32, RelaySideId), bool>,
route_weights: BTreeMap<(Option<RelaySideId>, RelaySideId), u32>,
route_priorities: BTreeMap<(Option<RelaySideId>, RelaySideId), u32>,
source_route_modes: BTreeMap<Option<RelaySideId>, RouteSelectionMode>,
route_selection_cursors: BTreeMap<Option<RelaySideId>, u64>,
adaptive_route_stats: BTreeMap<RelaySideId, AdaptiveRouteStats>,
side_runtime_stats: BTreeMap<RelaySideId, SideRuntimeStatsInner>,
side_transport: BTreeMap<RelaySideId, SideTransportState>,
rx_queue: BoundedDeque<RelayRxItem>,
tx_queue: BoundedDeque<RelayTxItem>,
tx_priority_burst: u8,
replay_queue: BoundedDeque<RelayReplayItem>,
recent_rx: BoundedDeque<u64>,
reliable_tx: BTreeMap<(RelaySideId, u32), ReliableTxState>,
reliable_rx: BTreeMap<(RelaySideId, u32), ReliableRxState>,
reliable_return_routes: BTreeMap<u64, ReliableReturnRouteState>,
reliable_return_route_order: VecDeque<u64>,
end_to_end_acked_destinations: BTreeMap<u64, BTreeSet<u64>>,
end_to_end_acked_destination_order: VecDeque<u64>,
total_handler_failures: u64,
total_handler_retries: u64,
#[cfg(feature = "discovery")]
discovery_routes: BTreeMap<RelaySideId, DiscoverySideState>,
#[cfg(feature = "discovery")]
discovery_cadence: DiscoveryCadenceState,
#[cfg(feature = "discovery")]
discovery_side_throttle: BTreeMap<RelaySideId, DiscoverySideThrottleState>,
#[cfg(all(feature = "discovery", feature = "timesync"))]
timesync_side_throttle: BTreeMap<RelaySideId, TimeSyncSideThrottleState>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RelayQueueKind {
Rx,
Tx,
Replay,
Recent,
ReliableRxBuffer,
}
impl RelayInner {
#[cfg(feature = "discovery")]
fn topology_board_byte_cost(board: &TopologyBoardNode) -> usize {
board
.sender_id
.len()
.saturating_add(board.reachable_endpoints.len() * size_of::<crate::DataEndpoint>())
.saturating_add(
board
.reachable_timesync_sources
.iter()
.map(|s| s.len())
.sum::<usize>(),
)
.saturating_add(board.connections.iter().map(|s| s.len()).sum::<usize>())
}
#[cfg(feature = "discovery")]
fn discovery_sender_byte_cost(sender: &str, state: &DiscoverySenderState) -> usize {
sender
.len()
.saturating_add(state.reachable.len() * size_of::<crate::DataEndpoint>())
.saturating_add(state.advertised_reachable.len() * size_of::<crate::DataEndpoint>())
.saturating_add(
state
.reachable_timesync_sources
.iter()
.map(|s| s.len())
.sum::<usize>(),
)
.saturating_add(
state
.advertised_reachable_timesync_sources
.iter()
.map(|s| s.len())
.sum::<usize>(),
)
.saturating_add(
state
.topology_boards
.iter()
.map(Self::topology_board_byte_cost)
.sum::<usize>(),
)
.saturating_add(size_of::<DiscoverySenderState>())
}
#[cfg(feature = "discovery")]
fn discovery_route_byte_cost(route: &DiscoverySideState) -> usize {
size_of::<DiscoverySideState>()
.saturating_add(route.reachable.len() * size_of::<crate::DataEndpoint>())
.saturating_add(
route
.reachable_timesync_sources
.iter()
.map(|s| s.len())
.sum::<usize>(),
)
.saturating_add(
route
.announcers
.iter()
.map(|(sender, state)| Self::discovery_sender_byte_cost(sender, state))
.sum::<usize>(),
)
}
#[cfg(feature = "discovery")]
fn discovery_bytes_used(&self) -> usize {
self.discovery_routes
.values()
.map(Self::discovery_route_byte_cost)
.sum()
}
#[inline]
fn reliable_rx_buffered_bytes(&self) -> usize {
self.reliable_rx
.values()
.flat_map(|state| state.buffered.values())
.map(|bytes| size_of::<crate::SharedBytes>() + bytes.len())
.sum()
}
#[inline]
fn shared_queue_bytes_used(&self) -> usize {
self.rx_queue
.bytes_used()
.saturating_add(self.tx_queue.bytes_used())
.saturating_add(self.replay_queue.bytes_used())
.saturating_add(self.recent_rx.max_bytes())
.saturating_add(self.reliable_rx_buffered_bytes())
.saturating_add(crate::config::schema_bytes_used())
.saturating_add({
#[cfg(feature = "discovery")]
{
self.discovery_bytes_used()
}
#[cfg(not(feature = "discovery"))]
{
0
}
})
}
fn reliable_rx_buffer_len(&self) -> usize {
self.reliable_rx
.values()
.map(|state| state.buffered.len())
.sum()
}
fn pop_reliable_rx_buffered(&mut self) -> Option<crate::SharedBytes> {
let key = self
.reliable_rx
.iter()
.find_map(|(key, state)| (!state.buffered.is_empty()).then_some(*key))?;
self.reliable_rx
.get_mut(&key)?
.buffered
.pop_first()
.map(|(_, v)| v)
}
fn pop_lowest_packet_queue_item(&mut self, below: usize) -> bool {
let victim = [
(
RelayQueueKind::Rx,
self.rx_queue.lowest_priority(|item| item.priority),
),
(
RelayQueueKind::Tx,
self.tx_queue.lowest_priority(|item| item.priority),
),
(
RelayQueueKind::Replay,
self.replay_queue.lowest_priority(|item| item.priority),
),
]
.into_iter()
.filter_map(|(kind, priority)| priority.map(|priority| (kind, priority)))
.filter(|(_, priority)| usize::from(*priority) < below)
.min_by_key(|(_, priority)| *priority)
.map(|(kind, _)| kind);
match victim {
Some(RelayQueueKind::Rx) => self.rx_queue.pop_lowest_priority().is_some(),
Some(RelayQueueKind::Tx) => self.tx_queue.pop_lowest_priority().is_some(),
Some(RelayQueueKind::Replay) => self.replay_queue.pop_lowest_priority().is_some(),
_ => false,
}
}
fn make_shared_queue_room(
&mut self,
incoming_cost: usize,
_preferred: RelayQueueKind,
) -> TelemetryResult<()> {
if incoming_cost > self.memory.max_queue_budget {
return Err(TelemetryError::PacketTooLarge(
"Item exceeds maximum shared queue budget",
));
}
while self.shared_queue_bytes_used().saturating_add(incoming_cost)
> self.memory.max_queue_budget
{
if !self.pop_lowest_packet_queue_item(256) && self.pop_reliable_rx_buffered().is_none()
{
return Err(TelemetryError::PacketTooLarge(
"Item exceeds maximum shared queue budget",
));
}
}
Ok(())
}
#[cfg(feature = "discovery")]
fn schema_growth_budget(&self) -> usize {
let protected = self
.recent_rx
.max_bytes()
.saturating_add(self.discovery_bytes_used())
.saturating_add(self.reliable_rx_buffered_bytes());
self.memory.max_queue_budget.saturating_sub(protected)
}
#[cfg(feature = "discovery")]
fn commit_discovery_route(
&mut self,
side: RelaySideId,
route: DiscoverySideState,
) -> TelemetryResult<()> {
let previous = self
.discovery_routes
.get(&side)
.map_or(0, Self::discovery_route_byte_cost);
let incoming = Self::discovery_route_byte_cost(&route);
let growth = incoming.saturating_sub(previous);
while self.shared_queue_bytes_used().saturating_add(growth) > self.memory.max_queue_budget {
if !self.pop_lowest_packet_queue_item(256) {
return Err(TelemetryError::Io("shared priority queue saturated"));
}
}
self.discovery_routes.insert(side, route);
Ok(())
}
#[cfg(feature = "discovery")]
fn fit_discovery_budget(&mut self) {
while self.shared_queue_bytes_used() > self.memory.max_queue_budget {
if self.pop_lowest_packet_queue_item(256) {
continue;
}
break;
}
}
fn push_rx(&mut self, item: RelayRxItem) -> TelemetryResult<()> {
self.make_shared_queue_room(item.byte_cost(), RelayQueueKind::Rx)?;
self.rx_queue
.push_back_prioritized(item, |queued| queued.priority)
}
fn push_tx(&mut self, item: RelayTxItem) -> TelemetryResult<()> {
#[cfg(feature = "discovery")]
{
if let RelayItem::Packet(incoming) = &item.data
&& self.tx_queue.iter().any(|queued| {
queued.dst == item.dst && matches!(&queued.data,
RelayItem::Packet(existing) if discovery::same_schema_snapshot(existing, incoming))
})
{
return Ok(());
}
let incoming = match &item.data {
RelayItem::Packet(pkt) if discovery::is_discovery_type(pkt.data_type()) => {
Some((pkt.data_type(), pkt.sender().to_owned()))
}
RelayItem::Packet(_) | RelayItem::Packed(_) => None,
};
if let Some((incoming_ty, incoming_sender)) = incoming {
self.tx_queue.retain(|queued| {
let same_snapshot = match &queued.data {
RelayItem::Packet(pkt) => {
pkt.data_type() == incoming_ty && pkt.sender() == incoming_sender
}
RelayItem::Packed(_) => false,
};
queued.dst != item.dst || !same_snapshot
});
}
}
self.make_shared_queue_room(item.byte_cost(), RelayQueueKind::Tx)?;
self.tx_queue
.push_back_prioritized(item, |queued| queued.priority)
}
fn push_replay(&mut self, item: RelayReplayItem) -> TelemetryResult<()> {
self.make_shared_queue_room(item.byte_cost(), RelayQueueKind::Replay)?;
self.replay_queue
.push_back_prioritized(item, |queued| queued.priority)
}
fn push_recent_rx(&mut self, id: u64) -> TelemetryResult<()> {
while self.recent_rx.len() >= self.memory.max_recent_rx_ids {
let _ = self.recent_rx.pop_front();
}
self.make_shared_queue_room(0, RelayQueueKind::Recent)?;
self.recent_rx.push_back(id)
}
#[allow(unused_mut)]
fn buffer_reliable_rx(
&mut self,
side: RelaySideId,
ty: crate::DataType,
seq: u32,
mut bytes: crate::SharedBytes,
) -> TelemetryResult<()> {
let key = Relay::reliable_key(side, ty);
if self
.reliable_rx
.get(&key)
.is_some_and(|state| state.buffered.contains_key(&seq))
{
return Ok(());
}
let cost = size_of::<crate::SharedBytes>() + bytes.len();
self.make_shared_queue_room(cost, RelayQueueKind::ReliableRxBuffer)?;
#[cfg(feature = "compact-packet-store")]
bytes.park_for_queue()?;
let rx_state = self
.reliable_rx
.entry(key)
.or_insert_with(|| ReliableRxState {
expected_seq: 1,
buffered: BTreeMap::new(),
});
if rx_state.buffered.len() >= runtime_reliable_max_pending() {
let _ = rx_state.buffered.pop_first();
}
rx_state.buffered.insert(seq, bytes);
Ok(())
}
}
pub struct Relay {
sender: RouterMutex<Arc<str>>,
state: RouterMutex<RelayInner>,
side_tx_gate: ReentryGate,
clock: Box<dyn Clock + Send + Sync>,
}
enum RemoteSidePlan {
Target(Vec<RelaySideId>),
}
impl Relay {
const END_TO_END_ACK_SENDER: &'static str = "E2EACK";
const END_TO_END_ACK_PREFIX: &'static str = "E2EACK:";
fn relay_item_priority(data: &RelayItem) -> TelemetryResult<u8> {
let ty = match data {
RelayItem::Packet(pkt) => pkt.data_type(),
RelayItem::Packed(bytes) => wire_format::peek_routing_envelope(bytes.as_ref())?.ty,
};
Ok(crate::scheduler_priority(ty))
}
#[inline]
fn is_side_tx_busy(err: &TelemetryError) -> bool {
matches!(err, TelemetryError::Io("side tx busy"))
}
fn process_replay_queue_item(&self) -> TelemetryResult<bool> {
let Some(mut item) = ({
let mut st = self.state.lock();
let ready = st
.replay_queue
.iter()
.position(|item| item.retry_after_ms <= self.clock.now_ms());
ready.and_then(|index| st.replay_queue.remove_pos(index))
}) else {
return Ok(false);
};
let frame = wire_format::peek_frame_info(item.bytes.as_ref())?;
let ty = frame.envelope.ty;
let Some(hdr) = frame.reliable else {
return Ok(false);
};
{
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, item.dst, ty);
if !tx_state.sent.contains_key(&hdr.seq) {
return Ok(false);
}
}
if let Err(e) = self.send_reliable_raw_to_side(item.dst, item.bytes.clone()) {
if Self::is_side_tx_busy(&e) {
let mut st = self.state.lock();
item.retry_after_ms = self.clock.now_ms().saturating_add(1);
st.push_replay(item)?;
return Ok(true);
}
return Err(e);
}
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, item.dst, ty);
if let Some(sent) = tx_state.sent.get_mut(&hdr.seq) {
sent.last_send_ms = self.clock.now_ms();
sent.queued = false;
}
Ok(true)
}
fn pop_ready_tx_item(
&self,
) -> Option<(
Option<RelaySideId>,
RelaySideId,
RelayTxHandlerFn,
RelaySideOptions,
RelayItem,
)> {
let mut st = self.state.lock();
let mut priority_burst = st.tx_priority_burst;
let item = st.tx_queue.pop_front_fair_if(
&mut priority_burst,
MAX_PRIORITY_BURST,
|queued| queued.priority,
|queued| queued.retry_after_ms <= self.clock.now_ms(),
);
st.tx_priority_burst = priority_burst;
if let Some(item) = item {
let side = st.sides.get(item.dst).and_then(|side| side.clone());
side.map(|s| (item.src, item.dst, s.tx_handler, s.opts, item.data))
} else {
None
}
}
fn send_tx_item(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
handler: RelayTxHandlerFn,
opts: RelaySideOptions,
data: RelayItem,
) -> TelemetryResult<bool> {
let allowed = {
let mut st = self.state.lock();
let ty = match &data {
RelayItem::Packet(pkt) => Some(pkt.data_type()),
RelayItem::Packed(bytes) => {
Some(wire_format::peek_routing_envelope(bytes.as_ref())?.ty)
}
};
let route_allowed = self.route_allowed_locked(&st, src, ty, dst);
#[cfg(all(feature = "discovery", feature = "timesync"))]
let timesync_allowed = ty
.map(|ty| {
Self::timesync_allowed_for_side_locked(&mut st, dst, ty, self.clock.now_ms())
})
.unwrap_or(true);
#[cfg(not(all(feature = "discovery", feature = "timesync")))]
let timesync_allowed = true;
route_allowed && timesync_allowed
};
if !allowed {
return Ok(false);
}
if opts.reliable_enabled && matches!(handler, RelayTxHandlerFn::Packed(_)) {
self.send_reliable_to_side(dst, data)?;
Ok(true)
} else if let Some(adjusted) = self.adjust_reliable_for_side(opts, data)? {
self.call_tx_handler(dst, &handler, &adjusted)?;
Ok(true)
} else {
Ok(false)
}
}
pub fn new(clock: Box<dyn Clock + Send + Sync>) -> Self {
Self::new_with_config(RelayConfig::default(), clock)
}
pub fn new_with_config(cfg: RelayConfig, clock: Box<dyn Clock + Send + Sync>) -> Self {
let memory = cfg.memory_config();
Self {
sender: RouterMutex::new(cfg.sender()),
state: RouterMutex::new(RelayInner {
memory,
sides: Vec::new(),
route_overrides: BTreeMap::new(),
typed_route_overrides: BTreeMap::new(),
route_weights: BTreeMap::new(),
route_priorities: BTreeMap::new(),
source_route_modes: BTreeMap::new(),
route_selection_cursors: BTreeMap::new(),
adaptive_route_stats: BTreeMap::new(),
side_runtime_stats: BTreeMap::new(),
side_transport: BTreeMap::new(),
rx_queue: BoundedDeque::new(
memory.max_queue_budget,
memory.starting_queue_size,
memory.queue_grow_step,
),
tx_queue: BoundedDeque::new(
memory.max_queue_budget,
memory.starting_queue_size,
memory.queue_grow_step,
),
tx_priority_burst: 0,
replay_queue: BoundedDeque::new(
memory.max_queue_budget,
memory.starting_queue_size,
memory.queue_grow_step,
),
recent_rx: BoundedDeque::new(
memory.recent_rx_queue_bytes(),
memory.recent_rx_queue_bytes(),
memory.queue_grow_step,
),
reliable_tx: BTreeMap::new(),
reliable_rx: BTreeMap::new(),
reliable_return_routes: BTreeMap::new(),
reliable_return_route_order: VecDeque::new(),
end_to_end_acked_destinations: BTreeMap::new(),
end_to_end_acked_destination_order: VecDeque::new(),
total_handler_failures: 0,
total_handler_retries: 0,
#[cfg(feature = "discovery")]
discovery_routes: BTreeMap::new(),
#[cfg(feature = "discovery")]
discovery_cadence: DiscoveryCadenceState::default(),
#[cfg(feature = "discovery")]
discovery_side_throttle: BTreeMap::new(),
#[cfg(all(feature = "discovery", feature = "timesync"))]
timesync_side_throttle: BTreeMap::new(),
}),
side_tx_gate: ReentryGate::new(),
clock,
}
}
#[inline]
fn sender_arc(&self) -> Arc<str> {
self.sender.lock().clone()
}
#[inline]
pub fn sender(&self) -> Arc<str> {
self.sender_arc()
}
pub fn set_sender<S: AsRef<str>>(&self, sender: S) {
*self.sender.lock() = Arc::from(sender.as_ref());
}
#[inline]
fn try_enter_side_tx(&self) -> Option<ReentryGuard<'_>> {
self.side_tx_gate.try_enter()
}
#[inline]
fn side_tx_active(&self) -> bool {
self.side_tx_gate.is_active()
}
#[inline]
fn side_ref(st: &RelayInner, side: RelaySideId) -> TelemetryResult<&RelaySide> {
st.sides
.get(side)
.and_then(|side| side.as_ref())
.ok_or(TelemetryError::HandlerError("relay: invalid side id"))
}
fn note_side_tx_success(
&self,
side: RelaySideId,
ty: crate::DataType,
bytes: usize,
attempts: usize,
) {
let mut st = self.state.lock();
#[cfg(feature = "discovery")]
if ty == crate::DataType::DiscoveryAnnounce
&& let Some(throttle) = st.discovery_side_throttle.get_mut(&side)
{
throttle.last_liveness_tx_ms = self.clock.now_ms();
}
let entry = st.side_runtime_stats.entry(side).or_default();
entry.note_tx(ty, bytes, attempts.saturating_sub(1));
}
fn note_side_tx_failure(&self, side: RelaySideId, ty: crate::DataType, attempts: usize) {
let mut st = self.state.lock();
st.total_handler_failures = st.total_handler_failures.saturating_add(1);
st.total_handler_retries = st.total_handler_retries.saturating_add(attempts as u64);
let entry = st.side_runtime_stats.entry(side).or_default();
entry.note_tx_failure(ty, attempts);
}
fn note_side_rx(&self, side: RelaySideId, ty: crate::DataType, bytes: usize) {
let mut st = self.state.lock();
let entry = st.side_runtime_stats.entry(side).or_default();
entry.note_rx(ty, bytes);
}
#[inline]
fn ensure_side_ingress_enabled(&self, side: RelaySideId) -> TelemetryResult<()> {
let st = self.state.lock();
let side_ref = Self::side_ref(&st, side)?;
if side_ref.opts.ingress_enabled {
Ok(())
} else {
Err(TelemetryError::HandlerError(
"relay: ingress disabled for side id",
))
}
}
#[inline]
fn route_allowed_locked(
&self,
st: &RelayInner,
src: Option<RelaySideId>,
ty: Option<crate::DataType>,
dst: RelaySideId,
) -> bool {
let Ok(dst_side) = Self::side_ref(st, dst) else {
return false;
};
if !dst_side.opts.egress_enabled {
return false;
}
if let Some(src_id) = src {
let Ok(src_side) = Self::side_ref(st, src_id) else {
return false;
};
if !src_side.opts.ingress_enabled || src_id == dst {
return false;
}
}
let base_allowed = st.route_overrides.get(&(src, dst)).copied().unwrap_or(true);
if !base_allowed {
return false;
}
let Some(ty) = ty else {
return true;
};
if st
.typed_route_overrides
.keys()
.any(|(typed_src, typed_ty, _)| *typed_src == src && *typed_ty == ty.as_u32())
{
return st
.typed_route_overrides
.get(&(src, ty.as_u32(), dst))
.copied()
.unwrap_or(false);
}
true
}
fn has_typed_route_overrides_locked(
st: &RelayInner,
src: Option<RelaySideId>,
ty: crate::DataType,
) -> bool {
st.typed_route_overrides
.keys()
.any(|(typed_src, typed_ty, _)| *typed_src == src && *typed_ty == ty.as_u32())
}
fn eligible_side_ids_locked(
&self,
st: &RelayInner,
src: Option<RelaySideId>,
ty: Option<crate::DataType>,
restrict_link_local: bool,
) -> Vec<RelaySideId> {
st.sides
.iter()
.enumerate()
.filter_map(|(side_id, side)| {
let side = side.as_ref()?;
if restrict_link_local && !side.opts.link_local_enabled {
return None;
}
if self.route_allowed_locked(st, src, ty, side_id) {
Some(side_id)
} else {
None
}
})
.collect()
}
fn apply_route_selection_locked(
&self,
st: &mut RelayInner,
src: Option<RelaySideId>,
mut sides: Vec<RelaySideId>,
origin: RouteSelectionOrigin,
) -> Vec<RelaySideId> {
if sides.len() <= 1 {
return sides;
}
let selection_mode = st.source_route_modes.get(&src).copied();
if selection_mode.is_none() && origin == RouteSelectionOrigin::Discovered {
return self.apply_adaptive_discovery_selection_locked(st, src, sides);
}
match selection_mode.unwrap_or(RouteSelectionMode::Fanout) {
RouteSelectionMode::Fanout => sides,
RouteSelectionMode::Weighted => {
crate::small_sort::sort(&mut sides);
let total_weight = sides.iter().fold(0_u64, |acc, side| {
acc + u64::from(st.route_weights.get(&(src, *side)).copied().unwrap_or(1))
});
if total_weight == 0 {
return Vec::new();
}
let cursor = st.route_selection_cursors.entry(src).or_insert(0);
let pick = *cursor % total_weight;
*cursor = cursor.wrapping_add(1);
let mut remaining = pick;
for side in sides {
let weight =
u64::from(st.route_weights.get(&(src, side)).copied().unwrap_or(1));
if remaining < weight {
return vec![side];
}
remaining -= weight;
}
Vec::new()
}
RouteSelectionMode::Failover => {
sides.sort_by_key(|side| {
(
st.route_priorities.get(&(src, *side)).copied().unwrap_or(0),
*side,
)
});
sides.truncate(1);
sides
}
}
}
fn apply_adaptive_discovery_selection_locked(
&self,
st: &mut RelayInner,
src: Option<RelaySideId>,
mut sides: Vec<RelaySideId>,
) -> Vec<RelaySideId> {
crate::small_sort::sort(&mut sides);
let mut unmeasured: Vec<_> = sides
.iter()
.copied()
.filter(|side| !st.adaptive_route_stats.contains_key(side))
.collect();
if !unmeasured.is_empty() {
let cursor = st.route_selection_cursors.entry(src).or_insert(0);
let pick = (*cursor as usize) % unmeasured.len();
*cursor = cursor.wrapping_add(1);
return vec![unmeasured.swap_remove(pick)];
}
let now_ms = self.clock.now_ms();
let total_weight = sides.iter().fold(0_u64, |acc, side| {
acc + st
.adaptive_route_stats
.get(side)
.map(|stats| stats.weight(now_ms))
.unwrap_or(1)
});
if total_weight == 0 {
sides.truncate(1);
return sides;
}
let cursor = st.route_selection_cursors.entry(src).or_insert(0);
let pick = *cursor % total_weight;
*cursor = cursor.wrapping_add(1);
let mut remaining = pick;
for side in sides {
let weight = st
.adaptive_route_stats
.get(&side)
.map(|stats| stats.weight(now_ms))
.unwrap_or(1);
if remaining < weight {
return vec![side];
}
remaining -= weight;
}
Vec::new()
}
#[cfg(feature = "discovery")]
fn retain_shortest_discovery_routes_locked(
st: &RelayInner,
sides: &mut Vec<RelaySideId>,
endpoints: &[crate::DataEndpoint],
now_ms: u64,
) {
if sides.len() <= 1 || endpoints.is_empty() {
return;
}
let route_distance = |side: RelaySideId| -> Option<usize> {
let route = st.discovery_routes.get(&side)?;
route
.announcers
.iter()
.filter(|(_, state)| {
now_ms.saturating_sub(state.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
})
.filter_map(|(announcer, state)| {
let targets: BTreeSet<&str> = state
.topology_boards
.iter()
.filter(|board| {
endpoints
.iter()
.any(|endpoint| board.reachable_endpoints.contains(endpoint))
})
.map(|board| board.sender_id.as_str())
.collect();
if targets.is_empty() {
return state
.reachable
.iter()
.any(|endpoint| endpoints.contains(endpoint))
.then_some(usize::MAX / 2);
}
let mut distances: BTreeMap<&str, usize> = BTreeMap::new();
let mut pending = VecDeque::from([(announcer.as_str(), 0usize)]);
while let Some((sender, distance)) = pending.pop_front() {
if distances.contains_key(sender) {
continue;
}
distances.insert(sender, distance);
if targets.contains(sender) {
return Some(distance);
}
for board in &state.topology_boards {
if board.sender_id == sender {
for peer in &board.connections {
if !distances.contains_key(peer.as_str()) {
pending.push_back((peer.as_str(), distance + 1));
}
}
} else if board.connections.iter().any(|peer| peer == sender)
&& !distances.contains_key(board.sender_id.as_str())
{
pending.push_back((board.sender_id.as_str(), distance + 1));
}
}
}
None
})
.min()
};
let distances: Vec<(RelaySideId, usize)> = sides
.iter()
.filter_map(|side| route_distance(*side).map(|distance| (*side, distance)))
.collect();
let Some(best) = distances.iter().map(|(_, distance)| *distance).min() else {
return;
};
sides.retain(|side| {
distances
.iter()
.any(|(candidate, distance)| candidate == side && *distance == best)
});
}
fn record_side_tx_sample(
&self,
side: RelaySideId,
bytes: usize,
started_ms: u64,
ended_ms: u64,
) {
let sample_ms = ended_ms.saturating_sub(started_ms).max(1);
let sample_bps = ((bytes as u128).saturating_mul(1000) / u128::from(sample_ms))
.min(u128::from(u64::MAX)) as u64;
let mut st = self.state.lock();
st.adaptive_route_stats
.entry(side)
.or_default()
.observe(bytes, sample_bps, ended_ms);
}
pub fn note_side_link_probe_sample(
&self,
side: RelaySideId,
bytes: usize,
duration_ms: u64,
) -> TelemetryResult<()> {
{
let st = self.state.lock();
let _ = Self::side_ref(&st, side).map_err(|_| TelemetryError::BadArg)?;
}
let ended_ms = self.clock.now_ms();
self.record_side_tx_sample(side, bytes, ended_ms.saturating_sub(duration_ms), ended_ms);
Ok(())
}
fn relay_item_wire_len(data: &RelayItem) -> TelemetryResult<usize> {
match data {
RelayItem::Packet(pkt) => Ok(wire_format::packet_wire_size(pkt)),
RelayItem::Packed(bytes) => Ok(bytes.len()),
}
}
#[inline]
fn decode_end_to_end_reliable_ack(payload: &[u8]) -> TelemetryResult<u64> {
if payload.len() != 8 {
return Err(TelemetryError::Unpack("bad reliable e2e ack payload"));
}
Ok(u64::from_le_bytes(payload[0..8].try_into().unwrap()))
}
#[inline]
fn is_end_to_end_ack_sender(sender: &str) -> bool {
sender == Self::END_TO_END_ACK_SENDER || sender.starts_with(Self::END_TO_END_ACK_PREFIX)
}
#[inline]
fn sender_hash(sender: &str) -> u64 {
if let Some(address) = sender
.strip_prefix("@addr:")
.and_then(|value| value.parse::<u32>().ok())
{
return u64::from(address);
}
hash_bytes_u64(0x517C_C1B7_2722_0A95, sender.as_bytes())
}
#[cfg(feature = "discovery")]
fn canonical_sender_locked(st: &RelayInner, sender: &str) -> String {
let Some(address) = sender
.strip_prefix("@addr:")
.and_then(|value| value.parse::<u32>().ok())
else {
return sender.to_string();
};
Self::hostname_for_wire_address_locked(st, address).unwrap_or_else(|| sender.to_string())
}
#[cfg(feature = "discovery")]
fn hostname_for_wire_address_locked(st: &RelayInner, address: u32) -> Option<String> {
let mut found: Option<String> = None;
let mut collision = false;
let mut consider = |candidate: &str| {
if sender_address_u32(candidate) != address {
return;
}
match found.as_deref() {
None => found = Some(candidate.to_string()),
Some(existing) if existing == candidate => {}
Some(_) => collision = true,
}
};
for route in st.discovery_routes.values() {
for (announcer, state) in &route.announcers {
consider(announcer);
for board in &state.topology_boards {
consider(&board.sender_id);
}
}
}
if collision { None } else { found }
}
fn decode_end_to_end_ack_sender_hash(sender: &str) -> Option<u64> {
sender
.strip_prefix(Self::END_TO_END_ACK_PREFIX)
.filter(|sender| !sender.is_empty())
.map(Self::sender_hash)
}
fn end_to_end_ack_sender_hash(pkt: &Packet) -> Option<u64> {
if pkt.data_type() != crate::DataType::ReliableAck {
return None;
}
pkt.wire_target_senders()
.first()
.copied()
.or_else(|| Self::decode_end_to_end_ack_sender_hash(pkt.sender()))
}
#[cfg(feature = "discovery")]
fn is_end_to_end_destination_sender(&self, sender: &str) -> bool {
sender != self.sender_arc().as_ref() && !Self::is_end_to_end_ack_sender(sender)
}
fn reliable_control_target_packet_id(data: &RelayItem) -> TelemetryResult<Option<u64>> {
match data {
RelayItem::Packet(pkt) => {
if Self::end_to_end_ack_sender_hash(pkt).is_none() {
return Ok(None);
}
Self::decode_end_to_end_reliable_ack(pkt.payload()).map(Some)
}
RelayItem::Packed(bytes) => {
if wire_format::peek_routing_frame_info(bytes.as_ref())
.ok()
.is_some_and(|frame| {
frame.envelope.ty != crate::DataType::ReliableAck || frame.ack_only()
})
{
return Ok(None);
}
let pkt = wire_format::unpack_packet(bytes.as_ref())?;
if Self::end_to_end_ack_sender_hash(&pkt).is_none() {
return Ok(None);
}
Self::decode_end_to_end_reliable_ack(pkt.payload()).map(Some)
}
}
}
fn note_reliable_return_route(&self, side: RelaySideId, packet_id: u64) {
let mut st = self.state.lock();
let last_replay_ms = if st.recent_rx.contains(&packet_id) {
self.clock
.now_ms()
.saturating_sub(runtime_reliable_retransmit_ms())
} else {
self.clock.now_ms()
};
Self::remember_reliable_return_route_locked(&mut st, packet_id);
st.reliable_return_routes
.entry(packet_id)
.and_modify(|route| route.side = side)
.or_insert(ReliableReturnRouteState {
side,
last_replay_ms,
});
}
fn remember_reliable_return_route_locked(st: &mut RelayInner, packet_id: u64) {
let cap = runtime_reliable_max_return_routes().max(1);
st.reliable_return_route_order
.retain(|id| st.reliable_return_routes.contains_key(id) && *id != packet_id);
while st.reliable_return_route_order.len() >= cap {
if let Some(oldest) = st.reliable_return_route_order.pop_front() {
st.reliable_return_routes.remove(&oldest);
} else {
break;
}
}
st.reliable_return_route_order.push_back(packet_id);
}
fn note_end_to_end_acked_destination_locked(
st: &mut RelayInner,
packet_id: u64,
sender_hash: u64,
) {
let entry_cap = runtime_reliable_max_end_to_end_ack_cache().max(1);
st.end_to_end_acked_destination_order
.retain(|id| st.end_to_end_acked_destinations.contains_key(id) && *id != packet_id);
while st.end_to_end_acked_destination_order.len() >= entry_cap {
if let Some(oldest) = st.end_to_end_acked_destination_order.pop_front() {
st.end_to_end_acked_destinations.remove(&oldest);
} else {
break;
}
}
st.end_to_end_acked_destination_order.push_back(packet_id);
let acked = st
.end_to_end_acked_destinations
.entry(packet_id)
.or_default();
let sender_cap = runtime_reliable_max_end_to_end_pending().max(1);
if acked.len() < sender_cap || acked.contains(&sender_hash) {
acked.insert(sender_hash);
}
}
#[inline]
fn reliable_key(side: RelaySideId, ty: crate::DataType) -> (RelaySideId, u32) {
(side, ty.as_u32())
}
fn reliable_tx_state_mut<'a>(
&'a self,
st: &'a mut RelayInner,
side: RelaySideId,
ty: crate::DataType,
) -> &'a mut ReliableTxState {
let key = Self::reliable_key(side, ty);
st.reliable_tx
.entry(key)
.or_insert_with(|| ReliableTxState {
next_seq: 1,
sent_order: VecDeque::new(),
sent: BTreeMap::new(),
})
}
fn reliable_rx_state_mut<'a>(
&'a self,
st: &'a mut RelayInner,
side: RelaySideId,
ty: crate::DataType,
) -> &'a mut ReliableRxState {
let key = Self::reliable_key(side, ty);
st.reliable_rx
.entry(key)
.or_insert_with(|| ReliableRxState {
expected_seq: 1,
buffered: BTreeMap::new(),
})
}
fn handle_reliable_ack(&self, side: RelaySideId, ty: crate::DataType, ack: u32) {
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, side, ty);
if matches!(reliable_mode(ty), crate::ReliableMode::Unordered) {
tx_state.sent.remove(&ack);
tx_state.sent_order.retain(|seq| *seq != ack);
return;
}
while let Some(seq) = tx_state.sent_order.front().copied() {
if seq > ack {
break;
}
tx_state.sent_order.pop_front();
tx_state.sent.remove(&seq);
}
}
fn handle_reliable_partial_ack(&self, side: RelaySideId, ty: crate::DataType, seq: u32) {
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, side, ty);
if let Some(sent) = tx_state.sent.get_mut(&seq) {
sent.partial_acked = true;
}
}
fn reliable_control_packet(
&self,
control_ty: crate::DataType,
ty: crate::DataType,
seq: u32,
) -> TelemetryResult<Packet> {
let sender = self.sender_arc();
Packet::new(
control_ty,
message_meta(control_ty).endpoints_ref(),
sender.as_ref(),
self.clock.now_ms(),
crate::router::encode_slice_le(&[ty.as_u32(), seq]),
)
}
fn queue_reliable_ack(
&self,
side: RelaySideId,
ty: crate::DataType,
seq: u32,
) -> TelemetryResult<()> {
let pkt = self.reliable_control_packet(crate::DataType::ReliableAck, ty, seq)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
let mut st = self.state.lock();
st.push_tx(RelayTxItem {
src: None,
dst: side,
data,
priority,
retry_after_ms: 0,
})?;
Ok(())
}
fn queue_reliable_packet_request(
&self,
side: RelaySideId,
ty: crate::DataType,
seq: u32,
) -> TelemetryResult<()> {
let pkt = self.reliable_control_packet(crate::DataType::ReliablePacketRequest, ty, seq)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
let mut st = self.state.lock();
st.push_tx(RelayTxItem {
src: None,
dst: side,
data,
priority,
retry_after_ms: 0,
})?;
Ok(())
}
fn queue_reliable_partial_ack(
&self,
side: RelaySideId,
ty: crate::DataType,
seq: u32,
) -> TelemetryResult<()> {
let pkt = self.reliable_control_packet(crate::DataType::ReliablePartialAck, ty, seq)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
let mut st = self.state.lock();
st.push_tx(RelayTxItem {
src: None,
dst: side,
data,
priority,
retry_after_ms: 0,
})?;
Ok(())
}
fn queue_reliable_retransmit(
&self,
side: RelaySideId,
ty: crate::DataType,
seq: u32,
) -> TelemetryResult<()> {
let mut queued = None;
{
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, side, ty);
if let Some(sent) = tx_state.sent.get_mut(&seq)
&& !sent.queued
{
sent.queued = true;
sent.partial_acked = false;
queued = Some(sent.bytes.clone());
}
}
if let Some(bytes) = queued {
let mut st = self.state.lock();
st.push_replay(RelayReplayItem {
dst: side,
bytes,
priority: message_priority(ty).saturating_add(16),
retry_after_ms: 0,
})?;
}
Ok(())
}
fn send_reliable_raw_to_side(
&self,
side: RelaySideId,
bytes: crate::SharedBytes,
) -> TelemetryResult<()> {
let (handler, opts) = {
let st = self.state.lock();
let side_ref = Self::side_ref(&st, side)?;
if !side_ref.opts.egress_enabled {
return Ok(());
}
(side_ref.tx_handler.clone(), side_ref.opts.clone())
};
let Some(_side_tx_guard) = self.try_enter_side_tx() else {
return Err(TelemetryError::Io("side tx busy"));
};
let started_ms = self.clock.now_ms();
let ty = wire_format::peek_routing_envelope(bytes.as_ref())
.map(|env| env.ty)
.unwrap_or(crate::DataType::ReliableAck);
let result = match handler {
RelayTxHandlerFn::Packed(f) => {
let frames = self.encode_side_transport_frames(side, opts, bytes.clone())?;
let mut sent_bytes = 0usize;
self.send_side_frames_resumable(side, frames, |frame| {
f(frame)?;
sent_bytes = sent_bytes.saturating_add(frame.len());
Ok(())
})?;
self.record_side_tx_sample(side, sent_bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, sent_bytes, 1);
return Ok(());
}
RelayTxHandlerFn::Packet(f) => {
if wire_format::peek_frame_info(bytes.as_ref())
.ok()
.is_some_and(|frame| frame.ack_only())
{
return Ok(());
}
let pkt = wire_format::unpack_packet(bytes.as_ref())?;
f(&pkt)
}
};
if result.is_ok() {
self.record_side_tx_sample(side, bytes.len(), started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, bytes.len(), 1);
} else {
self.note_side_tx_failure(side, ty, 1);
}
result
}
#[allow(unused_mut)]
fn send_reliable_to_side(&self, side: RelaySideId, data: RelayItem) -> TelemetryResult<()> {
let ty = match &data {
RelayItem::Packet(pkt) => pkt.data_type(),
RelayItem::Packed(bytes) => wire_format::peek_routing_envelope(bytes)?.ty,
};
let end_to_end_ordered = !self.item_target_senders(&data)?.is_empty()
&& matches!(reliable_mode(ty), crate::ReliableMode::Ordered);
let (handler, opts, hop_reliable_enabled) = {
let st = self.state.lock();
let side_ref = Self::side_ref(&st, side)?;
let opts = side_ref.opts.clone();
let hop_reliable_enabled = opts.reliable_enabled
&& !end_to_end_ordered
&& !self.side_has_multiple_announcers_locked(&st, side, self.clock.now_ms());
(side_ref.tx_handler.clone(), opts, hop_reliable_enabled)
};
let RelayTxHandlerFn::Packed(f) = &handler else {
return self.call_tx_handler(side, &handler, &data);
};
if !hop_reliable_enabled {
let mut adjusted_opts = opts;
adjusted_opts.reliable_enabled = false;
if let Some(adjusted) = self.adjust_reliable_for_side(adjusted_opts, data)? {
return self.call_tx_handler(side, &handler, &adjusted);
}
return Ok(());
}
let ty = match &data {
RelayItem::Packet(pkt) => pkt.data_type(),
RelayItem::Packed(bytes) => {
let Ok(frame) = wire_format::peek_frame_info(bytes.as_ref()) else {
return self.call_tx_handler(side, &handler, &data);
};
frame.envelope.ty
}
};
if !is_reliable_type(ty) {
if let Some(adjusted) = self.adjust_reliable_for_side(opts, data)? {
self.call_tx_handler(side, &handler, &adjusted)?;
}
return Ok(());
}
let (seq, flags) = {
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, side, ty);
if tx_state.sent.len() >= runtime_reliable_max_pending() {
return Err(TelemetryError::PacketTooLarge(
"relay reliable history full",
));
}
let seq = tx_state.next_seq;
let next = tx_state.next_seq.wrapping_add(1);
tx_state.next_seq = if next == 0 { 1 } else { next };
let flags = match reliable_mode(ty) {
crate::ReliableMode::Unordered => wire_format::RELIABLE_FLAG_UNORDERED,
_ => 0,
};
(seq, flags)
};
let mut bytes: crate::SharedBytes = match data {
RelayItem::Packet(pkt) => {
crate::shared_bytes::convert(wire_format::pack_packet_with_reliable(
&pkt,
wire_format::ReliableHeader { flags, seq, ack: 0 },
))
}
RelayItem::Packed(bytes) => {
let Some(rewritten) =
wire_format::rewrite_reliable_header_owned(bytes.as_ref(), flags, seq, 0)?
else {
let Some(_side_tx_guard) = self.try_enter_side_tx() else {
return Err(TelemetryError::Io("side tx busy"));
};
let started_ms = self.clock.now_ms();
let frames = self.encode_side_transport_frames(side, opts, bytes.clone())?;
let mut sent_bytes = 0usize;
self.send_side_frames_resumable(side, frames, |frame| {
f(frame)?;
sent_bytes = sent_bytes.saturating_add(frame.len());
Ok(())
})?;
self.record_side_tx_sample(side, sent_bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, sent_bytes, 1);
return Ok(());
};
crate::shared_bytes::convert(rewritten)
}
};
let Some(_side_tx_guard) = self.try_enter_side_tx() else {
return Err(TelemetryError::Io("side tx busy"));
};
let started_ms = self.clock.now_ms();
let frames = self.encode_side_transport_frames(side, opts, bytes.clone())?;
let mut sent_bytes = 0usize;
self.send_side_frames_resumable(side, frames, |frame| {
f(frame)?;
sent_bytes = sent_bytes.saturating_add(frame.len());
Ok(())
})?;
self.record_side_tx_sample(side, sent_bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, sent_bytes, 1);
#[cfg(feature = "compact-packet-store")]
{
let _ = bytes.park_for_queue();
}
{
let mut st = self.state.lock();
let tx_state = self.reliable_tx_state_mut(&mut st, side, ty);
tx_state.sent_order.push_back(seq);
tx_state.sent.insert(
seq,
ReliableSent {
bytes: bytes.clone(),
last_send_ms: self.clock.now_ms(),
retries: 0,
queued: false,
partial_acked: false,
},
);
}
Ok(())
}
fn item_route_info(
&self,
data: &RelayItem,
) -> TelemetryResult<(Vec<crate::DataEndpoint>, crate::DataType)> {
match data {
RelayItem::Packet(pkt) => {
let mut eps = pkt.endpoints().to_vec();
crate::small_sort::sort(&mut eps);
eps.dedup();
Ok((eps, pkt.data_type()))
}
RelayItem::Packed(bytes) => {
let env = wire_format::peek_envelope(bytes.as_ref())?;
let mut eps: Vec<crate::DataEndpoint> = env.endpoints.iter().copied().collect();
crate::small_sort::sort(&mut eps);
eps.dedup();
Ok((eps, env.ty))
}
}
}
fn endpoints_are_link_local_only(eps: &[crate::DataEndpoint]) -> bool {
!eps.is_empty() && eps.iter().all(|ep| ep.is_link_local_only())
}
fn item_target_senders<'a>(
&self,
data: &'a RelayItem,
) -> TelemetryResult<alloc::borrow::Cow<'a, [u64]>> {
match data {
RelayItem::Packet(pkt) => Ok(alloc::borrow::Cow::Borrowed(pkt.wire_target_senders())),
RelayItem::Packed(bytes) => {
let env = wire_format::peek_routing_envelope(bytes.as_ref())?;
if env.target_senders().len() == 0 {
Ok(alloc::borrow::Cow::Borrowed(&[]))
} else {
Ok(alloc::borrow::Cow::Owned(env.target_senders().collect()))
}
}
}
}
#[cfg(feature = "discovery")]
fn has_explicit_route_policy_locked(
st: &RelayInner,
src: Option<RelaySideId>,
ty: crate::DataType,
) -> bool {
st.route_overrides
.keys()
.any(|(route_src, _)| *route_src == src)
|| Self::has_typed_route_overrides_locked(st, src, ty)
}
#[cfg(feature = "discovery")]
fn side_matches_target_senders_locked(
st: &RelayInner,
side: RelaySideId,
target_senders: &[u64],
now_ms: u64,
) -> bool {
st.discovery_routes
.get(&side)
.map(|route| {
if now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
return false;
}
route.announcers.iter().any(|(announcer, sender_state)| {
if now_ms.saturating_sub(sender_state.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
return false;
}
target_senders.contains(&Self::sender_hash(announcer))
|| sender_state
.advertised_address
.is_some_and(|address| target_senders.contains(&u64::from(address)))
|| sender_state.topology_boards.iter().any(|board| {
target_senders.contains(&Self::sender_hash(&board.sender_id))
})
})
})
.unwrap_or(false)
}
fn remote_side_plan(
&self,
data: &RelayItem,
exclude: RelaySideId,
) -> TelemetryResult<RemoteSidePlan> {
#[cfg(feature = "discovery")]
{
let (eps, ty) = self.item_route_info(data)?;
let target_senders = self.item_target_senders(data)?;
let preferred_packet_id = Self::reliable_control_target_packet_id(data)?;
if discovery::is_discovery_type(ty) {
if matches!(
ty,
crate::DataType::DiscoveryAnnounce
| crate::DataType::DiscoveryTimeSyncSources
| crate::DataType::DiscoveryTopology
| crate::DataType::DiscoveryLinkCapabilities
) {
return Ok(RemoteSidePlan::Target(Vec::new()));
}
let mut st = self.state.lock();
let sides = self.eligible_side_ids_locked(&st, Some(exclude), Some(ty), false);
return Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
sides,
RouteSelectionOrigin::Flood,
)));
}
#[cfg(feature = "timesync")]
let preferred_timesync_source = self.preferred_timesync_route_source(data, ty)?;
#[cfg(not(feature = "timesync"))]
let preferred_timesync_source: Option<String> = None;
let mut st = self.state.lock();
if let Some(packet_id) = preferred_packet_id {
let target_side = self.allowed_target_side_locked(
&st,
exclude,
ty,
st.reliable_return_routes
.get(&packet_id)
.map(|route| route.side),
);
if let Some(side) = target_side {
#[cfg(feature = "timesync")]
if !Self::timesync_allowed_for_side_locked(
&mut st,
side,
ty,
self.clock.now_ms(),
) {
return Ok(RemoteSidePlan::Target(Vec::new()));
}
return Ok(RemoteSidePlan::Target(vec![side]));
}
let eligible = self.eligible_side_ids_locked(&st, Some(exclude), Some(ty), false);
let original = target_senders.get(1..).unwrap_or(&[]);
let exact: Vec<_> = eligible
.iter()
.copied()
.filter(|side| {
!original.is_empty()
&& Self::side_matches_target_senders_locked(
&st,
*side,
original,
self.clock.now_ms(),
)
})
.collect();
if !exact.is_empty() {
return Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
exact,
RouteSelectionOrigin::Discovered,
)));
}
return Ok(RemoteSidePlan::Target(if eligible.len() == 1 {
eligible
} else {
Vec::new()
}));
}
let restrict_link_local = Self::endpoints_are_link_local_only(&eps);
let discovered_origin = if is_reliable_type(ty) {
RouteSelectionOrigin::Flood
} else {
RouteSelectionOrigin::Discovered
};
if st.discovery_routes.is_empty() {
let mut fallback = self.eligible_side_ids_locked(
&st,
Some(exclude),
Some(ty),
restrict_link_local,
);
#[cfg(feature = "timesync")]
{
fallback = Self::filter_timesync_sides_locked(
&mut st,
ty,
self.clock.now_ms(),
fallback,
);
}
return Ok(RemoteSidePlan::Target(if fallback.len() == 1 {
fallback
} else {
Vec::new()
}));
}
let now_ms = self.clock.now_ms();
let mut had_exact = false;
let mut exact_targets = Vec::new();
let mut had_known = false;
let mut generic_targets = Vec::new();
let mut unique_endpoint_fallbacks = Vec::new();
for (&side, route) in st.discovery_routes.iter() {
if side == exclude
|| now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS
{
continue;
}
if restrict_link_local
&& st
.sides
.get(side)
.and_then(|side| side.as_ref())
.map(|s| !s.opts.link_local_enabled)
.unwrap_or(true)
{
continue;
}
if !self.route_allowed_locked(&st, Some(exclude), Some(ty), side) {
continue;
}
if !target_senders.is_empty() {
if !Self::side_matches_target_senders_locked(&st, side, &target_senders, now_ms)
{
if eps.iter().copied().any(|ep| route.reachable.contains(&ep)) {
unique_endpoint_fallbacks.push(side);
}
continue;
}
had_known = true;
generic_targets.push(side);
continue;
}
if preferred_timesync_source.as_deref().is_some_and(|source| {
route.reachable_timesync_sources.iter().any(|s| s == source)
}) {
had_exact = true;
exact_targets.push(side);
continue;
}
if eps.iter().copied().any(|ep| route.reachable.contains(&ep)) {
had_known = true;
generic_targets.push(side);
}
}
if !target_senders.is_empty() && !had_known && unique_endpoint_fallbacks.len() == 1 {
had_known = true;
generic_targets = unique_endpoint_fallbacks;
}
if had_exact {
#[cfg(feature = "timesync")]
{
exact_targets = Self::filter_timesync_sides_locked(
&mut st,
ty,
self.clock.now_ms(),
exact_targets,
);
}
let targets = self.filter_end_to_end_satisfied_sides_locked(
&st,
data,
exact_targets,
&eps,
ty,
)?;
Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
targets,
discovered_origin,
)))
} else if had_known {
Self::retain_shortest_discovery_routes_locked(
&st,
&mut generic_targets,
&eps,
now_ms,
);
#[cfg(feature = "timesync")]
{
generic_targets = Self::filter_timesync_sides_locked(
&mut st,
ty,
self.clock.now_ms(),
generic_targets,
);
}
let targets = self.filter_end_to_end_satisfied_sides_locked(
&st,
data,
generic_targets,
&eps,
ty,
)?;
Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
targets,
discovered_origin,
)))
} else {
let fallback = self.eligible_side_ids_locked(
&st,
Some(exclude),
Some(ty),
restrict_link_local,
);
if fallback.len() == 1 {
return Ok(RemoteSidePlan::Target(fallback));
}
if Self::has_explicit_route_policy_locked(&st, Some(exclude), ty) {
let mut sides = self.eligible_side_ids_locked(
&st,
Some(exclude),
Some(ty),
restrict_link_local,
);
#[cfg(feature = "timesync")]
{
sides = Self::filter_timesync_sides_locked(
&mut st,
ty,
self.clock.now_ms(),
sides,
);
}
Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
sides,
RouteSelectionOrigin::Flood,
)))
} else {
Ok(RemoteSidePlan::Target(Vec::new()))
}
}
}
#[cfg(not(feature = "discovery"))]
{
let (_, ty) = self.item_route_info(data)?;
let mut st = self.state.lock();
if let Some(packet_id) = Self::reliable_control_target_packet_id(data)? {
let target_side = self.allowed_target_side_locked(
&st,
exclude,
ty,
st.reliable_return_routes
.get(&packet_id)
.map(|route| route.side),
);
if let Some(side) = target_side {
return Ok(RemoteSidePlan::Target(vec![side]));
}
let sides = self.eligible_side_ids_locked(&st, Some(exclude), Some(ty), false);
return Ok(RemoteSidePlan::Target(if sides.len() == 1 {
sides
} else {
Vec::new()
}));
}
let sides = self.eligible_side_ids_locked(&st, Some(exclude), Some(ty), false);
Ok(RemoteSidePlan::Target(self.apply_route_selection_locked(
&mut st,
Some(exclude),
sides,
RouteSelectionOrigin::Flood,
)))
}
}
#[inline]
fn allowed_target_side_locked(
&self,
st: &RelayInner,
exclude: RelaySideId,
ty: crate::DataType,
target_side: Option<RelaySideId>,
) -> Option<RelaySideId> {
target_side.filter(|side| self.route_allowed_locked(st, Some(exclude), Some(ty), *side))
}
#[cfg(feature = "discovery")]
fn filter_end_to_end_satisfied_sides_locked(
&self,
st: &RelayInner,
data: &RelayItem,
sides: Vec<RelaySideId>,
eps: &[crate::DataEndpoint],
ty: crate::DataType,
) -> TelemetryResult<Vec<RelaySideId>> {
if !is_reliable_type(ty) || Self::reliable_control_target_packet_id(data)?.is_some() {
return Ok(sides);
}
let packet_id = match data {
RelayItem::Packet(pkt) => pkt.packet_id(),
RelayItem::Packed(bytes) => match wire_format::packet_id_from_wire(bytes.as_ref()) {
Ok(packet_id) => packet_id,
Err(TelemetryError::Unpack("reliable control frame")) => return Ok(sides),
Err(err) => return Err(err),
},
};
let Some(acked) = st.end_to_end_acked_destinations.get(&packet_id) else {
return Ok(sides);
};
let now_ms = self.clock.now_ms();
let mut filtered = Vec::new();
for side in sides {
let Some(route) = st.discovery_routes.get(&side) else {
filtered.push(side);
continue;
};
let mut still_pending = false;
let mut had_destination_board = false;
for sender_state in route.announcers.values() {
if now_ms.saturating_sub(sender_state.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
continue;
}
for board in sender_state.topology_boards.iter() {
if !self.is_end_to_end_destination_sender(&board.sender_id) {
continue;
}
had_destination_board = true;
let sender_hash = Self::sender_hash(&board.sender_id);
if acked.contains(&sender_hash) {
continue;
}
if eps
.iter()
.copied()
.any(|ep| board.reachable_endpoints.contains(&ep))
{
still_pending = true;
break;
}
still_pending = true;
break;
}
if still_pending {
break;
}
}
if still_pending || !had_destination_board {
filtered.push(side);
}
}
Ok(filtered)
}
#[cfg(not(feature = "discovery"))]
fn filter_end_to_end_satisfied_sides_locked(
&self,
_st: &RelayInner,
_data: &RelayItem,
sides: Vec<RelaySideId>,
_eps: &[crate::DataEndpoint],
_ty: crate::DataType,
) -> TelemetryResult<Vec<RelaySideId>> {
Ok(sides)
}
#[cfg(feature = "discovery")]
fn side_has_multiple_announcers_locked(
&self,
st: &RelayInner,
side: RelaySideId,
now_ms: u64,
) -> bool {
st.discovery_routes
.get(&side)
.map(|route| {
route
.announcers
.values()
.filter(|sender| {
now_ms.saturating_sub(sender.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
})
.take(2)
.count()
> 1
})
.unwrap_or(false)
}
#[cfg(not(feature = "discovery"))]
fn side_has_multiple_announcers_locked(
&self,
_st: &RelayInner,
_side: RelaySideId,
_now_ms: u64,
) -> bool {
false
}
#[cfg(feature = "discovery")]
fn sender_topology_board_mut<'a>(
sender_state: &'a mut DiscoverySenderState,
sender_id: &str,
) -> &'a mut TopologyBoardNode {
if let Some(idx) = sender_state
.topology_boards
.iter()
.position(|board| board.sender_id == sender_id)
{
return &mut sender_state.topology_boards[idx];
}
sender_state.topology_boards.push(TopologyBoardNode {
sender_id: sender_id.to_string(),
reachable_endpoints: Vec::new(),
reachable_timesync_sources: Vec::new(),
connections: Vec::new(),
});
sender_state
.topology_boards
.last_mut()
.expect("board inserted above")
}
#[cfg(feature = "discovery")]
fn refresh_sender_topology_state(sender_state: &mut DiscoverySenderState) {
discovery::normalize_topology_boards(&mut sender_state.topology_boards);
let (mut reachable, mut reachable_timesync_sources) =
discovery::summarize_topology_boards(&sender_state.topology_boards);
reachable.extend(sender_state.advertised_reachable.iter().copied());
crate::small_sort::sort(&mut reachable);
reachable.dedup();
reachable_timesync_sources.extend(
sender_state
.advertised_reachable_timesync_sources
.iter()
.cloned(),
);
reachable_timesync_sources.sort_unstable();
reachable_timesync_sources.dedup();
sender_state.reachable = reachable;
sender_state.reachable_timesync_sources = reachable_timesync_sources;
}
#[cfg(feature = "discovery")]
fn recompute_discovery_side_state(route: &mut DiscoverySideState) {
let mut reachable = Vec::new();
let mut reachable_timesync_sources = Vec::new();
let mut last_seen_ms = 0u64;
for sender in route.announcers.values() {
reachable.extend(sender.reachable.iter().copied());
reachable_timesync_sources.extend(sender.reachable_timesync_sources.iter().cloned());
last_seen_ms = last_seen_ms.max(sender.last_seen_ms);
}
crate::small_sort::sort(&mut reachable);
reachable.dedup();
reachable_timesync_sources.sort_unstable();
reachable_timesync_sources.dedup();
route.reachable = reachable;
route.reachable_timesync_sources = reachable_timesync_sources;
route.last_seen_ms = last_seen_ms;
}
#[cfg(feature = "discovery")]
fn local_discovery_topology_board(&self, st: &RelayInner, now_ms: u64) -> TopologyBoardNode {
let mut connections = Vec::new();
for route in st.discovery_routes.values() {
if now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
continue;
}
for (sender, sender_state) in route.announcers.iter() {
if now_ms.saturating_sub(sender_state.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS {
connections.push(sender.clone());
}
}
}
connections.sort_unstable();
connections.dedup();
let sender = self.sender_arc();
TopologyBoardNode {
sender_id: sender.to_string(),
reachable_endpoints: Vec::new(),
reachable_timesync_sources: Vec::new(),
connections,
}
}
#[cfg(feature = "discovery")]
fn advertised_discovery_topology_for_link_locked(
&self,
st: &RelayInner,
now_ms: u64,
link_local_enabled: bool,
exclude_side: Option<RelaySideId>,
) -> Vec<TopologyBoardNode> {
let mut boards = vec![self.local_discovery_topology_board(st, now_ms)];
for (&route_side, route) in st.discovery_routes.iter() {
if exclude_side == Some(route_side) {
continue;
}
if now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
continue;
}
for (announcer, sender_state) in route.announcers.iter() {
if now_ms.saturating_sub(sender_state.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
continue;
}
if sender_state.topology_boards.is_empty() {
let sender = self.sender_arc();
let fallback = TopologyBoardNode {
sender_id: announcer.clone(),
reachable_endpoints: sender_state.reachable.clone(),
reachable_timesync_sources: sender_state.reachable_timesync_sources.clone(),
connections: vec![sender.to_string()],
};
discovery::merge_topology_boards(&mut boards, core::slice::from_ref(&fallback));
} else {
discovery::merge_topology_boards(&mut boards, &sender_state.topology_boards);
if let Some(board) = boards
.iter_mut()
.find(|board| board.sender_id == *announcer)
{
let sender = self.sender_arc();
if !board.connections.iter().any(|peer| peer == sender.as_ref()) {
board.connections.push(sender.to_string());
}
}
}
}
}
if !link_local_enabled {
for board in &mut boards {
board
.reachable_endpoints
.retain(|ep| !ep.is_link_local_only());
}
}
discovery::normalize_topology_boards(&mut boards);
boards
}
#[cfg(feature = "discovery")]
fn note_discovery_topology_change_locked(st: &mut RelayInner, now_ms: u64) {
st.discovery_cadence.on_topology_change(now_ms);
for side_id in st
.sides
.iter()
.enumerate()
.filter_map(|(side_id, side)| side.as_ref().map(|_| side_id))
{
let throttle = st.discovery_side_throttle.entry(side_id).or_default();
throttle.pending_incremental = true;
throttle.incremental_retries_remaining = DISCOVERY_INCREMENTAL_RETRY_COUNT;
}
for transport in st.side_transport.values_mut() {
transport.refresh_transmit_header_templates();
}
}
#[cfg(feature = "discovery")]
fn prune_discovery_routes_locked(st: &mut RelayInner, now_ms: u64) -> bool {
let mut changed = false;
st.discovery_routes.retain(|_, route| {
let before_peers = route.announcers.len();
route.announcers.retain(|_, sender| {
now_ms.saturating_sub(sender.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
});
if before_peers != route.announcers.len() {
Self::recompute_discovery_side_state(route);
changed = true;
}
let keep = !route.announcers.is_empty();
changed |= !keep;
keep
});
changed
}
#[cfg(feature = "discovery")]
fn reconcile_end_to_end_acked_destinations_locked(&self, st: &mut RelayInner) {
if st.end_to_end_acked_destinations.is_empty() {
return;
}
let mut active_senders = BTreeSet::new();
for route in st.discovery_routes.values() {
for sender_state in route.announcers.values() {
for board in sender_state.topology_boards.iter() {
if self.is_end_to_end_destination_sender(&board.sender_id) {
active_senders.insert(Self::sender_hash(&board.sender_id));
}
}
}
}
st.end_to_end_acked_destinations.retain(|_, acked| {
acked.retain(|sender_hash| active_senders.contains(sender_hash));
!acked.is_empty()
});
}
#[cfg(feature = "discovery")]
fn advertised_summaries_from_topology_locked(
&self,
st: &RelayInner,
now_ms: u64,
link_local_enabled: bool,
exclude_side: Option<RelaySideId>,
topology: &[TopologyBoardNode],
) -> (Vec<crate::DataEndpoint>, Vec<String>) {
let (mut endpoints, mut sources) = discovery::summarize_topology_boards(topology);
for (&side, route) in &st.discovery_routes {
if exclude_side == Some(side)
|| now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS
{
continue;
}
for peer in route.announcers.values() {
if now_ms.saturating_sub(peer.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS {
continue;
}
for endpoint in &peer.advertised_reachable {
if !discovery::is_discovery_endpoint(*endpoint)
&& (link_local_enabled || !endpoint.is_link_local_only())
&& !endpoints.contains(endpoint)
{
endpoints.push(*endpoint);
}
}
for source in &peer.advertised_reachable_timesync_sources {
if !sources.contains(source) {
sources.push(source.clone());
}
}
}
}
endpoints.retain(|endpoint| {
!discovery::is_discovery_endpoint(*endpoint)
&& (link_local_enabled || !endpoint.is_link_local_only())
});
crate::small_sort::sort(&mut endpoints);
sources.sort_unstable();
(endpoints, sources)
}
#[cfg(feature = "discovery")]
fn advertised_discovery_endpoints_for_link_locked(
&self,
st: &RelayInner,
now_ms: u64,
link_local_enabled: bool,
exclude_side: Option<RelaySideId>,
) -> Vec<crate::DataEndpoint> {
let (mut reachable_endpoints, _) = discovery::summarize_topology_boards(
&self.advertised_discovery_topology_for_link_locked(
st,
now_ms,
link_local_enabled,
exclude_side,
),
);
for (&side, route) in &st.discovery_routes {
if exclude_side == Some(side)
|| now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS
{
continue;
}
for sender in route.announcers.values() {
if now_ms.saturating_sub(sender.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS {
reachable_endpoints.extend(sender.advertised_reachable.iter().copied());
}
}
}
crate::small_sort::sort(&mut reachable_endpoints);
reachable_endpoints.dedup();
reachable_endpoints
.into_iter()
.filter(|ep| {
!discovery::is_discovery_endpoint(*ep)
&& (link_local_enabled || !ep.is_link_local_only())
})
.collect()
}
#[cfg(feature = "discovery")]
fn advertised_discovery_timesync_sources_for_link_locked(
&self,
st: &RelayInner,
now_ms: u64,
exclude_side: Option<RelaySideId>,
) -> Vec<String> {
let (_, mut sources) = discovery::summarize_topology_boards(
&self.advertised_discovery_topology_for_link_locked(st, now_ms, true, exclude_side),
);
for (&side, route) in &st.discovery_routes {
if exclude_side == Some(side)
|| now_ms.saturating_sub(route.last_seen_ms) > DISCOVERY_ROUTE_TTL_MS
{
continue;
}
for sender in route.announcers.values() {
if now_ms.saturating_sub(sender.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS {
sources.extend(sender.advertised_reachable_timesync_sources.iter().cloned());
}
}
}
sources.sort_unstable();
sources.dedup();
sources
}
#[cfg(feature = "discovery")]
#[cfg(feature = "timesync")]
fn preferred_timesync_route_source(
&self,
data: &RelayItem,
ty: crate::DataType,
) -> TelemetryResult<Option<String>> {
if !matches!(
ty,
crate::DataType::TimeSyncAnnounce | crate::DataType::TimeSyncResponse
) {
return Ok(None);
}
let sender = match data {
RelayItem::Packet(pkt) => pkt.sender().to_owned(),
RelayItem::Packed(bytes) => {
if wire_format::peek_frame_info(bytes.as_ref())
.ok()
.is_some_and(|frame| frame.ack_only())
{
return Ok(None);
}
wire_format::unpack_packet(bytes.as_ref())?
.sender()
.to_owned()
}
};
Ok(Some(sender))
}
#[cfg(all(feature = "discovery", feature = "timesync"))]
#[inline]
fn side_is_slow_control_link_locked(
st: &RelayInner,
side_id: RelaySideId,
now_ms: u64,
) -> bool {
st.adaptive_route_stats.get(&side_id).is_some_and(|stats| {
let recent_slow = stats.last_slow_observed_ms > 0
&& now_ms.saturating_sub(stats.last_slow_observed_ms)
<= discovery::DISCOVERY_SLOW_LINK_FULL_INTERVAL_MS;
stats.sample_count > 0
&& ((stats.estimated_bandwidth_bps > 0
&& stats.estimated_bandwidth_bps <= CONTROL_SLOW_LINK_CAPACITY_BPS)
|| recent_slow)
})
}
#[cfg(feature = "discovery")]
fn discovery_level_for_side_locked(
st: &mut RelayInner,
side_id: RelaySideId,
now_ms: u64,
) -> Option<DiscoveryAdvertiseLevel> {
let throttle = st.discovery_side_throttle.entry(side_id).or_default();
if !throttle.has_sent_full {
throttle.has_sent_full = true;
throttle.pending_incremental = false;
throttle.incremental_retries_remaining = 0;
throttle.next_ping_ms = now_ms.saturating_add(DISCOVERY_SLOW_LINK_PING_INTERVAL_MS);
return Some(DiscoveryAdvertiseLevel::Full);
}
if throttle.pending_incremental {
throttle.incremental_retries_remaining =
throttle.incremental_retries_remaining.saturating_sub(1);
throttle.pending_incremental = throttle.incremental_retries_remaining > 0;
return Some(DiscoveryAdvertiseLevel::Incremental);
}
if throttle.liveness_due(now_ms) {
throttle.next_ping_ms = now_ms.saturating_add(DISCOVERY_SLOW_LINK_PING_INTERVAL_MS);
return Some(DiscoveryAdvertiseLevel::MinimalPing);
}
None
}
#[cfg(all(feature = "discovery", feature = "timesync"))]
#[inline]
fn is_timesync_type(ty: crate::DataType) -> bool {
matches!(
ty,
crate::DataType::TimeSyncAnnounce
| crate::DataType::TimeSyncRequest
| crate::DataType::TimeSyncResponse
)
}
#[cfg(all(feature = "discovery", feature = "timesync"))]
fn timesync_allowed_for_side_locked(
st: &mut RelayInner,
side_id: RelaySideId,
ty: crate::DataType,
now_ms: u64,
) -> bool {
if !Self::is_timesync_type(ty) {
return true;
}
if !Self::side_is_slow_control_link_locked(st, side_id, now_ms) {
st.timesync_side_throttle.remove(&side_id);
return true;
}
let throttle = st.timesync_side_throttle.entry(side_id).or_default();
if now_ms >= throttle.next_allowed_ms {
throttle.next_allowed_ms = now_ms.saturating_add(TIMESYNC_SLOW_LINK_MIN_INTERVAL_MS);
return true;
}
false
}
#[cfg(all(feature = "discovery", feature = "timesync"))]
fn filter_timesync_sides_locked(
st: &mut RelayInner,
ty: crate::DataType,
now_ms: u64,
sides: Vec<RelaySideId>,
) -> Vec<RelaySideId> {
sides
.into_iter()
.filter(|side| Self::timesync_allowed_for_side_locked(st, *side, ty, now_ms))
.collect()
}
#[cfg(feature = "discovery")]
fn queue_discovery_announce(
&self,
include_schema: bool,
skip_in_flight_sides: bool,
) -> TelemetryResult<()> {
self.queue_discovery_announce_on_side(include_schema, skip_in_flight_sides, None)
}
#[cfg(feature = "discovery")]
fn refresh_known_discovery_liveness(
&self,
pkt: &Packet,
side: RelaySideId,
) -> TelemetryResult<bool> {
let empty_ping =
pkt.data_type() == crate::DataType::DiscoveryAnnounce && pkt.payload().is_empty();
if !empty_ping && pkt.data_type() != crate::DataType::DiscoveryAddress {
return Ok(false);
}
crate::memory_admission::check(
if empty_ping {
512
} else {
1024 + pkt.payload().len().saturating_mul(4)
},
pkt.payload().len().saturating_mul(2).max(128),
)?;
let mut address = if empty_ping {
None
} else {
Some(discovery::decode_discovery_address(pkt)?)
};
let now = self.clock.now_ms();
let mut st = self.state.lock();
if let Some(ad) = &mut address {
let link_local = st
.sides
.get(side)
.and_then(|entry| entry.as_ref())
.is_some_and(|entry| entry.opts.link_local_enabled);
if !link_local {
ad.reachable_endpoints.retain(|ep| !ep.is_link_local_only());
}
}
let canonical;
let sender = if let Some(ad) = &address {
ad.hostname.as_str()
} else {
canonical = Self::canonical_sender_locked(&st, pkt.sender());
canonical.as_str()
};
let Some(route) = st.discovery_routes.get_mut(&side) else {
return Ok(false);
};
let Some(peer) = route.announcers.get_mut(sender) else {
return Ok(false);
};
if let Some(ad) = &address
&& (peer.advertised_address != Some(ad.address)
|| peer.advertised_reachable != ad.reachable_endpoints
|| peer.advertised_reachable_timesync_sources != ad.reachable_timesync_sources)
{
return Ok(false);
}
peer.last_seen_ms = now;
route.last_seen_ms = route.last_seen_ms.max(now);
Ok(true)
}
#[cfg(feature = "discovery")]
fn queue_pressure_keepalives(
&self,
requested_side: Option<RelaySideId>,
) -> TelemetryResult<()> {
let now = self.clock.now_ms();
let sender = self.sender_arc();
crate::memory_admission::check(
256 + sender.len().saturating_mul(4),
sender.len().max(256),
)?;
let side_count = self.state.lock().sides.len();
for side in 0..side_count {
let due = {
let st = self.state.lock();
requested_side.is_none_or(|wanted| wanted == side)
&& st.sides[side].is_some()
&& self.route_allowed_locked(
&st,
None,
Some(crate::DataType::DiscoveryAnnounce),
side,
)
&& st
.discovery_side_throttle
.get(&side)
.is_none_or(|throttle| now >= throttle.next_ping_ms)
};
if !due {
continue;
}
let pkt = discovery::build_discovery_announce(sender.as_ref(), now, &[])?;
self.state.lock().push_tx(RelayTxItem {
src: None,
dst: side,
data: RelayItem::Packet(Arc::new(pkt)),
priority: 255,
retry_after_ms: 0,
})?;
self.state
.lock()
.discovery_side_throttle
.entry(side)
.or_default()
.next_ping_ms = now.saturating_add(discovery::DISCOVERY_SLOW_INTERVAL_MS);
}
Ok(())
}
#[cfg(feature = "discovery")]
fn queue_discovery_announce_on_side(
&self,
include_schema: bool,
skip_in_flight_sides: bool,
requested_side: Option<RelaySideId>,
) -> TelemetryResult<()> {
#[cfg(not(feature = "std"))]
let _ = include_schema;
if let Some(side) = requested_side {
return self.queue_discovery_announce_one_side(
include_schema,
skip_in_flight_sides,
Some(side),
true,
);
}
let count = self.state.lock().sides.len();
let mut advertised = false;
for side in 0..count {
if self
.state
.lock()
.sides
.get(side)
.is_some_and(|side| side.is_some())
{
self.queue_discovery_announce_one_side(
include_schema,
skip_in_flight_sides,
Some(side),
false,
)?;
advertised = true;
}
}
if advertised {
self.state
.lock()
.discovery_cadence
.on_announce_sent(self.clock.now_ms());
}
Ok(())
}
#[cfg(feature = "discovery")]
fn queue_discovery_announce_one_side(
&self,
include_schema: bool,
skip_in_flight_sides: bool,
requested_side: Option<RelaySideId>,
advance_cadence: bool,
) -> TelemetryResult<()> {
let (scratch, largest) = if include_schema {
(8192, 2048)
} else {
let now = self.clock.now_ms();
let sender = self.sender_arc();
let st = self.state.lock();
discovery::topology_snapshot_scratch(
sender.as_ref(),
0,
0,
4,
true,
st.discovery_routes
.iter()
.filter(|(side, route)| {
Some(**side) != requested_side
&& now.saturating_sub(route.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
})
.flat_map(|(_, route)| route.announcers.iter())
.filter(|(_, peer)| {
now.saturating_sub(peer.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
})
.map(|(name, peer)| {
(
name.as_str(),
peer.reachable.as_slice(),
peer.reachable_timesync_sources.as_slice(),
peer.topology_boards.as_slice(),
)
}),
)
};
while let Err(err) = crate::memory_admission::check(scratch, largest) {
if matches!(err, TelemetryError::Io("memory pressure"))
&& crate::memory_admission::check(512, 256).is_ok()
{
let mut st = self.state.lock();
if st.pop_lowest_packet_queue_item(200) {
continue;
}
}
if !include_schema && matches!(err, TelemetryError::Io("memory pressure")) {
return self.queue_pressure_keepalives(requested_side);
}
return Err(err);
}
let now_ms = self.clock.now_ms();
let per_side = {
let mut st = self.state.lock();
if Self::prune_discovery_routes_locked(&mut st, now_ms) {
self.reconcile_end_to_end_acked_destinations_locked(&mut st);
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
st.fit_discovery_budget();
if !st.sides.iter().any(|side| side.is_some()) {
return Ok(());
}
if include_schema {
for (side, throttle) in st.discovery_side_throttle.iter_mut() {
if requested_side.is_none_or(|requested| requested == *side) {
throttle.has_sent_full = false;
}
}
}
if advance_cadence {
st.discovery_cadence.on_announce_sent(now_ms);
}
let mut per_side = Vec::new();
for side_id in 0..st.sides.len() {
let Some(side) = st.sides[side_id].as_ref() else {
continue;
};
let opts = side.opts.clone();
let link_local_enabled = opts.link_local_enabled;
if requested_side.is_some_and(|requested| requested != side_id) {
continue;
}
let discovery_in_flight = skip_in_flight_sides
&& st.reliable_tx.iter().any(|((pending_side, ty), tx_state)| {
*pending_side == side_id
&& !tx_state.sent.is_empty()
&& crate::DataType::try_from_u32(*ty)
.is_some_and(discovery::is_discovery_type)
});
if !self.route_allowed_locked(
&st,
None,
Some(crate::DataType::DiscoveryAnnounce),
side_id,
) {
continue;
}
let level = if discovery_in_flight {
let throttle = st.discovery_side_throttle.entry(side_id).or_default();
if !throttle.liveness_due(now_ms) {
None
} else {
throttle.next_ping_ms =
now_ms.saturating_add(DISCOVERY_SLOW_LINK_PING_INTERVAL_MS);
Some(DiscoveryAdvertiseLevel::MinimalPing)
}
} else {
Self::discovery_level_for_side_locked(&mut st, side_id, now_ms)
};
let Some(level) = level else {
continue;
};
let capabilities = opts.link_capabilities();
if level == DiscoveryAdvertiseLevel::MinimalPing {
per_side.push((
side_id,
level,
Vec::new(),
Vec::new(),
Vec::new(),
Vec::new(),
capabilities,
));
continue;
}
let current_topology = self.advertised_discovery_topology_for_link_locked(
&st,
now_ms,
link_local_enabled,
Some(side_id),
);
let (endpoints, timesync_sources) = self.advertised_summaries_from_topology_locked(
&st,
now_ms,
link_local_enabled,
Some(side_id),
¤t_topology,
);
let (topology, removed_topology) = match level {
DiscoveryAdvertiseLevel::Full => (current_topology.clone(), Vec::new()),
DiscoveryAdvertiseLevel::Incremental => {
let previous = &st
.discovery_side_throttle
.get(&side_id)
.expect("discovery throttle exists")
.last_topology;
discovery::topology_delta(previous, ¤t_topology)
}
DiscoveryAdvertiseLevel::MinimalPing => (Vec::new(), Vec::new()),
};
let throttle = st
.discovery_side_throttle
.get_mut(&side_id)
.expect("discovery throttle exists");
if level == DiscoveryAdvertiseLevel::Full
|| (level == DiscoveryAdvertiseLevel::Incremental
&& throttle.incremental_retries_remaining == 0)
{
throttle.last_topology = current_topology;
}
per_side.push((
side_id,
level,
endpoints,
timesync_sources,
topology,
removed_topology,
capabilities,
));
}
per_side
};
let mut st = self.state.lock();
for (dst, level, endpoints, timesync_sources, topology, removed_topology, capabilities) in
per_side
{
let sender = self.sender_arc();
if include_schema && level == DiscoveryAdvertiseLevel::Full {
let pkt = discovery::build_discovery_schema(sender.as_ref(), now_ms)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
if level == DiscoveryAdvertiseLevel::Full {
let pkt = discovery::build_discovery_link_capabilities(
sender.as_ref(),
now_ms,
capabilities,
)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
if level != DiscoveryAdvertiseLevel::Full || !endpoints.is_empty() {
let pkt = discovery::build_discovery_announce(
sender.as_ref(),
now_ms,
endpoints.as_slice(),
)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
if level != DiscoveryAdvertiseLevel::MinimalPing && !timesync_sources.is_empty() {
let pkt = discovery::build_discovery_timesync_sources(
sender.as_ref(),
now_ms,
timesync_sources.as_slice(),
)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
if level != DiscoveryAdvertiseLevel::MinimalPing
&& (!topology.is_empty() || !removed_topology.is_empty())
{
let pkt = if level == DiscoveryAdvertiseLevel::Full {
discovery::build_discovery_topology(sender.as_ref(), now_ms, &topology)?
} else {
discovery::build_discovery_topology_delta(
sender.as_ref(),
now_ms,
&topology,
&removed_topology,
)?
};
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
}
Ok(())
}
#[cfg(feature = "discovery")]
fn note_schema_received(&self, side: RelaySideId, sender: &str) {
let mut st = self.state.lock();
let address = sender
.strip_prefix("@addr:")
.and_then(|value| value.parse::<u32>().ok());
if let Some(route) = st.discovery_routes.get_mut(&side) {
for (name, peer) in &mut route.announcers {
if name == sender
|| address.is_some_and(|address| {
peer.advertised_address == Some(address)
|| sender_address_u32(name) == address
})
{
peer.has_schema = true;
break;
}
}
}
}
#[cfg(feature = "discovery")]
fn poll_discovery_announce(&self) -> TelemetryResult<bool> {
let now_ms = self.clock.now_ms();
let recovering = {
let mut st = self.state.lock();
let mut recovering = false;
for side in 0..st.sides.len() {
let missing = st.discovery_routes.get(&side).is_some_and(|route| {
route.announcers.iter().any(|(name, peer)| {
!peer.has_full_topology
&& now_ms.saturating_sub(peer.last_seen_ms) < DISCOVERY_ROUTE_TTL_MS
&& !route.announcers.values().any(|bridge| {
bridge.has_full_topology
&& now_ms.saturating_sub(bridge.last_seen_ms)
< DISCOVERY_ROUTE_TTL_MS
&& bridge
.topology_boards
.iter()
.any(|board| board.sender_id == *name)
})
})
});
if !missing {
continue;
}
let throttle = st.discovery_side_throttle.entry(side).or_default();
if throttle.next_topology_request_ms == 0 {
throttle.next_topology_request_ms =
now_ms.saturating_add(discovery::DISCOVERY_BASELINE_GRACE_MS);
continue;
}
if now_ms < throttle.next_topology_request_ms {
continue;
}
throttle.next_topology_request_ms =
now_ms.saturating_add(discovery::DISCOVERY_BASELINE_RETRY_MS);
let pkt = discovery::build_discovery_topology_request(
self.sender_arc().as_ref(),
now_ms,
)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst: side,
data,
priority,
retry_after_ms: 0,
})?;
recovering = true;
}
recovering
};
let schema_recovery_sides = {
let mut st = self.state.lock();
let mut due = Vec::new();
for side in 0..st.sides.len() {
let missing = st.discovery_routes.get(&side).is_some_and(|route| {
route.announcers.values().any(|peer| {
peer.has_full_topology
&& !peer.has_schema
&& now_ms.saturating_sub(peer.last_seen_ms) < DISCOVERY_ROUTE_TTL_MS
})
});
if !missing {
continue;
}
let throttle = st.discovery_side_throttle.entry(side).or_default();
if throttle.next_schema_request_ms == 0 {
throttle.next_schema_request_ms =
now_ms.saturating_add(discovery::DISCOVERY_BASELINE_GRACE_MS);
} else if now_ms >= throttle.next_schema_request_ms {
throttle.next_schema_request_ms =
now_ms.saturating_add(discovery::DISCOVERY_BASELINE_RETRY_MS);
due.push(side);
}
}
due
};
let recovering = recovering || !schema_recovery_sides.is_empty();
for side in schema_recovery_sides {
let pkt =
discovery::build_discovery_schema_request(self.sender_arc().as_ref(), now_ms)?;
let data = RelayItem::Packet(Arc::new(pkt));
let priority = Self::relay_item_priority(&data)?;
self.state.lock().push_tx(RelayTxItem {
src: None,
dst: side,
data,
priority,
retry_after_ms: 0,
})?;
}
let due = {
let mut st = self.state.lock();
let removed = Self::prune_discovery_routes_locked(&mut st, now_ms);
if removed {
self.reconcile_end_to_end_acked_destinations_locked(&mut st);
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
st.fit_discovery_budget();
let has_any = st.sides.iter().enumerate().any(|(side_id, side)| {
let Some(side) = side.as_ref() else {
return false;
};
if !self.route_allowed_locked(
&st,
None,
Some(crate::DataType::DiscoveryAnnounce),
side_id,
) {
return false;
}
let _ = side;
let in_flight = st.reliable_tx.iter().any(|((pending_side, ty), tx_state)| {
*pending_side == side_id
&& !tx_state.sent.is_empty()
&& crate::DataType::try_from_u32(*ty)
.is_some_and(discovery::is_discovery_type)
});
st.discovery_side_throttle
.get(&side_id)
.is_none_or(|throttle| {
(!in_flight && (!throttle.has_sent_full || throttle.pending_incremental))
|| throttle.liveness_due(now_ms)
})
});
if !st.sides.iter().any(|side| side.is_some()) || !has_any {
return Ok(recovering);
}
st.discovery_cadence.due(now_ms)
};
if !due {
return Ok(recovering);
}
self.queue_discovery_announce(false, true)?;
Ok(true)
}
#[cfg(feature = "discovery")]
fn learn_discovery_item(&self, src: RelaySideId, data: &RelayItem) -> TelemetryResult<()> {
let pkt = match data {
RelayItem::Packet(pkt) => {
if !discovery::is_discovery_type(pkt.data_type()) {
return Ok(());
}
pkt.as_ref().clone()
}
RelayItem::Packed(bytes) => {
let env = wire_format::peek_routing_envelope(bytes.as_ref())?;
if !discovery::is_discovery_type(env.ty) {
return Ok(());
}
if wire_format::peek_frame_info(bytes.as_ref())
.ok()
.is_some_and(|frame| frame.ack_only())
{
return Ok(());
}
wire_format::unpack_packet(bytes.as_ref())?
}
};
if self.refresh_known_discovery_liveness(&pkt, src)? {
return Ok(());
}
#[cfg(not(feature = "std"))]
if pkt.data_type() == crate::DataType::DiscoverySchema
&& discovery::discovery_schema_payload_is_fully_known(pkt.payload())?
{
self.note_schema_received(src, pkt.sender());
return Ok(());
}
if matches!(
pkt.data_type(),
crate::DataType::DiscoveryTopologyRequest | crate::DataType::DiscoverySchemaRequest
) {
crate::memory_admission::check(1024, 256)?;
if let Some(transport) = self.state.lock().side_transport.get_mut(&src) {
transport.refresh_transmit_header_templates();
}
self.state
.lock()
.discovery_side_throttle
.entry(src)
.or_default()
.has_sent_full = false;
return self.queue_discovery_announce_on_side(
pkt.data_type() == crate::DataType::DiscoverySchemaRequest,
false,
Some(src),
);
}
if pkt.data_type() == crate::DataType::ManagedVariableRequest {
discovery::decode_managed_variable_request(&pkt)?;
return Ok(());
}
let retained_summary = if matches!(
pkt.data_type(),
crate::DataType::DiscoveryAddress
| crate::DataType::DiscoveryTopology
| crate::DataType::DiscoveryAnnounce
| crate::DataType::DiscoveryTimeSyncSources
| crate::DataType::ManagedVariableRequest
) {
self.state
.lock()
.discovery_routes
.get(&src)
.map_or(0, RelayInner::discovery_route_byte_cost)
} else {
0
};
let scratch = if pkt.data_type() == crate::DataType::DiscoveryAddress {
4096usize.saturating_add(retained_summary)
} else {
8192usize.saturating_add(retained_summary)
};
let decode_bytes = pkt
.payload()
.len()
.saturating_mul(4)
.saturating_add(scratch);
let largest_block = pkt.payload().len().saturating_mul(2).max(2048);
while let Err(err) = crate::memory_admission::check(decode_bytes, largest_block) {
if !matches!(err, TelemetryError::Io("memory pressure")) {
return Err(err);
}
let mut st = self.state.lock();
if !st.pop_lowest_packet_queue_item(256) {
return Err(err);
}
}
let now_ms = self.clock.now_ms();
if pkt.data_type() == crate::DataType::DiscoverySchema {
let snapshot = discovery::decode_discovery_schema(&pkt)?;
let mut st = self.state.lock();
let budget = st.schema_growth_budget();
let report = crate::config::merge_owned_schema_snapshot_with_budget(snapshot, budget)?;
if report.changed() {
st.fit_discovery_budget();
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
drop(st);
self.note_schema_received(src, pkt.sender());
return Ok(());
}
if pkt.data_type() == crate::DataType::DiscoveryLinkCapabilities {
let _ = discovery::decode_discovery_link_capabilities(&pkt)?;
return Ok(());
}
let mut st = self.state.lock();
if pkt.data_type() == crate::DataType::DiscoveryLeave {
let leaving = pkt.sender();
let before = st.discovery_routes.clone();
for route in st.discovery_routes.values_mut() {
route.announcers.remove(leaving);
for sender_state in route.announcers.values_mut() {
sender_state
.topology_boards
.retain(|board| board.sender_id != leaving);
for board in sender_state.topology_boards.iter_mut() {
board.connections.retain(|peer| peer != leaving);
}
Self::refresh_sender_topology_state(sender_state);
}
Self::recompute_discovery_side_state(route);
}
st.discovery_routes
.retain(|_, route| !route.announcers.is_empty());
if st.discovery_routes != before {
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
let _ = Self::prune_discovery_routes_locked(&mut st, now_ms);
self.reconcile_end_to_end_acked_destinations_locked(&mut st);
return Ok(());
}
let address_ad = if pkt.data_type() == crate::DataType::DiscoveryAddress {
Some(discovery::decode_discovery_address(&pkt)?)
} else {
None
};
let mut topology_ad = if pkt.data_type() == crate::DataType::DiscoveryTopology {
Some(discovery::decode_discovery_topology_update(&pkt)?)
} else {
None
};
let announcer_id = if let Some(ad) = address_ad.as_ref() {
ad.hostname.clone()
} else {
let canonical = Self::canonical_sender_locked(&st, pkt.sender());
if canonical != pkt.sender() {
canonical
} else if let Some(address) = pkt
.sender()
.strip_prefix("@addr:")
.and_then(|value| value.parse::<u32>().ok())
{
topology_ad
.as_ref()
.and_then(|update| {
update
.boards
.iter()
.find(|board| sender_address_u32(&board.sender_id) == address)
})
.map(|board| board.sender_id.clone())
.unwrap_or(canonical)
} else {
canonical
}
};
let announced = if pkt.data_type() == crate::DataType::DiscoveryAnnounce {
Some(discovery::decode_discovery_announce(&pkt)?)
} else {
None
};
let timesync_sources = if pkt.data_type() == crate::DataType::DiscoveryTimeSyncSources {
Some(discovery::decode_discovery_timesync_sources(&pkt)?)
} else {
None
};
if let Some(update) = topology_ad.as_mut() {
for board in &mut update.boards {
board.sender_id = Self::canonical_sender_locked(&st, &board.sender_id);
for peer in &mut board.connections {
*peer = Self::canonical_sender_locked(&st, peer);
}
}
}
let mut route = st.discovery_routes.get(&src).cloned().unwrap_or_default();
if pkt.sender() != announcer_id {
route.announcers.remove(pkt.sender());
}
let side_link_local_enabled = st
.sides
.get(src)
.and_then(|entry| entry.as_ref())
.map(|side_ref| side_ref.opts.link_local_enabled)
.unwrap_or(false);
let mut sender_state =
core::mem::take(route.announcers.entry(announcer_id.clone()).or_default());
let changed = match pkt.data_type() {
crate::DataType::DiscoveryAddress => {
let ad = address_ad.expect("decoded above");
let advertised_address = ad.address;
let mut reachable = ad.reachable_endpoints;
if !side_link_local_enabled {
reachable.retain(|ep| !ep.is_link_local_only());
}
let changed = sender_state.advertised_address != Some(advertised_address)
|| sender_state.advertised_reachable != reachable
|| sender_state.advertised_reachable_timesync_sources
!= ad.reachable_timesync_sources;
sender_state.advertised_address = Some(advertised_address);
sender_state.advertised_reachable = reachable;
sender_state.advertised_reachable_timesync_sources = ad.reachable_timesync_sources;
Self::refresh_sender_topology_state(&mut sender_state);
changed
}
crate::DataType::DiscoveryAnnounce => {
let mut reachable = announced.expect("parsed before moving route state");
if reachable.is_empty() {
false
} else {
if !side_link_local_enabled {
reachable.retain(|ep| !ep.is_link_local_only());
}
let board = Self::sender_topology_board_mut(&mut sender_state, &announcer_id);
let changed = board.reachable_endpoints != reachable;
board.reachable_endpoints = reachable;
Self::refresh_sender_topology_state(&mut sender_state);
changed
}
}
crate::DataType::DiscoveryTimeSyncSources => {
let sources = timesync_sources.expect("parsed before moving route state");
let board = Self::sender_topology_board_mut(&mut sender_state, &announcer_id);
let changed = board.reachable_timesync_sources != sources;
board.reachable_timesync_sources = sources;
Self::refresh_sender_topology_state(&mut sender_state);
changed
}
crate::DataType::DiscoveryTopology => {
let mut update = topology_ad
.take()
.expect("topology packet was decoded before route selection");
if !side_link_local_enabled {
for board in update.boards.iter_mut() {
board
.reachable_endpoints
.retain(|ep| !ep.is_link_local_only());
}
}
if !update.incremental {
sender_state.has_full_topology = true;
}
let changed =
discovery::apply_topology_update(&mut sender_state.topology_boards, update);
Self::refresh_sender_topology_state(&mut sender_state);
changed
}
crate::DataType::DiscoverySchema => false,
_ => false,
};
sender_state.last_seen_ms = now_ms;
route.announcers.insert(announcer_id, sender_state);
Self::recompute_discovery_side_state(&mut route);
st.commit_discovery_route(src, route)?;
if changed {
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
let _ = Self::prune_discovery_routes_locked(&mut st, now_ms);
self.reconcile_end_to_end_acked_destinations_locked(&mut st);
Ok(())
}
#[cfg(not(feature = "discovery"))]
fn learn_discovery_item(&self, _src: RelaySideId, _data: &RelayItem) -> TelemetryResult<()> {
Ok(())
}
#[cfg(not(feature = "discovery"))]
fn queue_discovery_announce(&self) -> TelemetryResult<()> {
Ok(())
}
#[cfg(not(feature = "discovery"))]
fn poll_discovery_announce(&self) -> TelemetryResult<bool> {
Ok(false)
}
fn process_reliable_timeouts(&self) -> TelemetryResult<()> {
let now = self.clock.now_ms();
let mut requeue: Vec<(RelaySideId, crate::DataType, u32)> = Vec::new();
{
let mut st = self.state.lock();
if st.reliable_tx.is_empty() {
return Ok(());
}
for ((side, ty_u32), tx_state) in st.reliable_tx.iter_mut() {
let Some(ty) = crate::DataType::try_from_u32(*ty_u32) else {
continue;
};
let sent_order: Vec<u32> = tx_state.sent_order.iter().copied().collect();
for seq in sent_order {
let Some(sent) = tx_state.sent.get_mut(&seq) else {
continue;
};
if sent.queued
|| now.wrapping_sub(sent.last_send_ms) < runtime_reliable_retransmit_ms()
{
continue;
}
if sent.partial_acked {
sent.partial_acked = false;
sent.last_send_ms = now;
continue;
}
if sent.retries >= runtime_reliable_max_retries() {
tx_state.sent.remove(&seq);
tx_state.sent_order.retain(|existing| *existing != seq);
continue;
}
sent.retries += 1;
requeue.push((*side, ty, seq));
}
}
}
for (side, ty, seq) in requeue {
self.queue_reliable_retransmit(side, ty, seq)?;
}
Ok(())
}
fn get_hash(item: &RelayRxItem) -> u64 {
match &item.data {
RelayItem::Packet(pkt) => pkt.packet_id(),
RelayItem::Packed(bytes) => {
let reliable_seq = wire_format::peek_frame_info(bytes.as_ref())
.ok()
.and_then(|frame| frame.reliable)
.and_then(|hdr| {
if (hdr.flags & wire_format::RELIABLE_FLAG_ACK_ONLY) != 0 {
None
} else {
Some(hdr.seq)
}
});
match wire_format::packet_id_from_wire(bytes.as_ref()) {
Ok(id) => {
if let Some(seq) = reliable_seq {
hash_bytes_u64(id, &seq.to_le_bytes())
} else {
id
}
}
Err(_e) => {
let h: u64 = 0x9E37_79B9_7F4A_7C15;
hash_bytes_u64(h, bytes.as_ref())
}
}
}
}
}
fn is_duplicate_pkt(&self, item: &RelayRxItem) -> TelemetryResult<bool> {
let id = Self::get_hash(item);
let mut st = self.state.lock();
if st.recent_rx.contains(&id) {
Ok(true)
} else {
st.push_recent_rx(id)?;
Ok(false)
}
}
fn should_forward_duplicate_reliable_item(&self, item: &RelayRxItem) -> TelemetryResult<bool> {
let (_, ty) = self.item_route_info(&item.data)?;
if !is_reliable_type(ty)
|| matches!(
ty,
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
)
{
return Ok(false);
}
if !self.item_target_senders(&item.data)?.is_empty() {
let id = match &item.data {
RelayItem::Packet(pkt) => pkt.packet_id(),
RelayItem::Packed(bytes) => wire_format::packet_id_from_wire(bytes)?,
};
let mut st = self.state.lock();
return Ok(st.reliable_return_routes.get_mut(&id).is_some_and(|route| {
let now = self.clock.now_ms();
if now.saturating_sub(route.last_replay_ms) < runtime_reliable_retransmit_ms() {
return false;
}
route.last_replay_ms = now;
true
}));
}
let RemoteSidePlan::Target(sides) = self.remote_side_plan(&item.data, item.src)?;
let st = self.state.lock();
let now_ms = self.clock.now_ms();
Ok(sides
.into_iter()
.any(|side| self.side_has_multiple_announcers_locked(&st, side, now_ms)))
}
pub fn add_side_packed<N, F>(&self, name: N, tx: F) -> RelaySideId
where
N: AsRef<str>,
F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
{
self.add_side_packed_with_options(name, tx, RelaySideOptions::default())
}
pub fn add_side_packed_small_packets<N, F>(
&self,
name: N,
tx: F,
max_frame_bytes: usize,
) -> RelaySideId
where
N: AsRef<str>,
F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
{
self.add_side_packed_with_options(
name,
tx,
RelaySideOptions::default().with_small_packet_transport(max_frame_bytes),
)
}
pub fn add_side_packed_with_options<N, F>(
&self,
name: N,
tx: F,
opts: RelaySideOptions,
) -> RelaySideId
where
N: AsRef<str>,
F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
{
let mut st = self.state.lock();
let side = Some(RelaySide {
name: Arc::from(name.as_ref()),
tx_handler: RelayTxHandlerFn::Packed(Arc::new(tx)),
opts,
});
let id = if let Some(id) = st.sides.iter().position(Option::is_none) {
st.sides[id] = side;
id
} else {
let id = st.sides.len();
st.sides.push(side);
id
};
st.side_runtime_stats
.insert(id, SideRuntimeStatsInner::default());
st.side_transport.insert(id, SideTransportState::default());
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, self.clock.now_ms());
id
}
pub fn add_side_packet<N, F>(&self, name: N, tx: F) -> RelaySideId
where
N: AsRef<str>,
F: Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static,
{
self.add_side_packet_with_options(name, tx, RelaySideOptions::default())
}
pub fn add_side_packet_with_options<N, F>(
&self,
name: N,
tx: F,
opts: RelaySideOptions,
) -> RelaySideId
where
N: AsRef<str>,
F: Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static,
{
let mut st = self.state.lock();
let side = Some(RelaySide {
name: Arc::from(name.as_ref()),
tx_handler: RelayTxHandlerFn::Packet(Arc::new(tx)),
opts,
});
let id = if let Some(id) = st.sides.iter().position(Option::is_none) {
st.sides[id] = side;
id
} else {
let id = st.sides.len();
st.sides.push(side);
id
};
st.side_runtime_stats
.insert(id, SideRuntimeStatsInner::default());
st.side_transport.insert(id, SideTransportState::default());
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, self.clock.now_ms());
id
}
pub fn remove_side(&self, side: RelaySideId) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let slot = st.sides.get_mut(side).ok_or(TelemetryError::BadArg)?;
if slot.is_none() {
return Err(TelemetryError::BadArg);
}
*slot = None;
while st.sides.last().is_some_and(Option::is_none) {
st.sides.pop();
}
st.route_overrides
.retain(|(src_side, dst_side), _| *src_side != Some(side) && *dst_side != side);
st.typed_route_overrides
.retain(|(src_side, _, dst_side), _| *src_side != Some(side) && *dst_side != side);
st.route_weights
.retain(|(src_side, dst_side), _| *src_side != Some(side) && *dst_side != side);
st.route_priorities
.retain(|(src_side, dst_side), _| *src_side != Some(side) && *dst_side != side);
st.source_route_modes.remove(&Some(side));
st.route_selection_cursors.remove(&Some(side));
st.adaptive_route_stats.remove(&side);
#[cfg(feature = "discovery")]
st.discovery_side_throttle.remove(&side);
#[cfg(all(feature = "discovery", feature = "timesync"))]
st.timesync_side_throttle.remove(&side);
st.side_runtime_stats.remove(&side);
st.reliable_return_routes
.retain(|_, route| route.side != side);
st.rx_queue.retain(|queued| queued.src != side);
st.tx_queue
.retain(|queued| queued.dst != side && queued.src != Some(side));
st.replay_queue.retain(|queued| queued.dst != side);
st.reliable_tx.retain(|(side_id, _), _| *side_id != side);
st.reliable_rx.retain(|(side_id, _), _| *side_id != side);
#[cfg(feature = "discovery")]
{
st.discovery_routes.remove(&side);
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
Ok(())
}
pub fn set_side_ingress_enabled(
&self,
side: RelaySideId,
enabled: bool,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let side_ref = st
.sides
.get_mut(side)
.and_then(|side| side.as_mut())
.ok_or(TelemetryError::BadArg)?;
side_ref.opts.ingress_enabled = enabled;
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_side_egress_enabled(&self, side: RelaySideId, enabled: bool) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let side_ref = st
.sides
.get_mut(side)
.and_then(|side| side.as_mut())
.ok_or(TelemetryError::BadArg)?;
side_ref.opts.egress_enabled = enabled;
if !enabled {
st.tx_queue.retain(|queued| queued.dst != side);
st.replay_queue.retain(|queued| queued.dst != side);
}
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_source_route_mode(
&self,
src: Option<RelaySideId>,
mode: RouteSelectionMode,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.source_route_modes.insert(src, mode);
st.route_selection_cursors.remove(&src);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn clear_source_route_mode(&self, src: Option<RelaySideId>) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.source_route_modes.remove(&src);
st.route_selection_cursors.remove(&src);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_route_weight(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
weight: u32,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_weights.insert((src, dst), weight);
st.route_selection_cursors.remove(&src);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn clear_route_weight(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_weights.remove(&(src, dst));
st.route_selection_cursors.remove(&src);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_route_priority(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
priority: u32,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_priorities.insert((src, dst), priority);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn clear_route_priority(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_priorities.remove(&(src, dst));
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_route(
&self,
src: Option<RelaySideId>,
dst: RelaySideId,
enabled: bool,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_overrides.insert((src, dst), enabled);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn set_typed_route(
&self,
src: Option<RelaySideId>,
ty: crate::DataType,
dst: RelaySideId,
enabled: bool,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.typed_route_overrides
.insert((src, ty.as_u32(), dst), enabled);
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn clear_typed_route(
&self,
src: Option<RelaySideId>,
ty: crate::DataType,
dst: RelaySideId,
) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.typed_route_overrides.remove(&(src, ty.as_u32(), dst));
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
pub fn clear_route(&self, src: Option<RelaySideId>, dst: RelaySideId) -> TelemetryResult<()> {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
let _ = Self::side_ref(&st, dst).map_err(|_| TelemetryError::BadArg)?;
if let Some(src) = src {
let _ = Self::side_ref(&st, src).map_err(|_| TelemetryError::BadArg)?;
}
st.route_overrides.remove(&(src, dst));
#[cfg(feature = "discovery")]
Self::note_discovery_topology_change_locked(&mut st, now_ms);
Ok(())
}
#[cfg(feature = "discovery")]
pub fn announce_discovery(&self) -> TelemetryResult<()> {
self.queue_discovery_announce(true, false)
}
#[cfg(feature = "discovery")]
pub fn announce_leave(&self) -> TelemetryResult<()> {
let pkt = discovery::build_discovery_leave("relay", self.clock.now_ms())?;
let mut st = self.state.lock();
let dsts: Vec<usize> = st
.sides
.iter()
.enumerate()
.filter_map(|(idx, side)| side.as_ref().map(|_| idx))
.collect();
for dst in dsts {
let data = RelayItem::Packet(Arc::new(pkt.clone()));
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: None,
dst,
data,
priority,
retry_after_ms: 0,
})?;
}
Ok(())
}
#[cfg(feature = "discovery")]
pub fn poll_discovery(&self) -> TelemetryResult<bool> {
self.poll_discovery_announce()
}
#[cfg(feature = "discovery")]
pub fn export_topology(&self) -> TopologySnapshot {
let now_ms = self.clock.now_ms();
let mut st = self.state.lock();
if Self::prune_discovery_routes_locked(&mut st, now_ms) {
self.reconcile_end_to_end_acked_destinations_locked(&mut st);
Self::note_discovery_topology_change_locked(&mut st, now_ms);
}
let routes = st
.discovery_routes
.iter()
.filter_map(|(&side_id, route)| {
let side = st.sides.get(side_id).and_then(|side| side.as_ref())?;
let announcers = route
.announcers
.iter()
.map(|(sender_id, sender_state)| TopologyAnnouncerRoute {
sender_id: sender_id.clone(),
reachable_endpoints: sender_state
.reachable
.iter()
.copied()
.filter(|ep| !discovery::is_router_control_endpoint(*ep))
.collect(),
reachable_network_variables: Vec::new(),
reachable_timesync_sources: sender_state.reachable_timesync_sources.clone(),
routers: sender_state.topology_boards.clone(),
last_seen_ms: sender_state.last_seen_ms,
age_ms: now_ms.saturating_sub(sender_state.last_seen_ms),
})
.collect();
Some(TopologySideRoute {
side_id,
side_name: side.name.to_string(),
reachable_endpoints: route
.reachable
.iter()
.copied()
.filter(|ep| !discovery::is_router_control_endpoint(*ep))
.collect(),
reachable_network_variables: Vec::new(),
reachable_timesync_sources: route.reachable_timesync_sources.clone(),
announcers,
last_seen_ms: route.last_seen_ms,
age_ms: now_ms.saturating_sub(route.last_seen_ms),
})
})
.collect();
let routers = self.advertised_discovery_topology_for_link_locked(&st, now_ms, true, None);
let advertised_endpoints =
self.advertised_discovery_endpoints_for_link_locked(&st, now_ms, true, None);
let advertised_timesync_sources =
self.advertised_discovery_timesync_sources_for_link_locked(&st, now_ms, None);
let links = discovery::topology_links_from_boards(&routers);
TopologySnapshot {
advertised_endpoints,
advertised_timesync_sources,
routers,
links,
routes,
current_announce_interval_ms: st.discovery_cadence.current_interval_ms,
next_announce_ms: st.discovery_cadence.next_announce_ms,
}
}
#[cfg(feature = "discovery")]
pub fn client_stats(&self, sender_id: &str) -> Option<ClientStatsSnapshot> {
let now_ms = self.clock.now_ms();
let st = self.state.lock();
let mut side_ids = Vec::new();
let mut side_names = Vec::new();
let mut last_seen_ms = None::<u64>;
let mut reachable_endpoints = Vec::new();
let mut reachable_timesync_sources = Vec::new();
let mut packets_sent = 0u64;
let mut packets_received = 0u64;
let mut bytes_sent = 0u64;
let mut bytes_received = 0u64;
for (side_id, route) in &st.discovery_routes {
let Some(sender_state) = route.announcers.get(sender_id) else {
continue;
};
side_ids.push(*side_id);
if let Some(side_name) = st
.sides
.get(*side_id)
.and_then(|side| side.as_ref())
.map(|side| side.name.clone())
{
side_names.push(side_name.to_string());
}
last_seen_ms = Some(last_seen_ms.unwrap_or(0).max(sender_state.last_seen_ms));
reachable_endpoints.extend(sender_state.reachable.iter().copied());
reachable_timesync_sources
.extend(sender_state.reachable_timesync_sources.iter().cloned());
if let Some(stats) = st.side_runtime_stats.get(side_id) {
packets_sent = packets_sent.saturating_add(stats.tx_packets);
packets_received = packets_received.saturating_add(stats.rx_packets);
bytes_sent = bytes_sent.saturating_add(stats.tx_bytes);
bytes_received = bytes_received.saturating_add(stats.rx_bytes);
}
}
if side_ids.is_empty() {
return None;
}
reachable_endpoints.retain(|ep| !discovery::is_router_control_endpoint(*ep));
crate::small_sort::sort(&mut reachable_endpoints);
reachable_endpoints.dedup();
reachable_timesync_sources.sort_unstable();
reachable_timesync_sources.dedup();
crate::small_sort::sort(&mut side_ids);
side_ids.dedup();
side_names.sort_unstable();
side_names.dedup();
let age_ms = last_seen_ms.map(|seen| now_ms.saturating_sub(seen));
Some(ClientStatsSnapshot {
sender_id: sender_id.to_string(),
connected: age_ms.is_some_and(|age| age <= DISCOVERY_ROUTE_TTL_MS),
side_ids,
side_names,
last_seen_ms,
age_ms,
reachable_endpoints,
reachable_timesync_sources,
packets_sent,
packets_received,
bytes_sent,
bytes_received,
})
}
pub fn export_runtime_stats(&self) -> RuntimeStatsSnapshot {
let now_ms = self.clock.now_ms();
let st = self.state.lock();
let mut sides = Vec::new();
for (side_id, side) in st.sides.iter().enumerate() {
let Some(side) = side.as_ref() else { continue };
let stats = st
.side_runtime_stats
.get(&side_id)
.cloned()
.unwrap_or_default();
let adaptive = st
.adaptive_route_stats
.get(&side_id)
.cloned()
.unwrap_or_default()
.snapshot(now_ms, true);
let (tx_template_count, rx_template_count) = st
.side_transport
.get(&side_id)
.map(|state| (state.tx_template_count(), state.rx_template_count()))
.unwrap_or((0, 0));
let mut data_types: Vec<RuntimeTypeStats> = stats
.data_types
.into_iter()
.map(|(ty, item)| RuntimeTypeStats {
data_type: crate::DataType(ty),
tx_packets: item.tx_packets,
tx_bytes: item.tx_bytes,
rx_packets: item.rx_packets,
rx_bytes: item.rx_bytes,
relayed_tx_packets: item.relayed_tx_packets,
relayed_tx_bytes: item.relayed_tx_bytes,
relayed_rx_packets: item.relayed_rx_packets,
relayed_rx_bytes: item.relayed_rx_bytes,
tx_retries: item.tx_retries,
handler_failures: item.handler_failures,
})
.collect();
data_types.sort_unstable_by_key(|item| item.data_type.as_u32());
sides.push(RuntimeSideStats {
side_id,
side_name: side.name.to_string(),
reliable_enabled: side.opts.reliable_enabled,
link_local_enabled: side.opts.link_local_enabled,
header_template_enabled: side.opts.header_template_enabled,
max_frame_bytes: side.opts.max_frame_bytes,
compact_header_target_bytes: side.opts.compact_header_target_bytes,
side_transport_profile: side.opts.effective_transport_profile().as_str(),
ingress_enabled: side.opts.ingress_enabled,
egress_enabled: side.opts.egress_enabled,
tx_packets: stats.tx_packets,
tx_bytes: stats.tx_bytes,
rx_packets: stats.rx_packets,
rx_bytes: stats.rx_bytes,
relayed_tx_packets: stats.relayed_tx_packets,
relayed_tx_bytes: stats.relayed_tx_bytes,
relayed_rx_packets: stats.relayed_rx_packets,
relayed_rx_bytes: stats.relayed_rx_bytes,
local_delivery_packets: 0,
tx_retries: stats.tx_retries,
tx_handler_failures: stats.tx_handler_failures,
local_handler_failures: 0,
total_handler_retries: stats.total_handler_retries,
side_transport_full_frames: stats.side_transport_full_frames,
side_transport_compact_frames: stats.side_transport_compact_frames,
side_transport_compact_delta_frames: stats.side_transport_compact_delta_frames,
side_transport_compact_omitted_timestamp_frames: stats
.side_transport_compact_omitted_timestamp_frames,
side_transport_chunk_frames: stats.side_transport_chunk_frames,
side_transport_raw_bytes: stats.side_transport_raw_bytes,
side_transport_wire_bytes: stats.side_transport_wire_bytes,
side_transport_bytes_saved: stats.side_transport_bytes_saved,
side_transport_min_compact_overhead_bytes: stats
.side_transport_min_compact_overhead_bytes,
side_transport_max_compact_overhead_bytes: stats
.side_transport_max_compact_overhead_bytes,
side_transport_compact_target_misses: stats.side_transport_compact_target_misses,
side_transport_template_evictions: stats.side_transport_template_evictions,
side_transport_tx_template_count: tx_template_count,
side_transport_rx_template_count: rx_template_count,
max_side_transport_templates: side.opts.max_side_transport_templates,
adaptive,
data_types,
});
}
let mut route_modes: Vec<RouteModeStats> = st
.route_selection_cursors
.iter()
.map(|(src, cursor)| RouteModeStats {
src_side_id: *src,
selection_mode: st.source_route_modes.get(src).copied(),
cursor: *cursor,
})
.collect();
for src in st.source_route_modes.keys() {
if !route_modes.iter().any(|mode| mode.src_side_id == *src) {
route_modes.push(RouteModeStats {
src_side_id: *src,
selection_mode: st.source_route_modes.get(src).copied(),
cursor: 0,
});
}
}
route_modes.sort_unstable_by_key(|mode| mode.src_side_id.unwrap_or(usize::MAX));
let mut route_overrides: Vec<RouteOverrideStats> = st
.route_overrides
.iter()
.map(|((src, dst), enabled)| RouteOverrideStats {
src_side_id: *src,
dst_side_id: *dst,
enabled: *enabled,
})
.collect();
route_overrides.sort_unstable_by_key(|item| {
(item.src_side_id.unwrap_or(usize::MAX), item.dst_side_id)
});
let mut typed_route_overrides: Vec<TypedRouteOverrideStats> = st
.typed_route_overrides
.iter()
.map(|((src, ty, dst), enabled)| TypedRouteOverrideStats {
src_side_id: *src,
data_type: crate::DataType(*ty),
dst_side_id: *dst,
enabled: *enabled,
})
.collect();
typed_route_overrides.sort_unstable_by_key(|item| {
(
item.src_side_id.unwrap_or(usize::MAX),
item.data_type.as_u32(),
item.dst_side_id,
)
});
let mut route_weights: Vec<RouteWeightStats> = st
.route_weights
.iter()
.map(|((src, dst), weight)| RouteWeightStats {
src_side_id: *src,
dst_side_id: *dst,
weight: *weight,
})
.collect();
route_weights.sort_unstable_by_key(|item| {
(item.src_side_id.unwrap_or(usize::MAX), item.dst_side_id)
});
let mut route_priorities: Vec<RoutePriorityStats> = st
.route_priorities
.iter()
.map(|((src, dst), priority)| RoutePriorityStats {
src_side_id: *src,
dst_side_id: *dst,
priority: *priority,
})
.collect();
route_priorities.sort_unstable_by_key(|item| {
(item.src_side_id.unwrap_or(usize::MAX), item.dst_side_id)
});
#[cfg(feature = "discovery")]
let discovery = DiscoveryRuntimeStats {
route_count: st.discovery_routes.len(),
announcer_count: st
.discovery_routes
.values()
.map(|route| route.announcers.len())
.sum(),
current_announce_interval_ms: Some(st.discovery_cadence.current_interval_ms),
next_announce_ms: Some(st.discovery_cadence.next_announce_ms),
};
#[cfg(not(feature = "discovery"))]
let discovery = DiscoveryRuntimeStats {
route_count: 0,
announcer_count: 0,
current_announce_interval_ms: None,
next_announce_ms: None,
};
RuntimeStatsSnapshot {
sides,
route_modes,
route_overrides,
typed_route_overrides,
route_weights,
route_priorities,
queues: QueueRuntimeStats {
rx_len: st.rx_queue.len(),
rx_bytes: st.rx_queue.bytes_used(),
tx_len: st.tx_queue.len(),
tx_bytes: st.tx_queue.bytes_used(),
replay_len: st.replay_queue.len(),
replay_bytes: st.replay_queue.bytes_used(),
recent_rx_len: st.recent_rx.len(),
recent_rx_bytes: st.recent_rx.bytes_used(),
reliable_rx_buffered_len: st.reliable_rx_buffer_len(),
reliable_rx_buffered_bytes: st.reliable_rx_buffered_bytes(),
shared_queue_bytes_used: st.shared_queue_bytes_used(),
},
reliable: ReliableRuntimeStats {
reliable_return_route_count: st.reliable_return_routes.len(),
end_to_end_pending_count: 0,
end_to_end_pending_destination_count: 0,
end_to_end_acked_cache_count: st.end_to_end_acked_destinations.len(),
},
discovery,
total_handler_failures: st.total_handler_failures,
total_handler_retries: st.total_handler_retries,
}
}
pub fn export_memory_layout_json(&self) -> String {
let st = self.state.lock();
#[cfg(feature = "discovery")]
let discovery_bytes = st.discovery_bytes_used();
#[cfg(not(feature = "discovery"))]
let discovery_bytes = 0usize;
let schema_bytes = crate::config::schema_bytes_used();
let memory = st.memory;
let mut out = String::new();
let _ = core::fmt::Write::write_fmt(
&mut out,
format_args!(
"{{\"kind\":\"relay\",\
\"shared_queue_bytes_used\":{},\"shared_queue_bytes_allocated\":{},\
\"rx_queue_bytes_used\":{},\"rx_queue_bytes_allocated\":{},\"rx_queue_len\":{},\
\"tx_queue_bytes_used\":{},\"tx_queue_bytes_allocated\":{},\"tx_queue_len\":{},\
\"replay_queue_bytes_used\":{},\"replay_queue_bytes_allocated\":{},\"replay_queue_len\":{},\
\"recent_rx_bytes_used\":{},\"recent_rx_bytes_allocated\":{},\"recent_rx_len\":{},\
\"reliable_rx_buffer_bytes_used\":{},\"reliable_rx_buffer_bytes_allocated\":{},\"reliable_rx_buffer_len\":{},\
\"discovery_bytes_used\":{},\"discovery_bytes_allocated\":{},\
\"schema_bytes_used\":{},\"schema_bytes_allocated\":{}}}",
st.shared_queue_bytes_used(),
memory.max_queue_budget,
st.rx_queue.bytes_used(),
st.rx_queue.max_bytes(),
st.rx_queue.len(),
st.tx_queue.bytes_used(),
st.tx_queue.max_bytes(),
st.tx_queue.len(),
st.replay_queue.bytes_used(),
st.replay_queue.max_bytes(),
st.replay_queue.len(),
st.recent_rx.bytes_used(),
st.recent_rx.max_bytes(),
st.recent_rx.len(),
st.reliable_rx_buffered_bytes(),
memory.max_queue_budget,
st.reliable_rx_buffer_len(),
discovery_bytes,
memory.max_queue_budget,
schema_bytes,
memory.max_queue_budget,
),
);
out
}
#[cfg(test)]
pub(crate) fn debug_end_to_end_acked_destination_count(&self, packet_id: u64) -> Option<usize> {
let st = self.state.lock();
st.end_to_end_acked_destinations
.get(&packet_id)
.map(BTreeSet::len)
}
#[cfg(test)]
pub(crate) fn debug_end_to_end_acked_packet_count(&self) -> usize {
let st = self.state.lock();
st.end_to_end_acked_destinations.len()
}
#[cfg(test)]
pub(crate) fn debug_reliable_return_route_count(&self) -> usize {
let st = self.state.lock();
st.reliable_return_routes.len()
}
#[cfg(test)]
pub(crate) fn debug_side_storage(&self) -> (usize, usize) {
let st = self.state.lock();
(st.sides.len(), st.sides.capacity())
}
pub fn rx_packed_from_side(&self, src: RelaySideId, bytes: &[u8]) -> TelemetryResult<()> {
self.ensure_side_ingress_enabled(src)?;
let Some(bytes) = self.decode_side_transport_frame(src, bytes)? else {
return Ok(());
};
let mut st = self.state.lock();
let data = RelayItem::Packed(bytes);
let priority = Self::relay_item_priority(&data)?;
st.push_rx(RelayRxItem {
src,
data,
priority,
})
}
pub fn rx_from_side(&self, src: RelaySideId, packet: Packet) -> TelemetryResult<()> {
self.ensure_side_ingress_enabled(src)?;
let mut st = self.state.lock();
let data = RelayItem::Packet(Arc::new(packet));
let priority = Self::relay_item_priority(&data)?;
st.push_rx(RelayRxItem {
src,
data,
priority,
})
}
pub fn clear_queues(&self) {
let mut st = self.state.lock();
st.rx_queue.clear();
st.tx_queue.clear();
st.tx_priority_burst = 0;
}
pub fn clear_rx_queue(&self) {
let mut st = self.state.lock();
st.rx_queue.clear();
}
pub fn clear_tx_queue(&self) {
let mut st = self.state.lock();
st.tx_queue.clear();
st.tx_priority_burst = 0;
st.replay_queue.clear();
}
fn drain_queued_receipts_before_timeouts(
&self,
start: u64,
timeout_ms: u32,
) -> TelemetryResult<bool> {
let mut did_any = false;
loop {
let item = {
let mut st = self.state.lock();
let index = st.rx_queue.iter().position(|item| {
let ty = match &item.data {
RelayItem::Packet(pkt) => Some(pkt.data_type()),
RelayItem::Packed(bytes) => wire_format::scheduling_type(bytes.as_ref()),
};
matches!(
ty,
Some(
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
)
)
});
index.and_then(|index| st.rx_queue.remove_pos(index))
};
let Some(item) = item else {
return Ok(did_any);
};
self.process_rx_queue_item(item)?;
did_any = true;
if timeout_ms != 0 && self.clock.now_ms().wrapping_sub(start) >= u64::from(timeout_ms) {
return Ok(did_any);
}
}
}
fn process_rx_queue_item(&self, item: RelayRxItem) -> TelemetryResult<()> {
if crate::memory_admission::supports_work_kinds()
&& let RelayItem::Packed(bytes) = &item.data
{
self.admit_received_frame(item.src, bytes)?;
}
self.ensure_side_ingress_enabled(item.src)?;
match &item.data {
RelayItem::Packet(pkt) => {
pkt.validate()?;
#[cfg(feature = "discovery")]
self.refresh_peer_liveness(item.src, sender_address_u32(pkt.sender()));
let bytes = wire_format::packet_wire_size(pkt);
self.note_side_rx(item.src, pkt.data_type(), bytes);
}
RelayItem::Packed(bytes) => {
if let Ok(env) = wire_format::peek_routing_envelope(bytes.as_ref()) {
self.note_side_rx(item.src, env.ty, bytes.len());
}
}
}
match &item.data {
RelayItem::Packet(pkt) => {
if is_reliable_type(pkt.data_type()) && !is_internal_control_type(pkt.data_type()) {
self.note_reliable_return_route(item.src, pkt.packet_id());
}
}
RelayItem::Packed(bytes) => {
if let Ok(env) = wire_format::peek_routing_envelope(bytes.as_ref())
&& is_reliable_type(env.ty)
&& !is_internal_control_type(env.ty)
&& let Ok(packet_id) = wire_format::packet_id_from_wire(bytes.as_ref())
{
self.note_reliable_return_route(item.src, packet_id);
}
}
}
let mut released_buffered: Vec<crate::SharedBytes> = Vec::new();
if let RelayItem::Packed(bytes) = &item.data {
let (_opts, handler_is_packed, hop_reliable_enabled) = {
let st = self.state.lock();
let side_ref = Self::side_ref(&st, item.src)?;
let opts = side_ref.opts.clone();
(
opts.clone(),
matches!(side_ref.tx_handler, RelayTxHandlerFn::Packed(_)),
opts.reliable_enabled
&& !self.side_has_multiple_announcers_locked(
&st,
item.src,
self.clock.now_ms(),
),
)
};
let frame = match wire_format::peek_frame_info(bytes.as_ref()) {
Ok(frame) => frame,
Err(e) => {
if matches!(e, TelemetryError::Unpack(msg) if msg == "crc32 mismatch")
&& hop_reliable_enabled
&& handler_is_packed
&& let Ok(frame) = wire_format::peek_frame_info_unchecked(bytes.as_ref())
{
if is_reliable_type(frame.envelope.ty)
&& let Some(hdr) = frame.reliable
{
let unordered = (hdr.flags & wire_format::RELIABLE_FLAG_UNORDERED) != 0;
let unsequenced =
(hdr.flags & wire_format::RELIABLE_FLAG_UNSEQUENCED) != 0;
if !unsequenced {
let requested = if unordered {
hdr.seq
} else {
let mut st = self.state.lock();
let rx_state = self.reliable_rx_state_mut(
&mut st,
item.src,
frame.envelope.ty,
);
rx_state.expected_seq.min(hdr.seq)
};
self.queue_reliable_packet_request(
item.src,
frame.envelope.ty,
requested,
)?;
}
}
return Ok(());
}
return Err(e);
}
};
if hop_reliable_enabled
&& handler_is_packed
&& is_reliable_type(frame.envelope.ty)
&& let Some(hdr) = frame.reliable
{
if frame.ack_only() {
self.handle_reliable_ack(item.src, frame.envelope.ty, hdr.ack);
return Ok(());
}
let unordered = (hdr.flags & wire_format::RELIABLE_FLAG_UNORDERED) != 0;
let unsequenced = (hdr.flags & wire_format::RELIABLE_FLAG_UNSEQUENCED) != 0;
if !unsequenced {
if unordered {
self.queue_reliable_ack(item.src, frame.envelope.ty, hdr.seq)?;
} else {
let mut release: Vec<crate::SharedBytes> = Vec::new();
let mut last_delivered = None;
let mut ack_old = None;
let mut request_missing = None;
let mut partial_ack = None;
{
let mut st = self.state.lock();
let rx_state =
self.reliable_rx_state_mut(&mut st, item.src, frame.envelope.ty);
let expected_seq = rx_state.expected_seq;
if hdr.seq < expected_seq {
ack_old = Some(expected_seq.saturating_sub(1));
} else if hdr.seq > expected_seq {
request_missing = Some(expected_seq);
partial_ack = Some(hdr.seq);
st.buffer_reliable_rx(
item.src,
frame.envelope.ty,
hdr.seq,
bytes.clone(),
)?;
} else {
release.push(bytes.clone());
last_delivered = Some(hdr.seq);
let mut next_expected = hdr.seq.wrapping_add(1);
while let Some(buf) = rx_state.buffered.remove(&next_expected) {
release.push(buf);
last_delivered = Some(next_expected);
let next = next_expected.wrapping_add(1);
next_expected = if next == 0 { 1 } else { next };
}
rx_state.expected_seq = next_expected;
}
}
if let Some(ack_seq) = ack_old {
self.queue_reliable_ack(item.src, frame.envelope.ty, ack_seq)?;
return Ok(());
}
if let Some(request_seq) = request_missing {
if let Some(partial_seq) = partial_ack {
self.queue_reliable_partial_ack(
item.src,
frame.envelope.ty,
partial_seq,
)?;
}
self.queue_reliable_packet_request(
item.src,
frame.envelope.ty,
request_seq,
)?;
return Ok(());
}
if let Some(ack_seq) = last_delivered {
self.queue_reliable_ack(item.src, frame.envelope.ty, ack_seq)?;
}
released_buffered.extend(release.into_iter().skip(1));
}
}
}
}
if self.is_duplicate_pkt(&item)? && !self.should_forward_duplicate_reliable_item(&item)? {
return Ok(());
}
self.dispatch_relay_rx_item(&item)?;
for release_bytes in released_buffered {
let release_item = RelayRxItem {
src: item.src,
priority: Self::relay_item_priority(&RelayItem::Packed(release_bytes.clone()))?,
data: RelayItem::Packed(release_bytes),
};
if self.is_duplicate_pkt(&release_item)?
&& !self.should_forward_duplicate_reliable_item(&release_item)?
{
continue;
}
self.dispatch_relay_rx_item(&release_item)?;
}
Ok(())
}
fn dispatch_relay_rx_item(&self, item: &RelayRxItem) -> TelemetryResult<()> {
match &item.data {
RelayItem::Packet(pkt) => {
if matches!(
pkt.data_type(),
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
) {
if Self::end_to_end_ack_sender_hash(pkt).is_some()
&& Self::decode_end_to_end_reliable_ack(pkt.payload()).is_ok()
{
if let Ok(packet_id) = Self::decode_end_to_end_reliable_ack(pkt.payload())
&& let Some(sender_hash) = Self::end_to_end_ack_sender_hash(pkt)
{
let mut st = self.state.lock();
Self::note_end_to_end_acked_destination_locked(
&mut st,
packet_id,
sender_hash,
);
}
} else {
let vals = pkt.data_as_u32()?;
if vals.len() != 2 {
return Err(TelemetryError::Unpack("bad reliable control payload"));
}
let ty = crate::DataType::try_from_u32(vals[0])
.ok_or(TelemetryError::InvalidType)?;
let seq = vals[1];
match pkt.data_type() {
crate::DataType::ReliableAck => {
self.handle_reliable_ack(item.src, ty, seq)
}
crate::DataType::ReliablePartialAck => {
self.handle_reliable_partial_ack(item.src, ty, seq)
}
crate::DataType::ReliablePacketRequest => {
self.queue_reliable_retransmit(item.src, ty, seq)?
}
_ => {}
}
return Ok(());
}
}
}
RelayItem::Packed(bytes) => {
let env = wire_format::peek_routing_envelope(bytes.as_ref())?;
if matches!(
env.ty,
crate::DataType::ReliableAck
| crate::DataType::ReliablePacketRequest
| crate::DataType::ReliablePartialAck
) {
let pkt = wire_format::unpack_packet(bytes.as_ref())?;
return self.dispatch_relay_rx_item(&RelayRxItem {
src: item.src,
data: RelayItem::Packet(Arc::new(pkt)),
priority: item.priority,
});
}
}
}
let src = item.src;
let data = item.data.clone();
self.learn_discovery_item(src, &data)?;
let plan = self.remote_side_plan(&data, src)?;
let mut st = self.state.lock();
let RemoteSidePlan::Target(sides) = plan;
for dst in sides {
let priority = Self::relay_item_priority(&data)?;
st.push_tx(RelayTxItem {
src: Some(src),
dst,
data: data.clone(),
priority,
retry_after_ms: 0,
})?;
}
Ok(())
}
fn run_side_chunk_transfer<T>(
&self,
side: RelaySideId,
key: u64,
send: impl FnOnce(&mut usize) -> TelemetryResult<T>,
) -> TelemetryResult<T> {
let mut next = {
let mut st = self.state.lock();
let transport = st
.side_transport
.get_mut(&side)
.ok_or(TelemetryError::BadArg)?;
if transport.tx_chunk_progress.is_none() {
crate::memory_admission::check(
512,
core::mem::size_of::<crate::side_transport::SideTxProgress>(),
)?;
transport.tx_chunk_progress =
Some(Box::new(crate::side_transport::SideTxProgress::default()));
}
transport
.tx_chunk_progress
.as_mut()
.expect("initialized")
.begin(key, self.clock.now_ms())?
};
let result = send(&mut next);
if let Some(progress) = self
.state
.lock()
.side_transport
.get_mut(&side)
.and_then(|transport| transport.tx_chunk_progress.as_mut())
{
progress.finish(key, next, self.clock.now_ms(), &result);
}
result
}
fn send_side_frames_resumable(
&self,
side: RelaySideId,
mut frames: SideTransportFrames,
mut send: impl FnMut(&[u8]) -> TelemetryResult<()>,
) -> TelemetryResult<()> {
let bootstraps_template = frames.bootstraps_template();
let result = if frames.len() == 1 {
frames.send_resuming(&mut 0, send)
} else {
let key = frames.progress_key();
self.run_side_chunk_transfer(side, key, |next| frames.send_resuming(next, &mut send))
};
if result.is_err() && bootstraps_template {
if let Some(state) = self.state.lock().side_transport.get_mut(&side) {
state.refresh_transmit_header_templates();
}
}
result
}
fn split_side_transport_frame(
&self,
_side: RelaySideId,
frame: crate::SharedBytes,
max_frame_bytes: usize,
) -> TelemetryResult<SideTransportFrames> {
SideTransportFrames::split(frame, max_frame_bytes, self.sender_arc().as_bytes())
}
fn encode_side_transport_frames(
&self,
side: RelaySideId,
opts: RelaySideOptions,
raw: crate::SharedBytes,
) -> TelemetryResult<SideTransportFrames> {
if !opts.header_template_enabled && opts.max_frame_bytes == 0 {
return Ok(SideTransportFrames::single(raw));
}
let raw_len = raw.len();
if opts.max_frame_bytes != 0 && raw_len > opts.max_frame_bytes {
let frames = self.split_side_transport_frame(side, raw, opts.max_frame_bytes)?;
let mut st = self.state.lock();
let stats = st.side_runtime_stats.entry(side).or_default();
stats.note_side_transport_full(raw_len, frames.wire_len());
stats.note_side_transport_chunks(frames.len());
return Ok(frames);
}
if opts.header_template_enabled {
let frame = wire_format::peek_routing_frame_info(raw.as_ref())?;
let without_dictionary = opts.max_side_transport_templates == 0 || {
let st = self.state.lock();
self.side_has_multiple_announcers_locked(&st, side, self.clock.now_ms())
};
let reliable = is_reliable_type(frame.envelope.ty)
|| frame.reliable.is_some()
|| matches!(
frame.envelope.ty,
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
);
if without_dictionary || reliable {
let mut st = self.state.lock();
if without_dictionary {
let removed = st
.side_transport
.get_mut(&side)
.ok_or(TelemetryError::BadArg)?
.trim_tx_templates_to(0);
for _ in 0..removed {
st.side_runtime_stats
.entry(side)
.or_default()
.note_side_transport_template_eviction();
}
}
st.side_runtime_stats
.entry(side)
.or_default()
.note_side_transport_full(raw_len, raw_len);
return Ok(SideTransportFrames::single(raw));
}
}
let mut compact_payload_len = None;
let mut used_compact = false;
let used_timestamp_delta = false;
let omitted_timestamp = false;
let (template, _ty, flags, data_size, timestamp, nonce, reliable_seq_ack, payload) =
extract_side_header_template_view(raw.as_ref())?;
if is_reliable_type(_ty)
|| reliable_seq_ack.is_some()
|| matches!(
_ty,
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
)
{
let frames = if opts.max_frame_bytes != 0 && raw_len > opts.max_frame_bytes {
self.split_side_transport_frame(side, raw, opts.max_frame_bytes)?
} else {
SideTransportFrames::single(raw)
};
let wire_len = frames.wire_len();
let mut st = self.state.lock();
let stats = st.side_runtime_stats.entry(side).or_default();
stats.note_side_transport_full(raw_len, wire_len);
if frames.len() > 1 {
stats.note_side_transport_chunks(frames.len());
}
return Ok(frames);
}
let (template_id, use_compact, _previous_timestamp) = {
let mut st = self.state.lock();
let side_state = st
.side_transport
.get_mut(&side)
.ok_or(TelemetryError::BadArg)?;
if let Some(id) = side_state.tx_template_ids.get(&template.hash).copied() {
let previous = side_state.tx_last_timestamps.get(&id).copied();
const FULL_REFRESH_AFTER_COMPACT_FRAMES: u8 = 8;
let compact_uses = side_state.tx_compact_uses.entry(id).or_default();
#[cfg(feature = "discovery")]
let discovery_refresh = discovery::is_discovery_type(_ty);
#[cfg(not(feature = "discovery"))]
let discovery_refresh = false;
let use_compact =
*compact_uses < FULL_REFRESH_AFTER_COMPACT_FRAMES && !discovery_refresh;
if use_compact {
*compact_uses = compact_uses.saturating_add(1);
} else {
*compact_uses = 0;
}
(id, use_compact, previous)
} else {
let next = side_state.next_template_id.wrapping_add(1).max(1);
side_state.next_template_id = next;
let evicted = side_state.insert_tx_template(
template.into_owned(),
next,
opts.max_side_transport_templates,
);
if evicted {
st.side_runtime_stats
.entry(side)
.or_default()
.note_side_transport_template_eviction();
}
if let Some(side_state) = st.side_transport.get_mut(&side)
&& opts.max_side_transport_templates > 0
{
side_state.tx_last_timestamps.insert(next, timestamp);
}
(next, false, None)
}
};
let wrapped = if use_compact {
used_compact = true;
compact_payload_len = Some(payload.len());
let timestamp_field = Some(timestamp);
let mut body = Vec::with_capacity(payload.len() + 32);
body.push(flags);
write_uleb128_local(u64::from(template_id), &mut body);
write_uleb128_local(data_size, &mut body);
if let Some(timestamp_field) = timestamp_field {
write_uleb128_local(timestamp_field, &mut body);
}
if (flags & SIDE_TRANSPORT_FLAG_PACKET_NONCE) != 0 {
write_uleb128_local(u64::from(nonce), &mut body);
}
if let Some((seq, ack)) = reliable_seq_ack {
write_uleb128_local(u64::from(seq), &mut body);
write_uleb128_local(u64::from(ack), &mut body);
}
body.extend_from_slice(payload);
{
let mut st = self.state.lock();
if let Some(side_state) = st.side_transport.get_mut(&side) {
side_state.tx_last_timestamps.insert(template_id, timestamp);
}
}
let kind = if omitted_timestamp {
SIDE_TRANSPORT_KIND_COMPACT_SAME_TIMESTAMP
} else if used_timestamp_delta {
SIDE_TRANSPORT_KIND_COMPACT_DELTA
} else {
SIDE_TRANSPORT_KIND_COMPACT
};
wrap_side_transport_frame(kind, &body)
} else {
let mut body = Vec::with_capacity(raw.len() + 4);
write_uleb128_local(u64::from(template_id), &mut body);
body.extend_from_slice(raw.as_ref());
wrap_side_transport_frame(SIDE_TRANSPORT_KIND_FULL, &body)
};
let frames = if opts.max_frame_bytes != 0 && wrapped.len() > opts.max_frame_bytes {
self.split_side_transport_frame(side, wrapped, opts.max_frame_bytes)
} else {
Ok(SideTransportFrames::single(wrapped))
}?;
let wire_len = frames.wire_len();
let mut st = self.state.lock();
let stats = st.side_runtime_stats.entry(side).or_default();
if used_compact {
let overhead = compact_payload_len
.map(|payload_len| wire_len.saturating_sub(payload_len))
.unwrap_or(wire_len);
stats.note_side_transport_compact(
raw_len,
wire_len,
overhead,
used_timestamp_delta,
omitted_timestamp,
);
if opts.compact_header_target_bytes != 0 && overhead > opts.compact_header_target_bytes
{
stats.note_side_transport_compact_target_miss();
}
} else {
stats.note_side_transport_full(raw_len, wire_len);
}
if frames.len() > 1 {
stats.note_side_transport_chunks(frames.len());
}
Ok(frames)
}
fn expire_side_chunks(&self) {
let now = self.clock.now_ms();
let mut st = self.state.lock();
for side in st.side_transport.values_mut() {
side.rx_chunks
.retain(|_, assembly| now.saturating_sub(assembly.last_seen_ms) <= 2000);
}
}
#[cfg(feature = "discovery")]
fn refresh_ingress_peer_liveness(&self, side: RelaySideId, bytes: &[u8]) {
let Ok(frame) = wire_format::peek_routing_frame_info(bytes) else {
return;
};
let address = frame.envelope.source_address;
self.refresh_peer_liveness(side, address);
}
#[cfg(feature = "discovery")]
#[cfg_attr(not(feature = "std"), inline(never))]
fn refresh_peer_liveness(&self, side: RelaySideId, address: u32) {
let now = self.clock.now_ms();
let mut st = self.state.lock();
let Some(route) = st.discovery_routes.get_mut(&side) else {
return;
};
let mut found = None;
for (name, peer) in &mut route.announcers {
if peer.advertised_address == Some(address) || sender_address_u32(name) == address {
if found.is_some() {
return;
}
found = Some(peer);
}
}
let Some(peer) = found else {
return;
};
peer.last_seen_ms = now;
route.last_seen_ms = route.last_seen_ms.max(now);
}
#[cfg(all(feature = "discovery", not(feature = "std")))]
fn consume_known_packed_schema(
&self,
side: RelaySideId,
bytes: &[u8],
) -> TelemetryResult<bool> {
let Some((address, payload)) = wire_format::peek_plain_schema_payload(bytes)? else {
return Ok(false);
};
if !discovery::discovery_schema_payload_is_fully_known(payload)? {
return Ok(false);
}
let mut st = self.state.lock();
if let Some(route) = st.discovery_routes.get_mut(&side) {
for (name, peer) in &mut route.announcers {
if peer.advertised_address == Some(address) || sender_address_u32(name) == address {
peer.has_schema = true;
break;
}
}
}
Ok(true)
}
#[cfg(feature = "discovery")]
fn cached_transit_ready(&self, side: RelaySideId, bytes: &[u8]) -> TelemetryResult<bool> {
if bytes.len() > 128 || bytes.first().is_none_or(|flags| flags & 0x15 != 0) {
return Ok(false);
}
let frame = wire_format::peek_routing_frame_info(bytes)?;
let ty = frame.envelope.ty;
if is_internal_control_type(ty)
|| frame.envelope.is_reliable
|| frame.reliable.is_some()
|| frame.envelope.target_senders().len() != 0
|| !(1..=3).contains(&frame.envelope.endpoints().take(4).count())
{
return Ok(false);
}
let (template, ..) = extract_side_header_template_view(bytes)?;
let st = self.state.lock();
let active: usize = st.sides.iter().filter(|entry| entry.is_some()).count();
if active != 2 {
return Ok(false);
}
for (id, entry) in st
.sides
.iter()
.enumerate()
.filter_map(|(id, entry)| entry.as_ref().map(|e| (id, e)))
{
if entry.opts.reliable_enabled
|| !matches!(entry.tx_handler, RelayTxHandlerFn::Packed(_))
|| !st
.side_runtime_stats
.get(&id)
.is_some_and(|stats| stats.data_types.contains_key(&ty.as_u32()))
{
return Ok(false);
}
if id == side {
continue;
}
#[cfg(feature = "discovery")]
if !st.discovery_routes.get(&id).is_some_and(|route| {
self.clock.now_ms().saturating_sub(route.last_seen_ms) <= DISCOVERY_ROUTE_TTL_MS
&& frame
.envelope
.endpoints()
.any(|ep| route.reachable.contains(&ep))
}) {
return Ok(false);
}
if entry.opts.header_template_enabled
&& entry.opts.max_frame_bytes != 0
&& entry.opts.max_side_transport_templates != 0
&& !self.side_has_multiple_announcers_locked(&st, id, self.clock.now_ms())
&& !st.side_transport.get(&id).is_some_and(|transport| {
transport.tx_template_ids.contains_key(&template.hash)
&& transport.tx_templates.contains_key(&template.hash)
})
{
return Ok(false);
}
}
Ok(true)
}
#[cfg(not(feature = "discovery"))]
fn cached_transit_ready(&self, _side: RelaySideId, _bytes: &[u8]) -> TelemetryResult<bool> {
Ok(false)
}
fn admit_received_frame(&self, side: RelaySideId, bytes: &[u8]) -> TelemetryResult<()> {
if crate::memory_admission::supports_work_kinds()
&& self.cached_transit_ready(side, bytes)?
{
crate::memory_admission::check_cached_transit(bytes.len())
} else {
let mut protected = None;
loop {
match crate::memory_admission::check_receive(bytes) {
Err(TelemetryError::Io("memory pressure")) => {
let protected = *protected.get_or_insert_with(|| {
wire_format::peek_routing_frame_info(bytes)
.ok()
.is_some_and(|frame| message_priority(frame.envelope.ty) >= 200)
});
if !protected {
return Err(TelemetryError::Io("memory pressure"));
}
let mut st = self.state.lock();
let reclaimed = st.pop_lowest_packet_queue_item(200);
if !reclaimed {
return Err(TelemetryError::Io("memory pressure"));
}
}
result => return result,
}
}
}
}
fn decode_side_transport_frame(
&self,
side: RelaySideId,
bytes: &[u8],
) -> TelemetryResult<Option<crate::SharedBytes>> {
self.expire_side_chunks();
let Some((kind, body)) = parse_side_transport_wrapper(bytes)? else {
#[cfg(feature = "discovery")]
self.refresh_ingress_peer_liveness(side, bytes);
#[cfg(all(feature = "discovery", not(feature = "std")))]
if self.consume_known_packed_schema(side, bytes)? {
return Ok(None);
}
self.admit_received_frame(side, bytes)?;
return Ok(Some(crate::shared_bytes::convert(Arc::<[u8]>::from(bytes))));
};
match kind {
SIDE_TRANSPORT_KIND_FULL => {
let mut off = 0usize;
let template_id = u32::try_from(read_uleb128_local(body, &mut off)?)
.map_err(|_| TelemetryError::Unpack("side template id too large"))?;
#[cfg(feature = "discovery")]
self.refresh_ingress_peer_liveness(side, &body[off..]);
#[cfg(all(feature = "discovery", not(feature = "std")))]
if self.consume_known_packed_schema(side, &body[off..])? {
return Ok(None);
}
self.admit_received_frame(side, &body[off..])?;
let raw = Arc::<[u8]>::from(&body[off..]);
if let Ok((template, _, _, _, timestamp, _, _, _)) =
extract_side_header_template(raw.as_ref())
{
let mut st = self.state.lock();
let max_templates = st
.sides
.get(side)
.and_then(|side| side.as_ref())
.map(|side| side.opts.max_side_transport_templates)
.unwrap_or(DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT);
let evicted = st.side_transport.get_mut(&side).is_some_and(|side_state| {
let evicted =
side_state.insert_rx_template(template_id, template, max_templates);
if max_templates > 0 {
side_state.rx_last_timestamps.insert(template_id, timestamp);
}
evicted
});
if evicted {
st.side_runtime_stats
.entry(side)
.or_default()
.note_side_transport_template_eviction();
}
}
Ok(Some(crate::shared_bytes::convert(raw)))
}
SIDE_TRANSPORT_KIND_COMPACT
| SIDE_TRANSPORT_KIND_COMPACT_DELTA
| SIDE_TRANSPORT_KIND_COMPACT_SAME_TIMESTAMP => {
if body.is_empty() {
return Err(TelemetryError::Unpack("short side compact frame"));
}
let mut off = 1usize;
let template_id = u32::try_from(read_uleb128_local(body, &mut off)?)
.map_err(|_| TelemetryError::Unpack("side template id too large"))?;
let (template, timestamp_base) = {
let st = self.state.lock();
let state = st.side_transport.get(&side);
let template = state
.and_then(|state| state.rx_templates_by_id.get(&template_id))
.cloned();
let timestamp_base = if matches!(
kind,
SIDE_TRANSPORT_KIND_COMPACT_DELTA
| SIDE_TRANSPORT_KIND_COMPACT_SAME_TIMESTAMP
) {
state
.and_then(|state| state.rx_last_timestamps.get(&template_id))
.copied()
} else {
None
};
(template, timestamp_base)
};
let template =
template.ok_or(TelemetryError::Unpack("unknown side compact template"))?;
#[cfg(feature = "discovery")]
if let Some(address) = template.source_address() {
self.refresh_peer_liveness(side, address);
}
let reconstructed_bound = body
.len()
.saturating_add(template.prefix.len())
.saturating_add(template.between.len())
.saturating_add(32);
let protected = template
.data_type()
.is_some_and(|ty| message_priority(ty) >= 200);
let ack = template.is_ack();
loop {
match crate::memory_admission::check_frame(reconstructed_bound, ack) {
Err(TelemetryError::Io("memory pressure")) if protected => {
let mut st = self.state.lock();
let reclaimed = st.pop_lowest_packet_queue_item(200);
if !reclaimed {
return Err(TelemetryError::Io("memory pressure"));
}
}
result => {
result?;
break;
}
}
}
let timestamp_mode = match kind {
SIDE_TRANSPORT_KIND_COMPACT_DELTA => SideCompactTimestampMode::Delta,
SIDE_TRANSPORT_KIND_COMPACT_SAME_TIMESTAMP => SideCompactTimestampMode::Omitted,
_ => SideCompactTimestampMode::Absolute,
};
let (frame, timestamp) = reconstruct_side_compact_frame(
&template,
body[0],
&body[off..],
timestamp_mode,
timestamp_base,
)?;
let mut st = self.state.lock();
if let Some(side_state) = st.side_transport.get_mut(&side) {
side_state.rx_last_timestamps.insert(template_id, timestamp);
}
Ok(Some(frame))
}
SIDE_TRANSPORT_KIND_CHUNK => {
crate::memory_admission::check_receive(bytes)?;
if body.len() < 8 {
return Err(TelemetryError::Unpack("short side chunk frame"));
}
let transfer_id = u32::from_le_bytes([body[0], body[1], body[2], body[3]]);
let index = u16::from_le_bytes([body[4], body[5]]);
let total = u16::from_le_bytes([body[6], body[7]]);
if total == 0 || index >= total {
return Err(TelemetryError::Unpack("invalid side chunk index"));
}
let assembled = {
let mut st = self.state.lock();
let budget = st.memory.max_queue_budget;
let now_ms = self.clock.now_ms();
let side_state = st
.side_transport
.get_mut(&side)
.ok_or(TelemetryError::BadArg)?;
side_state
.rx_chunks
.retain(|_, assembly| now_ms.saturating_sub(assembly.last_seen_ms) <= 2000);
if !side_state.rx_chunks.contains_key(&transfer_id)
&& side_state.rx_chunks.len() >= 4
&& let Some(oldest) = side_state
.rx_chunks
.iter()
.min_by_key(|(_, a)| a.last_seen_ms)
.map(|(id, _)| *id)
{
side_state.rx_chunks.remove(&oldest);
}
SideChunkAssembly::make_room(
&mut side_state.rx_chunks,
transfer_id,
index,
total,
body.len() - 8,
budget,
)?;
let entry = side_state.rx_chunks.entry(transfer_id).or_default();
entry.last_seen_ms = now_ms;
if entry.total == 0 {
entry.total = total;
} else if entry.total != total {
side_state.rx_chunks.remove(&transfer_id);
return Err(TelemetryError::Unpack("side chunk total mismatch"));
}
entry.received.entry(index).or_insert_with(|| {
crate::shared_bytes::convert(Arc::<[u8]>::from(&body[8..]))
});
if entry.received.len() == usize::from(total) {
let entry = side_state
.rx_chunks
.remove(&transfer_id)
.ok_or(TelemetryError::Unpack("side chunk missing"))?;
Some(entry.assemble()?)
} else {
None
}
};
match assembled {
Some(frame) => self.decode_side_transport_frame(side, frame.as_ref()),
None => Ok(None),
}
}
_ => Err(TelemetryError::Unpack("unknown side transport frame")),
}
}
fn call_tx_handler(
&self,
side: RelaySideId,
handler: &RelayTxHandlerFn,
data: &RelayItem,
) -> TelemetryResult<()> {
let opts = {
let st = self.state.lock();
Self::side_ref(&st, side)?.opts.clone()
};
let Some(_side_tx_guard) = self.try_enter_side_tx() else {
return Err(TelemetryError::Io("side tx busy"));
};
let started_ms = self.clock.now_ms();
let ty = match data {
RelayItem::Packet(pkt) => pkt.data_type(),
RelayItem::Packed(bytes) => wire_format::peek_routing_envelope(bytes.as_ref())?.ty,
};
let result = match (handler, data) {
(RelayTxHandlerFn::Packed(f), RelayItem::Packed(bytes)) => {
let frames = self.encode_side_transport_frames(side, opts, bytes.clone())?;
let mut sent_bytes = 0usize;
self.send_side_frames_resumable(side, frames, |frame| {
f(frame)?;
sent_bytes = sent_bytes.saturating_add(frame.len());
Ok(())
})?;
self.record_side_tx_sample(side, sent_bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, sent_bytes, 1);
return Ok(());
}
(RelayTxHandlerFn::Packet(f), RelayItem::Packet(pkt)) => f(pkt),
(RelayTxHandlerFn::Packed(f), RelayItem::Packet(pkt)) => {
if ty == crate::DataType::DiscoverySchema
&& opts.max_frame_bytes != 0
&& pkt.payload().len() > opts.max_frame_bytes
{
let key = crate::packet::hash_bytes_u64(
pkt.packet_id(),
&opts.max_frame_bytes.to_le_bytes(),
);
let result = self.run_side_chunk_transfer(side, key, |next| {
crate::side_transport::send_packet_chunks(
pkt,
opts.max_frame_bytes,
self.sender_arc().as_bytes(),
next,
|frame| f(frame),
)
});
let (raw_len, wire_len, chunks) = match result {
Ok(stats) => stats,
Err(err) => {
self.note_side_tx_failure(side, ty, 1);
return Err(err);
}
};
{
let mut st = self.state.lock();
let stats = st.side_runtime_stats.entry(side).or_default();
stats.note_side_transport_full(raw_len, wire_len);
if chunks > 1 {
stats.note_side_transport_chunks(chunks);
}
}
self.record_side_tx_sample(side, wire_len, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, wire_len, 1);
return Ok(());
}
let owned = wire_format::pack_packet(pkt);
let frames = self.encode_side_transport_frames(
side,
opts.clone(),
crate::shared_bytes::convert(owned),
)?;
let mut sent_bytes = 0usize;
self.send_side_frames_resumable(side, frames, |frame| {
f(frame)?;
sent_bytes = sent_bytes.saturating_add(frame.len());
Ok(())
})?;
self.record_side_tx_sample(side, sent_bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, sent_bytes, 1);
return Ok(());
}
(RelayTxHandlerFn::Packet(f), RelayItem::Packed(bytes)) => {
if wire_format::peek_frame_info(bytes.as_ref())
.ok()
.is_some_and(|frame| frame.ack_only())
{
return Ok(());
}
let pkt = wire_format::unpack_packet(bytes.as_ref())?;
f(&pkt)
}
};
if result.is_ok()
&& let Ok(bytes) = Self::relay_item_wire_len(data)
{
self.record_side_tx_sample(side, bytes, started_ms, self.clock.now_ms());
self.note_side_tx_success(side, ty, bytes, 1);
} else if result.is_err() {
self.note_side_tx_failure(side, ty, 1);
}
result
}
fn adjust_reliable_for_side(
&self,
opts: RelaySideOptions,
data: RelayItem,
) -> TelemetryResult<Option<RelayItem>> {
if opts.reliable_enabled {
return Ok(Some(data));
}
match data {
RelayItem::Packed(bytes) => {
let frame = match wire_format::peek_routing_frame_info(bytes.as_ref()) {
Ok(frame) => frame,
Err(_) => return Ok(Some(RelayItem::Packed(bytes))),
};
if is_reliable_type(frame.envelope.ty)
&& let Some(hdr) = frame.reliable
{
if (hdr.flags & wire_format::RELIABLE_FLAG_ACK_ONLY) != 0 {
return Ok(None);
}
if (hdr.flags & wire_format::RELIABLE_FLAG_UNSEQUENCED) == 0 {
let Some(rewritten) = wire_format::rewrite_reliable_header_owned(
bytes.as_ref(),
wire_format::RELIABLE_FLAG_UNSEQUENCED,
hdr.seq,
0,
)?
else {
return Ok(Some(RelayItem::Packed(bytes)));
};
return Ok(Some(RelayItem::Packed(crate::shared_bytes::convert(
rewritten,
))));
}
}
Ok(Some(RelayItem::Packed(bytes)))
}
RelayItem::Packet(pkt) => {
if matches!(
pkt.data_type(),
crate::DataType::ReliableAck
| crate::DataType::ReliablePartialAck
| crate::DataType::ReliablePacketRequest
) {
if Self::end_to_end_ack_sender_hash(&pkt).is_some()
&& Self::decode_end_to_end_reliable_ack(pkt.payload()).is_ok()
{
return Ok(Some(RelayItem::Packet(pkt)));
}
return Ok(None);
}
Ok(Some(RelayItem::Packet(pkt)))
}
}
}
#[inline]
pub fn process_rx_queue(&self) -> TelemetryResult<()> {
self.process_rx_queue_with_timeout(0)
}
#[inline]
pub fn process_tx_queue(&self) -> TelemetryResult<()> {
self.process_tx_queue_with_timeout(0)
}
#[inline]
pub fn process_all_queues(&self) -> TelemetryResult<()> {
self.process_all_queues_with_timeout(0)
}
pub fn process_tx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
self.expire_side_chunks();
if self.side_tx_active() {
return Ok(());
}
#[cfg(feature = "discovery")]
{
let _ = crate::memory_admission::allow_drain(self.poll_discovery())?;
}
let start = self.clock.now_ms();
loop {
let received_control = crate::memory_admission::allow_drain(
self.drain_queued_receipts_before_timeouts(start, timeout_ms),
)?;
let control_budget_used = received_control
&& timeout_ms != 0
&& self.clock.now_ms().wrapping_sub(start) >= u64::from(timeout_ms);
if !control_budget_used {
crate::memory_admission::allow_drain(self.process_reliable_timeouts())?;
}
if !control_budget_used && self.process_replay_queue_item()? {
if timeout_ms != 0 && self.clock.now_ms().wrapping_sub(start) >= timeout_ms as u64 {
break;
}
continue;
}
let Some((src, dst, handler, opts, data)) = self.pop_ready_tx_item() else {
break;
};
match self.send_tx_item(src, dst, handler, opts, data.clone()) {
Ok(sent) => {
if sent
&& timeout_ms != 0
&& self.clock.now_ms().wrapping_sub(start) >= timeout_ms as u64
{
break;
}
}
Err(e) if Self::is_side_tx_busy(&e) => {
let priority = Self::relay_item_priority(&data)?;
let mut st = self.state.lock();
st.push_tx(RelayTxItem {
src,
dst,
data,
priority,
retry_after_ms: self.clock.now_ms().saturating_add(1),
})?;
continue;
}
Err(e) => return Err(e),
}
}
Ok(())
}
pub fn process_rx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
self.expire_side_chunks();
#[cfg(feature = "discovery")]
{
let _ = crate::memory_admission::allow_drain(self.poll_discovery())?;
}
let start = self.clock.now_ms();
loop {
let item_opt = {
let mut st = self.state.lock();
st.rx_queue.pop_front()
};
let Some(item) = item_opt else { break };
self.process_rx_queue_item(item)?;
if timeout_ms != 0 && self.clock.now_ms().wrapping_sub(start) >= timeout_ms as u64 {
break;
}
}
Ok(())
}
pub fn process_all_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
if self.side_tx_active() {
return Ok(());
}
#[cfg(feature = "discovery")]
let _ = crate::memory_admission::allow_drain(self.poll_discovery())?;
self.drain_queues_with_timeout(timeout_ms)
}
pub fn drain_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
self.expire_side_chunks();
if self.side_tx_active() {
return Ok(());
}
let drain_fully = timeout_ms == 0;
let start = if drain_fully { 0 } else { self.clock.now_ms() };
loop {
let mut did_any;
let received_control = crate::memory_admission::allow_drain(
self.drain_queued_receipts_before_timeouts(start, timeout_ms),
)?;
did_any = received_control;
let control_budget_used = received_control
&& timeout_ms != 0
&& self.clock.now_ms().wrapping_sub(start) >= u64::from(timeout_ms);
if !control_budget_used {
crate::memory_admission::allow_drain(self.process_reliable_timeouts())?;
}
if !control_budget_used
&& let Some(item) = {
let mut st = self.state.lock();
st.rx_queue.pop_front()
}
{
self.process_rx_queue_item(item)?;
did_any = true;
}
if !control_budget_used && self.process_replay_queue_item()? {
did_any = true;
}
let sent_one = if let Some((src, dst, handler, opts, data)) = self.pop_ready_tx_item() {
match self.send_tx_item(src, dst, handler, opts, data.clone()) {
Ok(sent) => sent,
Err(error) if Self::is_side_tx_busy(&error) => {
let priority = Self::relay_item_priority(&data)?;
self.state.lock().push_tx(RelayTxItem {
src,
dst,
data,
priority,
retry_after_ms: self.clock.now_ms().saturating_add(1),
})?;
did_any = true;
false
}
Err(error) => return Err(error),
}
} else {
false
};
if sent_one {
did_any = true;
}
if !drain_fully && self.clock.now_ms().wrapping_sub(start) >= timeout_ms as u64 {
break;
}
if !did_any {
break;
}
}
Ok(())
}
pub fn periodic(&self, timeout_ms: u32) -> TelemetryResult<()> {
#[cfg(feature = "discovery")]
{
let _ = crate::memory_admission::allow_drain(self.poll_discovery())?;
}
self.drain_queues_with_timeout(timeout_ms)
}
}
#[cfg(all(test, feature = "std", feature = "discovery"))]
mod discovery_reclaim_tests {
use super::*;
use crate::DataType;
use core::sync::atomic::{AtomicUsize, Ordering};
#[test]
fn relay_discovery_reclaims_telemetry_without_discarding_controls() {
crate::tests::ensure_common_test_schema();
static REFUSALS: AtomicUsize = AtomicUsize::new(0);
extern "C" fn scratch_after_reclaim(additional: usize, _: usize) -> bool {
additional < 2048 || REFUSALS.fetch_add(1, Ordering::SeqCst) >= 2
}
struct Reset;
impl Drop for Reset {
fn drop(&mut self) {
crate::memory_admission::set_probe(None);
}
}
let relay = Relay::new(Box::new(|| 1));
let side = relay.add_side_packet("CAN", |_| Ok(()));
for priority in [1, 2, 200, 254, 255] {
relay
.state
.lock()
.push_tx(RelayTxItem {
src: None,
dst: side,
data: RelayItem::Packed(crate::shared_bytes::convert(Arc::<[u8]>::from(
[0; 32],
))),
priority,
retry_after_ms: 0,
})
.unwrap();
}
REFUSALS.store(0, Ordering::SeqCst);
let _reset = Reset;
crate::memory_admission::set_probe(Some(scratch_after_reclaim));
relay
.queue_discovery_announce_on_side(false, true, None)
.unwrap();
let st = relay.state.lock();
assert_eq!(st.tx_queue.lowest_priority(|item| item.priority), Some(200));
for priority in [200, 254] {
assert_eq!(
st.tx_queue
.iter()
.filter(|item| item.priority == priority)
.count(),
1
);
}
assert!(
st.tx_queue
.iter()
.any(|item| matches!(&item.data, RelayItem::Packet(pkt)
if pkt.data_type() == DataType::DiscoveryLinkCapabilities))
);
}
}
#[cfg(all(test, feature = "std", feature = "discovery"))]
mod queued_schema_identity_tests {
use super::*;
fn catalog(timestamp: u64, byte: u8) -> Packet {
Packet::new(
crate::DataType::DiscoverySchema,
&[crate::DataEndpoint::Discovery],
"SCHEMA_PEER",
timestamp,
crate::shared_bytes::convert(Arc::<[u8]>::from(vec![byte; 1024])),
)
.unwrap()
}
#[test]
fn repeated_schema_preserves_partial_transfer_but_updates_and_other_sides_do_not_coalesce() {
let relay = Relay::new(Box::new(|| 0u64));
let side = relay.add_side_packed("A", |_| Ok(()));
let other_side = relay.add_side_packed("B", |_| Ok(()));
let original = catalog(1, 42);
let duplicate = catalog(2, 42);
assert_ne!(original.packet_id(), duplicate.packet_id());
let mut st = relay.state.lock();
st.tx_queue.clear();
let queued = |dst, pkt| RelayTxItem {
src: None,
dst,
data: RelayItem::Packet(Arc::new(pkt)),
priority: 255,
retry_after_ms: 123,
};
st.push_tx(queued(side, original.clone())).unwrap();
st.push_tx(queued(side, duplicate.clone())).unwrap();
assert_eq!(st.tx_queue.len(), 1);
let pending = st.tx_queue.iter().next().unwrap();
assert_eq!(pending.retry_after_ms, 123);
assert!(matches!(&pending.data, RelayItem::Packet(pkt)
if pkt.packet_id() == original.packet_id()));
st.push_tx(queued(other_side, duplicate)).unwrap();
assert_eq!(st.tx_queue.len(), 2);
let changed = catalog(3, 43);
st.push_tx(queued(side, changed.clone())).unwrap();
assert_eq!(st.tx_queue.len(), 2);
assert!(st.tx_queue.iter().any(|queued| queued.dst == side
&& matches!(&queued.data,
RelayItem::Packet(pkt) if pkt.packet_id() == changed.packet_id())));
}
#[test]
fn repeated_announcements_complete_one_fragment_per_dispatch() {
use core::sync::atomic::{AtomicUsize, Ordering};
let allowance = Arc::new(AtomicUsize::new(0));
let accepted = Arc::new(std::sync::Mutex::new(Vec::<Arc<[u8]>>::new()));
let available = allowance.clone();
let frames = accepted.clone();
let node = Relay::new(Box::new(|| 1u64));
let side = node.add_side_packed_with_options(
"bounded CAN",
move |bytes| {
if available
.try_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
count.checked_sub(1)
})
.is_err()
{
return Err(TelemetryError::Io("side tx busy"));
}
frames.lock().unwrap().push(Arc::<[u8]>::from(bytes));
Ok(())
},
RelaySideOptions::default().with_small_packet_transport(128),
);
let handler = node.state.lock().sides[side]
.as_ref()
.unwrap()
.tx_handler
.clone();
let mut complete = false;
for timestamp in 1..100 {
let mut state = 0x517cc1b7u32;
let payload: Vec<u8> = (0..3600)
.map(|_| {
state ^= state << 13;
state ^= state >> 17;
state ^= state << 5;
state as u8
})
.collect();
let packet = Packet::new(
crate::DataType::DiscoverySchema,
&[crate::DataEndpoint::Discovery],
"SCHEMA_PEER",
timestamp,
crate::shared_bytes::convert(Arc::<[u8]>::from(payload)),
)
.unwrap();
node.state
.lock()
.push_tx(RelayTxItem {
src: None,
dst: side,
data: RelayItem::Packet(Arc::new(packet)),
priority: 255,
retry_after_ms: 0,
})
.unwrap();
let data = node
.state
.lock()
.tx_queue
.iter()
.next()
.unwrap()
.data
.clone();
allowance.store(1, Ordering::Relaxed);
match node.call_tx_handler(side, &handler, &data) {
Ok(()) => {
complete = true;
break;
}
Err(TelemetryError::Io(_)) => {}
other => panic!("unexpected result: {other:?}"),
}
}
assert!(
complete,
"new announcement timestamps must not replay the prefix forever"
);
let received = accepted.lock().unwrap();
assert!(
received.len() > 1,
"fixture must exercise fragmentation, not a compressed single frame"
);
let first = &received[0];
assert_eq!(&first[..4], &[83, 68, 84, 3]);
assert_eq!(
u16::from_le_bytes([first[10], first[11]]) as usize,
received.len()
);
let transfer_id = &first[4..8];
for (index, frame) in received.iter().enumerate() {
assert_eq!(&frame[4..8], transfer_id);
assert_eq!(u16::from_le_bytes([frame[8], frame[9]]) as usize, index);
}
}
#[test]
fn schema_receipt_resolves_assigned_numeric_address_without_adding_peers() {
let relay = Relay::new(Box::new(|| 0u64));
let side = relay.add_side_packed("A", |_| Ok(()));
let assigned = 0x1234u32;
assert_ne!(sender_address_u32("SCHEMA_PEER"), assigned);
{
let mut st = relay.state.lock();
st.discovery_routes
.entry(side)
.or_default()
.announcers
.insert(
"SCHEMA_PEER".into(),
DiscoverySenderState {
advertised_address: Some(assigned),
..DiscoverySenderState::default()
},
);
}
relay.note_schema_received(side, "@addr:4660");
relay.note_schema_received(side, "@addr:4661");
let st = relay.state.lock();
let peers = &st.discovery_routes.get(&side).unwrap().announcers;
assert_eq!(peers.len(), 1);
assert!(peers["SCHEMA_PEER"].has_schema);
}
}
#[cfg(all(test, feature = "discovery", feature = "std"))]
fn untemplated_test_node() -> Relay {
Relay::new(Box::new(|| 0))
}
#[cfg(all(test, feature = "discovery", feature = "std"))]
fn untemplated_test_options() -> RelaySideOptions {
RelaySideOptions::default().with_small_packet_transport(64)
}
#[cfg(all(test, feature = "discovery", feature = "std"))]
#[allow(clippy::duplicate_mod)]
#[path = "tests/untemplated_transport.rs"]
mod untemplated_transport_tests;