rmux_core/events/
ring.rs

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

rmux_core/events/
ring.rs
