perf(eval): select percentile and median ranks instead of sorting the window

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to RSigma are documented in this file. Each entry correspond
 
 ## [Unreleased]
 
+### Percentile and median checks no longer sort the window (#619)
+
+`value_percentile` and `value_median` correlations sorted every finite value in the group's window on each event fed into the group. They now select the needed ranks with `select_nth_unstable_by`, which is O(n) on average and returns exactly the same values, including interpolated percentiles and even-count medians. In the new `correlation_numeric_window` benchmark, a group checked against a 5,000-value window processes events about 4x faster. Chained correlations also stop cloning every fired result, including its retained events in full event mode, just to read which correlation fired and for which group.
+
 ### OTLP log decoding is fuzzed (#617)
 
 OTLP log ingestion decodes untrusted protobuf and flattens each record to JSON, but it was the only untrusted input surface without a `cargo-fuzz` harness. The new `fuzz_otlp_logs_decode` target decodes arbitrary bytes as `ExportLogsServiceRequest`, converts whatever decodes with `logs_request_to_raw_events`, and asserts each event is valid JSON. It runs in the weekly fuzz workflow from a committed seed corpus. Conversion recurses once per nested `AnyValue`, so its depth depends on prost's decode recursion limit; a new unit test fails if a dependency turns that limit off through prost's `no-recursion-limit` feature.

crates/rsigma-eval/benches/correlation.rs

@@ -1,8 +1,8 @@
 //! Correlation engine benchmarks for rsigma-eval.
 //!
 //! Measures event_count and temporal correlation performance, end-to-end
-//! throughput with mixed detection + correlation, and state map pressure
-//! from many unique group keys.
+//! throughput with mixed detection + correlation, state map pressure from
+//! many unique group keys, and percentile/median checks over large windows.
 
 mod datagen;
 
@@ -375,6 +375,88 @@ level: high
     group.finish();
 }
 
+// ---------------------------------------------------------------------------
+// Benchmark: percentile and median over large windows
+// ---------------------------------------------------------------------------
+
+fn bench_correlation_numeric_window(c: &mut Criterion) {
+    let mut group = c.benchmark_group("correlation_numeric_window");
+    group.sample_size(20);
+
+    // One group fed one event per second into a sliding window of `size`
+    // seconds, so after the warm-up every measured event is checked against
+    // a full window of `size` values.
+    let yaml_for = |corr_type: &str, size: usize| {
+        format!(
+            r#"
+title: Latency Sample
+id: bench-latency-001
+logsource:
+    product: web
+detection:
+    selection:
+        EventType: 'request'
+    condition: selection
+level: low
+---
+title: Latency Window
+id: corr-latency-001
+correlation:
+    type: {corr_type}
+    rules:
+        - bench-latency-001
+    group-by:
+        - Host
+    timespan: {size}s
+    condition:
+        field: latency
+        percentile: 95
+        gte: 1000000000
+level: high
+"#
+        )
+    };
+
+    let n_events = 1_000usize;
+    let mut rng = datagen::rng();
+    let event_values: Vec<serde_json::Value> = (0..n_events)
+        .map(|_| {
+            serde_json::json!({
+                "EventType": "request",
+                "Host": "web-01",
+                "latency": rng.random_range(0.0..5_000.0),
+            })
+        })
+        .collect();
+    let events: Vec<JsonEvent> = event_values.iter().map(JsonEvent::borrow).collect();
+
+    for corr_type in ["value_percentile", "value_median"] {
+        for size in [1_000usize, 5_000] {
+            let collection = parse_sigma_yaml(&yaml_for(corr_type, size)).unwrap();
+            let mut engine = CorrelationEngine::new(CorrelationConfig::default());
+            engine.add_collection(&collection).unwrap();
+            let mut ts = 1_000_000i64;
+            for i in 0..size {
+                let _ = engine.process_event_at(&events[i % n_events], ts);
+                ts += 1;
+            }
+
+            group.throughput(criterion::Throughput::Elements(n_events as u64));
+            group.bench_with_input(BenchmarkId::new(corr_type, size), &events, |b, events| {
+                b.iter(|| {
+                    for event in events {
+                        let result = engine.process_event_at(black_box(event), ts);
+                        black_box(&result);
+                        ts += 1;
+                    }
+                });
+            });
+        }
+    }
+
+    group.finish();
+}
+
 // ---------------------------------------------------------------------------
 // Criterion harness
 // ---------------------------------------------------------------------------
@@ -387,5 +469,6 @@ criterion_group!(
     bench_correlation_batch,
     bench_correlation_state_pressure,
     bench_correlation_window_modes,
+    bench_correlation_numeric_window,
 );
 criterion_main!(benches);

crates/rsigma-eval/src/correlation/tests.rs

