rns_net/
event.rs

1//! Event types for the driver loop — concrete sync instantiation.
2
3use std::collections::{HashMap, VecDeque};
4use std::sync::atomic::{AtomicUsize, Ordering};
5use std::sync::{Arc, Condvar, Mutex};
6use std::time::{Duration, Instant};
7
8use rns_core::packet::RawPacket;
9use rns_core::transport::types::InterfaceId;
10
11pub use crate::common::event::{
12    BackboneInterfaceEntry, BackbonePeerHookEvent, BackbonePeerPoolMemberStatus,
13    BackbonePeerPoolStatus, BackbonePeerStateEntry, BlackholeInfo, DrainStatus,
14    DynamicInterfaceRegistration, HolePunchPolicy, HookInfo, InterfaceStatsResponse,
15    InterfaceTelemetry, KnownDestinationEntry, LifecycleState, LinkDatagramError, LinkInfoEntry,
16    LocalDestinationEntry, NextHopResponse, PathTableEntry, ProviderBridgeConsumerStats,
17    ProviderBridgeStats, QueryRequest, QueryResponse, RateTableEntry, ResourceInfoEntry,
18    RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError, RuntimeConfigErrorCode,
19    RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat, TrafficDetail,
20};
21
22/// Concrete Event type using boxed sync Writer.
23pub type Event = crate::common::event::Event<Box<dyn crate::interface::Writer>>;
24
25pub const DEFAULT_EVENT_QUEUE_CAPACITY: usize = 1024;
26pub const DEFAULT_ANNOUNCE_QUEUE_CAPACITY: usize = 128;
27pub const DEFAULT_PATH_REQUEST_QUEUE_CAPACITY: usize = 128;
28pub const DEFAULT_INGRESS_LIMITED_QUEUE_CAPACITY: usize = 8;
29
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub struct InboundQueueCapacities {
32    pub data: usize,
33    pub announce: usize,
34    pub path_request: usize,
35    pub ingress_limited: usize,
36}
37
38impl Default for InboundQueueCapacities {
39    fn default() -> Self {
40        Self {
41            data: DEFAULT_EVENT_QUEUE_CAPACITY,
42            announce: DEFAULT_ANNOUNCE_QUEUE_CAPACITY,
43            path_request: DEFAULT_PATH_REQUEST_QUEUE_CAPACITY,
44            ingress_limited: DEFAULT_INGRESS_LIMITED_QUEUE_CAPACITY,
45        }
46    }
47}
48
49impl InboundQueueCapacities {
50    pub(crate) fn from_shared_capacity(capacity: usize) -> Self {
51        let capacity = capacity.max(1);
52        Self {
53            data: capacity,
54            announce: capacity.min(DEFAULT_ANNOUNCE_QUEUE_CAPACITY),
55            path_request: capacity.min(DEFAULT_PATH_REQUEST_QUEUE_CAPACITY),
56            ingress_limited: capacity.min(DEFAULT_INGRESS_LIMITED_QUEUE_CAPACITY),
57        }
58    }
59
60    fn as_array(self) -> [usize; INBOUND_QUEUE_COUNT] {
61        [
62            self.data,
63            self.announce,
64            self.path_request,
65            self.ingress_limited,
66        ]
67    }
68}
69
70#[derive(Clone, Copy, Debug, PartialEq, Eq)]
71enum QueueClass {
72    Data = 0,
73    Announce = 1,
74    PathRequest = 2,
75    IngressLimited = 3,
76}
77
78const INBOUND_QUEUE_COUNT: usize = 4;
79
80struct QueuedEvent {
81    sequence: u64,
82    event: Event,
83}
84
85struct QueueState {
86    control: VecDeque<QueuedEvent>,
87    inbound: [VecDeque<QueuedEvent>; INBOUND_QUEUE_COUNT],
88    inbound_dropped: [usize; INBOUND_QUEUE_COUNT],
89    next_sequence: u64,
90    receiver_alive: bool,
91    announce_bursts: HashMap<InterfaceId, f64>,
92    path_request_bursts: HashMap<InterfaceId, f64>,
93    dynamic_interface_parents: HashMap<InterfaceId, InterfaceId>,
94}
95
96pub(crate) struct ReceivedEvent {
97    pub event: Event,
98    pub ingress_limited: bool,
99}
100
101impl QueueState {
102    fn new() -> Self {
103        Self {
104            control: VecDeque::new(),
105            inbound: std::array::from_fn(|_| VecDeque::new()),
106            inbound_dropped: [0; INBOUND_QUEUE_COUNT],
107            next_sequence: 0,
108            receiver_alive: true,
109            announce_bursts: HashMap::new(),
110            path_request_bursts: HashMap::new(),
111            dynamic_interface_parents: HashMap::new(),
112        }
113    }
114
115    fn has_events(&self) -> bool {
116        !self.control.is_empty() || self.inbound.iter().any(|queue| !queue.is_empty())
117    }
118
119    fn pop_next(&mut self) -> Option<ReceivedEvent> {
120        let control_barrier = self.control.front().map(|queued| queued.sequence);
121        for (index, queue) in self.inbound.iter_mut().enumerate() {
122            if queue.front().is_some_and(|queued| {
123                control_barrier.is_none_or(|barrier| queued.sequence < barrier)
124            }) {
125                return queue.pop_front().map(|queued| ReceivedEvent {
126                    event: queued.event,
127                    ingress_limited: index == QueueClass::IngressLimited as usize,
128                });
129            }
130        }
131        self.control.pop_front().map(|queued| ReceivedEvent {
132            event: queued.event,
133            ingress_limited: false,
134        })
135    }
136}
137
138struct QueueShared {
139    state: Mutex<QueueState>,
140    changed: Condvar,
141    sender_count: AtomicUsize,
142    control_capacity: usize,
143    inbound_capacities: [usize; INBOUND_QUEUE_COUNT],
144    path_request_dest: [u8; 16],
145}
146
147/// Sender for the prioritized driver event queue.
148pub struct EventSender {
149    shared: Arc<QueueShared>,
150}
151
152#[derive(Clone, Copy, Debug, PartialEq)]
153pub(crate) struct InboundQueueSnapshot {
154    pub heights: [usize; INBOUND_QUEUE_COUNT],
155    pub capacities: [usize; INBOUND_QUEUE_COUNT],
156    pub dropped: [usize; INBOUND_QUEUE_COUNT],
157}
158
159impl InboundQueueSnapshot {
160    pub fn total_height(self) -> usize {
161        self.heights.iter().sum()
162    }
163
164    pub fn total_capacity(self) -> usize {
165        self.capacities.iter().sum()
166    }
167
168    pub fn total_dropped(self) -> usize {
169        self.dropped.iter().sum()
170    }
171}
172
173#[derive(Clone, Copy, Debug, PartialEq)]
174pub(crate) struct BurstClassStats {
175    pub(crate) count: usize,
176    pub(crate) activated: Option<f64>,
177}
178
179#[derive(Clone, Copy, Debug, PartialEq)]
180pub(crate) struct DynamicBurstStats {
181    pub(crate) announce: BurstClassStats,
182    pub(crate) path_request: BurstClassStats,
183}
184
185impl Clone for EventSender {
186    fn clone(&self) -> Self {
187        self.shared.sender_count.fetch_add(1, Ordering::Relaxed);
188        Self {
189            shared: Arc::clone(&self.shared),
190        }
191    }
192}
193
194impl Drop for EventSender {
195    fn drop(&mut self) {
196        if self.shared.sender_count.fetch_sub(1, Ordering::AcqRel) == 1 {
197            self.shared.changed.notify_all();
198        }
199    }
200}
201
202impl EventSender {
203    fn classify(&self, event: &Event, state: &QueueState) -> Option<QueueClass> {
204        let Event::Frame {
205            interface_id, data, ..
206        } = event
207        else {
208            return None;
209        };
210        let packet = RawPacket::unpack(data).ok();
211        if packet.as_ref().is_some_and(|packet| {
212            packet.flags.packet_type == rns_core::constants::PACKET_TYPE_ANNOUNCE
213        }) {
214            return Some(if state.announce_bursts.contains_key(interface_id) {
215                QueueClass::IngressLimited
216            } else {
217                QueueClass::Announce
218            });
219        }
220        if packet
221            .as_ref()
222            .is_some_and(|packet| packet.destination_hash == self.shared.path_request_dest)
223        {
224            return Some(if state.path_request_bursts.contains_key(interface_id) {
225                QueueClass::IngressLimited
226            } else {
227                QueueClass::PathRequest
228            });
229        }
230        Some(QueueClass::Data)
231    }
232
233    // Preserve the former SyncSender-compatible error contract at this public boundary.
234    #[allow(clippy::result_large_err)]
235    fn enqueue(
236        &self,
237        event: Event,
238        block_control: bool,
239        drop_full_inbound: bool,
240    ) -> Result<(), std::sync::mpsc::TrySendError<Event>> {
241        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
242        if !state.receiver_alive {
243            return Err(std::sync::mpsc::TrySendError::Disconnected(event));
244        }
245        let class = self.classify(&event, &state);
246        if let Some(class) = class {
247            let index = class as usize;
248            if state.inbound[index].len() >= self.shared.inbound_capacities[index] {
249                state.inbound_dropped[index] = state.inbound_dropped[index].saturating_add(1);
250                return if drop_full_inbound {
251                    Ok(())
252                } else {
253                    Err(std::sync::mpsc::TrySendError::Full(event))
254                };
255            }
256            let sequence = state.next_sequence;
257            state.next_sequence = state.next_sequence.wrapping_add(1);
258            state.inbound[index].push_back(QueuedEvent { sequence, event });
259            drop(state);
260            self.shared.changed.notify_one();
261            return Ok(());
262        }
263
264        while state.control.len() >= self.shared.control_capacity {
265            if !block_control {
266                return Err(std::sync::mpsc::TrySendError::Full(event));
267            }
268            state = self
269                .shared
270                .changed
271                .wait(state)
272                .unwrap_or_else(|p| p.into_inner());
273            if !state.receiver_alive {
274                return Err(std::sync::mpsc::TrySendError::Disconnected(event));
275            }
276        }
277        let sequence = state.next_sequence;
278        state.next_sequence = state.next_sequence.wrapping_add(1);
279        state.control.push_back(QueuedEvent { sequence, event });
280        drop(state);
281        self.shared.changed.notify_one();
282        Ok(())
283    }
284
285    #[allow(clippy::result_large_err)]
286    pub fn send(&self, event: Event) -> Result<(), std::sync::mpsc::SendError<Event>> {
287        self.enqueue(event, true, true)
288            .map_err(|error| match error {
289                std::sync::mpsc::TrySendError::Full(event)
290                | std::sync::mpsc::TrySendError::Disconnected(event) => {
291                    std::sync::mpsc::SendError(event)
292                }
293            })
294    }
295
296    #[allow(clippy::result_large_err)]
297    pub fn try_send(&self, event: Event) -> Result<(), std::sync::mpsc::TrySendError<Event>> {
298        self.enqueue(event, false, false)
299    }
300
301    pub(crate) fn set_ingress_bursts(
302        &self,
303        interface_id: InterfaceId,
304        announce_activated: Option<f64>,
305        path_request_activated: Option<f64>,
306    ) {
307        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
308        if let Some(activated) = announce_activated {
309            state.announce_bursts.insert(interface_id, activated);
310        } else {
311            state.announce_bursts.remove(&interface_id);
312        }
313        if let Some(activated) = path_request_activated {
314            state.path_request_bursts.insert(interface_id, activated);
315        } else {
316            state.path_request_bursts.remove(&interface_id);
317        }
318    }
319
320    pub(crate) fn register_dynamic_parent(
321        &self,
322        interface_id: InterfaceId,
323        parent_id: InterfaceId,
324    ) {
325        self.shared
326            .state
327            .lock()
328            .unwrap_or_else(|p| p.into_inner())
329            .dynamic_interface_parents
330            .insert(interface_id, parent_id);
331    }
332
333    pub(crate) fn remove_interface(&self, interface_id: InterfaceId) {
334        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
335        state.announce_bursts.remove(&interface_id);
336        state.path_request_bursts.remove(&interface_id);
337        state.dynamic_interface_parents.remove(&interface_id);
338    }
339
340    pub(crate) fn inbound_queue_snapshot(&self) -> InboundQueueSnapshot {
341        let state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
342        InboundQueueSnapshot {
343            heights: std::array::from_fn(|index| state.inbound[index].len()),
344            capacities: self.shared.inbound_capacities,
345            dropped: state.inbound_dropped,
346        }
347    }
348
349    pub(crate) fn dynamic_burst_stats(&self, parent_id: InterfaceId) -> Option<DynamicBurstStats> {
350        let state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
351        let children = state
352            .dynamic_interface_parents
353            .iter()
354            .filter(|(_, candidate_parent)| **candidate_parent == parent_id);
355        let mut found = false;
356        let mut announce = (0, None::<f64>);
357        let mut path_request = (0, None::<f64>);
358        for (child, _) in children {
359            found = true;
360            if let Some(activated) = state.announce_bursts.get(child) {
361                announce.0 += 1;
362                announce.1 = Some(
363                    announce
364                        .1
365                        .map_or(*activated, |current| current.min(*activated)),
366                );
367            }
368            if let Some(activated) = state.path_request_bursts.get(child) {
369                path_request.0 += 1;
370                path_request.1 = Some(
371                    path_request
372                        .1
373                        .map_or(*activated, |current| current.min(*activated)),
374                );
375            }
376        }
377        found.then_some(DynamicBurstStats {
378            announce: BurstClassStats {
379                count: announce.0,
380                activated: announce.1,
381            },
382            path_request: BurstClassStats {
383                count: path_request.0,
384                activated: path_request.1,
385            },
386        })
387    }
388}
389
390/// Receiver for the prioritized driver event queue.
391pub struct EventReceiver {
392    shared: Arc<QueueShared>,
393}
394
395impl Drop for EventReceiver {
396    fn drop(&mut self) {
397        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
398        state.receiver_alive = false;
399        drop(state);
400        self.shared.changed.notify_all();
401    }
402}
403
404impl EventReceiver {
405    fn pop_locked(&self, state: &mut QueueState) -> Option<ReceivedEvent> {
406        let event = state.pop_next();
407        if event.is_some() {
408            self.shared.changed.notify_all();
409        }
410        event
411    }
412
413    pub fn recv(&self) -> Result<Event, std::sync::mpsc::RecvError> {
414        self.recv_classified().map(|received| received.event)
415    }
416
417    pub(crate) fn recv_classified(&self) -> Result<ReceivedEvent, std::sync::mpsc::RecvError> {
418        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
419        loop {
420            if let Some(event) = self.pop_locked(&mut state) {
421                return Ok(event);
422            }
423            if self.shared.sender_count.load(Ordering::Acquire) == 0 {
424                return Err(std::sync::mpsc::RecvError);
425            }
426            state = self
427                .shared
428                .changed
429                .wait(state)
430                .unwrap_or_else(|p| p.into_inner());
431        }
432    }
433
434    pub fn try_recv(&self) -> Result<Event, std::sync::mpsc::TryRecvError> {
435        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
436        if let Some(event) = self.pop_locked(&mut state) {
437            return Ok(event.event);
438        }
439        if self.shared.sender_count.load(Ordering::Acquire) == 0 {
440            Err(std::sync::mpsc::TryRecvError::Disconnected)
441        } else {
442            Err(std::sync::mpsc::TryRecvError::Empty)
443        }
444    }
445
446    pub fn recv_timeout(
447        &self,
448        timeout: Duration,
449    ) -> Result<Event, std::sync::mpsc::RecvTimeoutError> {
450        let deadline = Instant::now() + timeout;
451        let mut state = self.shared.state.lock().unwrap_or_else(|p| p.into_inner());
452        loop {
453            if let Some(event) = self.pop_locked(&mut state) {
454                return Ok(event.event);
455            }
456            if self.shared.sender_count.load(Ordering::Acquire) == 0 {
457                return Err(std::sync::mpsc::RecvTimeoutError::Disconnected);
458            }
459            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
460                return Err(std::sync::mpsc::RecvTimeoutError::Timeout);
461            };
462            let (next_state, result) = self
463                .shared
464                .changed
465                .wait_timeout(state, remaining)
466                .unwrap_or_else(|p| p.into_inner());
467            state = next_state;
468            if result.timed_out() && !state.has_events() {
469                return Err(std::sync::mpsc::RecvTimeoutError::Timeout);
470            }
471        }
472    }
473}
474
475pub fn channel() -> (EventSender, EventReceiver) {
476    channel_with_queue_capacities(
477        DEFAULT_EVENT_QUEUE_CAPACITY,
478        InboundQueueCapacities::default(),
479    )
480}
481
482pub fn channel_with_capacity(capacity: usize) -> (EventSender, EventReceiver) {
483    let capacity = capacity.max(1);
484    channel_with_queue_capacities(
485        capacity,
486        InboundQueueCapacities::from_shared_capacity(capacity),
487    )
488}
489
490pub(crate) fn channel_with_queue_capacities(
491    control_capacity: usize,
492    inbound_capacities: InboundQueueCapacities,
493) -> (EventSender, EventReceiver) {
494    let shared = Arc::new(QueueShared {
495        state: Mutex::new(QueueState::new()),
496        changed: Condvar::new(),
497        sender_count: AtomicUsize::new(1),
498        control_capacity: control_capacity.max(1),
499        inbound_capacities: inbound_capacities
500            .as_array()
501            .map(|capacity| capacity.max(1)),
502        path_request_dest: rns_core::destination::destination_hash(
503            "rnstransport",
504            &["path", "request"],
505            None,
506        ),
507    });
508    (
509        EventSender {
510            shared: Arc::clone(&shared),
511        },
512        EventReceiver { shared },
513    )
514}
515
516#[cfg(test)]
517mod tests {
518    use super::*;
519    use rns_core::packet::{PacketFlags, RawPacket};
520    use std::sync::mpsc::TrySendError;
521    use std::time::Duration;
522
523    fn frame(interface_id: u64, destination: [u8; 16], packet_type: u8) -> Event {
524        let raw = RawPacket::pack(
525            PacketFlags {
526                header_type: rns_core::constants::HEADER_1,
527                context_flag: rns_core::constants::FLAG_UNSET,
528                transport_type: rns_core::constants::TRANSPORT_BROADCAST,
529                destination_type: rns_core::constants::DESTINATION_PLAIN,
530                packet_type,
531            },
532            0,
533            &destination,
534            None,
535            rns_core::constants::CONTEXT_NONE,
536            b"queue-test",
537        )
538        .unwrap()
539        .raw;
540        Event::Frame {
541            interface_id: InterfaceId(interface_id),
542            data: raw,
543            rssi: None,
544            snr: None,
545        }
546    }
547
548    fn frame_interface(event: Event) -> u64 {
549        match event {
550            Event::Frame { interface_id, .. } => interface_id.0,
551            other => panic!("expected frame, got {other:?}"),
552        }
553    }
554
555    #[test]
556    fn inbound_queue_defaults_match_upstream() {
557        assert_eq!(
558            InboundQueueCapacities::default(),
559            InboundQueueCapacities {
560                data: 1024,
561                announce: 128,
562                path_request: 128,
563                ingress_limited: 8,
564            }
565        );
566    }
567
568    #[test]
569    fn bounded_event_queue_backpressures_when_full() {
570        let (tx, rx) = channel_with_capacity(1);
571
572        tx.try_send(Event::Tick).unwrap();
573        match tx.try_send(Event::Shutdown) {
574            Err(TrySendError::Full(Event::Shutdown)) => {}
575            other => panic!("expected full queue for second event, got {other:?}"),
576        }
577
578        assert!(matches!(
579            rx.recv_timeout(Duration::from_secs(1)).unwrap(),
580            Event::Tick
581        ));
582        tx.try_send(Event::Shutdown).unwrap();
583        assert!(matches!(
584            rx.recv_timeout(Duration::from_secs(1)).unwrap(),
585            Event::Shutdown
586        ));
587    }
588
589    #[test]
590    fn inbound_frames_are_drained_in_traffic_class_priority_order() {
591        let (tx, rx) = channel_with_capacity(8);
592        let path_dest =
593            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
594        tx.set_ingress_bursts(InterfaceId(4), None, Some(4.0));
595
596        tx.send(frame(4, path_dest, rns_core::constants::PACKET_TYPE_DATA))
597            .unwrap();
598        tx.send(frame(3, path_dest, rns_core::constants::PACKET_TYPE_DATA))
599            .unwrap();
600        tx.send(frame(
601            2,
602            [0xA2; 16],
603            rns_core::constants::PACKET_TYPE_ANNOUNCE,
604        ))
605        .unwrap();
606        tx.send(frame(1, [0xD1; 16], rns_core::constants::PACKET_TYPE_DATA))
607            .unwrap();
608        tx.send(Event::Shutdown).unwrap();
609
610        assert_eq!(tx.inbound_queue_snapshot().heights, [1, 1, 1, 1]);
611
612        assert_eq!(frame_interface(rx.recv().unwrap()), 1);
613        assert_eq!(frame_interface(rx.recv().unwrap()), 2);
614        assert_eq!(frame_interface(rx.recv().unwrap()), 3);
615        assert_eq!(frame_interface(rx.recv().unwrap()), 4);
616        assert!(matches!(rx.recv().unwrap(), Event::Shutdown));
617    }
618
619    #[test]
620    fn dynamic_parent_burst_counts_track_independent_states_and_cleanup() {
621        let (tx, _rx) = channel();
622        tx.register_dynamic_parent(InterfaceId(11), InterfaceId(10));
623        tx.register_dynamic_parent(InterfaceId(12), InterfaceId(10));
624        tx.register_dynamic_parent(InterfaceId(13), InterfaceId(10));
625        tx.register_dynamic_parent(InterfaceId(21), InterfaceId(20));
626        tx.set_ingress_bursts(InterfaceId(11), Some(11.0), None);
627        tx.set_ingress_bursts(InterfaceId(12), None, Some(12.0));
628        tx.set_ingress_bursts(InterfaceId(13), Some(9.0), None);
629        tx.set_ingress_bursts(InterfaceId(21), Some(21.0), Some(22.0));
630
631        assert_eq!(
632            tx.dynamic_burst_stats(InterfaceId(10)),
633            Some(DynamicBurstStats {
634                announce: BurstClassStats {
635                    count: 2,
636                    activated: Some(9.0),
637                },
638                path_request: BurstClassStats {
639                    count: 1,
640                    activated: Some(12.0),
641                },
642            })
643        );
644        assert_eq!(
645            tx.dynamic_burst_stats(InterfaceId(20)),
646            Some(DynamicBurstStats {
647                announce: BurstClassStats {
648                    count: 1,
649                    activated: Some(21.0),
650                },
651                path_request: BurstClassStats {
652                    count: 1,
653                    activated: Some(22.0),
654                },
655            })
656        );
657        assert_eq!(tx.dynamic_burst_stats(InterfaceId(30)), None);
658
659        tx.remove_interface(InterfaceId(12));
660        assert_eq!(
661            tx.dynamic_burst_stats(InterfaceId(10)),
662            Some(DynamicBurstStats {
663                announce: BurstClassStats {
664                    count: 2,
665                    activated: Some(9.0),
666                },
667                path_request: BurstClassStats {
668                    count: 0,
669                    activated: None,
670                },
671            })
672        );
673    }
674
675    #[test]
676    fn saturated_data_queue_does_not_consume_path_request_capacity() {
677        let (tx, rx) = channel_with_capacity(1);
678        let path_dest =
679            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
680        tx.try_send(frame(1, [0xD1; 16], rns_core::constants::PACKET_TYPE_DATA))
681            .unwrap();
682        assert!(matches!(
683            tx.try_send(frame(2, [0xD2; 16], rns_core::constants::PACKET_TYPE_DATA)),
684            Err(TrySendError::Full(_))
685        ));
686        tx.try_send(frame(3, path_dest, rns_core::constants::PACKET_TYPE_DATA))
687            .unwrap();
688        tx.try_send(Event::Shutdown).unwrap();
689
690        assert_eq!(frame_interface(rx.recv().unwrap()), 1);
691        assert_eq!(frame_interface(rx.recv().unwrap()), 3);
692        assert!(matches!(rx.recv().unwrap(), Event::Shutdown));
693    }
694
695    #[test]
696    fn custom_traffic_class_capacities_are_enforced_independently() {
697        let capacities = InboundQueueCapacities {
698            data: 1,
699            announce: 2,
700            path_request: 3,
701            ingress_limited: 4,
702        };
703        let (tx, _rx) = channel_with_queue_capacities(9, capacities);
704        let path_dest =
705            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
706        tx.set_ingress_bursts(InterfaceId(4), None, Some(1.0));
707
708        let classes = [
709            (1, [0xD1; 16], rns_core::constants::PACKET_TYPE_DATA, 1),
710            (2, [0xA2; 16], rns_core::constants::PACKET_TYPE_ANNOUNCE, 2),
711            (3, path_dest, rns_core::constants::PACKET_TYPE_DATA, 3),
712            (4, path_dest, rns_core::constants::PACKET_TYPE_DATA, 4),
713        ];
714        for (interface_id, destination, packet_type, capacity) in classes {
715            for _ in 0..capacity {
716                tx.try_send(frame(interface_id, destination, packet_type))
717                    .unwrap();
718            }
719            assert!(matches!(
720                tx.try_send(frame(interface_id, destination, packet_type)),
721                Err(TrySendError::Full(_))
722            ));
723        }
724
725        assert_eq!(
726            tx.inbound_queue_snapshot().capacities,
727            capacities.as_array()
728        );
729        assert_eq!(tx.inbound_queue_snapshot().heights, [1, 2, 3, 4]);
730        assert_eq!(tx.inbound_queue_snapshot().dropped, [1, 1, 1, 1]);
731        assert_eq!(tx.inbound_queue_snapshot().total_dropped(), 4);
732    }
733
734    #[test]
735    fn ingress_limited_classification_survives_limiter_state_change() {
736        let (tx, rx) = channel_with_capacity(2);
737        let path_dest =
738            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
739        tx.set_ingress_bursts(InterfaceId(4), None, Some(1.0));
740        tx.send(frame(4, path_dest, rns_core::constants::PACKET_TYPE_DATA))
741            .unwrap();
742
743        tx.set_ingress_bursts(InterfaceId(4), None, None);
744        let received = rx.recv_classified().unwrap();
745        assert!(received.ingress_limited);
746        assert_eq!(frame_interface(received.event), 4);
747    }
748
749    #[test]
750    fn ingress_limited_announce_keeps_higher_traffic_class() {
751        let (tx, rx) = channel_with_capacity(2);
752        tx.set_ingress_bursts(InterfaceId(4), Some(1.0), None);
753        tx.send(frame(
754            4,
755            [0xA4; 16],
756            rns_core::constants::PACKET_TYPE_ANNOUNCE,
757        ))
758        .unwrap();
759
760        tx.set_ingress_bursts(InterfaceId(4), None, None);
761        let received = rx.recv_classified().unwrap();
762        assert!(received.ingress_limited);
763        assert_eq!(frame_interface(received.event), 4);
764    }
765
766    #[test]
767    fn blocking_send_drops_only_the_full_inbound_class() {
768        let (tx, rx) = channel_with_capacity(1);
769        tx.send(frame(1, [0xD1; 16], rns_core::constants::PACKET_TYPE_DATA))
770            .unwrap();
771        tx.send(frame(2, [0xD2; 16], rns_core::constants::PACKET_TYPE_DATA))
772            .expect("a full inbound class must drop instead of blocking its reader");
773        assert_eq!(tx.inbound_queue_snapshot().dropped, [1, 0, 0, 0]);
774        assert_eq!(frame_interface(rx.recv().unwrap()), 1);
775
776        tx.send(frame(3, [0xD3; 16], rns_core::constants::PACKET_TYPE_DATA))
777            .unwrap();
778        tx.send(frame(4, [0xD4; 16], rns_core::constants::PACKET_TYPE_DATA))
779            .unwrap();
780        assert_eq!(tx.inbound_queue_snapshot().dropped, [2, 0, 0, 0]);
781
782        tx.send(Event::Shutdown).unwrap();
783        assert_eq!(frame_interface(rx.recv().unwrap()), 3);
784        assert!(matches!(rx.recv().unwrap(), Event::Shutdown));
785    }
786}
rns_net/event.rs

rns_net/
event.rs
