fix(runtime): cap inhibition sources and warn about unbounded long correlation windows

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to RSigma are documented in this file. Each entry correspond
 
 ## [Unreleased]
 
+### The inhibit store is capped, and long unbounded correlation windows warn at load (#616)
+
+`InhibitStore` was the only alert-pipeline store without a count cap: sources were pruned by `duration` only, so `equal` selectors over attacker-influenced values (source IPs, usernames) could grow it without bound. A new top-level `max_inhibit_sources` key (default 100000) caps tracked sources across all `inhibit_rules`. At the cap, a result that would open a new source is still checked as a target but is not tracked, and the refusal is counted on the new `rsigma_inhibit_sources_refused_total{rule}` counter; already-tracked sources keep refreshing. Restore and hot-reload trim the store to the cap, evicting the least recently seen sources, and `CapTrim` reports them as `inhibit_sources`. For library users, `AlertPipelineFile` gains `max_inhibit_sources`, and `MetricsHook` gains `on_inhibit_source_refused`, which defaults to a no-op.
+
+`max_group_entries` stays unbounded by default, because a finite default would silently stop `event_count` rules whose threshold exceeds it. Instead, a correlation whose `timespan` is longer than one hour and that has no per-group cap, from `--max-group-entries` or `rsigma.max_group_entries`, now logs a warning at load, since a busy group's window grows with `timespan` x event rate.
+
 ### Delivery failures without a DLQ sink are logged (#615)
 
 When a sink exhausted its retries and no `--dlq` was set, the daemon counted the dropped result on `rsigma_dlq_events_total` and logged it once, at `debug`, so every later dropped alert left no trace in the logs. Each drop now produces a `warn` log, rate-limited to one every 10 seconds; each warning carries the number of results dropped since the previous one and the last delivery error. A failed write to a configured DLQ sink is now counted on the new `rsigma_dlq_write_failures_total` counter, and the delivery layer and the daemon's DLQ bridge log a failure they could not queue instead of discarding it silently. The docs now say that a sink commits a failed result by handing it to the DLQ writer, so the source ack does not wait for the DLQ sink's write.

crates/rsigma-cli/src/daemon/metrics.rs

@@ -86,6 +86,7 @@ pub struct Metrics {
     pub silences_active: IntGauge,
     pub inhibited_total: IntCounterVec,
     pub inhibit_sources_active: IntGauge,
+    pub inhibit_sources_refused_total: IntCounterVec,
     pub rule_false_positive_ratio: GaugeVec,
     pub dispositions_total: IntCounterVec,
     pub disposition_ingest_total: IntCounterVec,
@@ -852,12 +853,23 @@ impl Metrics {
         ))
         .unwrap();
         inhibit_sources_active.set(0);
+        let inhibit_sources_refused_total = IntCounterVec::new(
+            Opts::new(
+                "rsigma_inhibit_sources_refused_total",
+                "Inhibition sources not tracked because max_inhibit_sources was reached, by rule name",
+            ),
+            &["rule"],
+        )
+        .unwrap();
         registry
             .register(Box::new(inhibited_total.clone()))
             .unwrap();
         registry
             .register(Box::new(inhibit_sources_active.clone()))
             .unwrap();
+        registry
+            .register(Box::new(inhibit_sources_refused_total.clone()))
+            .unwrap();
 
         // Triage feedback loop (#70). The gauge and the per-rule disposition
         // counter label on `rule_title` to line up with
