SEDSnet 4.2.3

//! Cold-path pressure arbitration. No retained state, payload copies or heap scratch.
use crate::{DataType, packet::Packet, wire_format};

pub(crate) fn packet_owner(packet: &Packet, priority: u8) -> Option<u32> {
    let meta = crate::config::try_get_message_meta(packet.data_type())?;
    eligible(
        packet.data_type(),
        priority,
        meta.reliable != crate::ReliableMode::None,
        !packet.wire_target_senders().is_empty(),
    )
    .then(|| wire_format::source_address_for_sender(packet.sender()))
}
pub(crate) fn packed_owner(bytes: &[u8], priority: u8) -> Option<u32> {
    if priority >= 200 {
        return None;
    }
    let frame = wire_format::peek_routing_frame_info(bytes).ok()?;
    eligible(
        frame.envelope.ty,
        priority,
        frame.envelope.is_reliable || frame.reliable.is_some(),
        frame.envelope.target_senders().len() != 0,
    )
    .then_some(frame.envelope.source_address)
}
pub(crate) fn incoming(bytes: &[u8]) -> Option<(u32, u8)> {
    let ty = wire_format::scheduling_type(bytes)?;
    crate::config::try_get_message_meta(ty)?;
    let priority = crate::scheduler_priority(ty);
    packed_owner(bytes, priority).map(|owner| (owner, priority))
}
/// The full retained header and wrapper CRC have already been validated.
/// Classify before reconstruction allocates, using no new retained state.
pub(crate) fn compact_incoming(
    template: &crate::side_transport::SideHeaderTemplate,
) -> Option<(u32, u8)> {
    let ty = template.data_type()?;
    let meta = crate::config::try_get_message_meta(ty)?;
    let priority = crate::scheduler_priority(ty);
    if !eligible(
        ty,
        priority,
        meta.reliable != crate::ReliableMode::None || template.reliable_flags.is_some(),
        template.base_flags & crate::side_transport::SIDE_TRANSPORT_FLAG_WIRE_CONTRACT != 0,
    ) {
        return None;
    }
    Some((template.source_address()?, priority))
}

