fix(alerts): key alert dedup state by org so one org's state cannot suppress another's (#15169)

src/config/src/config.rs

@@ -94,7 +94,8 @@ pub type RwBTreeMap<K, V> = tokio::sync::RwLock<BTreeMap<K, V>>;
 // 95: add band settings to anomaly_detection_config.
 // 96: create rum_pa_named_events, rum_pa_funnels and rum_pa_tombstones.
 // 97: create query_history.
-pub const DB_SCHEMA_VERSION: u64 = 97;
+// 98: key alert_dedup_state by (org_id, fingerprint).
+pub const DB_SCHEMA_VERSION: u64 = 98;
 pub const DB_SCHEMA_KEY: &str = "/db_schema_version/";
 
 // global version variables

src/core/src/alerts/deduplication.rs

@@ -156,6 +156,7 @@ pub struct DeduplicationOutcome {
 /// through — no retry bookkeeping required.
 pub async fn confirm_notification_sent(
     db: &DatabaseConnection,
+    org_id: &str,
     fingerprints: &[String],
 ) -> Result<(), sea_orm::DbErr> {
     if fingerprints.is_empty() {
@@ -166,6 +167,7 @@ pub async fn confirm_notification_sent(
             alert_dedup_state::Column::NotificationSent,
             sea_orm::sea_query::Expr::value(true),
         )
+        .filter(alert_dedup_state::Column::OrgId.eq(org_id))
         .filter(alert_dedup_state::Column::Fingerprint.is_in(fingerprints.to_vec()))
         .exec(db)
         .await?;
@@ -175,9 +177,10 @@ pub async fn confirm_notification_sent(
 /// Get or create deduplication state
 pub async fn get_dedup_state(
     db: &DatabaseConnection,
+    org_id: &str,
     fingerprint: &str,
 ) -> Result<Option<alert_dedup_state::Model>, sea_orm::DbErr> {
-    alert_dedup_state::Entity::find_by_id(fingerprint)
+    alert_dedup_state::Entity::find_by_id((org_id.to_string(), fingerprint.to_string()))
         .one(db)
         .await
 }
@@ -198,7 +201,7 @@ pub async fn save_dedup_state(
     params: DedupStateParams<'_>,
 ) -> Result<alert_dedup_state::Model, sea_orm::DbErr> {
     // Try to find existing record
-    if let Some(existing) = get_dedup_state(db, params.fingerprint).await? {
+    if let Some(existing) = get_dedup_state(db, params.org_id, params.fingerprint).await? {
         // Update existing
         let mut active: alert_dedup_state::ActiveModel = existing.clone().into();
         active.last_seen_at = Set(params.last_seen_at);
@@ -250,6 +253,7 @@ pub async fn cleanup_expired_state(
 /// Used to suppress alerts that share semantic dimensions with recently fired alerts.
 pub async fn find_matching_semantic_fingerprints(
     db: &DatabaseConnection,
+    org_id: &str,
     semantic_dimensions: &std::collections::HashMap<String, String>,
     time_window_minutes: i64,
 ) -> Result<Vec<alert_dedup_state::Model>, sea_orm::DbErr> {
@@ -258,6 +262,7 @@ pub async fn find_matching_semantic_fingerprints(
 
     // Get all recent fingerprints
     let recent_states = alert_dedup_state::Entity::find()
+        .filter(alert_dedup_state::Column::OrgId.eq(org_id))
         .filter(alert_dedup_state::Column::LastSeenAt.gt(cutoff_time))
         .all(db)
         .await?;
@@ -370,7 +375,7 @@ async fn apply_deduplication_impl(
         // error swallows the page for the whole window. `notification_sent`
         // has existed on this table since the feature shipped and was never
         // read; it is the confirm flag.
-        let should_send = match get_dedup_state(db, &fingerprint).await? {
+        let should_send = match get_dedup_state(db, org_id, &fingerprint).await? {
             Some(existing_state)
                 if config::meta::alerts::deduplication::reservation_suppresses(
                     existing_state.notification_sent,
@@ -479,9 +484,27 @@ mod tests {
     use config::meta::alerts::{
         AggFunction, Aggregation, Condition, Operator, QueryType, alert::Alert,
     };
+    use sea_orm::{ConnectionTrait, Database, Schema};
     use serde_json::{Map, Value, json};
 
-    use super::row_group_key;
+    use super::{
+        DedupStateParams, confirm_notification_sent, get_dedup_state, row_group_key,
+        save_dedup_state,
+    };
+
+    async fn dedup_db() -> sea_orm::DatabaseConnection {
+        let db = Database::connect("sqlite::memory:").await.unwrap();
+        for sql in [
+            "CREATE TABLE alerts (id CHAR(27) PRIMARY KEY)",
+            "INSERT INTO alerts (id) VALUES ('alert_a'), ('alert_b')",
+        ] {
+            db.execute_unprepared(sql).await.unwrap();
+        }
+        let backend = db.get_database_backend();
+        let table = Schema::new(backend).create_table_from_entity(super::alert_dedup_state::Entity);
+        db.execute(backend.build(&table)).await.unwrap();
+        db
+    }
 
     fn row(pairs: &[(&str, Value)]) -> Map<String, Value> {
         pairs
@@ -600,4 +623,34 @@ mod tests {
         alert.query_condition.query_type = QueryType::Slo;
         assert_eq!(row_group_key(&alert, &row(&[("group", json!("g"))])), None);
     }
+
+    #[tokio::test]
+    async fn test_dedup_state_is_scoped_by_org() {
+        let db = dedup_db().await;
+        let params = |org_id, alert_id| DedupStateParams {
+            org_id,
+            fingerprint: "fp",
+            alert_id,
+            first_seen_at: 1,
+            last_seen_at: 1,
+            occurrence_count: 1,
+        };
+        save_dedup_state(&db, params("org_a", "alert_a"))
+            .await
+            .unwrap();
+        assert!(get_dedup_state(&db, "org_b", "fp").await.unwrap().is_none());
+
+        save_dedup_state(&db, params("org_b", "alert_b"))
+            .await
+            .unwrap();
+        confirm_notification_sent(&db, "org_a", &["fp".to_string()])
+            .await
+            .unwrap();
+
+        let org_a = get_dedup_state(&db, "org_a", "fp").await.unwrap().unwrap();
+        let org_b = get_dedup_state(&db, "org_b", "fp").await.unwrap().unwrap();
+        assert!(org_a.notification_sent);
+        assert!(!org_b.notification_sent);
+        assert_eq!(org_b.alert_id, "alert_b");
+    }
 }

src/core/src/alerts/scheduler/handlers.rs

@@ -565,21 +565,21 @@ async fn dispatch_per_group(
 /// exactly what allows the next evaluation through. Best-effort: failing to
 /// confirm costs at most one duplicate notification, while failing the
 /// evaluation over it would cost the alert.
-async fn confirm_dedup_reservations(fingerprints: &[String], delivered: bool) {
-    if !delivered || fingerprints.is_empty() {
+async fn confirm_dedup_reservations(org_id: &str, fingerprints: &[String]) {
+    if fingerprints.is_empty() {
         return;
     }
     #[cfg(feature = "enterprise")]
     {
         let db = infra::db::get_orm_client_rw().await;
         if let Err(e) =
-            crate::alerts::deduplication::confirm_notification_sent(db, fingerprints).await
+            crate::alerts::deduplication::confirm_notification_sent(db, org_id, fingerprints).await
         {
             log::warn!("[SCHEDULER] could not confirm dedup reservations: {e}");
         }
     }
     #[cfg(not(feature = "enterprise"))]
-    let _ = fingerprints;
+    let _ = (org_id, fingerprints);
 }
 
 pub async fn handle_triggers(
@@ -3363,7 +3363,7 @@ async fn handle_alert_triggers(
                 })
                 .map(|(_, fingerprint)| fingerprint.clone())
                 .collect();
-            confirm_dedup_reservations(&confirmable, !confirmable.is_empty()).await;
+            confirm_dedup_reservations(&alert.org_id, &confirmable).await;
             record_delivery(&mut trigger_data);
             episode_delivered = true;
             trigger_data.period_end_time = if should_store_last_end_time {
@@ -3412,7 +3412,7 @@ async fn handle_alert_triggers(
                         .iter()
                         .map(|(_, fingerprint)| fingerprint.clone())
                         .collect();
-                    confirm_dedup_reservations(&fingerprints, true).await;
+                    confirm_dedup_reservations(&alert.org_id, &fingerprints).await;
                     let success_msg = success_msg.trim().to_owned();
                     let err_msg = err_msg.trim().to_owned();
                     if !err_msg.is_empty() {

src/infra/src/table/entity/alert_dedup_state.rs

@@ -20,13 +20,14 @@ use sea_orm::entity::prelude::*;
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
 #[sea_orm(table_name = "alert_dedup_state")]
 pub struct Model {
-    /// Fingerprint hash (primary key)
+    /// Part of the key because the fingerprint itself carries no org.
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub org_id: String,
+    /// Fingerprint hash
     #[sea_orm(primary_key, auto_increment = false)]
     pub fingerprint: String,
     /// Alert ID reference
     pub alert_id: String,
-    /// Organization ID
-    pub org_id: String,
     /// First time this fingerprint was seen (microseconds)
     pub first_seen_at: i64,
     /// Last time this fingerprint was seen (microseconds)

src/infra/src/table/migration/m20261007_000001_scope_alert_dedup_state_by_org.rs

@@ -0,0 +1,325 @@
+// Copyright 2026 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/>.
+
+//! Keys `alert_dedup_state` by `(org_id, fingerprint)`: the fingerprint carries no org.
+
+use sea_orm::{ConnectionTrait, DbBackend, TransactionTrait};
+use sea_orm_migration::prelude::*;
+
+const OLD_TABLE: &str = "alert_dedup_state_old";
+const ORG_INDEX: &str = "idx_alert_dedup_org_id";
+const COLUMNS: &str = "fingerprint, alert_id, org_id, first_seen_at, last_seen_at, \
+                       occurrence_count, notification_sent, created_at";
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        set_primary_key(
+            manager,
+            &[AlertDedupState::OrgId, AlertDedupState::Fingerprint],
+        )
+        .await?;
+        // The key now leads with org_id, so this index only adds write cost.
+        manager
+            .drop_index(
+                Index::drop()
+                    .if_exists()
+                    .name(ORG_INDEX)
+                    .table(AlertDedupState::Table)
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // Orgs may now share a fingerprint; the rows are a short-lived cache, so drop them all.
+        manager
+            .get_connection()
+            .execute_unprepared("DELETE FROM alert_dedup_state")
+            .await?;
+        set_primary_key(manager, &[AlertDedupState::Fingerprint]).await?;
+        manager
+            .create_index(
+                Index::create()
+                    .if_not_exists()
+                    .name(ORG_INDEX)
+                    .table(AlertDedupState::Table)
+                    .col(AlertDedupState::OrgId)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden, Clone, Copy)]
+enum AlertDedupState {
+    Table,
+    Fingerprint,
+    AlertId,
+    OrgId,
+    FirstSeenAt,
+    LastSeenAt,
+    OccurrenceCount,
+    NotificationSent,
+    CreatedAt,
+}
+
+#[derive(DeriveIden)]
+enum Alerts {
+    Table,
+    Id,
+}
+
+async fn set_primary_key(
+    manager: &SchemaManager<'_>,
+    key: &[AlertDedupState],
+) -> Result<(), DbErr> {
+    if manager.get_database_backend() == DbBackend::Postgres {
+        let columns = key
+            .iter()
+            .map(|column| column.to_string())
+            .collect::<Vec<_>>()
+            .join(", ");
+        manager
+            .get_connection()
+            .execute_unprepared(&format!(
+                "ALTER TABLE alert_dedup_state DROP CONSTRAINT alert_dedup_state_pkey, \
+                 ADD PRIMARY KEY ({columns})"
+            ))
+            .await?;
+        return Ok(());
+    }
+
+    // SQLite can't alter a PK; rebuild in a transaction, which the migrator gives only Postgres.
+    let txn = manager.get_connection().begin().await?;
+    let schema = SchemaManager::new(&txn);
+    schema
+        .rename_table(
+            Table::rename()
+                .table(AlertDedupState::Table, Alias::new(OLD_TABLE))
+                .to_owned(),
+        )
+        .await?;
+    schema.create_table(create_table(key)).await?;
+    txn.execute_unprepared(&format!(
+        "INSERT INTO alert_dedup_state ({COLUMNS}) SELECT {COLUMNS} FROM {OLD_TABLE}"
+    ))
+    .await?;
+    schema
+        .drop_table(Table::drop().table(Alias::new(OLD_TABLE)).to_owned())
+        .await?;
+    for (name, column) in [
+        ("idx_alert_dedup_alert_id", AlertDedupState::AlertId),
+        ("idx_alert_dedup_last_seen", AlertDedupState::LastSeenAt),
+    ] {
+        schema
+            .create_index(
+                Index::create()
+                    .name(name)
+                    .table(AlertDedupState::Table)
+                    .col(column)
+                    .to_owned(),
+            )
+            .await?;
+    }
+    txn.commit().await
+}
+
+fn create_table(key: &[AlertDedupState]) -> TableCreateStatement {
+    let mut primary_key = Index::create();
+    for column in key {
+        primary_key.col(*column);
+    }
+    Table::create()
+        .table(AlertDedupState::Table)
+        .col(
+            ColumnDef::new(AlertDedupState::Fingerprint)
+                .string_len(64)
+                .not_null(),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::AlertId)
+                .char_len(27)
+                .not_null(),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::OrgId)
+                .string_len(100)
+                .not_null(),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::FirstSeenAt)
+                .big_integer()
+                .not_null(),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::LastSeenAt)
+                .big_integer()
+                .not_null(),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::OccurrenceCount)
+                .big_integer()
+                .not_null()
+                .default(1),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::NotificationSent)
+                .boolean()
+                .not_null()
+                .default(false),
+        )
+        .col(
+            ColumnDef::new(AlertDedupState::CreatedAt)
+                .big_integer()
+                .not_null(),
+        )
+        .primary_key(&mut primary_key)
+        .foreign_key(
+            ForeignKey::create()
+                .name("fk_alert_dedup_alert")
+                .from(AlertDedupState::Table, AlertDedupState::AlertId)
+                .to(Alerts::Table, Alerts::Id)
+                .on_delete(ForeignKeyAction::Cascade),
+        )
+        .to_owned()
+}
+
+#[cfg(test)]
+mod tests {
+    use sea_orm::{Database, DatabaseConnection, EntityTrait, Statement};
+
+    use super::*;
+    use crate::table::entity::alert_dedup_state;
+
+    type ColumnInfo = (String, String, i32, Option<String>);
+
+    async fn db_with_original_table() -> DatabaseConnection {
+        let db = Database::connect("sqlite::memory:").await.unwrap();
+        for sql in [
+            "CREATE TABLE alerts (id CHAR(27) PRIMARY KEY)",
+            "INSERT INTO alerts (id) VALUES ('alert_a'), ('alert_b')",
+        ] {
+            db.execute_unprepared(sql).await.unwrap();
+        }
+        super::super::m20251024_000001_add_alert_deduplication::Migration
+            .up(&SchemaManager::new(&db))
+            .await
+            .unwrap();
+        db
+    }
+
+    async fn insert(db: &DatabaseConnection, org: &str, alert: &str) -> Result<(), DbErr> {
+        db.execute_unprepared(&format!(
+            "INSERT INTO alert_dedup_state ({COLUMNS}) \
+             VALUES ('fp', '{alert}', '{org}', 1, 2, 3, true, 4)"
+        ))
+        .await
+        .map(|_| ())
+    }
+
+    async fn columns(db: &DatabaseConnection) -> Vec<ColumnInfo> {
+        db.query_all(Statement::from_string(
+            DbBackend::Sqlite,
+            "SELECT name, type, \"notnull\", dflt_value \
+             FROM pragma_table_info('alert_dedup_state') ORDER BY cid",
+        ))
+        .await
+        .unwrap()
+        .iter()
+        .map(|row| {
+            (
+                row.try_get_by_index(0).unwrap(),
+                row.try_get_by_index(1).unwrap(),
+                row.try_get_by_index(2).unwrap(),
+                row.try_get_by_index(3).unwrap(),
+            )
+        })
+        .collect()
+    }
+
+    #[tokio::test]
+    async fn test_up_lets_two_orgs_share_a_fingerprint() {
+        let db = db_with_original_table().await;
+        insert(&db, "org_a", "alert_a").await.unwrap();
+        assert!(insert(&db, "org_b", "alert_b").await.is_err());
+
+        Migration.up(&SchemaManager::new(&db)).await.unwrap();
+
+        insert(&db, "org_b", "alert_b").await.unwrap();
+        assert!(insert(&db, "org_a", "alert_a").await.is_err());
+        let kept = alert_dedup_state::Entity::find_by_id(("org_a".to_string(), "fp".to_string()))
+            .one(&db)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            (
+                kept.alert_id.as_str(),
+                kept.first_seen_at,
+                kept.last_seen_at,
+                kept.occurrence_count,
+                kept.notification_sent,
+                kept.created_at
+            ),
+            ("alert_a", 1, 2, 3, true, 4)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_up_keeps_columns_indexes_and_foreign_key() {
+        let db = db_with_original_table().await;
+        let manager = SchemaManager::new(&db);
+        let original = columns(&db).await;
+
+        Migration.up(&manager).await.unwrap();
+
+        assert_eq!(columns(&db).await, original);
+        for index in ["idx_alert_dedup_alert_id", "idx_alert_dedup_last_seen"] {
+            assert!(manager.has_index("alert_dedup_state", index).await.unwrap());
+        }
+        assert!(
+            !manager
+                .has_index("alert_dedup_state", ORG_INDEX)
+                .await
+                .unwrap()
+        );
+        assert!(!manager.has_table(OLD_TABLE).await.unwrap());
+        assert!(insert(&db, "org_a", "no_such_alert").await.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_down_restores_fingerprint_key() {
+        let db = db_with_original_table().await;
+        let manager = SchemaManager::new(&db);
+        Migration.up(&manager).await.unwrap();
+        insert(&db, "org_a", "alert_a").await.unwrap();
+        insert(&db, "org_b", "alert_b").await.unwrap();
+
+        Migration.down(&manager).await.unwrap();
+
+        insert(&db, "org_a", "alert_a").await.unwrap();
+        assert!(insert(&db, "org_b", "alert_b").await.is_err());
+        assert!(
+            manager
+                .has_index("alert_dedup_state", ORG_INDEX)
+                .await
+                .unwrap()
+        );
+    }
+}

src/infra/src/table/migration/mod.rs

@@ -208,6 +208,7 @@ mod m20260928_000001_add_alert_recovery_destinations;
 mod m20261001_000001_add_anomaly_band_settings;
 mod m20261003_000001_create_rum_pa_tables;
 mod m20261004_000001_create_query_history;
+mod m20261007_000001_scope_alert_dedup_state_by_org;
 /// Shared body of the two `folder_id` migrations above; not a migration itself.
 mod workflow_folder_id;
 
@@ -519,6 +520,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20261001_000001_add_anomaly_band_settings::Migration),
             Box::new(m20261003_000001_create_rum_pa_tables::Migration),
             Box::new(m20261004_000001_create_query_history::Migration),
+            Box::new(m20261007_000001_scope_alert_dedup_state_by_org::Migration),
         ]
     }
 }
@@ -575,6 +577,7 @@ mod tests {
         (95, "m20261001_000001_add_anomaly_band_settings"),
         (96, "m20261003_000001_create_rum_pa_tables"),
         (97, "m20261004_000001_create_query_history"),
+        (98, "m20261007_000001_scope_alert_dedup_state_by_org"),
     ];
 
     #[test]