rmux_core/events/
ring.rs

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

rmux_core/events/
ring.rs
