logicaffeine_data/
backpressure.rs

1//! `BoundedRing<T>` — a fixed-capacity queue that drops the OLDEST item on overflow.
2//!
3//! This is the backpressure primitive for the real-time relay's per-connection outbound queue.
4//! The relay hands each connection an UNBOUNDED channel today (`logicaffeine_system::relay`), so a
5//! single slow consumer under a fast producer grows without bound — a latent OOM. A live-media
6//! queue must instead be bounded, and when it fills it should shed the STALEST frames (a late video
7//! frame is worthless; FEC and the next keyframe cover the gap), never the freshest, and never
8//! stall the whole relay. Every shed frame is counted in [`BoundedRing::dropped`] — the exact
9//! signal the congestion controller (`adaptRate`) reads to back off. Bounded memory,
10//! freshest-data-wins, drop-aware: the correct real-time transport policy.
11
12use std::collections::VecDeque;
13
14/// A FIFO queue capped at `cap` items; a push into a full ring evicts the oldest.
15#[derive(Debug, Clone)]
16pub struct BoundedRing<T> {
17    buf: VecDeque<T>,
18    cap: usize,
19    dropped: u64,
20}
21
22impl<T> BoundedRing<T> {
23    /// A ring holding at most `cap` items. `cap` must be positive.
24    pub fn new(cap: usize) -> Self {
25        assert!(cap > 0, "a BoundedRing needs a positive capacity");
26        Self {
27            buf: VecDeque::with_capacity(cap),
28            cap,
29            dropped: 0,
30        }
31    }
32
33    /// Enqueue `item`. If the ring is already full, the OLDEST item is evicted first (and counted
34    /// in [`dropped`](Self::dropped)); the evicted item is returned so the caller can telemeter it.
35    pub fn push(&mut self, item: T) -> Option<T> {
36        let evicted = if self.buf.len() == self.cap {
37            self.dropped += 1;
38            self.buf.pop_front()
39        } else {
40            None
41        };
42        self.buf.push_back(item);
43        evicted
44    }
45
46    /// Dequeue the oldest surviving item (FIFO).
47    pub fn pop(&mut self) -> Option<T> {
48        self.buf.pop_front()
49    }
50
51    pub fn len(&self) -> usize {
52        self.buf.len()
53    }
54
55    pub fn is_empty(&self) -> bool {
56        self.buf.is_empty()
57    }
58
59    pub fn is_full(&self) -> bool {
60        self.buf.len() == self.cap
61    }
62
63    pub fn capacity(&self) -> usize {
64        self.cap
65    }
66
67    /// The number of items shed to overflow since construction — the drop telemetry the congestion
68    /// controller reads to lower the send rate.
69    pub fn dropped(&self) -> u64 {
70        self.dropped
71    }
72}
73
74#[cfg(test)]
75#[path = "../../logicaffeine_tests/unit/logicaffeine_data/backpressure/tests.rs"]
76mod tests;
logicaffeine_data/backpressure.rs

logicaffeine_data/
backpressure.rs
