sedsnet/

relay.rs

#[cfg(all(test, feature = "std", feature = "discovery"))]
#[path = "tests/shared_chunk_budget_relay.rs"]
mod shared_chunk_budget_tests;
// Apply the same cache contract independently to router and relay state.
#[cfg(all(test, feature = "discovery"))]
#[path = "tests/relay_ack_recovery.rs"]
mod ack_recovery_tests;
#[cfg(all(test, feature = "std"))]
#[path = "tests/relay_control_priority.rs"]
mod control_priority_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()));
        }
        // Establish template, lose frame 1, reorder and duplicate later frames.
        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: Arc::new(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;

/// Logical side index (CAN, UART, RADIO, etc.)
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;
/// Packet Handler function type
type PacketHandlerFn = dyn Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static;

/// Packed Handler function type
type PackedHandlerFn = dyn Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static;

/// TX handler for a relay side: either packed or packet-based.
#[derive(Clone)]
pub enum RelayTxHandlerFn {
    Packed(Arc<PackedHandlerFn>),
    Packet(Arc<PacketHandlerFn>),
}

#[derive(Clone, Debug)]
pub struct RelaySideOptions {
    /// Enables the relay's per-link reliable transport layer on this side.
    ///
    /// When `true` and the side uses a packed TX handler, reliable schema traffic on this hop
    /// gains relay-managed sequence numbers, ACKs, packet requests, and retransmits.
    /// Packet-output sides still receive decoded packets rather than packed reliable framing.
    pub reliable_enabled: bool,
    /// Marks the side as eligible for link-local-only endpoints and discovery routes.
    pub link_local_enabled: bool,
    /// Allows packets received from this side to enter relay processing.
    pub ingress_enabled: bool,
    /// Allows the relay to transmit packets toward this side.
    pub egress_enabled: bool,
    /// Enables side-local header-template reuse for packed transport.
    pub header_template_enabled: bool,
    /// Maximum number of bytes to emit per packed TX callback.
    ///
    /// When non-zero and a packed frame would exceed this size, the relay
    /// splits it into ordered side-transport chunks and reassembles them on RX
    /// before normal relay processing. This is intended for fixed-size links
    /// such as CAN or I2C while keeping the user API packet-oriented.
    pub max_frame_bytes: usize,
    /// Target total side-transport overhead for compact follow-up frames.
    pub compact_header_target_bytes: usize,
    /// Maximum side-local header templates retained for TX and RX dictionaries.
    pub max_side_transport_templates: usize,
    /// Omits the timestamp field from compact follow-up frames when it is unchanged.
    pub omit_unchanged_compact_timestamps: bool,
    /// Optional per-data-type timestamp omission policy for compact follow-up frames.
    pub compact_timestamp_omission_types: CompactTimestampOmissionPolicy,
    /// Declared compact-link profile for stats and future negotiation.
    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 {
    /// Convenience preset for bounded packed-side transport.
    ///
    /// `max_frame_bytes == 0` leaves packed frames unbounded. Values greater
    /// than zero enable relay-managed chunking/reassembly on this side.
    #[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,
        }
    }
}

/// One side of the relay – a name + TX handler.
#[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>),
}

/// Item that was received by the relay from some side.
#[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(),
        }
    }
}

/// Item that is ready to be transmitted out a destination side.
#[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>())
    }
}

// -------------------- Reliable delivery state (relay) --------------------

#[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 {
    /// Numeric identity advertised for this hostname. Compact frames may carry
    /// this address instead of the hostname, so routing must accept either.
    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: Arc<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>,
    rx_chunk_reclaims: u32,
    rx_chunk_bytes: usize,
    next_template_id: u32,
}

impl SideTransportState {
    #[inline(never)]
    fn take_chunk(&mut self, id: u32, reclaimed: bool) -> Option<SideChunkAssembly> {
        let value = SideChunkAssembly::remove(&mut self.rx_chunks, &mut self.rx_chunk_bytes, id)?;
        if reclaimed {
            self.rx_chunk_reclaims = self.rx_chunk_reclaims.saturating_add(1);
        }
        Some(value)
    }
    /// Seed fixtures that construct raw maps directly.
    #[cfg(test)]
    #[inline(never)]
    fn refresh_rx_chunk_bytes(&mut self) {
        self.rx_chunk_bytes = self
            .rx_chunks
            .values()
            .map(SideChunkAssembly::byte_cost)
            .sum();
    }
    fn refresh_transmit_header_templates(&mut self) {
        // Refresh on the wire, not by forgetting the eviction history. RX
        // retains its dictionary; clearing only TX makes bounded caches evict
        // different entries after the next insertion/capacity negotiation.
        for uses in self.tx_compact_uses.values_mut() {
            *uses = u8::MAX;
        }
        // Local topology changes do not invalidate dictionaries received from
        // peers. Dropping RX state here loses already-in-flight compact ACKs.
    }

    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;
        }
        // Keep one owned header. The hash index needs only its wire ID, not
        // another copy of the header and its shared-buffer owners.
        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)
            // Match TX's deterministic lowest-hash eviction on both routers
            // and relays; lowest wire ID can select a different live header.
            && 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,
    // Scheduled and accepted sends differ when a constrained link is backed up.
    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,
        }
    }
}

use crate::type_stats::TypeRuntimeStatsInner;

#[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);
        self.type_stats_mut(ty)
            .tx(bytes as u64, true, 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);
        self.type_stats_mut(ty).rx(bytes as u64, true);
    }

    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);
        self.type_stats_mut(ty).failure(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,
}

/// Internal state, protected by RouterMutex so all public methods can take &self.
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>,
    // Sparse side tables must not reserve eleven full metadata values in
    // every B-tree leaf. Own each side once; keep the index pointer-sized.
    side_runtime_stats: BTreeMap<RelaySideId, Box<SideRuntimeStatsInner>>,
    side_transport: BTreeMap<RelaySideId, Box<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,
}

