fix: evaluate realtime metrics alerts against every data point

Real-time alerts on metrics streams evaluated only the first data point of
each metric per ingest request. All three metrics paths gated evaluation on
"have I seen this stream yet" rather than "has this alert already fired",
and inserted into the trigger map whether or not anything matched, so every
later record of that stream was skipped. Metrics arrive as many rows per
metric name, so the matching row is rarely first and alerts effectively
never fired.

Extract the per-stream evaluation state into RealtimeAlertEvaluator, which
checks every record against every alert that has not yet fired while capping
each alert at one trigger per request. That cap is load-bearing:
evaluate_trigger sends one notification per trigger with no deduplication,
and the silence window is applied once up front in get_stream_alerts, so it
cannot suppress duplicates within a single request.

Two adjacent defects found while doing this:

- get_stream_alerts returned instead of continuing when a stream's alerts
  were all filtered out, so a silenced alert on one metric suppressed
  alerting on every stream after it in the same call. This is reachable on
  the remote-write path, which preloads all metrics in one call.
- json.rs guarded its alert fetch with the bare stream name while the map is
  keyed "{org}/{type}/{stream}", so the guard never matched and the fetch ran
  once per record.

stream_trigger_map was Option-wrapped but never assigned None; unwrap it.

Logs and traces already carry the equivalent fix from #4498 and are
untouched.

src/core/src/ingestion/mod.rs

@@ -58,6 +58,7 @@ use crate::{
 
 pub mod grpc;
 pub mod ingestion_service;
+pub mod realtime_alerts;
 
 pub type TriggerAlertData = Vec<(Alert, Vec<Map<String, Value>>)>;
 
@@ -103,13 +104,22 @@ pub async fn get_stream_alerts(
             continue;
         }
 
-        let stream_alerts_cacher = STREAM_ALERTS.read().await;
-        let alerts_id_list = stream_alerts_cacher.get(&key);
-        if alerts_id_list.is_none() {
-            continue;
-        }
+        // Copy the ids out and release this guard before the loop below.
+        // `get_alert_from_cache` acquires the alert cache read lock, while alert
+        // deletion acquires the two caches in the opposite order (alert cache
+        // write, then stream-alert cache write). Holding this guard across that
+        // await is a lock-order inversion, and because these are write-preferring
+        // locks a pending writer also blocks new readers -- so it can stall
+        // ingestion for every stream type, not just metrics.
+        let alert_ids: Vec<String> = {
+            let stream_alerts_cacher = STREAM_ALERTS.read().await;
+            match stream_alerts_cacher.get(&key) {
+                Some(ids) => ids.clone(),
+                None => continue,
+            }
+        };
         let mut alerts_list = vec![];
-        for alert_id in alerts_id_list.unwrap().iter() {
+        for alert_id in alert_ids.iter() {
             if let Some((_, alert)) = alert::get_alert_from_cache(&stream.org_id, alert_id).await {
                 alerts_list.push(alert);
             }
@@ -127,10 +137,12 @@ pub async fn get_stream_alerts(
                 }
             })
             .collect::<Vec<_>>();
-        if alerts.is_empty() {
-            return;
+        // A stream whose alerts are all filtered out (disabled, scheduled, or
+        // silenced) must not abort the loop -- the streams after it in this
+        // request still need their alerts loaded.
+        if !alerts.is_empty() {
+            stream_alerts_map.insert(key, alerts);
         }
-        stream_alerts_map.insert(key, alerts);
     }
 }
 
@@ -1091,4 +1103,245 @@ mod tests {
         let result = create_log_ingestion_req(99, data);
         assert!(result.is_err());
     }