fn eligible(ty: DataType, priority: u8, reliable: bool, frozen: bool) -> bool {
    // Unknown runtime fragments cannot allocate UNKNOWN_TYPE fallback metadata
    // while we are deciding which retained payload is safe to reclaim.
    crate::config::try_get_message_meta(ty).is_some()
        && priority < 200
        && crate::transport_priority(ty) < 200
        && ty != DataType::P2pMessage
        && !reliable
        && !frozen
}
#[derive(Clone, Copy, Default)]
struct Load {
    sender: u32,
    bytes: usize,
    count: usize,
}
pub(crate) struct Loads {
    peers: [Load; 32],
    len: usize,
    overflow: bool,
}
impl Loads {
    pub fn new() -> Self {
        Self {
            peers: [Load::default(); 32],
            len: 0,
            overflow: false,
        }
    }
    pub fn record(&mut self, sender: u32, bytes: usize) {
        let slot = if let Some(index) = self.peers[..self.len]
            .iter()
            .position(|p| p.sender == sender)
        {
            index
        } else {
            if self.len == self.peers.len() {
                self.overflow = true;
                return;
            }
            self.peers[self.len].sender = sender;
            self.len += 1;
            self.len - 1
        };
        let peer = &mut self.peers[slot];
        peer.bytes = peer.bytes.saturating_add(bytes);
        peer.count += 1;
    }
    /// Queue limits must shed the dominant sender even when that sender is
    /// the producer of the new frame. Otherwise its newest frame repeatedly
    /// evicts the only report retained for a slower peer.
    pub fn queue_victim(&self, incoming: Option<u32>) -> Option<u32> {
        if self.overflow {
            return None;
        }
        self.peers[..self.len]
            .iter()
            .filter(|peer| peer.count > 1)
            .max_by_key(|peer| peer.bytes)
            .map(|peer| peer.sender)
            .or_else(|| {
                incoming.filter(|source| {
                    self.peers[..self.len]
                        .iter()
                        .any(|peer| peer.sender == *source)
                })
            })
    }
    pub fn dominant_other(&self, incoming: u32) -> Option<u32> {
        // Beyond the bounded scratch catalog, refuse rather than guess which
        // peer owns the backlog. This is not a limit on network membership.
        if self.overflow {
            return None;
        }
        let incoming_bytes = self.peers[..self.len]
            .iter()
            .find(|p| p.sender == incoming)
            .map_or(0, |p| p.bytes);
        self.peers[..self.len]
            .iter()
            .filter(|p| p.sender != incoming && p.count > 1 && p.bytes > incoming_bytes)
            .max_by_key(|p| p.bytes)
            .map(|p| p.sender)
    }
}
/// No classification or scratch work until the bounded queue needs a victim.
/// Internal controls retain the ordinary priority selector.
pub(crate) fn queue_victim<T, O, P, S>(
    queue: &crate::queue::BoundedDeque<T>,
    incoming: &T,
    owner: O,
    priority: P,
    stream: S,
    selected: &mut Option<u32>,
) -> Option<usize>
where
    T: crate::queue::ByteCost,
    O: Fn(&T) -> Option<u32>,
    P: Fn(&T) -> u8,
    S: Fn(&T) -> Option<u32>,
{
    let Some(incoming_owner) = owner(incoming) else {
        return queue.len().checked_sub(1);
    };
    let incoming_priority = priority(incoming);
    let victim = if let Some(owner) = *selected {
        owner
    } else {
        let mut loads = Loads::new();
        for item in queue.iter() {
            if priority(item) <= incoming_priority
                && let Some(source) = owner(item)
            {
                loads.record(source, item.byte_cost());
            }
        }
        let victim = loads.queue_victim(Some(incoming_owner))?;
        *selected = Some(victim);
        victim
    };
    let mut streams = Loads::new();
    for item in queue.iter() {
        if priority(item) <= incoming_priority
            && owner(item) == Some(victim)
            && let Some(ty) = stream(item)
        {
            streams.record(ty, item.byte_cost());
        }
    }
    // Reuse the bounded selector within the selected owner: repeated message
    // types yield first, then this incoming type's previous value, then the
    // ordinary priority fallback. No per-type reservation is retained.
    let stream_victim = streams.queue_victim(
        (victim == incoming_owner)
            .then(|| stream(incoming))
            .flatten(),
    );
    queue
        .iter()
        .enumerate()
        .filter(|(_, item)| {
            priority(item) <= incoming_priority
                && owner(item) == Some(victim)
                && stream_victim.is_none_or(|ty| stream(item) == Some(ty))
        })
        .min_by_key(|(index, item)| (priority(item), *index))
        .map(|(index, _)| index)
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn bounded_queues_replace_dominant_producer_before_another_peers_only_report() {
        let mut loads = Loads::new();
        loads.record(1, 50);
        loads.record(2, 100);
        loads.record(2, 100);
        assert_eq!(loads.queue_victim(Some(2)), Some(2));
        assert_eq!(loads.queue_victim(Some(3)), Some(2));
        assert_eq!(loads.queue_victim(None), Some(2));
        let mut singles = Loads::new();
        singles.record(1, 50);
        assert_eq!(singles.queue_victim(Some(2)), None);
        assert_eq!(singles.queue_victim(Some(1)), Some(1));
        assert_eq!(singles.queue_victim(None), None);
    }

    #[test]
    fn protected_and_frozen_packets_never_become_bulk_victims() {
        use alloc::sync::Arc;
        let ep = crate::config::register_endpoint("FAIR_KEEP_ENDPOINT", false).unwrap();
        let register = |name, mode| {
            crate::config::register_data_type_with_description(
                name,
                "",
                crate::MessageElement::Static(
                    4,
                    crate::MessageDataType::UInt8,
                    crate::MessageClass::Data,
                ),
                &[ep],
                mode,
                0,
            )
            .unwrap()
        };
        let plain = register("FAIR_PLAIN_REPORT", crate::ReliableMode::None);
        let reliable = register("FAIR_KEEP_RELIABLE", crate::ReliableMode::Ordered);
        let make = |ty| Packet::new(ty, &[ep], "SOURCE", 1, Arc::<[u8]>::from([1; 4])).unwrap();
        let packet = make(plain);
        assert!(packet_owner(&packet, 0).is_some());
        assert!(packed_owner(&wire_format::pack_packet(&packet), 0).is_some());
        assert!(packed_owner(&wire_format::pack_packet(&packet), 200).is_none());
        let (template, ..) =
            crate::side_transport::extract_side_header_template(&wire_format::pack_packet(&packet))
                .unwrap();
        assert_eq!(
            compact_incoming(&template),
            Some((wire_format::source_address_for_sender("SOURCE"), 0))
        );
        let packet = make(reliable);
        assert!(packet_owner(&packet, 0).is_none());
        assert!(packed_owner(&wire_format::pack_packet(&packet), 0).is_none());
        let (template, ..) =
            crate::side_transport::extract_side_header_template(&wire_format::pack_packet(&packet))
                .unwrap();
        assert!(compact_incoming(&template).is_none());
        let frozen = Packet::new_with_wire_contract(
            plain,
            &[ep],
            "SOURCE",
            1,
            1,
            Arc::<[u8]>::from([1; 4]),
            None,
            Arc::from([42u64]),
        )
        .unwrap();
        assert!(packet_owner(&frozen, 0).is_none());
        assert!(packed_owner(&wire_format::pack_packet(&frozen), 0).is_none());
        let (template, ..) =
            crate::side_transport::extract_side_header_template(&wire_format::pack_packet(&frozen))
                .unwrap();
        assert!(compact_incoming(&template).is_none());
    }
    #[test]
    fn unknown_large_membership_does_not_guess_an_eviction_victim() {
        let mut loads = Loads::new();
        for source in 1..=33 {
            loads.record(source, 100);
            loads.record(source, 100);
        }
        assert_eq!(loads.dominant_other(100), None);
        assert_eq!(loads.queue_victim(Some(100)), None);
    }
    #[test]
    fn largest_backlog_yields_without_sacrificing_another_peers_only_report() {
        let mut loads = Loads::new();
        loads.record(1, 500);
        loads.record(1, 500);
        loads.record(2, 50);
        loads.record(3, 50);
        assert_eq!(loads.dominant_other(2), Some(1));
        assert_eq!(loads.dominant_other(1), None);
        let mut singles = Loads::new();
        singles.record(1, 500);
        assert_eq!(singles.dominant_other(2), None);
    }
}