@@ -1225,6 +1237,7 @@ impl Metrics {
             silences_active,
             inhibited_total,
             inhibit_sources_active,
+            inhibit_sources_refused_total,
             rule_false_positive_ratio,
             dispositions_total,
             disposition_ingest_total,
@@ -1427,6 +1440,12 @@ impl MetricsHook for Metrics {
         self.inhibit_sources_active.set(count);
     }
 
+    fn on_inhibit_source_refused(&self, rule: &str) {
+        self.inhibit_sources_refused_total
+            .with_label_values(&[rule])
+            .inc();
+    }
+
     fn on_capture_event_dropped(&self, reason: &str) {
         self.capture_events_dropped_total
             .with_label_values(&[reason])

crates/rsigma-eval/src/correlation_engine/mod.rs

@@ -56,6 +56,11 @@ const SNAPSHOT_VERSION: u32 = 1;
 /// with a `WARN`.
 const MAX_CHAIN_DEPTH: usize = 10;
 
+/// Timespan above which a correlation without a per-group entry cap is
+/// warned about at load: a busy group's window retains `timespan` x event
+/// rate entries, which `max_state_entries` does not see.
+const LONG_WINDOW_WARN_SECS: u64 = 3600;
+
 /// Stateful correlation engine.
 ///
 /// Wraps the stateless `Engine` for detection rules and adds time-windowed
@@ -285,6 +290,15 @@ impl<L> CorrelationEngine<L> {
         self.apply_custom_attributes(&effective.custom_attributes);
 
         let compiled = compile_correlation(effective)?;
+        if self.has_unbounded_long_window(&compiled) {
+            log::warn!(
+                "Correlation '{}' has a {}s timespan and no per-group entry cap; \
+                 a busy group retains timespan x event rate entries. \
+                 Set max_group_entries (--max-group-entries) or the rsigma.max_group_entries attribute to bound it.",
+                compiled.id.as_deref().unwrap_or(&compiled.title),
+                compiled.timespan_secs
+            );
+        }
         let idx = self.correlations.len();
 
         // Index by each referenced rule ID/name
@@ -310,6 +324,16 @@ impl<L> CorrelationEngine<L> {
         Ok(())
     }
 
+    /// Whether `corr` keeps a window longer than [`LONG_WINDOW_WARN_SECS`]
+    /// with no per-group entry cap from the rule or the engine config.
+    fn has_unbounded_long_window(&self, corr: &CompiledCorrelation) -> bool {
+        corr.timespan_secs > LONG_WINDOW_WARN_SECS
+            && corr
+                .max_group_entries
+                .or(self.config.max_group_entries)
+                .is_none()
+    }
+
     /// Add all rules and correlations from a parsed collection.
     ///
     /// Detection rules are added first (so they're available for correlation

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

@@ -573,6 +573,35 @@ fn windowed_engine_with_group_cap(window_lines: &str, cap: usize) -> Correlation
     engine
 }
 
+#[test]
+fn test_long_window_without_group_cap_is_flagged() {
+    let flagged = |window_lines: &str, config: CorrelationConfig| {
+        let collection = parse_sigma_yaml(&window_mode_yaml(window_lines)).unwrap();
+        let mut engine = CorrelationEngine::new(config);
+        engine.add_collection(&collection).unwrap();
+        engine.has_unbounded_long_window(&engine.correlations[0])
+    };
+    assert!(flagged("    timespan: 2h", CorrelationConfig::default()));
+    assert!(!flagged("    timespan: 1h", CorrelationConfig::default()));
+    assert!(!flagged(
+        "    timespan: 2h",
+        CorrelationConfig {
+            max_group_entries: Some(10_000),
+            ..Default::default()
+        }
+    ));
+
+    let yaml = window_mode_yaml("    timespan: 24h").replace(
+        "id: corr-win\n",
+        "id: corr-win\nrsigma.max_group_entries: '10000'\n",
+    );
+    let mut engine = CorrelationEngine::new(CorrelationConfig::default());
+    engine
+        .add_collection(&parse_sigma_yaml(&yaml).unwrap())
+        .unwrap();
+    assert!(!engine.has_unbounded_long_window(&engine.correlations[0]));
+}
+
 #[test]
 fn test_max_group_entries_caps_event_count() {
     // Cap of 2 retained entries: the gte-3 threshold can never be met,

crates/rsigma-runtime/src/alert_pipeline/config.rs

@@ -47,6 +47,11 @@ const DEFAULT_MAX_ACTIVE_ALERTS: usize = 100_000;
 /// surface against unbounded silence creation.
 pub const DEFAULT_MAX_DYNAMIC_SILENCES: usize = 1_000;
 
+/// Default ceiling on tracked inhibition sources across all rules. Once full,
+/// results that would open a new source are not tracked, bounding memory when
+/// the `equal` selectors carry high-cardinality values.
+const DEFAULT_MAX_INHIBIT_SOURCES: usize = 100_000;
+
 /// Top-level alert-pipeline config file.
 ///
 /// ```yaml
@@ -88,6 +93,10 @@ pub struct AlertPipelineFile {
     /// Inhibition rules. An active source mutes matching targets.
     #[serde(default)]
     pub inhibit_rules: Vec<InhibitRuleFile>,
+    /// Ceiling on tracked inhibition sources across all rules. Past this, a
+    /// result that would open a new source is not tracked. Defaults to 100000.
+    #[serde(default)]
+    pub max_inhibit_sources: Option<usize>,
 }
 
 /// `inhibit_rules:` entry.
@@ -349,7 +358,11 @@ pub fn build_alert_pipeline(
     let inhibit = if file.inhibit_rules.is_empty() {
         None
     } else {
-        Some(build_inhibit(file.inhibit_rules)?)
+        Some(build_inhibit(
+            file.inhibit_rules,
+            file.max_inhibit_sources
+                .unwrap_or(DEFAULT_MAX_INHIBIT_SOURCES),
+        )?)
     };
 
     let max_silences = file.max_silences.unwrap_or(DEFAULT_MAX_DYNAMIC_SILENCES);
@@ -366,7 +379,10 @@ pub fn build_alert_pipeline(
 }
 
 /// Validate the `inhibit_rules:` list into an [`InhibitConfig`].
-fn build_inhibit(rules: Vec<InhibitRuleFile>) -> Result<InhibitConfig, AlertPipelineConfigError> {
+fn build_inhibit(
+    rules: Vec<InhibitRuleFile>,
+    max_sources: usize,
+) -> Result<InhibitConfig, AlertPipelineConfigError> {
     let mut out = Vec::with_capacity(rules.len());
     for (i, rule) in rules.into_iter().enumerate() {
         if rule.source_match.is_empty() || rule.target_match.is_empty() {
@@ -388,7 +404,10 @@ fn build_inhibit(rules: Vec<InhibitRuleFile>) -> Result<InhibitConfig, AlertPipe
             duration: rule.duration.unwrap_or(DEFAULT_INHIBIT_DURATION),
         });
     }
-    Ok(InhibitConfig { rules: out })
+    Ok(InhibitConfig {
+        rules: out,
+        max_sources,
+    })
 }
 
 /// Validate a `group:` block into a [`GroupConfig`].

crates/rsigma-runtime/src/alert_pipeline/inhibit.rs

@@ -54,6 +54,10 @@ pub struct InhibitRule {
 #[derive(Debug, Clone)]
 pub struct InhibitConfig {
     pub rules: Vec<InhibitRule>,
+    /// Ceiling on tracked sources across all rules. At the cap, a result that
+    /// would open a new source is still evaluated as a target but is not
+    /// tracked; already-tracked sources keep refreshing.
+    pub max_sources: usize,
 }
 
 /// Active-source index: `(rule_name, equal_key) -> last_seen`.
@@ -71,12 +75,14 @@ impl InhibitStore {
     /// Evaluate a result. If it is an inhibited target, return the inhibiting
     /// rule name (and do not record it as a source). Otherwise record it as a
     /// source for every rule whose `source_match` it satisfies, and return
-    /// `None`.
+    /// `None`. A new source refused by `max_sources` is reported to
+    /// `on_refused` with its rule name.
     pub fn evaluate(
         &mut self,
         cfg: &InhibitConfig,
         result: &EvaluationResult,
         now: i64,
+        mut on_refused: impl FnMut(&str),
     ) -> Option<String> {
         // 1. Inhibited target? Checked against sources from prior results.
         for rule in &cfg.rules {
@@ -95,7 +101,13 @@ impl InhibitStore {
         for rule in &cfg.rules {
             if rule.source_match.matches(result) {
                 let key = (rule.name.clone(), equal_key(&rule.equal, result));
-                self.sources.insert(key, now);
+                if let Some(last) = self.sources.get_mut(&key) {
+                    *last = now;
+                } else if self.sources.len() < cfg.max_sources {
+                    self.sources.insert(key, now);
+                } else {
+                    on_refused(&rule.name);
+                }
             }
         }
         None
@@ -137,6 +149,25 @@ impl InhibitStore {
         }
     }
 
+    /// Evict the least recently seen sources beyond `max`, returning how many
+    /// were evicted.
+    pub(crate) fn enforce_cap(&mut self, max: usize) -> usize {
+        let excess = self.sources.len().saturating_sub(max);
+        if excess == 0 {
+            return 0;
+        }
+        let mut by_age: Vec<(i64, (String, String))> = self
+            .sources
+            .iter()
+            .map(|(key, &last_seen)| (last_seen, key.clone()))
+            .collect();
+        by_age.sort_unstable();
+        for (_, key) in by_age.into_iter().take(excess) {
+            self.sources.remove(&key);
+        }
+        excess
+    }
+
     /// Count of currently-active sources.
     pub fn active_count(&self, cfg: &InhibitConfig, now: i64) -> usize {
         self.sources
@@ -216,6 +247,7 @@ mod tests {
                 equal: vec![Selector::parse("match.SourceIp").unwrap()],
                 duration: Duration::from_secs(300),
             }],
+            max_sources: 100,
         }
     }
 
@@ -225,17 +257,17 @@ mod tests {
         let mut store = InhibitStore::default();
         // A critical source on 10.0.0.1 registers.
         assert_eq!(
-            store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0),
+            store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {}),
             None
         );
         // A high target on the same IP is inhibited.
         assert_eq!(
-            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1),
+            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1, |_| {}),
             Some("critical-inhibits-high".to_string())
         );
         // A high target on a different IP is not inhibited.
         assert_eq!(
-            store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 2),
+            store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 2, |_| {}),
             None
         );
     }