+
+    mod get_stream_alerts_tests {
+        use config::meta::folder::Folder;
+        use svix_ksuid::{Ksuid, KsuidLike};
+
+        use super::*;
+        use crate::common::infra::config::ALERTS;
+
+        /// Seed one real-time alert for `org`/`stream` into the in-memory caches
+        /// `get_stream_alerts` reads from, and return its id.
+        ///
+        /// Every test uses a unique org so the process-global caches cannot make
+        /// tests order-dependent or flaky under parallel execution.
+        async fn seed_alert(org: &str, stream: &str, enabled: bool, real_time: bool) -> Ksuid {
+            let id = Ksuid::new(None, None);
+            // `Alert` carries private fields, so struct-update syntax is unavailable
+            // outside its defining crate. It is `#[serde(default)]`, so building it
+            // from JSON is the supported route and mirrors how alerts really arrive.
+            let mut alert: Alert = from_value(json!({
+                "name": format!("alert_{stream}"),
+                "org_id": org,
+                "stream_type": "metrics",
+                "stream_name": stream,
+                "is_real_time": real_time,
+                "enabled": enabled,
+            }))
+            .expect("valid alert json");
+            alert.id = Some(id);
+            ALERTS.write().await.insert(
+                alert::cache_alert_key(org, &id.to_string()),
+                (Folder::default(), alert),
+            );
+            STREAM_ALERTS
+                .write()
+                .await
+                .entry(alert::cache_stream_key(org, StreamType::Metrics, stream))
+                .or_default()
+                .push(id.to_string());
+            id
+        }
+
+        fn params(org: &str, stream: &str) -> StreamParams {
+            StreamParams::new(org, stream, StreamType::Metrics)
+        }
+
+        fn key(org: &str, stream: &str) -> String {
+            alert::cache_stream_key(org, StreamType::Metrics, stream)
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_loads_a_stream_with_alerts() {
+            let org = "d2_g1";
+            seed_alert(org, "alerted", true, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "alerted")], &mut map).await;
+
+            assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1));
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_skips_a_stream_without_alerts() {
+            let org = "d2_g2";
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "quiet")], &mut map).await;
+
+            assert!(
+                map.is_empty(),
+                "a stream with no alerts must not be inserted"
+            );
+        }
+
+        /// Seed a stream that has alert ids registered but whose alerts are all
+        /// silenced, so the filter leaves an empty list. This is the state that
+        /// reaches the `alerts.is_empty()` branch.
+        async fn seed_silenced_alert(org: &str, stream: &str) {
+            let id = seed_alert(org, stream, true, true).await;
+            REALTIME_ALERT_TRIGGERS.write().await.insert(
+                format!("{org}/{id}"),
+                db::scheduler::Trigger {
+                    is_silenced: true,
+                    ..Default::default()
+                },
+            );
+        }
+
+        /// D2: a stream whose alerts are all filtered out must not abort the loop
+        /// for the streams after it.
+        #[tokio::test]
+        async fn test_get_stream_alerts_continues_past_a_fully_filtered_stream() {
+            let org = "d2_g3";
+            seed_silenced_alert(org, "silenced").await;
+            seed_alert(org, "alerted", true, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "silenced"), params(org, "alerted")], &mut map).await;
+
+            assert_eq!(
+                map.get(&key(org, "alerted")).map(|v| v.len()),
+                Some(1),
+                "a stream whose alerts are all silenced must not suppress later streams"
+            );
+        }
+
+        /// An unregistered stream already takes the `continue` path; pin it so the
+        /// fix does not disturb it.
+        #[tokio::test]
+        async fn test_get_stream_alerts_continues_past_an_unregistered_stream() {
+            let org = "d2_g3b";
+            seed_alert(org, "alerted", true, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "quiet"), params(org, "alerted")], &mut map).await;
+
+            assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1));
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_loads_when_alerted_stream_is_first() {
+            let org = "d2_g4";
+            seed_alert(org, "alerted", true, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "alerted"), params(org, "quiet")], &mut map).await;
+
+            assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1));
+        }
+
+        /// D2: several fully-filtered streams in a row must not abort the loop.
+        #[tokio::test]
+        async fn test_get_stream_alerts_continues_past_multiple_filtered_streams() {
+            let org = "d2_g5";
+            seed_silenced_alert(org, "silenced_a").await;
+            seed_alert(org, "disabled_b", false, true).await;
+            seed_alert(org, "scheduled_c", true, false).await;
+            seed_alert(org, "alerted", true, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(
+                &[
+                    params(org, "silenced_a"),
+                    params(org, "disabled_b"),
+                    params(org, "scheduled_c"),
+                    params(org, "alerted"),
+                ],
+                &mut map,
+            )
+            .await;
+
+            assert_eq!(
+                map.get(&key(org, "alerted")).map(|v| v.len()),
+                Some(1),
+                "alerts must load after any number of fully-filtered streams"
+            );
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_handles_all_streams_without_alerts() {
+            let org = "d2_g6";
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "a"), params(org, "b")], &mut map).await;
+
+            assert!(map.is_empty());
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_does_not_refetch_cached_keys() {
+            let org = "d2_g7";
+            seed_alert(org, "alerted", true, true).await;
+
+            // Pre-seed the map with a sentinel; the function must leave it alone.
+            let mut map = HashMap::new();
+            map.insert(key(org, "alerted"), Vec::new());
+            get_stream_alerts(&[params(org, "alerted")], &mut map).await;
+
+            assert_eq!(
+                map.get(&key(org, "alerted")).map(|v| v.len()),
+                Some(0),
+                "an already-cached key must not be refetched"
+            );
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_filters_disabled_alerts() {
+            let org = "d2_g8";
+            seed_alert(org, "alerted", false, true).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "alerted")], &mut map).await;
+
+            assert!(map.is_empty(), "disabled alerts must be filtered out");
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_filters_scheduled_alerts() {
+            let org = "d2_g9";
+            seed_alert(org, "alerted", true, false).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "alerted")], &mut map).await;
+
+            assert!(map.is_empty(), "non-realtime alerts must be filtered out");
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_filters_silenced_alerts() {
+            let org = "d2_g10";
+            let id = seed_alert(org, "alerted", true, true).await;
+            REALTIME_ALERT_TRIGGERS.write().await.insert(
+                format!("{org}/{id}"),
+                db::scheduler::Trigger {
+                    is_silenced: true,
+                    ..Default::default()
+                },
+            );
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "alerted")], &mut map).await;
+
+            assert!(map.is_empty(), "silenced alerts must be filtered out");
+        }
+
+        #[tokio::test]
+        async fn test_get_stream_alerts_keeps_only_enabled_realtime_alerts() {
+            let org = "d2_g11";
+            seed_alert(org, "mixed", true, true).await;
+            seed_alert(org, "mixed", false, true).await;
+            seed_alert(org, "mixed", true, false).await;
+
+            let mut map = HashMap::new();
+            get_stream_alerts(&[params(org, "mixed")], &mut map).await;
+
+            assert_eq!(
+                map.get(&key(org, "mixed")).map(|v| v.len()),
+                Some(1),
+                "only the enabled real-time alert survives"
+            );
+        }
+    }
 }