@@ -451,13 +451,13 @@ fn test_check_condition_with_extended_expr() {
 
 #[test]
 fn test_percentile_linear_interp_single() {
-    assert!((percentile_linear_interp(&[42.0], 50.0) - 42.0).abs() < f64::EPSILON);
+    assert!((percentile_linear_interp(&mut [42.0], 50.0) - 42.0).abs() < f64::EPSILON);
 }
 
 #[test]
 fn test_percentile_linear_interp_basic() {
     // Values: [1, 2, 3, 4, 5]
-    let values = &[1.0, 2.0, 3.0, 4.0, 5.0];
+    let values = &mut [1.0, 2.0, 3.0, 4.0, 5.0];
     // 0th percentile = 1.0
     assert!((percentile_linear_interp(values, 0.0) - 1.0).abs() < f64::EPSILON);
     // 25th percentile = 2.0
@@ -473,20 +473,100 @@ fn test_percentile_linear_interp_basic() {
 #[test]
 fn test_percentile_linear_interp_interpolation() {
     // Values: [10, 20, 30, 40]
-    let values = &[10.0, 20.0, 30.0, 40.0];
+    let values = &mut [10.0, 20.0, 30.0, 40.0];
     // 50th percentile: rank = 0.5 * 3 = 1.5, interp between 20 and 30 = 25
     assert!((percentile_linear_interp(values, 50.0) - 25.0).abs() < f64::EPSILON);
 }
 
 #[test]
 fn test_percentile_linear_interp_1st_percentile() {
     // Values: [1, 2, 3, ..., 100]
-    let values: Vec<f64> = (1..=100).map(|x| x as f64).collect();
+    let mut values: Vec<f64> = (1..=100).map(|x| x as f64).collect();
     // 1st percentile = 1.0 + 0.01 * 99 * (2.0 - 1.0) ~ 1.99
-    let p1 = percentile_linear_interp(&values, 1.0);
+    let p1 = percentile_linear_interp(&mut values, 1.0);
     assert!((p1 - 1.99).abs() < 0.01);
 }
 
+/// Percentile by full sort, the definition the selection-based version must
+/// reproduce exactly.
+fn sorted_percentile(values: &[f64], percentile: f64) -> f64 {
+    let mut sorted = values.to_vec();
+    sorted.sort_by(f64::total_cmp);
+    let n = sorted.len();
+    if n == 1 {
+        return sorted[0];
+    }
+    let rank = percentile.clamp(0.0, 100.0) / 100.0 * (n - 1) as f64;
+    let (lower, upper) = (rank.floor() as usize, rank.ceil() as usize);
+    if lower == upper || upper >= n {
+        sorted[lower.min(n - 1)]
+    } else {
+        sorted[lower] + (rank - lower as f64) * (sorted[upper] - sorted[lower])
+    }
+}
+
+/// Deterministic unsorted samples with repeats and signed zeros.
+fn scrambled_samples(n: usize) -> Vec<f64> {
+    let mut x: u64 = 0x9E37_79B9_7F4A_7C15 ^ n as u64;
+    (0..n)
+        .map(|i| {
+            x = x
+                .wrapping_mul(6_364_136_223_846_793_005)
+                .wrapping_add(1_442_695_040_888_963_407);
+            match i % 7 {
+                0 => 0.0,
+                1 => -0.0,
+                2 => ((x >> 40) % 16) as f64,
+                _ => ((x >> 11) as f64 / (1u64 << 53) as f64 - 0.5) * 1e6,
+            }
+        })
+        .collect()
+}
+
+#[test]
+fn test_percentile_selection_matches_sorted_reference() {
+    for n in (1..=64).chain([99, 100, 1_000, 5_001]) {
+        let samples = scrambled_samples(n);
+        for p in (0..=100).map(f64::from).chain([0.5, 33.3, 99.9]) {
+            let got = percentile_linear_interp(&mut samples.clone(), p);
+            let want = sorted_percentile(&samples, p);
+            assert_eq!(
+                got.to_bits(),
+                want.to_bits(),
+                "n={n} p={p}: {got} != {want}"
+            );
+        }
+    }
+}
+
+#[test]
+fn test_window_median_matches_sorted_reference() {
+    let cond = CompiledCondition {
+        field: Some(vec!["latency".to_string()]),
+        predicates: vec![(ConditionOperator::Gte, f64::MIN)],
+        percentile: None,
+    };
+    for n in (1..=64).chain([99, 100, 1_000, 5_000]) {
+        let samples = scrambled_samples(n);
+        let mut state = WindowState::new_for(CorrelationType::ValueMedian);
+        for (i, v) in samples.iter().enumerate() {
+            state.push_numeric(i as i64, *v);
+        }
+        let mut sorted = samples.clone();
+        sorted.sort_by(f64::total_cmp);
+        let mid = n / 2;
+        let want = if n % 2 == 0 {
+            (sorted[mid - 1] + sorted[mid]) / 2.0
+        } else {
+            sorted[mid]
+        };
+        let checked = state.check_condition(&cond, CorrelationType::ValueMedian, &[], None);
+        let current = state.current_value(CorrelationType::ValueMedian, &[], None);
+        assert_eq!(checked.map(f64::to_bits), Some(want.to_bits()), "n={n}");
+        assert_eq!(current.map(f64::to_bits), Some(want.to_bits()), "n={n}");
+    }
+}
+
 #[test]
 fn test_value_percentile_check_condition() {
     let mut state = WindowState::new_for(CorrelationType::ValuePercentile);
@@ -509,14 +589,14 @@ fn test_value_percentile_check_condition() {
 
 #[test]
 fn test_percentile_0th_and_100th() {
-    let values = &[5.0, 10.0, 15.0, 20.0];
+    let values = &mut [5.0, 10.0, 15.0, 20.0];
     assert!((percentile_linear_interp(values, 0.0) - 5.0).abs() < f64::EPSILON);
     assert!((percentile_linear_interp(values, 100.0) - 20.0).abs() < f64::EPSILON);
 }
 
 #[test]
 fn test_percentile_two_values() {
-    let values = &[10.0, 20.0];
+    let values = &mut [10.0, 20.0];
     // 50th percentile between 10 and 20 = 15
     assert!((percentile_linear_interp(values, 50.0) - 15.0).abs() < f64::EPSILON);
     // 25th percentile = 12.5
@@ -525,7 +605,7 @@ fn test_percentile_two_values() {
 
 #[test]
 fn test_percentile_clamps_out_of_range() {
-    let values = &[1.0, 2.0, 3.0];
+    let values = &mut [1.0, 2.0, 3.0];
     // Negative percentile clamps to 0
     assert!((percentile_linear_interp(values, -10.0) - 1.0).abs() < f64::EPSILON);
     // > 100 clamps to 100