rmux_core/events/
ring.rs

1use std::collections::VecDeque;
2
3use super::cursor::{OutputCursor, OutputCursorItem, OutputGap};
4
5/// Default retained pane-output events per pane.
6pub const DEFAULT_OUTPUT_RING_CAPACITY: usize = 1024;
7/// Default retained recent live output bytes per pane.
8pub const DEFAULT_RECENT_LIVE_BUFFER_CAPACITY: usize = 1024 * 1024;
9
10/// A single pane-output event retained by an [`OutputRing`].
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub struct OutputEvent {
13    sequence: u64,
14    bytes: Vec<u8>,
15}
16
17impl OutputEvent {
18    /// Returns this event's monotonic per-ring sequence.
19    #[must_use]
20    pub const fn sequence(&self) -> u64 {
21        self.sequence
22    }
23
24    /// Returns the raw bytes carried by this output event.
25    #[must_use]
26    pub fn bytes(&self) -> &[u8] {
27        &self.bytes
28    }
29
30    /// Consumes this event and returns its raw bytes.
31    #[must_use]
32    pub fn into_bytes(self) -> Vec<u8> {
33        self.bytes
34    }
35}
36
37/// Bounded recent live bytes retained alongside an output ring.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct RecentOutputSnapshot {
40    bytes: Vec<u8>,
41    oldest_sequence: Option<u64>,
42    newest_sequence: Option<u64>,
43    chunks: Vec<RecentOutputSnapshotChunk>,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47struct RecentOutputSnapshotChunk {
48    sequence: u64,
49    start: usize,
50    starts_at_event_start: bool,
51}
52
53impl RecentOutputSnapshot {
54    /// Returns the retained recent live bytes.
55    #[must_use]
56    pub fn bytes(&self) -> &[u8] {
57        &self.bytes
58    }
59
60    /// Returns retained bytes whose contributing output event sequence is at
61    /// least `min_sequence`.
62    #[must_use]
63    pub fn bytes_from_sequence(&self, min_sequence: u64) -> &[u8] {
64        let start = self
65            .chunks
66            .iter()
67            .find(|chunk| chunk.sequence >= min_sequence)
68            .map_or(self.bytes.len(), |chunk| chunk.start);
69        &self.bytes[start..]
70    }
71
72    /// Returns the oldest output sequence contributing retained bytes.
73    #[must_use]
74    pub const fn oldest_sequence(&self) -> Option<u64> {
75        self.oldest_sequence
76    }
77
78    /// Returns the newest output sequence contributing retained bytes.
79    #[must_use]
80    pub const fn newest_sequence(&self) -> Option<u64> {
81        self.newest_sequence
82    }
83
84    /// Returns the oldest retained contributing sequence at or after
85    /// `min_sequence`.
86    #[must_use]
87    pub fn oldest_sequence_at_or_after(&self, min_sequence: u64) -> Option<u64> {
88        self.chunks
89            .iter()
90            .find(|chunk| chunk.sequence >= min_sequence)
91            .map(|chunk| chunk.sequence)
92    }
93
94    /// Returns whether the retained bytes for `sequence` begin at that output
95    /// event's first byte.
96    #[must_use]
97    pub fn starts_at_event_start(&self, sequence: u64) -> bool {
98        self.chunks
99            .iter()
100            .find(|chunk| chunk.sequence == sequence)
101            .is_some_and(|chunk| chunk.starts_at_event_start)
102    }
103
104    /// Returns the retained byte count.
105    #[must_use]
106    pub fn len(&self) -> usize {
107        self.bytes.len()
108    }
109
110    /// Returns whether the snapshot contains no retained bytes.
111    #[must_use]
112    pub fn is_empty(&self) -> bool {
113        self.bytes.is_empty()
114    }
115}
116
117/// Per-pane bounded live output storage with independent cursor polling.
118#[derive(Debug, Clone)]
119pub struct OutputRing {
120    event_capacity: usize,
121    recent_byte_capacity: usize,
122    next_sequence: u64,
123    events: VecDeque<OutputEvent>,
124    recent: RecentLiveBuffer,
125}
126
127impl OutputRing {
128    /// Creates an empty output ring with explicit event and recent-byte limits.
129    ///
130    /// Both limits must be positive. A zero-sized ring would make every
131    /// subscriber permanently lagged and is rejected at construction.
132    #[must_use]
133    pub fn new(event_capacity: usize, recent_byte_capacity: usize) -> Self {
134        assert!(event_capacity > 0, "output ring capacity must be positive");
135        assert!(
136            recent_byte_capacity > 0,
137            "recent live buffer capacity must be positive"
138        );
139        Self {
140            event_capacity,
141            recent_byte_capacity,
142            next_sequence: 0,
143            events: VecDeque::with_capacity(event_capacity),
144            recent: RecentLiveBuffer::new(recent_byte_capacity),
145        }
146    }
147
148    /// Creates an empty output ring using the v1 defaults.
149    #[must_use]
150    pub fn with_default_capacities() -> Self {
151        Self::new(
152            DEFAULT_OUTPUT_RING_CAPACITY,
153            DEFAULT_RECENT_LIVE_BUFFER_CAPACITY,
154        )
155    }
156
157    /// Appends one output event, rotates the ring, and updates recent live bytes.
158    pub fn push(&mut self, bytes: Vec<u8>) -> OutputEvent {
159        let event = OutputEvent {
160            sequence: self.next_sequence,
161            bytes,
162        };
163        self.next_sequence = self
164            .next_sequence
165            .checked_add(1)
166            .expect("output ring sequence space exhausted");
167        self.recent.push(event.sequence, &event.bytes);
168        self.events.push_back(event.clone());
169        while self.events.len() > self.event_capacity {
170            let _ = self.events.pop_front();
171        }
172        event
173    }
174
175    /// Clears retained events and recent bytes without rewinding the sequence.
176    pub fn clear_retained(&mut self) {
177        self.events.clear();
178        self.recent.clear();
179    }
180
181    /// Returns a cursor that starts with the oldest retained event.
182    #[must_use]
183    pub fn cursor_from_oldest(&self) -> OutputCursor {
184        OutputCursor::new(self.oldest_sequence())
185    }
186
187    /// Returns a cursor that starts after the newest appended event.
188    #[must_use]
189    pub fn cursor_from_now(&self) -> OutputCursor {
190        OutputCursor::new(self.next_sequence)
191    }
192
193    /// Polls one item for `cursor`, reporting gaps before retained events.
194    pub fn poll_cursor(&self, cursor: &mut OutputCursor) -> Option<OutputCursorItem> {
195        let next = cursor.next_sequence();
196        let oldest = self.oldest_sequence();
197        if next < oldest {
198            let missed = oldest.saturating_sub(next);
199            cursor.record_gap(missed, oldest);
200            return Some(OutputCursorItem::Gap(OutputGap::new(
201                next,
202                oldest,
203                missed,
204                self.newest_sequence(),
205                self.recent_snapshot(),
206            )));
207        }
208
209        if next >= self.next_sequence {
210            return None;
211        }
212
213        let offset = usize::try_from(next.saturating_sub(oldest)).ok()?;
214        let event = self.events.get(offset).cloned()?;
215        cursor.advance_to(next.wrapping_add(1));
216        Some(OutputCursorItem::Event(event))
217    }
218
219    /// Polls up to `limit` items for `cursor` from one retained-ring snapshot.
220    ///
221    /// A lag gap is returned only as the first item. Once the cursor is inside
222    /// the retained range, the same immutable ring view cannot produce a later
223    /// gap in this batch; callers therefore never advance over an event and
224    /// then replace it with a lag response from a concurrently rotated ring.
225    pub fn poll_cursor_batch(
226        &self,
227        cursor: &mut OutputCursor,
228        limit: usize,
229    ) -> Vec<OutputCursorItem> {
230        let mut items = Vec::new();
231        for _ in 0..limit {
232            let Some(item) = self.poll_cursor(cursor) else {
233                break;
234            };
235            let is_gap = matches!(item, OutputCursorItem::Gap(_));
236            items.push(item);
237            if is_gap {
238                break;
239            }
240        }
241        items
242    }
243
244    /// Returns the oldest retained event sequence, or the next sequence if empty.
245    #[must_use]
246    pub fn oldest_sequence(&self) -> u64 {
247        self.events
248            .front()
249            .map_or(self.next_sequence, OutputEvent::sequence)
250    }
251
252    /// Returns the next sequence that will be assigned.
253    #[must_use]
254    pub const fn next_sequence(&self) -> u64 {
255        self.next_sequence
256    }
257
258    /// Returns the newest appended sequence, or zero before the first append.
259    #[must_use]
260    pub fn newest_sequence(&self) -> u64 {
261        self.next_sequence.saturating_sub(1)
262    }
263
264    /// Returns the configured event capacity.
265    #[must_use]
266    pub const fn event_capacity(&self) -> usize {
267        self.event_capacity
268    }
269
270    /// Returns the configured recent live byte capacity.
271    #[must_use]
272    pub const fn recent_byte_capacity(&self) -> usize {
273        self.recent_byte_capacity
274    }
275
276    /// Returns retained event count.
277    #[must_use]
278    pub fn retained_len(&self) -> usize {
279        self.events.len()
280    }
281
282    /// Returns the total bytes currently retained in recent live storage.
283    #[must_use]
284    pub fn recent_len(&self) -> usize {
285        self.recent.len()
286    }
287
288    /// Returns a bounded recent live output snapshot.
289    #[must_use]
290    pub fn recent_snapshot(&self) -> RecentOutputSnapshot {
291        self.recent.snapshot()
292    }
293
294    /// Returns retained events in sequence order.
295    #[must_use]
296    pub fn retained_events(&self) -> Vec<OutputEvent> {
297        self.events.iter().cloned().collect()
298    }
299}
300
301impl Default for OutputRing {
302    fn default() -> Self {
303        Self::with_default_capacities()
304    }
305}
306
307#[derive(Debug, Clone)]
308struct RecentLiveBuffer {
309    capacity: usize,
310    len: usize,
311    chunks: VecDeque<RecentLiveChunk>,
312}
313
314#[derive(Debug, Clone)]
315struct RecentLiveChunk {
316    sequence: u64,
317    bytes: Vec<u8>,
318    starts_at_event_start: bool,
319}
320
321impl RecentLiveBuffer {
322    fn new(capacity: usize) -> Self {
323        Self {
324            capacity,
325            len: 0,
326            chunks: VecDeque::new(),
327        }
328    }
329
330    fn push(&mut self, sequence: u64, bytes: &[u8]) {
331        if bytes.is_empty() {
332            return;
333        }
334        if bytes.len() >= self.capacity {
335            self.chunks.clear();
336            self.chunks.push_back(RecentLiveChunk {
337                sequence,
338                bytes: bytes[bytes.len() - self.capacity..].to_vec(),
339                starts_at_event_start: bytes.len() == self.capacity,
340            });
341            self.len = self.capacity;
342            return;
343        }
344        self.chunks.push_back(RecentLiveChunk {
345            sequence,
346            bytes: bytes.to_vec(),
347            starts_at_event_start: true,
348        });
349        self.len = self.len.saturating_add(bytes.len());
350        self.trim_front();
351    }
352
353    fn clear(&mut self) {
354        self.chunks.clear();
355        self.len = 0;
356    }
357
358    fn trim_front(&mut self) {
359        while self.len > self.capacity {
360            let overflow = self.len - self.capacity;
361            let Some(front) = self.chunks.front_mut() else {
362                self.len = 0;
363                return;
364            };
365            if front.bytes.len() <= overflow {
366                self.len -= front.bytes.len();
367                let _ = self.chunks.pop_front();
368            } else {
369                front.bytes = front.bytes.split_off(overflow);
370                front.starts_at_event_start = false;
371                self.len -= overflow;
372            }
373        }
374    }
375
376    const fn len(&self) -> usize {
377        self.len
378    }
379
380    fn oldest_sequence(&self) -> Option<u64> {
381        self.chunks.front().map(|chunk| chunk.sequence)
382    }
383
384    fn newest_sequence(&self) -> Option<u64> {
385        self.chunks.back().map(|chunk| chunk.sequence)
386    }
387
388    fn snapshot(&self) -> RecentOutputSnapshot {
389        let mut bytes = Vec::with_capacity(self.len);
390        let mut snapshot_chunks = Vec::with_capacity(self.chunks.len());
391        for chunk in &self.chunks {
392            let start = bytes.len();
393            bytes.extend_from_slice(&chunk.bytes);
394            snapshot_chunks.push(RecentOutputSnapshotChunk {
395                sequence: chunk.sequence,
396                start,
397                starts_at_event_start: chunk.starts_at_event_start,
398            });
399        }
400        RecentOutputSnapshot {
401            bytes,
402            oldest_sequence: self.oldest_sequence(),
403            newest_sequence: self.newest_sequence(),
404            chunks: snapshot_chunks,
405        }
406    }
407}
408
409#[cfg(test)]
410mod tests {
411    use super::{OutputRing, DEFAULT_OUTPUT_RING_CAPACITY, DEFAULT_RECENT_LIVE_BUFFER_CAPACITY};
412    use crate::events::{OutputCursor, OutputCursorItem};
413
414    #[test]
415    fn default_capacities_match_recorded_budget() {
416        let ring = OutputRing::default();
417        assert_eq!(ring.event_capacity(), DEFAULT_OUTPUT_RING_CAPACITY);
418        assert_eq!(
419            ring.recent_byte_capacity(),
420            DEFAULT_RECENT_LIVE_BUFFER_CAPACITY
421        );
422        assert_eq!(DEFAULT_OUTPUT_RING_CAPACITY, 1_024);
423        assert_eq!(DEFAULT_RECENT_LIVE_BUFFER_CAPACITY, 1_048_576);
424    }
425
426    #[test]
427    fn ring_rotation_keeps_only_most_recent_events() {
428        let mut ring = OutputRing::new(2, 64);
429        ring.push(b"zero".to_vec());
430        ring.push(b"one".to_vec());
431        ring.push(b"two".to_vec());
432
433        let sequences = ring
434            .retained_events()
435            .iter()
436            .map(|event| event.sequence())
437            .collect::<Vec<_>>();
438        assert_eq!(sequences, vec![1, 2]);
439        assert_eq!(ring.oldest_sequence(), 1);
440        assert_eq!(ring.next_sequence(), 3);
441    }
442
443    #[test]
444    fn recent_live_buffer_obeys_byte_bound() {
445        let mut ring = OutputRing::new(8, 5);
446        ring.push(b"abc".to_vec());
447        ring.push(b"defg".to_vec());
448        ring.push(b"hi".to_vec());
449
450        let snapshot = ring.recent_snapshot();
451        assert_eq!(snapshot.bytes(), b"efghi");
452        assert_eq!(snapshot.len(), 5);
453        assert_eq!(snapshot.oldest_sequence(), Some(1));
454        assert_eq!(snapshot.newest_sequence(), Some(2));
455        assert_eq!(ring.recent_len(), 5);
456    }
457
458    #[test]
459    fn recent_live_buffer_releases_trimmed_prefix_capacity() {
460        let mut ring = OutputRing::new(8, 4);
461        ring.push(b"abcd".to_vec());
462        ring.push(b"ef".to_vec());
463
464        assert_eq!(ring.recent_snapshot().bytes(), b"cdef");
465        assert_eq!(ring.recent_len(), 4);
466        let retained_capacity = ring
467            .recent
468            .chunks
469            .iter()
470            .map(|chunk| chunk.bytes.capacity())
471            .sum::<usize>();
472        assert!(
473            retained_capacity <= ring.recent_byte_capacity(),
474            "recent buffer retained capacity {retained_capacity} exceeds configured bound {}",
475            ring.recent_byte_capacity()
476        );
477    }
478
479    #[test]
480    fn recent_live_buffer_trims_oversized_single_event_to_bound() {
481        let mut ring = OutputRing::new(8, 4);
482        ring.push(b"012345".to_vec());
483
484        assert_eq!(ring.recent_snapshot().bytes(), b"2345");
485        assert_eq!(ring.recent_snapshot().oldest_sequence(), Some(0));
486        assert_eq!(ring.recent_snapshot().newest_sequence(), Some(0));
487        assert_eq!(ring.recent_len(), 4);
488        assert_eq!(ring.retained_events()[0].bytes(), b"012345");
489    }
490
491    #[test]
492    fn recent_snapshot_filters_bytes_by_contributing_sequence() {
493        let mut ring = OutputRing::new(8, 64);
494        ring.push(b"stale".to_vec());
495        ring.push(b"future".to_vec());
496        ring.push(b"tail".to_vec());
497
498        let snapshot = ring.recent_snapshot();
499
500        assert_eq!(snapshot.bytes_from_sequence(0), b"stalefuturetail");
501        assert_eq!(snapshot.bytes_from_sequence(1), b"futuretail");
502        assert_eq!(snapshot.bytes_from_sequence(2), b"tail");
503        assert_eq!(snapshot.bytes_from_sequence(3), b"");
504        assert_eq!(snapshot.oldest_sequence_at_or_after(1), Some(1));
505        assert_eq!(snapshot.oldest_sequence_at_or_after(3), None);
506        assert!(snapshot.starts_at_event_start(1));
507    }
508
509    #[test]
510    fn recent_snapshot_records_when_retained_event_prefix_was_trimmed() {
511        let mut ring = OutputRing::new(8, 4);
512        ring.push(b"012345".to_vec());
513
514        let snapshot = ring.recent_snapshot();
515
516        assert_eq!(snapshot.bytes_from_sequence(0), b"2345");
517        assert_eq!(snapshot.oldest_sequence_at_or_after(0), Some(0));
518        assert!(!snapshot.starts_at_event_start(0));
519    }
520
521    #[test]
522    fn cursor_lag_across_full_rotation_reports_all_missed_events() {
523        let mut ring = OutputRing::new(3, 16);
524        let mut cursor = OutputCursor::new(0);
525        for index in 0..6 {
526            ring.push(format!("{index}").into_bytes());
527        }
528
529        let Some(OutputCursorItem::Gap(gap)) = ring.poll_cursor(&mut cursor) else {
530            panic!("cursor should lag after ring rotation");
531        };
532        assert_eq!(gap.expected_sequence(), 0);
533        assert_eq!(gap.resume_sequence(), 3);
534        assert_eq!(gap.missed_events(), 3);
535        assert_eq!(gap.missed_range(), 0..3);
536        assert_eq!(gap.recent_snapshot().bytes(), b"012345");
537        assert_eq!(gap.recent_snapshot().oldest_sequence(), Some(0));
538        assert_eq!(gap.recent_snapshot().newest_sequence(), Some(5));
539        assert_eq!(cursor.missed_events(), 3);
540    }
541
542    #[test]
543    fn cursor_polls_rotated_ring_by_sequence_offset() {
544        let mut ring = OutputRing::new(3, 16);
545        for index in 0..6 {
546            ring.push(format!("{index}").into_bytes());
547        }
548        let mut cursor = OutputCursor::new(4);
549
550        let Some(OutputCursorItem::Event(event)) = ring.poll_cursor(&mut cursor) else {
551            panic!("cursor should read retained event from rotated ring");
552        };
553        assert_eq!(event.sequence(), 4);
554        assert_eq!(event.bytes(), b"4");
555        assert_eq!(cursor.next_sequence(), 5);
556    }
557
558    #[test]
559    fn batch_poll_reports_gap_only_for_lagged_cursor() {
560        let mut ring = OutputRing::new(2, 16);
561        let mut stale = OutputCursor::new(0);
562        let mut aligned = OutputCursor::new(2);
563        for index in 0..4 {
564            ring.push(format!("{index}").into_bytes());
565        }
566
567        let stale_batch = ring.poll_cursor_batch(&mut stale, 8);
568        assert_eq!(stale_batch.len(), 1);
569        let OutputCursorItem::Gap(gap) = &stale_batch[0] else {
570            panic!("stale cursor should report its own output gap");
571        };
572        assert_eq!(gap.expected_sequence(), 0);
573        assert_eq!(gap.resume_sequence(), 2);
574        assert_eq!(gap.missed_events(), 2);
575        assert_eq!(stale.next_sequence(), 2);
576
577        let aligned_batch = ring.poll_cursor_batch(&mut aligned, 8);
578        let sequences = aligned_batch
579            .iter()
580            .map(|item| match item {
581                OutputCursorItem::Event(event) => event.sequence(),
582                OutputCursorItem::Gap(gap) => {
583                    panic!("aligned cursor must not inherit stale cursor lag: {gap:?}")
584                }
585            })
586            .collect::<Vec<_>>();
587        assert_eq!(sequences, vec![2, 3]);
588        assert_eq!(aligned.missed_events(), 0);
589        assert_eq!(aligned.next_sequence(), ring.next_sequence());
590    }
591
592    #[test]
593    fn clear_retained_drops_recent_snapshot_range_without_rewinding_sequence() {
594        let mut ring = OutputRing::new(3, 16);
595        let mut cursor = OutputCursor::new(0);
596        ring.push(b"one".to_vec());
597        ring.push(b"two".to_vec());
598
599        ring.clear_retained();
600
601        assert_eq!(ring.next_sequence(), 2);
602        let snapshot = ring.recent_snapshot();
603        assert!(snapshot.is_empty());
604        assert_eq!(snapshot.oldest_sequence(), None);
605        assert_eq!(snapshot.newest_sequence(), None);
606
607        let Some(OutputCursorItem::Gap(gap)) = ring.poll_cursor(&mut cursor) else {
608            panic!("cursor should observe cleared retained output as a gap");
609        };
610        assert_eq!(gap.expected_sequence(), 0);
611        assert_eq!(gap.resume_sequence(), 2);
612        assert_eq!(gap.missed_events(), 2);
613        assert_eq!(gap.missed_range(), 0..2);
614        assert_eq!(gap.newest_sequence(), 1);
615        assert!(gap.recent_snapshot().is_empty());
616    }
617}
rmux_core/events/ring.rs

rmux_core/events/
ring.rs