src/core/src/ingestion/realtime_alerts.rs

@@ -0,0 +1,438 @@
+// Copyright 2025 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+use config::{
+    meta::alerts::alert::Alert,
+    utils::{
+        json::{Map, Value},
+        time::now_micros,
+    },
+};
+
+use crate::{alerts::alert::AlertExt, ingestion::TriggerAlertData};
+
+/// Accumulates real-time alert triggers for the records of a single stream
+/// within one ingestion request.
+///
+/// Every record is checked against every alert that has not yet matched, and
+/// each alert contributes at most one trigger. That cap matters:
+/// `evaluate_trigger` sends one notification per trigger and performs no
+/// deduplication of its own, and the silence window is applied once up front in
+/// `get_stream_alerts`, so a second trigger for the same alert within one
+/// request is a second notification.
+pub struct RealtimeAlertEvaluator<'a> {
+    alerts: &'a [Alert],
+    /// Index-aligned with `alerts`: whether that alert has already fired.
+    /// Indexing by position keeps the per-record path free of key building.
+    /// Repeated ids in the input are collapsed once in `new`, so a position that
+    /// repeats an earlier alert starts out marked as fired.
+    fired: Vec<bool>,
+    triggers: TriggerAlertData,
+    /// Alerts that have not yet fired. Reaching zero short-circuits `observe`,
+    /// which also covers the common "stream has no alerts" case at no cost.
+    remaining: usize,
+    /// Captured once per stream; see `observe`.
+    end_time: i64,
+}
+
+impl<'a> RealtimeAlertEvaluator<'a> {
+    pub fn new(alerts: &'a [Alert]) -> Self {
+        let mut fired = vec![false; alerts.len()];
+        let mut remaining = alerts.len();
+
+        // The same alert id can appear twice for one stream: the bulk cache load
+        // pushes ids unconditionally, while the incremental path checks first.
+        // Position alone would then treat the repeats as separate alerts and send
+        // two notifications for one alert from a single request, so mark repeats
+        // as already fired. Done once per stream, which keeps the per-record path
+        // free of any key building.
+        for i in 1..alerts.len() {
+            let id = alerts[i].id;
+            if id.is_some() && alerts[..i].iter().any(|prev| prev.id == id) {
+                fired[i] = true;
+                remaining -= 1;
+            }
+        }
+
+        Self {
+            alerts,
+            fired,
+            triggers: Vec::with_capacity(alerts.len()),
+            remaining,
+            end_time: now_micros(),
+        }
+    }
+
+    /// Check one record against every alert that has not yet fired.
+    pub async fn observe(&mut self, record: &Map<String, Value>) {
+        if self.remaining == 0 {
+            return;
+        }
+        // The real-time branch of `evaluate` ignores this time range entirely and
+        // takes its own timestamp, so it is computed once per stream rather than
+        // once per record.
+        let end_time = self.end_time;
+        let alerts = self.alerts;
+        for (i, alert) in alerts.iter().enumerate() {
+            if self.fired[i] {
+                continue;
+            }
+            match alert.evaluate(Some(record), (None, end_time), None).await {
+                Ok(res) if res.data.is_some() => {
+                    self.triggers.push((alert.clone(), res.data.unwrap()));
+                    self.fired[i] = true;
+                    self.remaining -= 1;
+                }
+                Ok(_) => {
+                    // the record does not satisfy the alert condition
+                }
+                Err(e) => {
+                    log::error!("[METRICS] Error while evaluating realtime alert: {e}");
+                }
+            }
+        }
+    }
+
+    pub fn finish(self) -> TriggerAlertData {
+        self.triggers
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use config::utils::json::{from_value, json};
+    use svix_ksuid::{Ksuid, KsuidLike};
+
+    use super::*;
+
+    /// Build a real-time alert whose single condition is `column == value`.
+    ///
+    /// `Alert` carries private fields, so struct-update syntax is unavailable
+    /// outside its defining crate. It is `#[serde(default)]`, so building it from
+    /// JSON is the supported route and mirrors how alerts really arrive.
+    fn alert_matching(name: &str, column: &str, value: Value) -> Alert {
+        from_value(json!({
+            "name": name,
+            "org_id": "test_org",
+            "stream_type": "metrics",
+            "stream_name": "test_stream",
+            "is_real_time": true,
+            "enabled": true,
+            "query_condition": {
+                "type": "custom",
+                "conditions": [
+                    {"column": column, "operator": "=", "value": value, "ignore_case": false}
+                ]
+            }
+        }))
+        .expect("valid alert json")
+    }
+
+    fn record(pairs: &[(&str, Value)]) -> Map<String, Value> {
+        pairs
+            .iter()
+            .map(|(k, v)| ((*k).to_string(), v.clone()))
+            .collect()
+    }
+
+    /// One row that matches `alert_matching("..", "condition", "DiskPressure")`.
+    fn matching_record() -> Map<String, Value> {
+        record(&[("condition", json!("DiskPressure")), ("value", json!(1))])
+    }
+
+    /// One row that does not match.
+    fn other_record() -> Map<String, Value> {
+        record(&[("condition", json!("Ready")), ("value", json!(1))])
+    }
+
+    fn disk_pressure_alert() -> Alert {
+        alert_matching("disk_pressure", "condition", json!("DiskPressure"))
+    }
+
+    // ---- match position ----
+
+    #[tokio::test]
+    async fn test_observe_fires_on_the_only_record() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert_eq!(ev.finish().len(), 1);
+    }
+
+    /// D1: the reported bug -- a match on any record after the first.
+    #[tokio::test]
+    async fn test_observe_fires_when_match_is_the_second_record() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&other_record()).await;
+        ev.observe(&matching_record()).await;
+        assert_eq!(
+            ev.finish().len(),
+            1,
+            "a match on the second record must still fire"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_observe_fires_when_match_is_mid_batch() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        for _ in 0..24 {
+            ev.observe(&other_record()).await;
+        }
+        ev.observe(&matching_record()).await;
+        for _ in 0..25 {
+            ev.observe(&other_record()).await;
+        }
+        assert_eq!(ev.finish().len(), 1, "a match mid-batch must fire");
+    }
+
+    #[tokio::test]
+    async fn test_observe_fires_when_match_is_the_last_record() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        for _ in 0..49 {
+            ev.observe(&other_record()).await;
+        }
+        ev.observe(&matching_record()).await;
+        assert_eq!(
+            ev.finish().len(),
+            1,
+            "a match on the last record of 50 must fire"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_observe_produces_no_trigger_when_nothing_matches() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        for _ in 0..10 {
+            ev.observe(&other_record()).await;
+        }
+        assert!(ev.finish().is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_finish_returns_empty_when_no_records_observed() {
+        let alerts = vec![disk_pressure_alert()];
+        let ev = RealtimeAlertEvaluator::new(&alerts);
+        assert!(ev.finish().is_empty());
+    }
+
+    // ---- multiplicity: at most one trigger per alert per request ----
+
+    #[tokio::test]
+    async fn test_repeated_matches_produce_exactly_one_trigger() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        for _ in 0..10 {
+            ev.observe(&matching_record()).await;
+        }
+        assert_eq!(
+            ev.finish().len(),
+            1,
+            "ten matching records must still yield exactly one trigger"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_trigger_payload_holds_only_the_first_matching_record() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        for _ in 0..10 {
+            ev.observe(&matching_record()).await;
+        }
+        let triggers = ev.finish();
+        assert_eq!(triggers.len(), 1);
+        assert_eq!(
+            triggers[0].1.len(),
+            1,
+            "the payload must carry one row, not every matching row"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_trigger_payload_holds_the_earliest_match_across_the_batch() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        let mut first = matching_record();
+        first.insert("node".to_string(), json!("node_first"));
+        let mut second = matching_record();
+        second.insert("node".to_string(), json!("node_second"));
+
+        ev.observe(&other_record()).await;
+        ev.observe(&first).await;
+        ev.observe(&second).await;
+
+        let triggers = ev.finish();
+        assert_eq!(triggers.len(), 1);
+        assert_eq!(
+            triggers[0].1[0].get("node"),
+            Some(&json!("node_first")),
+            "the payload must hold the earliest matching record"
+        );
+    }
+
+    // ---- multiple alerts ----
+
+    #[tokio::test]
+    async fn test_two_alerts_matching_the_same_record_each_fire_once() {
+        let alerts = vec![
+            disk_pressure_alert(),
+            alert_matching("value_one", "value", json!(1)),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert_eq!(ev.finish().len(), 2);
+    }
+
+    /// D1: a per-stream flag passes the second-record test but fails this one.
+    #[tokio::test]
+    async fn test_two_alerts_matching_different_records_both_fire() {
+        let alerts = vec![
+            disk_pressure_alert(),
+            alert_matching("condition", "condition", json!("MemoryPressure")),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        ev.observe(&record(&[("condition", json!("MemoryPressure"))]))
+            .await;
+        assert_eq!(
+            ev.finish().len(),
+            2,
+            "alerts matching different records must both fire"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_two_alerts_fire_regardless_of_match_order() {
+        let alerts = vec![
+            disk_pressure_alert(),
+            alert_matching("condition", "condition", json!("MemoryPressure")),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&record(&[("condition", json!("MemoryPressure"))]))
+            .await;
+        ev.observe(&matching_record()).await;
+        assert_eq!(ev.finish().len(), 2, "match order must not matter");
+    }
+
+    #[tokio::test]
+    async fn test_only_the_matching_alert_fires() {
+        let alerts = vec![
+            disk_pressure_alert(),
+            alert_matching("condition", "condition", json!("NeverMatches")),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&other_record()).await;
+        ev.observe(&matching_record()).await;
+
+        let triggers = ev.finish();
+        assert_eq!(triggers.len(), 1);
+        assert_eq!(triggers[0].0.name, "disk_pressure");
+    }
+
+    #[tokio::test]
+    async fn test_no_alert_fires_when_none_match() {
+        let alerts = vec![
+            disk_pressure_alert(),
+            alert_matching("condition", "condition", json!("MemoryPressure")),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&other_record()).await;
+        assert!(ev.finish().is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_alerts_with_identical_conditions_fire_independently() {
+        let alerts = vec![
+            alert_matching("copy_a", "condition", json!("DiskPressure")),
+            alert_matching("copy_b", "condition", json!("DiskPressure")),
+        ];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert_eq!(
+            ev.finish().len(),
+            2,
+            "two distinct alerts with the same condition must each fire"
+        );
+    }
+
+    // ---- short-circuit and degenerate input ----
+
+    #[tokio::test]
+    async fn test_observe_is_a_noop_without_alerts() {
+        let alerts: Vec<Alert> = vec![];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert!(ev.finish().is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_observe_after_saturation_does_not_add_triggers() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        for _ in 0..5 {
+            ev.observe(&matching_record()).await;
+        }
+        assert_eq!(ev.finish().len(), 1);
+    }
+
+    /// The alert cache can hold the same id twice for one stream, because the
+    /// bulk load pushes ids without checking while the incremental path checks
+    /// first. One alert must still produce one notification.
+    #[tokio::test]
+    async fn test_duplicate_alert_ids_produce_one_trigger() {
+        let mut a = disk_pressure_alert();
+        a.id = Some(Ksuid::new(None, None));
+        let alerts = vec![a.clone(), a];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert_eq!(
+            ev.finish().len(),
+            1,
+            "the same alert listed twice must fire once"
+        );
+    }
+
+    /// Distinct alerts that happen to share a name are still separate alerts.
+    #[tokio::test]
+    async fn test_distinct_ids_with_same_name_both_fire() {
+        let mut a = disk_pressure_alert();
+        let mut b = disk_pressure_alert();
+        a.id = Some(Ksuid::new(None, None));
+        b.id = Some(Ksuid::new(None, None));
+        let alerts = vec![a, b];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&matching_record()).await;
+        assert_eq!(ev.finish().len(), 2);
+    }
+
+    /// A record lacking the condition column must not consume the alert -- a
+    /// later record still has to be able to fire it.
+    #[tokio::test]
+    async fn test_record_missing_condition_column_leaves_alert_eligible() {
+        let alerts = vec![disk_pressure_alert()];
+        let mut ev = RealtimeAlertEvaluator::new(&alerts);
+        ev.observe(&record(&[("unrelated", json!("x"))])).await;
+        ev.observe(&matching_record()).await;
+        assert_eq!(
+            ev.finish().len(),
+            1,
+            "a record missing the column must not disqualify later records"
+        );
+    }
+}

src/core/src/metrics/json.rs

@@ -46,11 +46,10 @@ use schema::check_for_schema;
 
 use super::get_exclude_labels;
 use crate::{
-    alerts::alert::AlertExt,
     common::meta::{authz::Authz, stream::SchemaRecords},
     ingestion::{
         TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id,
-        get_write_partition_key, write_file,
+        get_write_partition_key, realtime_alerts::RealtimeAlertEvaluator, write_file,
     },
     pipeline::batch_execution::ExecutablePipeline,
 };
@@ -129,7 +128,7 @@ pub async fn ingest(
 
     // realtime alerts
     let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
 
     // records buffer
     let mut json_data_by_stream: HashMap<String, Vec<_>> = HashMap::new();
@@ -405,21 +404,29 @@ pub async fn ingest(
             .unwrap_or_default();
         let partition_time_level = get_partition_time_level(StreamType::Metrics);
 
-        for (mut record, metric_type) in json_data {
-            // Start get stream alerts
-            if !stream_alerts_map.contains_key(&stream_name) {
-                crate::ingestion::get_stream_alerts(
-                    &[StreamParams {
-                        org_id: org_id.to_owned().into(),
-                        stream_name: stream_name.to_owned().into(),
-                        stream_type: StreamType::Metrics,
-                    }],
-                    &mut stream_alerts_map,
-                )
-                .await;
-            }
-            // End get stream alert
+        // Start get stream alerts -- fetched once per stream, not once per record
+        let alerts_key =
+            db::alerts::alert::cache_stream_key(org_id, StreamType::Metrics, &stream_name);
+        if !stream_alerts_map.contains_key(&alerts_key) {
+            crate::ingestion::get_stream_alerts(
+                &[StreamParams {
+                    org_id: org_id.to_owned().into(),
+                    stream_name: stream_name.to_owned().into(),
+                    stream_type: StreamType::Metrics,
+                }],
+                &mut stream_alerts_map,
+            )
+            .await;
+        }
+        let mut alert_evaluator = RealtimeAlertEvaluator::new(
+            stream_alerts_map
+                .get(&alerts_key)
+                .map(Vec::as_slice)
+                .unwrap_or_default(),
+        );
+        // End get stream alert
 
+        for (mut record, metric_type) in json_data {
             // check value
             let value =
                 parse_metric_value(record.get(VALUE_LABEL).ok_or(anyhow!("missing value"))?)?;
@@ -536,27 +543,14 @@ pub async fn ingest(
                 .or_insert_with(|| StreamStatus::new(&stream_name));
             stream_status.status.successful += 1;
 
-            // realtime alert
-            let need_trigger = !stream_trigger_map.contains_key(&stream_name);
-            if need_trigger && !stream_alerts_map.is_empty() {
-                // start check for alert trigger
-                let key = format!("{}/{}/{}", org_id, StreamType::Metrics, stream_name);
-                if let Some(alerts) = stream_alerts_map.get(&key) {
-                    let mut trigger_alerts: TriggerAlertData = Vec::new();
-                    let alert_end_time = now_micros();
-                    for alert in alerts {
-                        if let Ok(Some(data)) = alert
-                            .evaluate(Some(&record), (None, alert_end_time), None)
-                            .await
-                            .map(|res| res.data)
-                        {
-                            trigger_alerts.push((alert.clone(), data))
-                        }
-                    }
-                    stream_trigger_map.insert(stream_name.clone(), Some(trigger_alerts));
-                }
-            }
-            // End check for alert trigger
+            // realtime alert -- every record is checked against every alert that
+            // has not yet fired in this request
+            alert_evaluator.observe(&record).await;
+        }
+
+        let triggers = alert_evaluator.finish();
+        if !triggers.is_empty() {
+            stream_trigger_map.insert(stream_name.clone(), triggers);
         }
     }
 
@@ -631,9 +625,9 @@ pub async fn ingest(
         .inc();
 
     // only one trigger per request
-    for (_, entry) in stream_trigger_map {
-        if let Some(entry) = entry {
-            evaluate_trigger(entry).await;
+    for (_, triggers) in stream_trigger_map {
+        if !triggers.is_empty() {
+            tokio::spawn(async move { evaluate_trigger(triggers).await });
         }
     }
 

src/core/src/metrics/otlp.rs

@@ -40,7 +40,6 @@ use config::{
         json,
         schema::format_stream_name,
         schema_ext::SchemaExt,
-        time::now_micros,
     },
 };
 use db;
@@ -57,11 +56,11 @@ use prost::Message;
 use schema::{check_for_schema, stream_schema_exists};
 
 use crate::{
-    alerts::alert::AlertExt,
     common::meta::{http::HttpResponse as MetaHttpResponse, stream::SchemaRecords},
     ingestion::{
         TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id,
         grpc::{get_exemplar_val, get_metric_val, get_val},
+        realtime_alerts::RealtimeAlertEvaluator,
         write_file,
     },
     metrics::get_exclude_labels,
@@ -173,7 +172,7 @@ pub async fn handle_otlp_request(
 
     // realtime alerts
     let mut stream_alerts_map: HashMap<String, Vec<alert::Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
 
     let mut partial_success = ExportMetricsPartialSuccess::default();
 
@@ -536,6 +535,20 @@ pub async fn handle_otlp_request(
             .unwrap_or_default();
         let partition_time_level = get_partition_time_level(StreamType::Metrics);
 
+        // Realtime alerts for this stream. Alerts were fetched while records were
+        // built, for the metric name and its derived streams; a stream that only
+        // appears as a pipeline destination has no entry here and evaluates nothing.
+        let mut alert_evaluator = RealtimeAlertEvaluator::new(
+            stream_alerts_map
+                .get(&db::alerts::alert::cache_stream_key(
+                    org_id,
+                    StreamType::Metrics,
+                    &local_metric_name,
+                ))
+                .map(Vec::as_slice)
+                .unwrap_or_default(),
+        );
+
         // check for schema evolution
         let min_timestamp = batch_min_timestamp(&json_data, Utc::now().timestamp_micros());
 
@@ -588,27 +601,14 @@ pub async fn handle_otlp_request(
                 .push(Arc::new(json::Value::Object(val_map.to_owned())));
             hour_buf.records_size += value_str.len();
 
-            // real time alert
-            let need_trigger = !stream_trigger_map.contains_key(&local_metric_name);
-            if need_trigger && !stream_alerts_map.is_empty() {
-                // Start check for alert trigger
-                let key = format!("{}/{}/{}", org_id, StreamType::Metrics, local_metric_name);
-                if let Some(alerts) = stream_alerts_map.get(&key) {
-                    let mut trigger_alerts: TriggerAlertData = Vec::new();
-                    let alert_end_time = now_micros();
-                    for alert in alerts {
-                        if let Ok(Some(data)) = alert
-                            .evaluate(Some(&val_map), (None, alert_end_time), None)
-                            .await
-                            .map(|res| res.data)
-                        {
-                            trigger_alerts.push((alert.clone(), data))
-                        }
-                    }
-                    stream_trigger_map.insert(local_metric_name.clone(), Some(trigger_alerts));
-                }
-            }
-            // End check for alert trigger
+            // real time alert -- every record is checked against every alert that
+            // has not yet fired in this request
+            alert_evaluator.observe(&val_map).await;
+        }
+
+        let triggers = alert_evaluator.finish();
+        if !triggers.is_empty() {
+            stream_trigger_map.insert(local_metric_name.clone(), triggers);
         }
     }
 
@@ -681,9 +681,9 @@ pub async fn handle_otlp_request(
         .inc();
 
     // only one trigger per request
-    for (_, entry) in stream_trigger_map {
-        if let Some(entry) = entry {
-            evaluate_trigger(entry).await;
+    for (_, triggers) in stream_trigger_map {
+        if !triggers.is_empty() {
+            tokio::spawn(async move { evaluate_trigger(triggers).await });
         }
     }
 

src/core/src/metrics/prom.rs

@@ -56,13 +56,13 @@ use search_service;
 
 use super::native_histogram::{CLASSIC_HISTOGRAM_SUFFIXES, expand_native_histogram};
 use crate::{
-    alerts::alert::AlertExt,
     common::{
         infra::config::{METRIC_CLUSTER_LEADER, METRIC_CLUSTER_MAP},
         meta::stream::SchemaRecords,
     },
     ingestion::{
-        TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id, write_file,
+        TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id,
+        realtime_alerts::RealtimeAlertEvaluator, write_file,
     },
     pipeline::batch_execution::ExecutablePipeline,
 };
@@ -99,7 +99,7 @@ pub async fn remote_write(
 
     // realtime alerts
     let mut stream_alerts_map: HashMap<String, Vec<alert::Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
 
     let decoded = snap::raw::Decoder::new()
         .decompress_vec(&body)
@@ -559,6 +559,20 @@ pub async fn remote_write(
             .unwrap_or_default();
         let partition_time_level = get_partition_time_level(StreamType::Metrics);
 
+        // Realtime alerts for this stream. Alerts were preloaded in bulk for the
+        // metric names present in the request; a stream that only appears as a
+        // pipeline destination has no entry here and evaluates nothing.
+        let mut alert_evaluator = RealtimeAlertEvaluator::new(
+            stream_alerts_map
+                .get(&db::alerts::alert::cache_stream_key(
+                    org_id,
+                    StreamType::Metrics,
+                    &stream_name,
+                ))
+                .map(Vec::as_slice)
+                .unwrap_or_default(),
+        );
+
         for (mut val_map, timestamp) in json_data {
             let hash = super::signature_without_labels(&val_map, &[VALUE_LABEL]);
             val_map.insert(HASH_LABEL.to_string(), json::Value::Number(hash.into()));
@@ -631,27 +645,14 @@ pub async fn remote_write(
                 .push(Arc::new(json::Value::Object(val_map.to_owned())));
             hour_buf.records_size += value_str.len();
 
-            // real time alert
-            let need_trigger = !stream_trigger_map.contains_key(&stream_name);
-            if need_trigger && !stream_alerts_map.is_empty() {
-                // Start check for alert trigger
-                let key = format!("{}/{}/{}", org_id, StreamType::Metrics, stream_name);
-                if let Some(alerts) = stream_alerts_map.get(&key) {
-                    let mut trigger_alerts: TriggerAlertData = Vec::new();
-                    let alert_end_time = now_micros();
-                    for alert in alerts {
-                        if let Ok(Some(data)) = alert
-                            .evaluate(Some(&val_map), (None, alert_end_time), None)
-                            .await
-                            .map(|res| res.data)
-                        {
-                            trigger_alerts.push((alert.clone(), data));
-                        }
-                    }
-                    stream_trigger_map.insert(stream_name.clone(), Some(trigger_alerts));
-                }
-            }
-            // End check for alert trigger
+            // real time alert -- every record is checked against every alert that
+            // has not yet fired in this request
+            alert_evaluator.observe(&val_map).await;
+        }
+
+        let triggers = alert_evaluator.finish();
+        if !triggers.is_empty() {
+            stream_trigger_map.insert(stream_name.clone(), triggers);
         }
     }
     let elapsed_ms = step_start.elapsed().as_millis();
@@ -775,9 +776,9 @@ pub async fn remote_write(
         .inc();
 
     // only one trigger per request
-    for (_, entry) in stream_trigger_map {
-        if let Some(entry) = entry {
-            evaluate_trigger(entry).await;
+    for (_, triggers) in stream_trigger_map {
+        if !triggers.is_empty() {
+            tokio::spawn(async move { evaluate_trigger(triggers).await });
         }
     }