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    /// Returns a cursor that starts at an explicit output sequence.
216    #[must_use]
217    pub const fn cursor_from_sequence(sequence: u64) -> OutputCursor {
218        OutputCursor::new(sequence)
219    }
220
221    /// Polls one item for `cursor`, reporting gaps before retained events.
222    pub fn poll_cursor(&self, cursor: &mut OutputCursor) -> Option<OutputCursorItem> {
223        let next = cursor.next_sequence();
224        let oldest = self.oldest_sequence();
225        if next < oldest {
226            let missed = oldest.saturating_sub(next);
227            cursor.record_gap(missed, oldest);
228            return Some(OutputCursorItem::Gap(OutputGap::new(
229                next,
230                oldest,
231                missed,
232                self.newest_sequence(),
233                self.recent_snapshot(),
234            )));
235        }
236
237        if next >= self.next_sequence {
238            return None;
239        }
240
241        let offset = usize::try_from(next.saturating_sub(oldest)).ok()?;
242        let event = self.events.get(offset).cloned()?;
243        cursor.advance_to(next.wrapping_add(1));
244        Some(OutputCursorItem::Event(event))
245    }
246
247    /// Polls up to `limit` items for `cursor` from one retained-ring snapshot.
248    ///
249    /// A lag gap is returned only as the first item. Once the cursor is inside
250    /// the retained range, the same immutable ring view cannot produce a later
251    /// gap in this batch; callers therefore never advance over an event and
252    /// then replace it with a lag response from a concurrently rotated ring.
253    pub fn poll_cursor_batch(
254        &self,
255        cursor: &mut OutputCursor,
256        limit: usize,
257    ) -> Vec<OutputCursorItem> {
258        let mut items = Vec::new();
259        for _ in 0..limit {
260            let Some(item) = self.poll_cursor(cursor) else {
261                break;
262            };
263            let is_gap = matches!(item, OutputCursorItem::Gap(_));
264            items.push(item);
265            if is_gap {
266                break;
267            }
268        }
269        items
270    }
271
272    /// Returns the oldest retained event sequence, or the next sequence if empty.
273    #[must_use]
274    pub fn oldest_sequence(&self) -> u64 {
275        self.events
276            .front()
277            .map_or(self.next_sequence, OutputEvent::sequence)
278    }
279
280    /// Returns the next sequence that will be assigned.
281    #[must_use]
282    pub const fn next_sequence(&self) -> u64 {
283        self.next_sequence
284    }
285
286    /// Returns the newest appended sequence, or zero before the first append.
287    #[must_use]
288    pub fn newest_sequence(&self) -> u64 {
289        self.next_sequence.saturating_sub(1)
290    }
291
292    /// Returns the configured event capacity.
293    #[must_use]
294    pub const fn event_capacity(&self) -> usize {
295        self.event_capacity
296    }
297
298    /// Returns the configured recent live byte capacity.
299    #[must_use]
300    pub const fn recent_byte_capacity(&self) -> usize {
301        self.recent_byte_capacity
302    }
303
304    /// Returns retained event count.
305    #[must_use]
306    pub fn retained_len(&self) -> usize {
307        self.events.len()
308    }
309
310    /// Returns the total bytes currently retained in recent live storage.
311    #[must_use]
312    pub fn recent_len(&self) -> usize {
313        self.recent.len()
314    }
315
316    /// Returns a bounded recent live output snapshot.
317    #[must_use]
318    pub fn recent_snapshot(&self) -> RecentOutputSnapshot {
319        self.recent.snapshot()
320    }
321
322    /// Returns retained events in sequence order.
323    #[must_use]
324    pub fn retained_events(&self) -> Vec<OutputEvent> {
325        self.events.iter().cloned().collect()
326    }
327}
328
329impl Default for OutputRing {
330    fn default() -> Self {
331        Self::with_default_capacities()
332    }
333}
334
335#[derive(Debug, Clone)]
336struct RecentLiveBuffer {
337    capacity: usize,
338    len: usize,
339    chunks: VecDeque<RecentLiveChunk>,
340}
341
342#[derive(Debug, Clone)]
343struct RecentLiveChunk {
344    sequence: u64,
345    bytes: Vec<u8>,
346    starts_at_event_start: bool,
347}
348
349impl RecentLiveBuffer {
350    fn new(capacity: usize) -> Self {
351        Self {
352            capacity,
353            len: 0,
354            chunks: VecDeque::new(),
355        }
356    }
357
358    fn push(&mut self, sequence: u64, bytes: &[u8]) {
359        if bytes.is_empty() {
360            return;
361        }
362        if bytes.len() >= self.capacity {
363            self.chunks.clear();
364            self.chunks.push_back(RecentLiveChunk {
365                sequence,
366                bytes: bytes[bytes.len() - self.capacity..].to_vec(),
367                starts_at_event_start: bytes.len() == self.capacity,
368            });
369            self.len = self.capacity;
370            return;
371        }
372        self.chunks.push_back(RecentLiveChunk {
373            sequence,
374            bytes: bytes.to_vec(),
375            starts_at_event_start: true,
376        });
377        self.len = self.len.saturating_add(bytes.len());
378        self.trim_front();
379    }
380
381    fn clear(&mut self) {
382        self.chunks.clear();
383        self.len = 0;
384    }
385
386    fn trim_front(&mut self) {
387        while self.len > self.capacity {
388            let overflow = self.len - self.capacity;
389            let Some(front) = self.chunks.front_mut() else {
390                self.len = 0;
391                return;
392            };
393            if front.bytes.len() <= overflow {
394                self.len -= front.bytes.len();
395                let _ = self.chunks.pop_front();
396            } else {
397                front.bytes = front.bytes.split_off(overflow);
398                front.starts_at_event_start = false;
399                self.len -= overflow;
400            }
401        }
402    }
403
404    const fn len(&self) -> usize {
405        self.len
406    }
407
408    fn oldest_sequence(&self) -> Option<u64> {
409        self.chunks.front().map(|chunk| chunk.sequence)
410    }
411
412    fn newest_sequence(&self) -> Option<u64> {
413        self.chunks.back().map(|chunk| chunk.sequence)
414    }
415
416    fn snapshot(&self) -> RecentOutputSnapshot {
417        let mut bytes = Vec::with_capacity(self.len);
418        let mut snapshot_chunks = Vec::with_capacity(self.chunks.len());
419        for chunk in &self.chunks {
420            let start = bytes.len();
421            bytes.extend_from_slice(&chunk.bytes);
422            snapshot_chunks.push(RecentOutputSnapshotChunk {
423                sequence: chunk.sequence,
424                start,
425                starts_at_event_start: chunk.starts_at_event_start,
426            });
427        }
428        RecentOutputSnapshot {
429            bytes,
430            oldest_sequence: self.oldest_sequence(),
431            newest_sequence: self.newest_sequence(),
432            chunks: snapshot_chunks,
433        }
434    }
435}
436
437#[cfg(test)]
438mod tests {
439    use super::{OutputRing, DEFAULT_OUTPUT_RING_CAPACITY, DEFAULT_RECENT_LIVE_BUFFER_CAPACITY};
440    use crate::events::{OutputCursor, OutputCursorItem};
441
442    #[test]
443    fn default_capacities_match_recorded_budget() {
444        let ring = OutputRing::default();
445        assert_eq!(ring.event_capacity(), DEFAULT_OUTPUT_RING_CAPACITY);
446        assert_eq!(
447            ring.recent_byte_capacity(),
448            DEFAULT_RECENT_LIVE_BUFFER_CAPACITY
449        );
450        assert_eq!(DEFAULT_OUTPUT_RING_CAPACITY, 1_024);
451        assert_eq!(DEFAULT_RECENT_LIVE_BUFFER_CAPACITY, 1_048_576);
452    }
453
454    #[test]
455    fn ring_rotation_keeps_only_most_recent_events() {
456        let mut ring = OutputRing::new(2, 64);
457        ring.push(b"zero".to_vec());
458        ring.push(b"one".to_vec());
459        ring.push(b"two".to_vec());
460
461        let sequences = ring
462            .retained_events()
463            .iter()
464            .map(|event| event.sequence())
465            .collect::<Vec<_>>();
466        assert_eq!(sequences, vec![1, 2]);
467        assert_eq!(ring.oldest_sequence(), 1);
468        assert_eq!(ring.next_sequence(), 3);
469    }
470
471    #[test]
472    fn recent_live_buffer_obeys_byte_bound() {
473        let mut ring = OutputRing::new(8, 5);
474        ring.push(b"abc".to_vec());
475        ring.push(b"defg".to_vec());
476        ring.push(b"hi".to_vec());
477
478        let snapshot = ring.recent_snapshot();
479        assert_eq!(snapshot.bytes(), b"efghi");
480        assert_eq!(snapshot.len(), 5);
481        assert_eq!(snapshot.oldest_sequence(), Some(1));
482        assert_eq!(snapshot.newest_sequence(), Some(2));
483        assert_eq!(ring.recent_len(), 5);
484    }
485
486    #[test]
487    fn recent_live_buffer_releases_trimmed_prefix_capacity() {
488        let mut ring = OutputRing::new(8, 4);
489        ring.push(b"abcd".to_vec());
490        ring.push(b"ef".to_vec());
491
492        assert_eq!(ring.recent_snapshot().bytes(), b"cdef");
493        assert_eq!(ring.recent_len(), 4);
494        let retained_capacity = ring
495            .recent
496            .chunks
497            .iter()
498            .map(|chunk| chunk.bytes.capacity())
499            .sum::<usize>();
500        assert!(
501            retained_capacity <= ring.recent_byte_capacity(),
502            "recent buffer retained capacity {retained_capacity} exceeds configured bound {}",
503            ring.recent_byte_capacity()
504        );
505    }
506
507    #[test]
508    fn recent_live_buffer_trims_oversized_single_event_to_bound() {
509        let mut ring = OutputRing::new(8, 4);
510        ring.push(b"012345".to_vec());
511
512        assert_eq!(ring.recent_snapshot().bytes(), b"2345");
513        assert_eq!(ring.recent_snapshot().oldest_sequence(), Some(0));
514        assert_eq!(ring.recent_snapshot().newest_sequence(), Some(0));
515        assert_eq!(ring.recent_len(), 4);
516        assert_eq!(ring.retained_events()[0].bytes(), b"012345");
517    }
518
519    #[test]
520    fn recent_snapshot_filters_bytes_by_contributing_sequence() {
521        let mut ring = OutputRing::new(8, 64);
522        ring.push(b"stale".to_vec());
523        ring.push(b"future".to_vec());
524        ring.push(b"tail".to_vec());
525
526        let snapshot = ring.recent_snapshot();
527
528        assert_eq!(snapshot.bytes_from_sequence(0), b"stalefuturetail");
529        assert_eq!(snapshot.bytes_from_sequence(1), b"futuretail");
530        assert_eq!(snapshot.bytes_from_sequence(2), b"tail");
531        assert_eq!(snapshot.bytes_from_sequence(3), b"");
532        assert_eq!(snapshot.oldest_sequence_at_or_after(1), Some(1));
533        assert_eq!(snapshot.oldest_sequence_at_or_after(3), None);
534        assert!(snapshot.starts_at_event_start(1));
535    }
536
537    #[test]
538    fn recent_snapshot_records_when_retained_event_prefix_was_trimmed() {
539        let mut ring = OutputRing::new(8, 4);
540        ring.push(b"012345".to_vec());
541
542        let snapshot = ring.recent_snapshot();
543
544        assert_eq!(snapshot.bytes_from_sequence(0), b"2345");
545        assert_eq!(snapshot.oldest_sequence_at_or_after(0), Some(0));
546        assert!(!snapshot.starts_at_event_start(0));
547    }
548
549    #[test]
550    fn cursor_lag_across_full_rotation_reports_all_missed_events() {
551        let mut ring = OutputRing::new(3, 16);
552        let mut cursor = OutputCursor::new(0);
553        for index in 0..6 {
554            ring.push(format!("{index}").into_bytes());
555        }
556
557        let Some(OutputCursorItem::Gap(gap)) = ring.poll_cursor(&mut cursor) else {
558            panic!("cursor should lag after ring rotation");
559        };
560        assert_eq!(gap.expected_sequence(), 0);
561        assert_eq!(gap.resume_sequence(), 3);
562        assert_eq!(gap.missed_events(), 3);
563        assert_eq!(gap.missed_range(), 0..3);
564        assert_eq!(gap.recent_snapshot().bytes(), b"012345");
565        assert_eq!(gap.recent_snapshot().oldest_sequence(), Some(0));
566        assert_eq!(gap.recent_snapshot().newest_sequence(), Some(5));
567        assert_eq!(cursor.missed_events(), 3);
568    }
569
570    #[test]
571    fn cursor_polls_rotated_ring_by_sequence_offset() {
572        let mut ring = OutputRing::new(3, 16);
573        for index in 0..6 {
574            ring.push(format!("{index}").into_bytes());
575        }
576        let mut cursor = OutputCursor::new(4);
577
578        let Some(OutputCursorItem::Event(event)) = ring.poll_cursor(&mut cursor) else {
579            panic!("cursor should read retained event from rotated ring");
580        };
581        assert_eq!(event.sequence(), 4);
582        assert_eq!(event.bytes(), b"4");
583        assert_eq!(cursor.next_sequence(), 5);
584    }
585
586    #[test]
587    fn batch_poll_reports_gap_only_for_lagged_cursor() {
588        let mut ring = OutputRing::new(2, 16);
589        let mut stale = OutputCursor::new(0);
590        let mut aligned = OutputCursor::new(2);
591        for index in 0..4 {
592            ring.push(format!("{index}").into_bytes());
593        }
594
595        let stale_batch = ring.poll_cursor_batch(&mut stale, 8);
596        assert_eq!(stale_batch.len(), 1);
597        let OutputCursorItem::Gap(gap) = &stale_batch[0] else {
598            panic!("stale cursor should report its own output gap");
599        };
600        assert_eq!(gap.expected_sequence(), 0);
601        assert_eq!(gap.resume_sequence(), 2);
602        assert_eq!(gap.missed_events(), 2);
603        assert_eq!(stale.next_sequence(), 2);
604
605        let aligned_batch = ring.poll_cursor_batch(&mut aligned, 8);
606        let sequences = aligned_batch
607            .iter()
608            .map(|item| match item {
609                OutputCursorItem::Event(event) => event.sequence(),
610                OutputCursorItem::Gap(gap) => {
611                    panic!("aligned cursor must not inherit stale cursor lag: {gap:?}")
612                }
613            })
614            .collect::<Vec<_>>();
615        assert_eq!(sequences, vec![2, 3]);
616        assert_eq!(aligned.missed_events(), 0);
617        assert_eq!(aligned.next_sequence(), ring.next_sequence());
618    }
619
620    #[test]
621    fn clear_retained_drops_recent_snapshot_range_without_rewinding_sequence() {
622        let mut ring = OutputRing::new(3, 16);
623        let mut cursor = OutputCursor::new(0);
624        ring.push(b"one".to_vec());
625        ring.push(b"two".to_vec());
626
627        ring.clear_retained();
628
629        assert_eq!(ring.next_sequence(), 2);
630        let snapshot = ring.recent_snapshot();
631        assert!(snapshot.is_empty());
632        assert_eq!(snapshot.oldest_sequence(), None);
633        assert_eq!(snapshot.newest_sequence(), None);
634
635        let Some(OutputCursorItem::Gap(gap)) = ring.poll_cursor(&mut cursor) else {
636            panic!("cursor should observe cleared retained output as a gap");
637        };
638        assert_eq!(gap.expected_sequence(), 0);
639        assert_eq!(gap.resume_sequence(), 2);
640        assert_eq!(gap.missed_events(), 2);
641        assert_eq!(gap.missed_range(), 0..2);
642        assert_eq!(gap.newest_sequence(), 1);
643        assert!(gap.recent_snapshot().is_empty());
644    }
645}
rmux_core/events/ring.rs

rmux_core/events/
ring.rs