impl RelayInner {
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn fair_bulk_owner(data: &RelayItem, priority: u8) -> Option<u32> {
        match data {
            RelayItem::Packet(packet) => crate::admission_fairness::packet_owner(packet, priority),
            RelayItem::Packed(bytes) => crate::shared_bytes::with_slice(bytes, |view| {
                crate::admission_fairness::packed_owner(view, priority)
            })
            .ok()
            .flatten(),
        }
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn bulk_stream(data: &RelayItem) -> Option<u32> {
        match data {
            RelayItem::Packet(packet) => Some(packet.data_type().as_u32()),
            RelayItem::Packed(bytes) => crate::shared_bytes::with_slice(bytes, |view| {
                wire_format::scheduling_type(view).map(|ty| ty.as_u32())
            })
            .ok()
            .flatten(),
        }
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn bulk_loads(&self, priority: u8) -> crate::admission_fairness::Loads {
        let mut loads = crate::admission_fairness::Loads::new();
        for item in self.rx_queue.iter() {
            if item.priority <= priority
                && let Some(owner) = Self::fair_bulk_owner(&item.data, item.priority)
            {
                loads.record(owner, item.byte_cost());
            }
        }
        for item in self.tx_queue.iter() {
            if item.priority <= priority
                && let Some(owner) = Self::fair_bulk_owner(&item.data, item.priority)
            {
                loads.record(owner, item.byte_cost());
            }
        }
        loads
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn competing_bulk_owner(&self, incoming: u32, priority: u8) -> Option<u32> {
        self.bulk_loads(priority).dominant_other(incoming)
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn reclaim_bulk_from(
        &mut self,
        owner: u32,
        priority: u8,
        incoming_stream: Option<u32>,
    ) -> bool {
        let mut streams = crate::admission_fairness::Loads::new();
        for item in self.rx_queue.iter() {
            if item.priority <= priority
                && Self::fair_bulk_owner(&item.data, item.priority) == Some(owner)
                && let Some(ty) = Self::bulk_stream(&item.data)
            {
                streams.record(ty, item.byte_cost());
            }
        }
        for item in self.tx_queue.iter() {
            if item.priority <= priority
                && Self::fair_bulk_owner(&item.data, item.priority) == Some(owner)
                && let Some(ty) = Self::bulk_stream(&item.data)
            {
                streams.record(ty, item.byte_cost());
            }
        }
        let stream = streams.queue_victim(incoming_stream);
        let rx = self
            .rx_queue
            .iter()
            .enumerate()
            .filter(|(_, item)| {
                item.priority <= priority
                    && Self::fair_bulk_owner(&item.data, item.priority) == Some(owner)
                    && stream.is_none_or(|ty| Self::bulk_stream(&item.data) == Some(ty))
            })
            .min_by_key(|(index, item)| (item.priority, *index))
            .map(|(index, item)| (index, item.priority));
        let tx = self
            .tx_queue
            .iter()
            .enumerate()
            .filter(|(_, item)| {
                item.priority <= priority
                    && Self::fair_bulk_owner(&item.data, item.priority) == Some(owner)
                    && stream.is_none_or(|ty| Self::bulk_stream(&item.data) == Some(ty))
            })
            .min_by_key(|(index, item)| (item.priority, *index))
            .map(|(index, item)| (index, item.priority));
        match (rx, tx) {
            (Some((index, rp)), Some((_, tp))) if rp <= tp => {
                self.rx_queue.remove_pos(index).is_some()
            }
            (Some((index, _)), None) => self.rx_queue.remove_pos(index).is_some(),
            (_, Some((index, _))) => self.tx_queue.remove_pos(index).is_some(),
            _ => false,
        }
    }

    #[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(never)]
    fn side_chunk_bytes_used(&self) -> usize {
        self.side_transport
            .values()
            .map(|side| side.rx_chunk_bytes)
            .sum()
    }
    fn discard_side_chunk(&mut self, side: RelaySideId, id: u32) -> bool {
        self.side_transport
            .get_mut(&side)
            .is_some_and(|state| state.take_chunk(id, true).is_some())
    }
    fn reclaim_oldest_side_chunk(&mut self, keep: Option<(RelaySideId, u32)>) -> bool {
        let mut victim = None;
        let mut time = u64::MAX;
        for (&side, state) in &self.side_transport {
            let preserved = match keep {
                Some((owner, id)) if owner == side => Some(id),
                _ => None,
            };
            if let Some((id, seen)) = SideChunkAssembly::oldest(&state.rx_chunks, preserved)
                && (victim.is_none() || seen < time)
            {
                time = seen;
                victim = Some((side, id));
            }
        }
        match victim {
            Some((side, id)) => self.discard_side_chunk(side, id),
            None => false,
        }
    }
    fn make_side_chunk_room(
        &mut self,
        side: RelaySideId,
        id: u32,
        index: u16,
        total: u16,
        payload_len: usize,
    ) -> TelemetryResult<()> {
        let budget = self
            .memory
            .max_queue_budget
            .saturating_sub(self.protected_metadata_bytes_used());
        let state = self
            .side_transport
            .get_mut(&side)
            .ok_or(TelemetryError::BadArg)?;
        let existing = state.rx_chunks.get(&id).map(|a| a.total);
        if existing.is_some_and(|old| old != 0 && old != total) {
            state.take_chunk(id, true);
            return Err(TelemetryError::Unpack("side chunk total mismatch"));
        }
        let before = state.rx_chunks.len();
        let room = SideChunkAssembly::make_room(
            &mut state.rx_chunks,
            &mut state.rx_chunk_bytes,
            id,
            index,
            total,
            payload_len,
            budget,
        );
        if before != state.rx_chunks.len() {
            state.rx_chunk_reclaims = state
                .rx_chunk_reclaims
                .saturating_add((before - state.rx_chunks.len()) as u32);
        }
        let cost = room?;
        if cost == 0 {
            return Ok(());
        }
        if existing.is_none()
            && state.rx_chunks.len() >= 4
            && let Some((oldest, _)) = SideChunkAssembly::oldest(&state.rx_chunks, None)
        {
            state.take_chunk(oldest, true);
        }
        while self.shared_queue_bytes_used().saturating_add(cost) > self.memory.max_queue_budget {
            #[cfg(any(feature = "std", feature = "fair-admission"))]
            if let Some(owner) = self.bulk_loads(199).queue_victim(None)
                && self.reclaim_bulk_from(owner, 199, None)
            {
                continue;
            }
            if self.reclaim_oldest_side_chunk(Some((side, id))) {
                continue;
            }
            // The sender retries this fragment, not the earlier accepted ones.
            // Keep the bounded partial assembly while protected queues drain.
            return Err(TelemetryError::Io("memory pressure"));
        }
        Ok(())
    }

    #[inline(never)]
    fn protected_metadata_bytes_used(&self) -> usize {
        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
                }
            })
    }
    #[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.side_chunk_bytes_used())
            .saturating_add(self.protected_metadata_bytes_used())
    }

    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,
        _incoming_priority: u8,
        _incoming: Option<&RelayItem>,
    ) -> TelemetryResult<()> {
        if incoming_cost > self.memory.max_queue_budget {
            return Err(TelemetryError::PacketTooLarge(
                "Item exceeds maximum shared queue budget",
            ));
        }

        #[cfg(any(feature = "std", feature = "fair-admission"))]
        let mut fair_victim = None;
        while self.shared_queue_bytes_used().saturating_add(incoming_cost)
            > self.memory.max_queue_budget
        {
            #[cfg(any(feature = "std", feature = "fair-admission"))]
            {
                let incoming =
                    _incoming.and_then(|data| Self::fair_bulk_owner(data, _incoming_priority));
                let victim = *fair_victim.get_or_insert_with(|| {
                    self.bulk_loads(_incoming_priority.min(199))
                        .queue_victim(incoming)
                });
                if let Some(victim) = victim
                    && self.reclaim_bulk_from(
                        victim,
                        _incoming_priority.min(199),
                        (incoming == Some(victim))
                            .then(|| _incoming.and_then(Self::bulk_stream))
                            .flatten(),
                    )
                {
                    continue;
                }
                if incoming.is_some() {
                    // Best effort cannot replace another peer's sole report,
                    // reliable replay, schema, or protected control.
                    return Err(TelemetryError::Io("shared fair queue saturated"));
                }
            }
            // Incoming work cannot evict the routes required to deliver it.
            // Prefer low-priority packets; refuse admission once only metadata
            // remains rather than treating topology as an ordinary queue.
            if !self.pop_lowest_packet_queue_item(256) && self.pop_reliable_rx_buffered().is_none()
            {
                if _incoming_priority >= 200 && self.reclaim_oldest_side_chunk(None) {
                    continue;
                }
                return Err(TelemetryError::PacketTooLarge(
                    "Item exceeds maximum shared queue budget",
                ));
            }
        }

        Ok(())
    }

    #[cfg(feature = "discovery")]
    fn schema_growth_budget(&self) -> usize {
        // Queued packets are reclaimable. Existing reachability and active
        // reliable reassembly remain protected when admitting a new schema.
        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;
            }
            // Live routes are protected. Growth is admitted transactionally;
            // external registry/config changes cannot justify erasing routes.
            break;
        }
    }

    fn push_rx(&mut self, item: RelayRxItem) -> TelemetryResult<()> {
        self.make_shared_queue_room(item.byte_cost(), item.priority, Some(&item.data))?;
        #[cfg(any(feature = "std", feature = "fair-admission"))]
        {
            #[cfg(any(feature = "std", feature = "fair-admission"))]
            let mut fair_victim = None;
            self.rx_queue.push_back_prioritized_select(
                item,
                |queued| queued.priority,
                |queue, _incoming| {
                    #[cfg(any(feature = "std", feature = "fair-admission"))]
                    {
                        crate::admission_fairness::queue_victim(
                            queue,
                            _incoming,
                            |item| Self::fair_bulk_owner(&item.data, item.priority),
                            |item| item.priority,
                            |item| Self::bulk_stream(&item.data),
                            &mut fair_victim,
                        )
                    }
                },
            )
        }
        #[cfg(not(any(feature = "std", feature = "fair-admission")))]
        {
            self.rx_queue
                .push_back_prioritized(item, |queued| queued.priority)
        }
    }

    fn push_tx(&mut self, item: RelayTxItem) -> TelemetryResult<()> {
        #[cfg(feature = "discovery")]
        {
            // Preserve the transfer identity of an unchanged queued catalog.
            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(), item.priority, Some(&item.data))?;
        #[cfg(any(feature = "std", feature = "fair-admission"))]
        {
            #[cfg(any(feature = "std", feature = "fair-admission"))]
            let mut fair_victim = None;
            self.tx_queue.push_back_prioritized_select(
                item,
                |queued| queued.priority,
                |queue, _incoming| {
                    #[cfg(any(feature = "std", feature = "fair-admission"))]
                    {
                        crate::admission_fairness::queue_victim(
                            queue,
                            _incoming,
                            |item| Self::fair_bulk_owner(&item.data, item.priority),
                            |item| item.priority,
                            |item| Self::bulk_stream(&item.data),
                            &mut fair_victim,
                        )
                    }
                },
            )
        }
        #[cfg(not(any(feature = "std", feature = "fair-admission")))]
        {
            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(), item.priority, None)?;
        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, 0, None)?;
        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, crate::transport_priority(ty), None)?;
        #[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(())
    }
}

