ferravex_net/
reliable.rs

1//! Bounded ordered event delivery over an unreliable transport. This is not encryption.
2use std::collections::{BTreeMap, VecDeque};
3
4pub const EVENT_LIMIT: usize = 256;
5pub const EVENT_BYTES: usize = 900;
6const RETRY_MS: u64 = 100;
7#[derive(Clone, Debug, PartialEq, Eq)]
8pub struct EventPacket {
9    pub sequence: u64,
10    pub payload: Vec<u8>,
11}
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum Error {
14    Full,
15    Oversized,
16    SequenceExhausted,
17    OutsideWindow,
18}
19struct Pending {
20    packet: EventPacket,
21    last_sent: Option<u64>,
22}
23#[derive(Default)]
24pub struct EventSender {
25    next: u64,
26    pending: VecDeque<Pending>,
27}
28impl EventSender {
29    pub fn queue(&mut self, payload: &[u8]) -> Result<u64, Error> {
30        if payload.len() > EVENT_BYTES {
31            return Err(Error::Oversized);
32        }
33        if self.pending.len() >= EVENT_LIMIT {
34            return Err(Error::Full);
35        }
36        let sequence = self.next;
37        self.next = self.next.checked_add(1).ok_or(Error::SequenceExhausted)?;
38        self.pending.push_back(Pending {
39            packet: EventPacket {
40                sequence,
41                payload: payload.to_vec(),
42            },
43            last_sent: None,
44        });
45        Ok(sequence)
46    }
47    /// Returns at most `budget` events. Call with a monotonic millisecond clock.
48    pub fn due(&mut self, now: u64, budget: usize) -> Vec<EventPacket> {
49        self.pending
50            .iter_mut()
51            .filter(|p| {
52                p.last_sent
53                    .is_none_or(|last| now.saturating_sub(last) >= RETRY_MS)
54            })
55            .take(budget.min(EVENT_LIMIT))
56            .map(|p| {
57                p.last_sent = Some(now);
58                p.packet.clone()
59            })
60            .collect()
61    }
62    /// Acknowledgments are individual, so loss of an earlier event cannot be hidden by a newer ACK.
63    pub fn acknowledge(&mut self, sequence: u64) -> bool {
64        if let Some(index) = self
65            .pending
66            .iter()
67            .position(|p| p.packet.sequence == sequence)
68        {
69            self.pending.remove(index);
70            true
71        } else {
72            false
73        }
74    }
75    pub fn pending(&self) -> usize {
76        self.pending.len()
77    }
78}
79#[derive(Default)]
80pub struct EventReceiver {
81    next: u64,
82    buffered: BTreeMap<u64, Vec<u8>>,
83}
84impl EventReceiver {
85    /// ACK only on success, including duplicates. Delivery is strictly once and ordered.
86    /// The receive window bounds memory even when event zero never arrives.
87    pub fn receive(&mut self, packet: &EventPacket) -> Result<Vec<EventPacket>, Error> {
88        if packet.payload.len() > EVENT_BYTES {
89            return Err(Error::Oversized);
90        }
91        if packet.sequence < self.next {
92            return Ok(Vec::new());
93        }
94        if packet.sequence - self.next >= EVENT_LIMIT as u64 {
95            return Err(Error::OutsideWindow);
96        }
97        self.buffered
98            .entry(packet.sequence)
99            .or_insert_with(|| packet.payload.clone());
100        let mut delivered = Vec::new();
101        while let Some(payload) = self.buffered.remove(&self.next) {
102            let sequence = self.next;
103            self.next = self.next.checked_add(1).ok_or(Error::SequenceExhausted)?;
104            delivered.push(EventPacket { sequence, payload });
105        }
106        Ok(delivered)
107    }
108    pub fn next_sequence(&self) -> u64 {
109        self.next
110    }
111    pub fn contains_buffered(&self, sequence: u64) -> bool {
112        self.buffered.contains_key(&sequence)
113    }
114    pub fn buffered(&self) -> usize {
115        self.buffered.len()
116    }
117}
118#[cfg(test)]
119mod tests {
120    use super::*;
121    #[test]
122    fn lost_reordered_and_duplicate_events_deliver_once_in_order() {
123        let (mut sender, mut receiver) = (EventSender::default(), EventReceiver::default());
124        for i in 0..8 {
125            sender.queue(&[i]).unwrap();
126        }
127        let packets = sender.due(0, 8);
128        let mut delivered = Vec::new();
129        for i in [7, 2, 2, 5, 1, 4, 6, 3] {
130            delivered.extend(receiver.receive(&packets[i]).unwrap());
131            sender.acknowledge(i as u64);
132        }
133        assert!(delivered.is_empty());
134        assert_eq!(sender.pending(), 1);
135        assert!(sender.due(99, 8).is_empty());
136        let retry = sender.due(100, 8);
137        delivered.extend(receiver.receive(&retry[0]).unwrap());
138        sender.acknowledge(0);
139        assert_eq!(
140            delivered.iter().map(|p| p.payload[0]).collect::<Vec<_>>(),
141            (0..8).collect::<Vec<_>>()
142        );
143        assert!(receiver.receive(&retry[0]).unwrap().is_empty());
144        assert_eq!(sender.pending(), 0);
145    }
146    #[test]
147    fn flow_control_bounds_memory_and_rejects_future_ack() {
148        let mut sender = EventSender::default();
149        for _ in 0..EVENT_LIMIT {
150            sender.queue(&[1]).unwrap();
151        }
152        assert_eq!(sender.queue(&[]), Err(Error::Full));
153        assert!(!sender.acknowledge(9999));
154        assert_eq!(sender.due(0, 3).len(), 3);
155        let mut receiver = EventReceiver::default();
156        assert_eq!(
157            receiver.receive(&EventPacket {
158                sequence: 9999,
159                payload: vec![]
160            }),
161            Err(Error::OutsideWindow)
162        );
163        assert_eq!(receiver.buffered(), 0);
164    }
165}
ferravex_net/reliable.rs

ferravex_net/
reliable.rs