@@ -244,10 +276,10 @@ mod tests {
     fn source_expires_after_duration() {
         let cfg = cfg();
         let mut store = InhibitStore::default();
-        store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0);
+        store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {});
         // After the 300s window, the source no longer inhibits.
         assert_eq!(
-            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 400),
+            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 400, |_| {}),
             None
         );
     }
@@ -264,12 +296,13 @@ mod tests {
                 equal: vec![Selector::parse("match.SourceIp").unwrap()],
                 duration: Duration::from_secs(300),
             }],
+            max_sources: 100,
         };
         let mut store = InhibitStore::default();
-        store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 0);
+        store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 0, |_| {});
         // The same kind of result is a source, not an inhibited target.
         assert_eq!(
-            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1),
+            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1, |_| {}),
             None
         );
     }
@@ -278,10 +311,64 @@ mod tests {
     fn gc_and_active_count() {
         let cfg = cfg();
         let mut store = InhibitStore::default();
-        store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0);
+        store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {});
         assert_eq!(store.active_count(&cfg, 100), 1);
         assert_eq!(store.active_count(&cfg, 400), 0, "past duration");
         store.gc(&cfg, 400);
         assert!(store.is_empty());
     }
+
+    #[test]
+    fn new_sources_past_the_cap_are_refused() {
+        let cfg = InhibitConfig {
+            max_sources: 2,
+            ..cfg()
+        };
+        let mut store = InhibitStore::default();
+        let mut refused = Vec::new();
+        for (i, ip) in ["10.0.0.1", "10.0.0.2", "10.0.0.3"].into_iter().enumerate() {
+            store.evaluate(&cfg, &detection(ip, Level::Critical), i as i64, |rule| {
+                refused.push(rule.to_string())
+            });
+        }
+        assert_eq!(refused, ["critical-inhibits-high"]);
+        assert_eq!(store.active_count(&cfg, 10), 2);
+
+        // The refused source does not inhibit; tracked ones still do and
+        // still refresh at the cap.
+        assert_eq!(
+            store.evaluate(&cfg, &detection("10.0.0.3", Level::High), 10, |_| {}),
+            None
+        );
+        store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 200, |_| {
+            panic!("refreshing a tracked source is not a refusal")
+        });
+        assert_eq!(
+            store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 450, |_| {}),
+            Some("critical-inhibits-high".to_string())
+        );
+    }
+
+    #[test]
+    fn enforce_cap_keeps_the_most_recently_seen() {
+        let cfg = cfg();
+        let mut store = InhibitStore::default();
+        for (ip, now) in [("10.0.0.1", 30), ("10.0.0.2", 10), ("10.0.0.3", 20)] {
+            store.evaluate(&cfg, &detection(ip, Level::Critical), now, |_| {});
+        }
+        assert_eq!(store.enforce_cap(5), 0);
+        assert_eq!(store.enforce_cap(2), 1);
+        assert_eq!(
+            store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 40, |_| {}),
+            None,
+            "the least recently seen source was evicted"
+        );
+        for ip in ["10.0.0.1", "10.0.0.3"] {
+            assert!(
+                store
+                    .evaluate(&cfg, &detection(ip, Level::High), 40, |_| {})
+                    .is_some()
+            );
+        }
+    }
 }