teksilo_telemetry/queue/
mem.rs

1// SPDX-License-Identifier: MPL-2.0
2// SPDX-FileCopyrightText: 2026 FernTech
3
4//! In-memory `EventQueue` implementation.
5
6use std::collections::VecDeque;
7use std::sync::Mutex;
8
9use teksilo_core::telemetry::OwnedEvent;
10
11use super::EventQueue;
12
13/// Bounded FIFO event buffer with `Send + Sync` access.
14///
15/// Capacity defaults to 10_000; oldest events are dropped past that.
16/// Use [`PersistentEventQueue`](super::PersistentEventQueue) when
17/// surviving process restart matters.
18pub struct InMemoryEventQueue {
19    inner: Mutex<VecDeque<OwnedEvent>>,
20    capacity: usize,
21}
22
23impl InMemoryEventQueue {
24    pub fn new() -> Self {
25        Self::with_capacity(10_000)
26    }
27
28    pub fn with_capacity(capacity: usize) -> Self {
29        Self {
30            inner: Mutex::new(VecDeque::with_capacity(capacity.min(256))),
31            capacity,
32        }
33    }
34}
35
36impl Default for InMemoryEventQueue {
37    fn default() -> Self {
38        Self::new()
39    }
40}
41
42impl EventQueue for InMemoryEventQueue {
43    fn push(&self, event: OwnedEvent) {
44        let mut q = self.inner.lock().expect("queue mutex poisoned");
45        if q.len() >= self.capacity {
46            q.pop_front();
47        }
48        q.push_back(event);
49    }
50
51    fn len(&self) -> usize {
52        self.inner.lock().expect("queue mutex poisoned").len()
53    }
54
55    fn drain_batch(&self, n: usize) -> Vec<OwnedEvent> {
56        let mut q = self.inner.lock().expect("queue mutex poisoned");
57        let take = n.min(q.len());
58        q.drain(..take).collect()
59    }
60
61    fn discard_all(&self) {
62        let mut q = self.inner.lock().expect("queue mutex poisoned");
63        q.clear();
64    }
65
66    fn peek_recent(&self, n: usize) -> Vec<OwnedEvent> {
67        let q = self.inner.lock().expect("queue mutex poisoned");
68        q.iter().rev().take(n).cloned().collect::<Vec<_>>()
69    }
70}
71
72impl std::fmt::Debug for InMemoryEventQueue {
73    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74        f.debug_struct("InMemoryEventQueue")
75            .field("len", &self.len())
76            .field("capacity", &self.capacity)
77            .finish()
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84    use std::time::SystemTime;
85    use teksilo_core::telemetry::EventCategory;
86
87    fn ev(name: &str) -> OwnedEvent {
88        OwnedEvent {
89            name: name.to_string(),
90            category: EventCategory::Intent,
91            timestamp: SystemTime::UNIX_EPOCH,
92            install_id: None,
93            session_id: "test".into(),
94            schema_version: 1,
95            props: vec![],
96        }
97    }
98
99    #[test]
100    fn push_pop_round_trip() {
101        let q = InMemoryEventQueue::new();
102        q.push(ev("a"));
103        q.push(ev("b"));
104        assert_eq!(q.len(), 2);
105        let batch = q.drain_batch(10);
106        assert_eq!(batch.len(), 2);
107        assert_eq!(batch[0].name, "a");
108        assert_eq!(batch[1].name, "b");
109        assert!(q.is_empty());
110    }
111
112    #[test]
113    fn capacity_drops_oldest() {
114        let q = InMemoryEventQueue::with_capacity(3);
115        for name in ["a", "b", "c", "d", "e"] {
116            q.push(ev(name));
117        }
118        assert_eq!(q.len(), 3);
119        let batch = q.drain_batch(10);
120        let names: Vec<String> = batch.iter().map(|e| e.name.clone()).collect();
121        assert_eq!(names, vec!["c".to_string(), "d".into(), "e".into()]);
122    }
123
124    #[test]
125    fn discard_all_empties() {
126        let q = InMemoryEventQueue::new();
127        q.push(ev("a"));
128        q.push(ev("b"));
129        q.discard_all();
130        assert!(q.is_empty());
131    }
132
133    #[test]
134    fn peek_recent_returns_newest_first() {
135        let q = InMemoryEventQueue::new();
136        q.push(ev("a"));
137        q.push(ev("b"));
138        q.push(ev("c"));
139        let recent = q.peek_recent(2);
140        assert_eq!(recent[0].name, "c");
141        assert_eq!(recent[1].name, "b");
142    }
143}
teksilo_telemetry/queue/mem.rs

teksilo_telemetry/queue/
mem.rs
