prns_runtime/runtime/node_introspection/impls/
heap.rs

1use alloc::collections::{BTreeMap, VecDeque};
2use alloc::string::String;
3use alloc::vec::Vec;
4use core::future::Future;
5
6use prns_core::engine::{AnnounceRateState, RouteSnapshot};
7use prns_core::interfaces::PacketPhyStats;
8use prns_core::routing::announce::AnnounceRateAccounting;
9use prns_core::routing::dedup::PacketHash;
10use prns_core::units::InstantMillis;
11use prns_core::wire::{DestinationHash, TRUNCATED_HASH_BYTE_LEN};
12
13use super::super::{fold_logical_interface_inventory, InterfaceInventoryEntry};
14
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct AnnounceRateSnapshot {
17    pub destination: DestinationHash,
18    pub last_allowed_announce_at: InstantMillis,
19    pub blocked_until: InstantMillis,
20    pub rate_violations: u16,
21    pub observed_at: Vec<InstantMillis>,
22}
23
24const MAX_ANNOUNCE_RATE_OBSERVATIONS: usize = 16;
25
26#[derive(Default)]
27pub struct HeapAnnounceRateHistory {
28    observed_at: BTreeMap<AnnounceRateHistoryKey, VecDeque<InstantMillis>>,
29}
30
31#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
32struct AnnounceRateHistoryKey([u8; TRUNCATED_HASH_BYTE_LEN]);
33
34impl From<DestinationHash> for AnnounceRateHistoryKey {
35    fn from(destination: DestinationHash) -> Self {
36        Self(*destination.as_bytes())
37    }
38}
39
40impl HeapAnnounceRateHistory {
41    pub fn record(
42        &mut self,
43        destination: DestinationHash,
44        observed_at: InstantMillis,
45        accounting: AnnounceRateAccounting,
46    ) {
47        match accounting {
48            AnnounceRateAccounting::NotApplied => return,
49            AnnounceRateAccounting::Started => {
50                self.observed_at.insert(destination.into(), VecDeque::new());
51            }
52            AnnounceRateAccounting::Continued => {}
53        }
54        let history = self.observed_at.entry(destination.into()).or_default();
55        if history.len() == MAX_ANNOUNCE_RATE_OBSERVATIONS {
56            history.pop_front();
57        }
58        history.push_back(observed_at);
59    }
60
61    #[must_use]
62    pub fn snapshot(&self, state: AnnounceRateState) -> AnnounceRateSnapshot {
63        AnnounceRateSnapshot {
64            destination: state.destination,
65            last_allowed_announce_at: state.last_allowed_announce_at,
66            blocked_until: state.blocked_until,
67            rate_violations: state.rate_violations,
68            observed_at: self
69                .observed_at
70                .get(&state.destination.into())
71                .map(|history| history.iter().copied().collect())
72                .unwrap_or_default(),
73        }
74    }
75}
76
77pub trait NodeIntrospection {
78    fn interface_inventory(&self) -> Vec<InterfaceInventoryEntry<String>>;
79
80    fn link_count(&self) -> impl Future<Output = u32> + Send;
81
82    fn packet_phy(&self, packet_hash: PacketHash) -> Option<PacketPhyStats>;
83
84    fn announce_rates(&self) -> impl Future<Output = Vec<AnnounceRateSnapshot>> + Send;
85
86    fn routes(&self) -> impl Future<Output = Vec<RouteSnapshot>> + Send;
87
88    fn route(
89        &self,
90        destination: DestinationHash,
91    ) -> impl Future<Output = Option<RouteSnapshot>> + Send;
92}
93
94#[must_use]
95pub fn logical_interface_inventory<Label: Ord>(
96    mut inventory: Vec<InterfaceInventoryEntry<Label>>,
97) -> Vec<InterfaceInventoryEntry<Label>> {
98    let logical_len = fold_logical_interface_inventory(&mut inventory).len();
99    inventory.truncate(logical_len);
100    inventory
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106
107    #[test]
108    fn announce_rate_history_is_bounded_and_restartable() {
109        let destination = DestinationHash::new([0x42; 16]);
110        let mut history = HeapAnnounceRateHistory::default();
111        history.record(
112            destination,
113            InstantMillis(99),
114            AnnounceRateAccounting::Started,
115        );
116        for observed_at in 0..20 {
117            history.record(
118                destination,
119                InstantMillis(observed_at),
120                AnnounceRateAccounting::Continued,
121            );
122        }
123
124        let snapshot = history.snapshot(AnnounceRateState {
125            destination,
126            last_allowed_announce_at: InstantMillis(19),
127            blocked_until: InstantMillis(0),
128            rate_violations: 0,
129        });
130
131        assert_eq!(
132            snapshot.observed_at,
133            (4..20).map(InstantMillis).collect::<Vec<_>>()
134        );
135
136        history.record(
137            destination,
138            InstantMillis(25),
139            AnnounceRateAccounting::Started,
140        );
141
142        assert_eq!(
143            history
144                .snapshot(AnnounceRateState {
145                    destination,
146                    last_allowed_announce_at: InstantMillis(25),
147                    blocked_until: InstantMillis(0),
148                    rate_violations: 0,
149                })
150                .observed_at,
151            vec![InstantMillis(25)]
152        );
153    }
154}
prns_runtime/runtime/node_introspection/impls/heap.rs

prns_runtime/runtime/node_introspection/impls/
heap.rs
