sentinel_core/correlate/
window.rs

1//! Sliding window correlator for streaming mode.
2//!
3//! Accumulates normalized events by `trace_id` with ring buffer, TTL eviction,
4//! and O(1) LRU eviction when max active traces is exceeded.
5
6use std::collections::VecDeque;
7use std::num::NonZeroUsize;
8
9use lru::LruCache;
10
11use crate::normalize::NormalizedEvent;
12
13/// Configuration for the trace window.
14#[derive(Debug, Clone)]
15pub struct WindowConfig {
16    /// Maximum events kept per trace (ring buffer).
17    pub max_events_per_trace: usize,
18    /// Trace time-to-live in milliseconds.
19    pub trace_ttl_ms: u64,
20    /// Maximum number of active traces before LRU eviction. Must be >= 1.
21    pub max_active_traces: NonZeroUsize,
22}
23
24/// Default LRU cap for the streaming correlator (compile-time non-zero).
25const DEFAULT_MAX_ACTIVE_TRACES: NonZeroUsize =
26    NonZeroUsize::new(10_000).expect("non-zero literal");
27
28impl Default for WindowConfig {
29    fn default() -> Self {
30        Self {
31            max_events_per_trace: 1000,
32            trace_ttl_ms: 30_000,
33            max_active_traces: DEFAULT_MAX_ACTIVE_TRACES,
34        }
35    }
36}
37
38/// Buffer for a single trace.
39struct TraceBuffer {
40    events: VecDeque<NormalizedEvent>,
41    /// Absolute timestamp (ms since epoch) of the last event pushed to this trace.
42    /// Used for TTL eviction: the LRU cache handles relative access ordering.
43    last_seen_ms: u64,
44}
45
46/// Sliding window that accumulates events by `trace_id`.
47///
48/// Uses an LRU cache for O(1) amortized eviction when at capacity.
49pub struct TraceWindow {
50    config: WindowConfig,
51    traces: LruCache<String, TraceBuffer>,
52}
53
54impl TraceWindow {
55    #[must_use]
56    pub fn new(config: WindowConfig) -> Self {
57        let cap = config.max_active_traces;
58        Self {
59            config,
60            traces: LruCache::new(cap),
61        }
62    }
63
64    /// Push a normalized event into the window.
65    ///
66    /// Returns the LRU-evicted trace (if any) so the caller can run detection
67    /// on it before discarding. Returns `None` if no eviction was needed.
68    pub fn push(
69        &mut self,
70        event: NormalizedEvent,
71        now_ms: u64,
72    ) -> Option<(String, Vec<NormalizedEvent>)> {
73        // Fast path: trace already exists: get_mut auto-promotes to MRU.
74        if let Some(buf) = self.traces.get_mut(event.event.trace_id.as_str()) {
75            buf.last_seen_ms = now_ms;
76            buf.events.push_back(event);
77            // Ring buffer: drop oldest if over capacity
78            if buf.events.len() > self.config.max_events_per_trace {
79                buf.events.pop_front();
80            }
81            return None;
82        }
83
84        // Slow path: new trace, clone trace_id; push evicts LRU if at cap.
85        let trace_id = event.event.trace_id.clone();
86        let mut events = VecDeque::with_capacity(8);
87        events.push_back(event);
88
89        self.traces
90            .push(
91                trace_id,
92                TraceBuffer {
93                    events,
94                    last_seen_ms: now_ms,
95                },
96            )
97            .map(|(id, buf)| (id, Vec::from(buf.events)))
98    }
99
100    /// Evict traces that have not been updated within the TTL.
101    ///
102    /// Scans the full LRU cache rather than stopping at the first non-expired
103    /// entry, because clock adjustments (NTP) can cause `last_seen_ms` and LRU
104    /// position to diverge, leaving expired traces behind non-expired ones.
105    ///
106    /// The key cloning into a temporary `Vec<String>` is required because
107    /// the `lru` crate does not expose `retain()` or `drain_filter()`.
108    /// At `max_active_traces = 10_000` the cost is bounded and runs at
109    /// most once per tick (~15s). If the `lru` crate adds in-place removal
110    /// in a future release, this can be simplified.
111    pub fn evict(&mut self, now_ms: u64) {
112        for key in self.collect_expired_keys(now_ms) {
113            self.traces.pop(&key);
114        }
115    }
116
117    /// Evict expired traces and return them for processing.
118    ///
119    /// Unlike `evict()` which silently drops expired traces, this method
120    /// returns them so the daemon can run detection before discarding.
121    /// Scans the full cache to handle clock skew (see `evict()`).
122    pub fn evict_expired(&mut self, now_ms: u64) -> Vec<(String, Vec<NormalizedEvent>)> {
123        let expired_keys = self.collect_expired_keys(now_ms);
124        let mut expired = Vec::with_capacity(expired_keys.len());
125        for key in expired_keys {
126            if let Some((_id, buf)) = self.traces.pop_entry(&key) {
127                expired.push((key, Vec::from(buf.events)));
128            }
129        }
130        expired
131    }
132
133    /// Collect trace IDs whose `last_seen_ms` is older than `trace_ttl_ms`.
134    /// Shared by `evict()` and `evict_expired()`.
135    fn collect_expired_keys(&self, now_ms: u64) -> Vec<String> {
136        let ttl = self.config.trace_ttl_ms;
137        self.traces
138            .iter()
139            .filter(|(_, buf)| now_ms.saturating_sub(buf.last_seen_ms) > ttl)
140            .map(|(id, _)| id.clone())
141            .collect()
142    }
143
144    /// Drain all traces, returning their events grouped by `trace_id`.
145    pub fn drain_all(&mut self) -> Vec<(String, Vec<NormalizedEvent>)> {
146        let mut result = Vec::with_capacity(self.traces.len());
147        while let Some((id, buf)) = self.traces.pop_lru() {
148            result.push((id, Vec::from(buf.events)));
149        }
150        result
151    }
152
153    /// Number of active traces.
154    #[must_use]
155    pub fn active_traces(&self) -> usize {
156        self.traces.len()
157    }
158
159    /// Clone a trace's spans without evicting or promoting it in the LRU.
160    /// Returns `None` if the trace is not in the window.
161    #[must_use]
162    pub fn peek_clone(&self, trace_id: &str) -> Option<Vec<NormalizedEvent>> {
163        self.traces
164            .peek(trace_id)
165            .map(|buf| buf.events.iter().cloned().collect())
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use std::sync::Arc;
172
173    use super::*;
174    use crate::event::{EventSource, EventType, SpanEvent};
175    use crate::normalize;
176
177    fn make_event(trace_id: &str, target: &str) -> NormalizedEvent {
178        let event = SpanEvent {
179            timestamp: "2025-07-10T14:32:01.123Z".to_string(),
180            trace_id: trace_id.to_string(),
181            span_id: "span-1".to_string(),
182            parent_span_id: None,
183            service: Arc::from("test"),
184            cloud_region: None,
185            event_type: EventType::Sql,
186            operation: "SELECT".to_string(),
187            target: target.to_string(),
188            duration_us: 100,
189            source: EventSource {
190                endpoint: "GET /test".to_string(),
191                method: "Test::test".to_string(),
192            },
193            status_code: None,
194            response_size_bytes: None,
195            code_function: None,
196            code_filepath: None,
197            code_lineno: None,
198            code_namespace: None,
199            instrumentation_scopes: Vec::new(),
200        };
201        normalize::normalize(event)
202    }
203
204    #[test]
205    fn accumulates_events_by_trace() {
206        let mut w = TraceWindow::new(WindowConfig::default());
207        w.push(make_event("t1", "SELECT 1"), 0);
208        w.push(make_event("t1", "SELECT 2"), 10);
209        w.push(make_event("t2", "SELECT 3"), 20);
210
211        assert_eq!(w.active_traces(), 2);
212        let drained = w.drain_all();
213        let t1 = drained.iter().find(|(id, _)| id == "t1").unwrap();
214        assert_eq!(t1.1.len(), 2);
215    }
216
217    #[test]
218    fn ring_buffer_overflow() {
219        let config = WindowConfig {
220            max_events_per_trace: 3,
221            ..Default::default()
222        };
223        let mut w = TraceWindow::new(config);
224        for i in 0..5 {
225            w.push(
226                make_event("t1", &format!("SELECT {i}")),
227                u64::try_from(i).unwrap(),
228            );
229        }
230
231        let drained = w.drain_all();
232        let t1 = drained.iter().find(|(id, _)| id == "t1").unwrap();
233        assert_eq!(t1.1.len(), 3);
234        // Should have the last 3 events (2, 3, 4)
235        assert_eq!(t1.1[0].event.target, "SELECT 2");
236        assert_eq!(t1.1[2].event.target, "SELECT 4");
237    }
238
239    #[test]
240    fn ttl_eviction() {
241        let config = WindowConfig {
242            trace_ttl_ms: 100,
243            ..Default::default()
244        };
245        let mut w = TraceWindow::new(config);
246        w.push(make_event("t1", "SELECT 1"), 0);
247        w.push(make_event("t2", "SELECT 2"), 50);
248
249        w.evict(150);
250        // t1 last_seen=0, now=150, diff=150 > 100 -> evicted
251        // t2 last_seen=50, now=150, diff=100 -> NOT evicted (100 <= 100)
252        assert_eq!(w.active_traces(), 1);
253        let drained = w.drain_all();
254        assert_eq!(drained[0].0, "t2");
255    }
256
257    #[test]
258    fn lru_eviction() {
259        let config = WindowConfig {
260            max_active_traces: NonZeroUsize::new(2).unwrap(),
261            ..Default::default()
262        };
263        let mut w = TraceWindow::new(config);
264        w.push(make_event("t1", "SELECT 1"), 0);
265        w.push(make_event("t2", "SELECT 2"), 10);
266        // This should evict t1 (LRU: oldest access)
267        let evicted = w.push(make_event("t3", "SELECT 3"), 20);
268
269        assert!(evicted.is_some());
270        assert_eq!(evicted.unwrap().0, "t1");
271        assert_eq!(w.active_traces(), 2);
272        assert!(w.traces.peek(&"t2".to_string()).is_some());
273        assert!(w.traces.peek(&"t3".to_string()).is_some());
274        assert!(w.traces.peek(&"t1".to_string()).is_none());
275    }
276
277    #[test]
278    fn drain_empties_window() {
279        let mut w = TraceWindow::new(WindowConfig::default());
280        w.push(make_event("t1", "SELECT 1"), 0);
281        let drained = w.drain_all();
282        assert_eq!(drained.len(), 1);
283        assert_eq!(w.active_traces(), 0);
284    }
285
286    #[test]
287    fn lru_touch_prevents_eviction() {
288        let config = WindowConfig {
289            max_active_traces: NonZeroUsize::new(2).unwrap(),
290            ..Default::default()
291        };
292        let mut w = TraceWindow::new(config);
293        w.push(make_event("t1", "SELECT 1"), 0);
294        w.push(make_event("t2", "SELECT 2"), 10);
295        // Touch t1 so it becomes more recent than t2 (get_mut promotes to MRU)
296        w.push(make_event("t1", "SELECT 1b"), 20);
297        // Insert t3: should evict t2 (LRU), not t1 (MRU)
298        let evicted = w.push(make_event("t3", "SELECT 3"), 30);
299
300        assert!(evicted.is_some());
301        assert_eq!(evicted.unwrap().0, "t2");
302        assert_eq!(w.active_traces(), 2);
303        assert!(w.traces.peek(&"t1".to_string()).is_some());
304        assert!(w.traces.peek(&"t3".to_string()).is_some());
305        assert!(w.traces.peek(&"t2".to_string()).is_none());
306    }
307
308    #[test]
309    fn evict_on_empty_window() {
310        let mut w = TraceWindow::new(WindowConfig::default());
311        w.evict(1000);
312        assert_eq!(w.active_traces(), 0);
313    }
314
315    #[test]
316    fn ttl_evicts_all_expired() {
317        let config = WindowConfig {
318            trace_ttl_ms: 50,
319            ..Default::default()
320        };
321        let mut w = TraceWindow::new(config);
322        w.push(make_event("t1", "SELECT 1"), 0);
323        w.push(make_event("t2", "SELECT 2"), 10);
324        // Both expired at now=200
325        w.evict(200);
326        assert_eq!(w.active_traces(), 0);
327    }
328
329    #[test]
330    fn drain_empty_window() {
331        let mut w = TraceWindow::new(WindowConfig::default());
332        let drained = w.drain_all();
333        assert!(drained.is_empty());
334    }
335
336    #[test]
337    fn lru_eviction_chain() {
338        let config = WindowConfig {
339            max_active_traces: NonZeroUsize::new(1).unwrap(),
340            ..Default::default()
341        };
342        let mut w = TraceWindow::new(config);
343
344        let evicted1 = w.push(make_event("t1", "SELECT 1"), 0);
345        assert!(evicted1.is_none()); // first insert, no eviction
346
347        let evicted2 = w.push(make_event("t2", "SELECT 2"), 10);
348        // t1 evicted, only t2 remains
349        assert!(evicted2.is_some());
350        assert_eq!(evicted2.unwrap().0, "t1");
351        assert_eq!(w.active_traces(), 1);
352        assert!(w.traces.peek(&"t2".to_string()).is_some());
353
354        let evicted3 = w.push(make_event("t3", "SELECT 3"), 20);
355        // t2 evicted, only t3 remains
356        assert!(evicted3.is_some());
357        assert_eq!(evicted3.unwrap().0, "t2");
358        assert_eq!(w.active_traces(), 1);
359        assert!(w.traces.peek(&"t3".to_string()).is_some());
360    }
361
362    #[test]
363    fn evict_expired_returns_traces() {
364        let config = WindowConfig {
365            trace_ttl_ms: 100,
366            ..Default::default()
367        };
368        let mut w = TraceWindow::new(config);
369        w.push(make_event("t1", "SELECT 1"), 0);
370        w.push(make_event("t2", "SELECT 2"), 50);
371
372        // Not yet expired
373        let expired = w.evict_expired(50);
374        assert!(expired.is_empty());
375        assert_eq!(w.active_traces(), 2);
376
377        // t1 expired (150 - 0 = 150 > 100), t2 not (150 - 50 = 100 <= 100)
378        let expired = w.evict_expired(150);
379        assert_eq!(expired.len(), 1);
380        assert_eq!(expired[0].0, "t1");
381        assert_eq!(w.active_traces(), 1);
382    }
383
384    #[test]
385    fn push_returns_evicted_events() {
386        let config = WindowConfig {
387            max_active_traces: NonZeroUsize::new(1).unwrap(),
388            ..Default::default()
389        };
390        let mut w = TraceWindow::new(config);
391        w.push(make_event("t1", "SELECT 1"), 0);
392        w.push(make_event("t1", "SELECT 2"), 5);
393
394        let evicted = w.push(make_event("t2", "SELECT 3"), 10);
395        assert!(evicted.is_some());
396        let (id, events) = evicted.unwrap();
397        assert_eq!(id, "t1");
398        assert_eq!(events.len(), 2); // both events from t1
399    }
400}
sentinel_core/correlate/window.rs

sentinel_core/correlate/
window.rs