/// Relay that fans out packets from one side to all others.
/// - Supports both packed bytes and full Packet.
/// - Has RX & TX queues, like Router.
/// - Uses a Clock for the *_with_timeout APIs, same style as Router.
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 item_type_priority(data: &RelayItem) -> TelemetryResult<(crate::DataType, u8)> {
        let (ty, control) = match data {
            RelayItem::Packet(pkt) => (
                pkt.data_type(),
                pkt.payload().len() <= 16
                    && pkt.endpoints().len() <= 3
                    && crate::small_frozen_control(
                        pkt.data_type(),
                        pkt.wire_target_senders().len(),
                        pkt.wire_shape(),
                    ),
            ),
            RelayItem::Packed(bytes) => {
                let env = wire_format::peek_routing_envelope(bytes.as_ref())?;
                let control = bytes.len() <= 128
                    && bytes.first().is_some_and(|flags| flags & 0x11 == 0)
                    && env.endpoints().take(4).count() <= 3
                    && crate::small_frozen_control(
                        env.ty,
                        env.target_senders().len(),
                        env.wire_shape,
                    );
                (env.ty, control)
            }
        };
        let priority = crate::scheduler_priority(ty);
        Ok((ty, if control { priority.max(254) } else { priority }))
    }

    fn relay_item_priority(data: &RelayItem) -> TelemetryResult<u8> {
        Self::item_type_priority(data).map(|(_, priority)| priority)
    }

    #[inline]
    fn is_transient_tx_error(err: &TelemetryError) -> bool {
        matches!(err, TelemetryError::Io("side tx busy" | "memory pressure"))
    }

    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_transient_tx_error(&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,
        u8,
    )> {
        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,
                    item.priority,
                )
            })
        } 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)
        }
    }

    /// Create a new relay with the given clock.
    pub fn new(clock: Box<dyn Clock + Send + Sync>) -> Self {
        Self::new_with_config(RelayConfig::default(), clock)
    }

    /// Create a new relay with explicit runtime configuration and 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()
    }

    /// Prefer the discovery side with the shortest advertised topology path
    /// to the packet's destination. This prevents a route reflected through
    /// another router from competing equally with the direct route.
    #[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.iter() {
                            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);
    }

    /// Seed adaptive route selection with a transport-measured link probe.
    ///
    /// Call this after a side-specific bring-up probe, or whenever the transport already knows the
    /// duration for a frame. The relay does not emit synthetic probe frames by itself.
    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> {
        Self::hostname_for_wire_address_borrowed_locked(st, address).map(str::to_string)
    }

    #[cfg(feature = "discovery")]
    fn hostname_for_wire_address_borrowed_locked<'a>(
        st: &'a RelayInner,
        address: u32,
    ) -> Option<&'a str> {
        let mut found: Option<&'a str> = None;
        let mut collision = false;
        let mut consider = |candidate: &'a str| {
            if sender_address_u32(candidate) != address {
                return;
            }
            match found {
                None => found = Some(candidate),
                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.iter() {
                    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)
    }

    /// Extract the logical packet ID targeted by an end-to-end reliable ACK item.
    ///
    /// Relay queues can hold either decoded packets or packed frames. This
    /// helper normalizes both forms so relay ACK-routing logic can treat them
    /// uniformly.
    ///
    /// Only relay-visible end-to-end `ReliableAck` packets qualify here.
    /// Unrelated traffic returns `Ok(None)`.
    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();
        // A retry can remain in RX dedupe after its smaller return-path cache
        // entry was evicted. Relearning that path must not restart the replay
        // delay on every retry, or sustained cache churn suppresses it forever.
        let now = self.clock.now_ms();
        let retry_ms = runtime_reliable_retransmit_ms();
        let duplicate = st.recent_rx.contains(&packet_id);
        let last_replay_ms = if duplicate {
            now.saturating_sub(retry_ms)
        } else {
            now
        };
        // A reflected copy is not evidence that the original publisher
        // moved. Keep the first live ingress during the current retry
        // window, otherwise an echo can redirect an already queued ACK into
        // the downstream bus. A retry-window expiry or disabled/removed side
        // still permits alternate-path recovery without explicit operator action.
        let replace_side = !duplicate
            || st
                .reliable_return_routes
                .get(&packet_id)
                .is_none_or(|route| {
                    now.saturating_sub(route.last_replay_ms) >= retry_ms
                        || !Self::side_ref(&st, route.side)
                            .is_ok_and(|side| side.opts.egress_enabled)
                });
        Self::remember_reliable_return_route_locked(&mut st, packet_id);
        st.reliable_return_routes
            .entry(packet_id)
            .and_modify(|route| {
                if replace_side {
                    route.side = side;
                }
            })
            .or_insert(ReliableReturnRouteState {
                side,
                last_replay_ms,
            });
    }

    /// Refresh or insert `packet_id` in the bounded reliable return-route cache.
    ///
    /// The relay uses this cache to route end-to-end acknowledgements back
    /// toward the original live ingress of the corresponding
    /// reliable data packet.
    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) {
                // Publish an aggregate split-horizon topology instead of
                // leaking far-side topology as direct neighbors. Address
                // allocation advertisements still cross transparent relays:
                // static/requested address conflicts must be resolved globally.
                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);
                }
            }

            // Compact/DHCP links can temporarily disagree on whether a frozen
            // destination is represented by hostname hash or assigned address.
            // Preserve autonomous endpoint routing only when discovery gives
            // one unambiguous non-ingress route; never fan an unknown target
            // out across multiple sides.
            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;
                    }
                    // Keep forwarding while any discovered destination sender for this packet
                    // remains unacked, even if topology/schema metadata changed for new packets.
                    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 {
        let boards = Arc::make_mut(&mut sender_state.topology_boards);
        if let Some(idx) = boards.iter().position(|board| board.sender_id == sender_id) {
            return &mut boards[idx];
        }
        boards.push(TopologyBoardNode {
            sender_id: sender_id.to_string(),
            reachable_endpoints: Vec::new(),
            reachable_timesync_sources: Vec::new(),
            connections: Vec::new(),
        });
        boards.last_mut().expect("inserted above")
    }

    #[cfg(feature = "discovery")]
    fn refresh_sender_topology_state(sender_state: &mut DiscoverySenderState) {
        if let Some(boards) = Arc::get_mut(&mut sender_state.topology_boards) {
            discovery::normalize_topology_boards(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 {
                    // Merge the retained view directly. Cloning it first and
                    // cloning again into the output doubled temporary strings.
                    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 {
        // Match router maintenance: unchanged routes require no heap work.
        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,
            ),
        );
        // Compact advertisements can know routes before detailed topology arrives.
        // Preserve that reachability without assigning downstream endpoints to the bridge.
        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);
        }
        // Stable links carry only a small liveness announcement. Route and
        // schema snapshots are emitted on changes or explicit recovery requests.
        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> {
        if pkt.data_type() == crate::DataType::DiscoveryTopology {
            // No decoded graph or retained-side clone is needed for a retry
            // whose payload already describes precisely the retained state.
            // Keep changed topology behind the full admission guard.
            crate::memory_admission::check(512, 128)?;
            let now = self.clock.now_ms();
            let mut st = self.state.lock();
            let canonical = Self::canonical_sender_locked(&st, pkt.sender());
            let Some(route) = st.discovery_routes.get_mut(&side) else {
                return Ok(false);
            };
            let Some(peer) = route.announcers.get_mut(canonical.as_str()) else {
                return Ok(false);
            };
            let Some(full) =
                discovery::topology_payload_matches_retained(pkt.payload(), &peer.topology_boards)
            else {
                return Ok(false);
            };
            peer.has_full_topology |= full;
            peer.last_seen_ms = now;
            route.last_seen_ms = route.last_seen_ms.max(now);
            return Ok(true);
        }
        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);
        }
        // A liveness update cannot add ownership, change identity, or enlarge
        // discovery state. Leave new/changed summaries behind the full guard.
        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;
            }
        }
        // The shared cadence advances once per announcement, regardless of
        // side count. Advancing it once per side silently doubled backoff.
        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) = {
            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(),
                        )
                    }),
            )
        };
        // Full metadata also carries the learned topology. Its working set
        // must never bypass the graph estimate merely because schema is due.
        let (scratch, largest) = if include_schema {
            (scratch.max(8192), largest.max(2048))
        } else {
            (scratch, largest)
        };
        // Snapshot construction may clone several route/schema summaries.
        // Reclaim ordinary queued telemetry before refusing discovery scratch.
        // Priority >= 200 commands, ACKs and metadata stay protected; do not
        // evict routes or reliable receive/reassembly state.
        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);
            }
            // The caller selects one side; keep its plan on the stack.
            let mut per_side = None;
            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;
                }
                // Preserve neighbor liveness even if a reliable discovery
                // baseline is still waiting for its hop ACK.
                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 = Some((
                        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),
                    &current_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, &current_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 = Some((
                    side_id,
                    level,
                    endpoints,
                    timesync_sources,
                    topology,
                    removed_topology,
                    capabilities,
                ));
            }
            per_side
        };
        let mut st = self.state.lock();
        if let Some((
            dst,
            level,
            endpoints,
            timesync_sources,
            topology,
            removed_topology,
            capabilities,
        )) = per_side
        {
            let sender = self.sender_arc();
            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,
                })?;
            }
            // Match the router's phase ordering: route information first,
            // then release snapshot buffers before constructing schema.
            drop((endpoints, timesync_sources, topology, removed_topology));
            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,
                })?;
            }
        }
        Ok(())
    }

    #[cfg(feature = "discovery")]
    fn note_schema_received(&self, side: RelaySideId, sender: &str) {
        let mut st = self.state.lock();
        // Schema snapshots can be forwarded. They must never introduce a new
        // link-local announcer or change shared-bus reliability semantics.
        // If schema precedes topology, bounded recovery will request it again.
        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);
        }
        // Keep periodic discovery lightweight so a hosted relay cannot fill a
        // constrained link with repeated schema snapshots.
        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(());
        }
        // A compact address summary does not clone the complete topology or
        // schema. Account for its ingress-side aggregates instead; charging
        // every new peer for a full snapshot can prevent restart recovery.
        if matches!(
            pkt.data_type(),
            crate::DataType::DiscoveryTopologyRequest | crate::DataType::DiscoverySchemaRequest
        ) {
            crate::memory_admission::check(1024, 256)?;
            // A restarted receiver has no compact-header dictionary even if
            // discovery still has exactly the same routes. Refresh TX only on
            // its ingress link; do not erase RX templates or routing state.
            if let Some(transport) = self.state.lock().side_transport.get_mut(&src) {
                transport.refresh_transmit_header_templates();
            }
            // Like routers, relays retain a split-horizon baseline and must
            // answer a restarting neighbor without waiting for master election.
            {
                let mut st = self.state.lock();
                let throttle = st.discovery_side_throttle.entry(src).or_default();
                throttle.has_sent_full = false;
                // A full reply replaces this derived delta baseline. Release
                // it before snapshot admission, preserving live routing state.
                throttle.last_topology = Vec::new();
            }
            return self.queue_discovery_announce_on_side(
                pkt.data_type() == crate::DataType::DiscoverySchemaRequest,
                false,
                Some(src),
            );
        }
        // Standalone relays forward variable requests; they do not own a
        // subscription registry or decode a schema for this four-byte payload.
        if pkt.data_type() == crate::DataType::ManagedVariableRequest {
            discovery::decode_managed_variable_request(&pkt)?;
            return Ok(());
        }
        let small_topology_scratch = if pkt.data_type() == crate::DataType::DiscoveryTopology
            && pkt.payload().len() <= 128
        {
            let st = self.state.lock();
            let sender = pkt
                .sender()
                .strip_prefix("@addr:")
                .and_then(|value| value.parse::<u32>().ok())
                .and_then(|address| Self::hostname_for_wire_address_borrowed_locked(&st, address))
                .unwrap_or(pkt.sender());
            st.discovery_routes.get(&src).and_then(|route| {
                let peer = route.announcers.get(sender)?;
                let mut shallow = discovery::CloneScratch::default();
                shallow.buffer(route.reachable.len().saturating_mul(4));
                shallow.strings(&route.reachable_timesync_sources);
                for (name, value) in &route.announcers {
                    shallow.buffer(name.len());
                    shallow.buffer(value.reachable.len().saturating_mul(4));
                    shallow.buffer(value.advertised_reachable.len().saturating_mul(4));
                    shallow.strings(&value.reachable_timesync_sources);
                    shallow.strings(&value.advertised_reachable_timesync_sources);
                }
                let entry =
                    core::mem::size_of::<String>() + core::mem::size_of::<DiscoverySenderState>();
                shallow.bytes = shallow
                    .bytes
                    .saturating_add(route.announcers.len().saturating_mul(entry));
                shallow.largest = shallow
                    .largest
                    .max(entry.saturating_mul(11).saturating_add(128));
                let mut deep = discovery::CloneScratch::default();
                if discovery::topology_payload_is_incremental(pkt.payload()) {
                    deep.graph(&peer.topology_boards, pkt.payload().len() / 16);
                }
                Some((
                    2048usize
                        .saturating_add(shallow.bytes.saturating_mul(3))
                        .saturating_add(deep.bytes)
                        .saturating_add(pkt.payload().len().saturating_mul(4)),
                    shallow.largest.max(deep.largest).max(2048),
                ))
            })
        } else {
            None
        };
        let retained_summary = if small_topology_scratch.is_none()
            && 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 (decode_bytes, largest_block) = if let Some(scratch) = small_topology_scratch {
            scratch
        } else {
            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);
            (decode_bytes, largest_block)
        };
        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() {
                    Arc::make_mut(&mut sender_state.topology_boards)
                        .retain(|board| board.sender_id != leaving);
                    for board in Arc::make_mut(&mut 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
            }
        };
        // Parse fallible payloads and resolve aliases before moving retained state.
        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());
                }
                // This packet is an aggregate route summary, not a claim that
                // the announcing bridge owns every endpoint behind it.
                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");
                // Empty announcements are slow-link keepalives, not endpoint
                // withdrawals. Preserve ownership until an explicit topology
                // update, leave, or expiry removes it. The common receive tail
                // still refreshes last_seen_ms without scheduling rediscovery.
                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_shared_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 {
                        // Do not let a lost packet-request pin a partially
                        // acknowledged entry forever. Allow one interval for
                        // the explicit request, then return to bounded replay.
                        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(())
    }

    /// Compute a de-dupe hash for a QueueItem.
    /// Uses packet ID for Packet items, and attempts to extract packet ID from
    /// packed bytes. If extraction fails, hashes raw bytes as a fallback.
    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) => {
                        // Fallback: if bytes are malformed (or compression feature mismatch),
                        // hash raw bytes so we can still dedupe identical network duplicates.
                        let h: u64 = 0x9E37_79B9_7F4A_7C15;
                        hash_bytes_u64(h, bytes.as_ref())
                    }
                }
            }
        }
    }

    /// Compute a dedupe ID for an incoming RelayRxItem.
    /// Note: we intentionally do *not* include `src` so that the same
    /// packet coming from multiple sides is only processed once.
    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) && self.item_target_senders(&item.data)?.is_empty())
            || 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)))
    }

    /// Register a side whose TX callback consumes packed packet bytes.
    ///
    /// Returns the side id later used for ingress APIs such as `rx_packed_from_side`.
    /// The default options disable the relay's per-link reliable framing on this side.
    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())
    }

    /// Register a packed side with bounded-frame transport enabled.
    ///
    /// `max_frame_bytes == 0` leaves frames unbounded.
    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),
        )
    }

    /// Register a packed-output side with explicit side options.
    ///
    /// `opts.reliable_enabled` enables relay-managed per-hop ACK/retransmit behavior on this side.
    /// `opts.link_local_enabled` gates link-local-only forwarding and discovery use of this side.
    /// `ingress_enabled` and `egress_enabled` set the initial directional policy.
    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, Box::<SideRuntimeStatsInner>::default());
        st.side_transport
            .insert(id, Box::<SideTransportState>::default());
        #[cfg(feature = "discovery")]
        Self::note_discovery_topology_change_locked(&mut st, self.clock.now_ms());
        id
    }

    /// Register a side whose TX callback receives decoded [`Packet`] values.
    ///
    /// Packet-output sides do not preserve the relay's packed reliable hop framing, so use a
    /// packed side when this hop should participate in relay-managed per-link reliability.
    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())
    }

    /// Register a packet-output side with explicit side options.
    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, Box::<SideRuntimeStatsInner>::default());
        st.side_transport
            .insert(id, Box::<SideTransportState>::default());
        #[cfg(feature = "discovery")]
        Self::note_discovery_topology_change_locked(&mut st, self.clock.now_ms());
        id
    }

    /// Remove a side while keeping existing side IDs stable.
    ///
    /// `side` must be an id returned by one of the `add_side_*` calls. Remaining side ids are not
    /// renumbered.
    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();
        }
        // Preserve capacity for link churn. It is bounded by the maximum number
        // of sides seen by this relay and is reclaimed when the relay drops.
        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(())
    }

    /// Enable or disable ingress processing for a registered side.
    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(())
    }

    /// Enable or disable egress toward a registered side.
    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(())
    }

    /// Set the route-selection policy for traffic originating from `src`.
    ///
    /// `src == None` targets locally-originated relay traffic such as discovery output.
    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)?;
        }
        // Preserve explicit Fanout so discovery does not replace it with its
        // adaptive single-path default. Clearing is a separate operation.
        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(())
    }

    /// Clear a source-specific route-selection override.
    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(())
    }

    /// Set the weighted-routing weight from `src` toward `dst`.
    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(())
    }

    /// Clear a previously configured weighted-routing weight override.
    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(())
    }

    /// Set the failover priority from `src` toward `dst`.
    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(())
    }

    /// Clear a previously configured failover priority override.
    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(())
    }

    /// Allow or block routing from `src` toward `dst`.
    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(())
    }

    /// Allow or block routing for a specific `DataType` from `src` toward `dst`.
    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(())
    }

    /// Clear a typed route override for the `(src, ty, dst)` triple.
    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(())
    }

    /// Clear a non-typed route override so the relay falls back to default behavior.
    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")]
    /// Queues an immediate discovery announcement for this relay.
    pub fn announce_discovery(&self) -> TelemetryResult<()> {
        self.queue_discovery_announce(true, false)
    }

    #[cfg(feature = "discovery")]
    /// Broadcast that this relay is leaving so peers can prune topology immediately.
    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")]
    /// Polls discovery state and queues an announce if the cadence says one is due.
    pub fn poll_discovery(&self) -> TelemetryResult<bool> {
        self.poll_discovery_announce()
    }

    #[cfg(feature = "discovery")]
    /// Exports the relay's current discovered topology snapshot.
    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.as_ref().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 empty_stats = SideRuntimeStatsInner::default();
            let stats = st
                .side_runtime_stats
                .get(&side_id)
                .map(Box::as_ref)
                .unwrap_or(&empty_stats);
            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
                .iter()
                .map(|(&ty, item)| {
                    let item = item.counters();
                    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_chunk_reclaims: st
                    .side_transport
                    .get(&side_id)
                    .map_or(0, |state| u64::from(state.rx_chunk_reclaims)),
                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,
        }
    }

    /// Export current relay memory usage/layout as JSON for profiling.
    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())
    }

    /// Enqueue packed bytes that originated from `src` into the relay RX queue.
    ///
    /// Note: `Arc::from(bytes)` allocates and copies `len` bytes into a new `crate::SharedBytes`.
    /// This is still “fast enough” for many cases, but it is not allocation-free / ISR-safe.
    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,
        })
    }

    /// Enqueue a full packet that originated from `src` into the relay RX queue.
    ///
    /// The packet is wrapped in `Arc<Packet>` so fanout can clone the pointer cheaply.
    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,
        })
    }

    /// Clear both RX and TX queues.
    pub fn clear_queues(&self) {
        let mut st = self.state.lock();
        st.rx_queue.clear();
        st.tx_queue.clear();
        st.tx_priority_burst = 0;
    }

    /// Clear only RX queue.
    pub fn clear_rx_queue(&self) {
        let mut st = self.state.lock();
        st.rx_queue.clear();
    }

    /// Clear only TX queue.
    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();
    }

    /// Internal: expand one RX item into TX items for all other sides.
    ///
    /// Fanout is cheap: the `RelayItem` is cloned (Arc bump) and reused across all destinations.
    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()) || !pkt.wire_target_senders().is_empty())
                    && !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) || env.target_senders().len() != 0)
                    && !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)? {
            // Already fanned out this packet recently; skip.
            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 {
            // Encoding has already inserted the TX template, but a busy or
            // failed handler did not accept its bootstrap. Keep dictionary
            // eviction history and force a full refresh on the next attempt.
            // Never clear RX templates or allocate during this recovery.
            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 {
            // Canonical bytes keep a partially accepted transfer independent
            // of dictionary churn while other traffic is serviced.
            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 {
            // Shared buses cannot retain one coherent dictionary across
            // independent producers. Do not build owned templates or wrap
            // canonical bytes when no dictionary can be used.
            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()
                || frame.envelope.target_senders().len() != 0
                || 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));
            }
        }

        #[cfg(any(feature = "std", feature = "fair-admission"))]
        if opts.header_template_enabled && crate::memory_admission::supports_work_kinds() {
            let (template, ..) = extract_side_header_template_view(raw.as_ref())?;
            let cached = self
                .state
                .lock()
                .side_transport
                .get(&side)
                .is_some_and(|transport| {
                    transport.tx_template_ids.contains_key(&template.hash)
                        && transport.tx_templates.contains_key(&template.hash)
                });
            if !cached
                && matches!(
                    crate::memory_admission::check_receive(raw.as_ref()),
                    Err(TelemetryError::Io("memory pressure"))
                )
            {
                // Compression is optional. Retain the canonical shared frame
                // instead of allocating a dictionary entry under pressure.
                // Oversized frames returned through the normal chunker above.
                self.state
                    .lock()
                    .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())?;
        // Reliable frames never use dictionary compression. Send the native,
        // self-describing wire packet without allocating/retaining an unused
        // template on either peer. Raw packets are accepted by both decoders.
        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();
                // Discovery must bootstrap even when a receiver lost its dictionary.
                #[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());
            // A previous transmitted frame is not a confirmed receive base.
            // Absolute timestamps preserve identity after loss or reordering.
            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() {
            let side = side.as_mut();
            let reclaimed =
                SideChunkAssembly::expire(&mut side.rx_chunks, &mut side.rx_chunk_bytes, now);
            side.rx_chunk_reclaims = side.rx_chunk_reclaims.saturating_add(reclaimed);
        }
    }

    // A validated frame from a previously discovered direct peer proves that
    // its link is alive, even when allocation admission refuses its payload.
    // Update existing timestamps only: no allocation, schema merge or new route.
    #[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 {
                // Ambiguous numeric identities must be resolved by discovery.
                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);
    }

    /// Consume only an exact unsequenced topology retry on its ingress link.
    /// CRC, shape, payload length and reliable-header restrictions are checked
    /// before this path; it cannot bypass receipts, update a graph or allocate.
    #[cfg(feature = "discovery")]
    fn consume_known_packed_topology(
        &self,
        side: RelaySideId,
        bytes: &[u8],
    ) -> TelemetryResult<bool> {
        let Some((address, payload)) =
            wire_format::peek_plain_discovery_payload(bytes, crate::DataType::DiscoveryTopology)?
        else {
            return Ok(false);
        };
        let now = self.clock.now_ms();
        let mut st = self.state.lock();
        let Some(route) = st.discovery_routes.get_mut(&side) else {
            return Ok(false);
        };
        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 Ok(false);
                }
                found = Some(peer);
            }
        }
        let Some(peer) = found else {
            return Ok(false);
        };
        let Some(full) =
            discovery::topology_payload_matches_retained(payload, &peer.topology_boards)
        else {
            return Ok(false);
        };
        peer.has_full_topology |= full;
        peer.last_seen_ms = now;
        route.last_seen_ms = route.last_seen_ms.max(now);
        Ok(true)
    }

    #[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_discovery_payload(bytes, crate::DataType::DiscoverySchema)?
        else {
            return Ok(false);
        };
        if !discovery::discovery_schema_payload_is_fully_known(payload)? {
            return Ok(false);
        }
        // Discovery schemas are always sent as full refreshes. No compact
        // dictionary entry or second owned payload is needed for this no-op.
        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)
    }

    // A two-sided, fully warmed packed bridge needs no new large metadata.
    // Re-check before dispatch: schemas/templates/routes can change while queued.
    #[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 & 0x11 != 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
            // A matching inline schema is immutable metadata on the existing
            // shared frame. It needs no owned decoder state. Unknown or changed
            // shapes still require full admission and payload decoding.
            || frame.envelope.wire_shape.is_some_and(|shape| {
                shape != crate::message_meta(ty).element
            })
            || !(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)
                })
            {
                // An optional compression entry may have been evicted while
                // routing/statistics remain warm. Canonical forwarding needs
                // no replacement entry, but must fit the transport and belong
                // to an already discovered ingress peer.
                #[cfg(any(feature = "std", feature = "fair-admission"))]
                if (entry.opts.max_frame_bytes == 0 || bytes.len() <= entry.opts.max_frame_bytes)
                    && st.discovery_routes.get(&side).is_some_and(|route| {
                        route.announcers.iter().any(|(name, peer)| {
                            peer.advertised_address == Some(frame.envelope.source_address)
                                || sender_address_u32(name) == frame.envelope.source_address
                        })
                    })
                {
                    continue;
                }
                return Ok(false);
            }
        }
        Ok(true)
    }
    #[cfg(not(feature = "discovery"))]
    fn cached_transit_ready(&self, _side: RelaySideId, _bytes: &[u8]) -> TelemetryResult<bool> {
        Ok(false)
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn reclaim_for_receive(&self, bytes: &[u8], victim: &mut Option<(u32, u8)>) -> bool {
        let Some((incoming, priority)) = crate::admission_fairness::incoming(bytes) else {
            return false;
        };
        self.reclaim_for_incoming(incoming, priority, victim)
    }
    #[cfg(any(feature = "std", feature = "fair-admission"))]
    fn reclaim_for_incoming(
        &self,
        incoming: u32,
        priority: u8,
        victim: &mut Option<(u32, u8)>,
    ) -> bool {
        let mut st = self.state.lock();
        let (owner, priority) = if let Some(selected) = *victim {
            selected
        } else {
            let Some(owner) = st.competing_bulk_owner(incoming, priority) else {
                return false;
            };
            *victim = Some((owner, priority));
            (owner, priority)
        };
        st.reclaim_bulk_from(owner, priority, None)
    }
    fn admit_received_frame(&self, side: RelaySideId, bytes: &[u8]) -> TelemetryResult<()> {
        #[cfg(any(feature = "std", feature = "fair-admission"))]
        let mut fair_victim = None;
        if crate::memory_admission::supports_work_kinds()
            && self.cached_transit_ready(side, bytes)?
        {
            let result = crate::memory_admission::check_cached_transit(bytes.len());
            #[cfg(any(feature = "std", feature = "fair-admission"))]
            let result = {
                let mut result = result;
                while matches!(result, Err(TelemetryError::Io("memory pressure"))) {
                    if !self.reclaim_for_receive(bytes, &mut fair_victim) {
                        break;
                    }
                    result = crate::memory_admission::check_cached_transit(bytes.len());
                }
                result
            };
            result
        } 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
                                        || crate::small_frozen_control(
                                            frame.envelope.ty,
                                            frame.envelope.target_senders().len(),
                                            frame.envelope.wire_shape,
                                        )
                                })
                        });
                        if !protected {
                            #[cfg(any(feature = "std", feature = "fair-admission"))]
                            if self.reclaim_for_receive(bytes, &mut fair_victim) {
                                continue;
                            }
                            return Err(TelemetryError::Io("memory pressure"));
                        }
                        // Queue priority cannot help before the wire copy is
                        // admitted. Reclaim bulk payloads, preserving control
                        // traffic and every learned route/schema.
                        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);
            }
            #[cfg(feature = "discovery")]
            if self.consume_known_packed_topology(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"))?;
                // The wrapper CRC and known template identify this peer even
                // if payload allocation is refused. No payload copy is needed.
                #[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);
                // Wrapper CRC and the retained full template are validated
                // before compact reconstruction. Preserve the same control
                // priority as canonical ingress while reclaiming only bulk.
                let protected = template
                    .data_type()
                    .is_some_and(|ty| message_priority(ty) >= 200);
                let ack = template.is_ack();
                #[cfg(any(feature = "std", feature = "fair-admission"))]
                let mut fair_incoming = None;
                #[cfg(any(feature = "std", feature = "fair-admission"))]
                let mut fair_victim = None;
                loop {
                    match crate::memory_admission::check_compact_reconstruction(
                        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"));
                            }
                        }
                        Err(TelemetryError::Io("memory pressure")) => {
                            // Compact reports need the same fairness before
                            // allocation as canonical ingress. Otherwise they
                            // never reach the canonical admission/reclaim path.
                            #[cfg(any(feature = "std", feature = "fair-admission"))]
                            if let Some((incoming, priority)) =
                                *fair_incoming.get_or_insert_with(|| {
                                    crate::admission_fairness::compact_incoming(&template)
                                })
                                && self.reclaim_for_incoming(incoming, priority, &mut fair_victim)
                            {
                                continue;
                            }
                            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);
                    }
                }
                // Byte reconstruction is not decoder/forwarding admission.
                // Check the canonical frame before queueing, and recheck at
                // dispatch if routes, schemas or available memory have changed.
                if crate::memory_admission::supports_work_kinds() {
                    self.admit_received_frame(side, frame.as_ref())?;
                }
                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 now_ms = self.clock.now_ms();
                    st.make_side_chunk_room(side, transfer_id, index, total, body.len() - 8)?;
                    let side_state = st
                        .side_transport
                        .get_mut(&side)
                        .ok_or(TelemetryError::BadArg)?;
                    let entry = side_state.rx_chunks.entry(transfer_id).or_default();
                    entry.last_seen_ms = now_ms;
                    entry.total = total;
                    let added = entry.insert_with(index, || {
                        crate::shared_bytes::convert(Arc::<[u8]>::from(&body[8..]))
                    });
                    side_state.rx_chunk_bytes = side_state.rx_chunk_bytes.saturating_add(added);
                    if entry.received.len() == usize::from(total) {
                        let entry = side_state
                            .take_chunk(transfer_id, false)
                            .ok_or(TelemetryError::Unpack("side chunk missing"))?;
                        Some(entry.assemble().inspect_err(|_| {
                            side_state.rx_chunk_reclaims =
                                side_state.rx_chunk_reclaims.saturating_add(1);
                        })?)
                    } 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")),
        }
    }

    /// Helper: call a TX handler with the best representation we have.
    /// - Packet handler + Packet item: direct.
    /// - Packed handler + Packed item: direct.
    /// - Packet handler + Packed item: unpack for this call.
    /// - Packed handler + Packet item: pack for this call.
    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) {
            // Fast paths
            (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),

            // Conversion paths
            (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
                ) {
                    // Hop reliability can be disabled on a physical bridge,
                    // but the publisher's end-to-end ACK must still cross it.
                    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)))
            }
        }
    }

    /// Drain the RX queue fully, expanding to TX items.
    #[inline]
    pub fn process_rx_queue(&self) -> TelemetryResult<()> {
        self.process_rx_queue_with_timeout(0)
    }

    /// Drain the TX queue fully, invoking per-side TX handlers.
    ///
    /// If called from inside a side TX callback, this becomes a no-op so relay TX handlers cannot
    /// recurse into nested queue drains on the same stack.
    #[inline]
    pub fn process_tx_queue(&self) -> TelemetryResult<()> {
        self.process_tx_queue_with_timeout(0)
    }

    /// Drain RX then TX queues fully (one pass).
    #[inline]
    pub fn process_all_queues(&self) -> TelemetryResult<()> {
        self.process_all_queues_with_timeout(0)
    }

    /// Process the TX queue for up to `timeout_ms` milliseconds.
    ///
    /// `timeout_ms == 0` drains fully. If called from inside a side TX callback, this becomes a
    /// no-op so relay TX handlers cannot recurse into nested queue drains on the same stack.
    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, priority)) = 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_transient_tx_error(&e) => {
                    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(())
    }

    /// Process RX queue with timeout.
    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(())
    }

    /// Process RX and TX queues interleaved for up to `timeout_ms` milliseconds.
    ///
    /// `timeout_ms == 0` drains fully. If called from inside a side TX callback, this becomes a
    /// no-op so relay TX handlers cannot recurse into nested queue drains on the same stack.
    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)
    }

    /// Drain RX/TX work without generating periodic discovery announcements.
    /// Reliable retries remain active. Call `periodic` or `poll_discovery`
    /// separately on a regular cadence. Nested side callbacks remain a no-op.
    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())?;
            }

            // One TX/RX pair is atomic for fairness under a short budget.
            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;
            }

            // Then send out TX
            let sent_one =
                if let Some((src, dst, handler, opts, data, priority)) = self.pop_ready_tx_item() {
                    match self.send_tx_item(src, dst, handler, opts, data.clone()) {
                        Ok(sent) => sent,
                        Err(error) if Self::is_transient_tx_error(&error) => {
                            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(())
    }

    /// Runs one application-loop maintenance cycle.
    ///
    /// This polls built-in discovery when that feature is compiled in, then drains queued RX/TX
    /// work for up to `timeout_ms` milliseconds.
    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 {
            // Incompressible bytes ensure this really uses several chunks.
            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();
        // One transfer ID and consecutive fragment indices prove prefix reuse
        // did not create incomplete transfers or duplicate accepted fragments.
        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;

#[cfg(all(test, feature = "std"))]
mod frozen_control_delivery_tests {
    use super::*;
    use crate::router::{NetworkVariablePermissions, Router, RouterConfig, RouterSideOptions};
    use core::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
    use std::sync::Mutex;

    fn control() -> Packet {
        let ep = crate::config::register_endpoint("FROZEN_RELAY_CONTROL", false).unwrap();
        let ty = crate::config::register_data_type_with_description(
            "FROZEN_RELAY_CONTROL",
            "",
            crate::MessageElement::Static(
                1,
                crate::MessageDataType::UInt8,
                crate::MessageClass::Data,
            ),
            &[ep],
            crate::ReliableMode::None,
            16,
        )
        .unwrap();
        Packet::new_with_wire_contract(
            ty,
            &[ep],
            "GS",
            1,
            7,
            Arc::<[u8]>::from([1]),
            None,
            Arc::from([Relay::sender_hash("FC")]),
        )
        .unwrap()
    }

    #[test]
    fn foreign_frozen_control_retains_priority_and_full_reserved_admission() {
        let packet = control();
        let bytes = wire_format::pack_packet(&packet);
        let relay = Relay::new(Box::new(|| 0u64));
        let side = relay.add_side_packed("radio", |_| Ok(()));
        relay.rx_packed_from_side(side, &bytes).unwrap();
        relay.clear_queues();
        let foreign = Router::new(RouterConfig::new([]).with_sender("FOREIGN"));
        let foreign_side = foreign.add_side_packed("radio", |_| Ok(()));
        foreign
            .rx_packed_queue_from_side(&bytes, foreign_side)
            .unwrap();
        foreign.clear_queues();
        assert_eq!(
            Relay::relay_item_priority(&RelayItem::Packet(Arc::new(packet))).unwrap(),
            254
        );
        assert_eq!(
            Relay::relay_item_priority(&RelayItem::Packed(crate::shared_bytes::convert(
                bytes.clone()
            )))
            .unwrap(),
            254
        );
        extern "C" fn protected_only(additional: usize, largest: usize, kind: u32) -> bool {
            kind == crate::memory_admission::WORK_SMALL_PROTECTED
                && (2048..=2432).contains(&additional)
                && largest <= 160
        }
        struct Reset;
        impl Drop for Reset {
            fn drop(&mut self) {
                crate::memory_admission::set_probe_v2(None);
            }
        }
        let _reset = Reset;
        crate::memory_admission::set_probe_v2(Some(protected_only));
        assert_eq!(wire_format::unpack_packet(&bytes).unwrap().payload(), &[1]);
        relay.rx_packed_from_side(side, &bytes).unwrap();
        foreign
            .rx_packed_queue_from_side(&bytes, foreign_side)
            .unwrap();
        let mut corrupt = bytes.to_vec();
        *corrupt.last_mut().unwrap() ^= 1;
        assert!(relay.rx_packed_from_side(side, &corrupt).is_err());
    }

    #[test]
    fn best_effort_schema_frozen_contract_retries_through_standalone_relay() {
        let packet = control();
        let bytes = wire_format::pack_packet(&packet);
        let now = Arc::new(AtomicU64::new(0));
        let clock = now.clone();
        let relay = Relay::new(Box::new(move || clock.load(Ordering::Relaxed)));
        let upstream = Arc::new(Mutex::new(Vec::<Vec<u8>>::new()));
        let output = upstream.clone();
        let radio = relay.add_side_packed_with_options(
            "radio",
            move |bytes| {
                output.lock().unwrap().push(bytes.to_vec());
                Ok(())
            },
            RelaySideOptions {
                header_template_enabled: false,
                reliable_enabled: false,
                ..Default::default()
            },
        );
        let downstream = Arc::new(Mutex::new(Vec::<Vec<u8>>::new()));
        let output = downstream.clone();
        let can = relay.add_side_packed_with_options(
            "CAN",
            move |bytes| {
                output.lock().unwrap().push(bytes.to_vec());
                Ok(())
            },
            RelaySideOptions {
                header_template_enabled: false,
                reliable_enabled: false,
                ..Default::default()
            },
        );
        relay.set_route(Some(radio), can, true).unwrap();
        relay.set_route(Some(can), radio, true).unwrap();
        let child = Router::new(RouterConfig::new([]).with_sender("FC"));
        child
            .enable_network_variable(packet.data_type(), NetworkVariablePermissions::READ_ONLY)
            .unwrap();
        let effects = Arc::new(AtomicUsize::new(0));
        let count = effects.clone();
        child
            .on_network_variable_update(packet.data_type(), move |_| {
                count.fetch_add(1, Ordering::Relaxed);
                Ok(())
            })
            .unwrap();
        let receipts = Arc::new(Mutex::new(Vec::<Vec<u8>>::new()));
        let output = receipts.clone();
        let child_side = child.add_side_packed_with_options(
            "CAN",
            move |bytes| {
                output.lock().unwrap().push(bytes.to_vec());
                Ok(())
            },
            RouterSideOptions {
                header_template_enabled: false,
                reliable_enabled: false,
                ..Default::default()
            },
        );
        relay.clear_queues();
        child.clear_queues();
        for attempt in 0..3 {
            now.store(
                u64::from(crate::config::runtime_tuning_config().reliable_retransmit_ms) * attempt,
                Ordering::Relaxed,
            );
            relay.rx_packed_from_side(radio, &bytes).unwrap();
            relay.drain_queues_with_timeout(0).unwrap();
            let frames = core::mem::take(&mut *downstream.lock().unwrap());
            let values: Vec<_> = frames
                .into_iter()
                .filter(|bytes| wire_format::peek_envelope(bytes).unwrap().ty == packet.data_type())
                .collect();
            assert!(
                !values.is_empty(),
                "frozen Off-schema retry was swallowed by relay dedupe"
            );
            if attempt == 0 {
                continue;
            } // The first data transmission is lost.
            for bytes in values {
                child.rx_packed_queue_from_side(&bytes, child_side).unwrap();
            }
            child.drain_queues_with_timeout(0).unwrap();
            for bytes in core::mem::take(&mut *receipts.lock().unwrap()) {
                relay.rx_packed_from_side(can, &bytes).unwrap();
            }
            relay.drain_queues_with_timeout(0).unwrap();
            assert_eq!(
                effects.load(Ordering::Relaxed),
                1,
                "receipt retry repeated application effect"
            );
            assert!(
                core::mem::take(&mut *upstream.lock().unwrap())
                    .iter()
                    .any(|bytes| {
                        let ack = wire_format::unpack_packet(bytes).unwrap();
                        ack.data_type() == crate::DataType::ReliableAck
                            && ack.payload() == packet.packet_id().to_le_bytes()
                    }),
                "the endpoint receipt did not return through the relay"
            );
        }
    }
}

#[cfg(all(test, feature = "std"))]
mod bounded_fair_queue_tests {
    use super::*;
    #[test]
    fn small_canonical_reports_survive_fast_senders_fixed_ring_bursts() {
        let ep = crate::config::register_endpoint("RELAY_SMALL_QUEUE_DEST", false).unwrap();
        let ty = crate::config::register_data_type_with_description(
            "RELAY_SMALL_QUEUE_REPORT",
            "",
            crate::MessageElement::Static(
                4,
                crate::MessageDataType::UInt8,
                crate::MessageClass::Data,
            ),
            &[ep],
            crate::ReliableMode::None,
            0,
        )
        .unwrap();
        for tx in [false, true] {
            for slow_last in [false, true] {
                let node = Relay::new(Box::new(|| 0));
                let make = |sender| {
                    RelayItem::Packed(crate::shared_bytes::convert(wire_format::pack_packet(
                        &Packet::new(ty, &[ep], sender, 1, Arc::<[u8]>::from([1; 4])).unwrap(),
                    )))
                };
                let rx = |data| RelayRxItem {
                    src: 0,
                    data,
                    priority: 0,
                };
                let tx_item = |data| RelayTxItem {
                    src: None,
                    dst: 1,
                    data,
                    priority: 0,
                    retry_after_ms: 0,
                };
                let mut st = node.state.lock();
                st.memory.max_queue_budget = usize::MAX;
                if tx {
                    st.tx_queue = BoundedDeque::new(8192, 3 * size_of::<RelayTxItem>(), 1.0);
                } else {
                    st.rx_queue = BoundedDeque::new(8192, 3 * size_of::<RelayRxItem>(), 1.0);
                }
                let order = if slow_last {
                    ["FAST", "FAST", "SLOW"]
                } else {
                    ["SLOW", "FAST", "FAST"]
                };
                for sender in order.into_iter().chain(core::iter::repeat_n("FAST", 32)) {
                    if tx {
                        st.push_tx(tx_item(make(sender))).unwrap();
                    } else {
                        st.push_rx(rx(make(sender))).unwrap();
                    }
                }
                let slow = wire_format::source_address_for_sender("SLOW");
                let retained = if tx {
                    st.tx_queue.iter().any(|item| {
                        RelayInner::fair_bulk_owner(&item.data, item.priority) == Some(slow)
                    })
                } else {
                    st.rx_queue.iter().any(|item| {
                        RelayInner::fair_bulk_owner(&item.data, item.priority) == Some(slow)
                    })
                };
                assert!(
                    retained,
                    "small report was displaced: tx={tx}, slow_last={slow_last}"
                );
            }
        }
    }

    #[test]
    fn slow_stream_survives_same_board_fast_stream_bursts() {
        let ep = crate::config::register_endpoint("RELAY_STREAM_QUEUE_DEST", false).unwrap();
        let ty = crate::config::register_data_type_with_description(
            "RELAY_STREAM_QUEUE_FAST",
            "",
            crate::MessageElement::Static(
                64,
                crate::MessageDataType::UInt8,
                crate::MessageClass::Data,
            ),
            &[ep],
            crate::ReliableMode::None,
            0,
        )
        .unwrap();
        let slow_ty = crate::config::register_data_type_with_description(
            "RELAY_STREAM_QUEUE_SLOW",
            "",
            crate::MessageElement::Static(
                64,
                crate::MessageDataType::UInt8,
                crate::MessageClass::Data,
            ),
            &[ep],
            crate::ReliableMode::None,
            0,
        )
        .unwrap();
        for mode in 0..3 {
            for tx in [false, true] {
                for slow_last in [false, true] {
                    let node = Relay::new(Box::new(|| 0));
                    let make = |sender| {
                        let packet = Packet::new(
                            if sender == "SLOW" { slow_ty } else { ty },
                            &[ep],
                            "SAME_BOARD",
                            1,
                            Arc::<[u8]>::from([1; 64]),
                        )
                        .unwrap();
                        if mode == 1 {
                            RelayItem::Packet(packet.into())
                        } else {
                            RelayItem::Packed(crate::shared_bytes::convert(
                                wire_format::pack_packet(&packet),
                            ))
                        }
                    };
                    let rx = |data| RelayRxItem {
                        src: 0,
                        data,
                        priority: 0,
                    };
                    let tx_item = |data| RelayTxItem {
                        src: None,
                        dst: 1,
                        data,
                        priority: 0,
                        retry_after_ms: 0,
                    };
                    let mut st = node.state.lock();
                    st.memory.max_queue_budget = usize::MAX;
                    if mode == 0 {
                        if tx {
                            st.tx_queue =
                                BoundedDeque::new(8192, 3 * size_of::<RelayTxItem>(), 1.0);
                        } else {
                            st.rx_queue =
                                BoundedDeque::new(8192, 3 * size_of::<RelayRxItem>(), 1.0);
                        }
                    } else if mode == 1 {
                        let cost = if tx {
                            tx_item(make("SLOW")).byte_cost()
                                + 2 * tx_item(make("FAST")).byte_cost()
                        } else {
                            rx(make("SLOW")).byte_cost() + 2 * rx(make("FAST")).byte_cost()
                        };
                        if tx {
                            st.tx_queue = BoundedDeque::new(cost, cost, 2.0);
                        } else {
                            st.rx_queue = BoundedDeque::new(cost, cost, 2.0);
                        }
                    }
                    let order = if slow_last {
                        ["FAST", "FAST", "SLOW"]
                    } else {
                        ["SLOW", "FAST", "FAST"]
                    };
                    for (index, sender) in order
                        .into_iter()
                        .chain(core::iter::repeat_n("FAST", 32))
                        .enumerate()
                    {
                        if mode == 2 && index == 3 {
                            st.memory.max_queue_budget = st.shared_queue_bytes_used();
                        }
                        if tx {
                            st.push_tx(tx_item(make(sender))).unwrap();
                        } else {
                            st.push_rx(rx(make(sender))).unwrap();
                        }
                    }
                    let is_slow = |data: &RelayItem| match data {
                        RelayItem::Packet(p) => p.data_type() == slow_ty,
                        RelayItem::Packed(b) => {
                            wire_format::scheduling_type(b.as_ref()) == Some(slow_ty)
                        }
                    };
                    let retained = if tx {
                        st.tx_queue.iter().any(|item| {
                            let data = &item.data;
                            is_slow(data)
                        })
                    } else {
                        st.rx_queue.iter().any(|item| is_slow(&item.data))
                    };
                    assert!(
                        retained,
                        "slow stream displaced: mode={mode}, tx={tx}, slow_last={slow_last}"
                    );
                }
            }
        }
    }

    #[test]
    fn fast_producer_preserves_slow_reports_at_each_queue_boundary() {
        let ep = crate::config::register_endpoint("RELAY_QUEUE_FAIR_DEST", false).unwrap();
        let ty = crate::config::register_data_type_with_description(
            "RELAY_QUEUE_FAIR_REPORT",
            "",
            crate::MessageElement::Static(
                64,
                crate::MessageDataType::UInt8,
                crate::MessageClass::Data,
            ),
            &[ep],
            crate::ReliableMode::None,
            0,
        )
        .unwrap();
        for mode in 0..3 {
            for tx in [false, true] {
                for slow_last in [false, true] {
                    for next_sender in ["FAST", "NEW_SLOW"] {
                        let router = Relay::new(Box::new(|| 0));
                        let make = |sender| {
                            RelayItem::Packet(Arc::new(
                                Packet::new(ty, &[ep], sender, 1, Arc::<[u8]>::from([1; 64]))
                                    .unwrap(),
                            ))
                        };
                        let rx = |data| RelayRxItem {
                            src: 0,
                            data,
                            priority: 0,
                        };
                        let tx_item = |data| RelayTxItem {
                            src: None,
                            dst: 1,
                            data,
                            priority: 0,
                            retry_after_ms: 0,
                        };
                        let order = if slow_last {
                            ["FAST", "FAST", "SLOW"]
                        } else {
                            ["SLOW", "FAST", "FAST"]
                        };
                        let mut st = router.state.lock();
                        st.memory.max_queue_budget = usize::MAX;
                        let cost = if tx {
                            order.iter().map(|s| tx_item(make(s)).byte_cost()).sum()
                        } else {
                            order.iter().map(|s| rx(make(s)).byte_cost()).sum()
                        };
                        if mode == 0 {
                            if tx {
                                st.tx_queue = BoundedDeque::new(cost, cost, 2.0);
                            } else {
                                st.rx_queue = BoundedDeque::new(cost, cost, 2.0);
                            }
                        } else if mode == 1 {
                            if tx {
                                st.tx_queue =
                                    BoundedDeque::new(cost * 10, 3 * size_of::<RelayTxItem>(), 1.0);
                            } else {
                                st.rx_queue =
                                    BoundedDeque::new(cost * 10, 3 * size_of::<RelayRxItem>(), 1.0);
                            }
                        }
                        for sender in order {
                            if tx {
                                st.push_tx(tx_item(make(sender))).unwrap();
                            } else {
                                st.push_rx(rx(make(sender))).unwrap();
                            }
                        }
                        if mode == 2 {
                            st.memory.max_queue_budget = st.shared_queue_bytes_used();
                        }
                        if tx {
                            st.push_tx(tx_item(make(next_sender))).unwrap_or_else(|error| panic!("admit error {error:?}: mode={mode} tx={tx} slow_last={slow_last} incoming={next_sender}"));
                        } else {
                            st.push_rx(rx(make(next_sender))).unwrap_or_else(|error| panic!("admit error {error:?}: mode={mode} tx={tx} slow_last={slow_last} incoming={next_sender}"));
                        }
                        let slow = wire_format::source_address_for_sender("SLOW");
                        let retained = if tx {
                            st.tx_queue.iter().any(|item| {
                                RelayInner::fair_bulk_owner(&item.data, item.priority) == Some(slow)
                            })
                        } else {
                            st.rx_queue.iter().any(|item| {
                                RelayInner::fair_bulk_owner(&item.data, item.priority) == Some(slow)
                            })
                        };
                        assert!(
                            retained,
                            "slow report evicted: mode={mode} tx={tx} slow_last={slow_last} incoming={next_sender}"
                        );
                    }
                }
            }
        }
    }
}

#[cfg(all(test, feature = "discovery"))]
#[test]
fn cow_discovery_rejected_growth_keeps_original_graph_and_routes() {
    crate::tests::ensure_common_test_schema();
    let node = Relay::new_with_config(RelayConfig::default().with_sender("GB"), Box::new(|| 0));
    let side = node.add_side_packed("can", |_| Ok(()));
    let baseline = discovery::build_discovery_topology(
        "PEER",
        0,
        &[TopologyBoardNode {
            sender_id: "PEER".into(),
            reachable_endpoints: Vec::new(),
            reachable_timesync_sources: Vec::new(),
            connections: Vec::new(),
        }],
    )
    .unwrap();
    node.rx_from_side(side, baseline).unwrap();
    node.process_all_queues().unwrap();
    node.clear_queues();
    let before = node.export_topology();
    let original = {
        let mut st = node.state.lock();
        let graph = st.discovery_routes[&side].announcers["PEER"]
            .topology_boards
            .clone();
        st.memory.max_queue_budget = st.shared_queue_bytes_used();
        graph
    };
    let update = discovery::build_discovery_topology_delta(
        "PEER",
        1,
        &[TopologyBoardNode {
            sender_id: "PEER".into(),
            reachable_endpoints: Vec::new(),
            reachable_timesync_sources: vec!["NEW_CLOCK".into()],
            connections: vec!["NEW_NEIGHBOR".into()],
        }],
        &[],
    )
    .unwrap();
    assert!(
        node.rx_from_side(side, update)
            .and_then(|_| node.process_all_queues())
            .is_err(),
        "growth must fail when no queue budget can be reclaimed"
    );
    assert_eq!(node.export_topology().routes, before.routes);
    let st = node.state.lock();
    assert!(Arc::ptr_eq(
        &original,
        &st.discovery_routes[&side].announcers["PEER"].topology_boards
    ));
    assert!(original[0].connections.is_empty());
    assert!(original[0].reachable_timesync_sources.is_empty());
}