From 78e91516c69d8d1455f20c0b10c7786f2659c968 Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Thu, 5 Nov 2026 11:49:30 +0530 Subject: [PATCH 1/8] fix: reject duplicate alert target titles --- src/alerts/mod.rs | 3 +++ src/alerts/target.rs | 23 +++++++++++++++++++++++ src/handlers/http/targets.rs | 9 ++++++--- 3 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 6e9422af1..5402fd6f1 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -1214,6 +1214,8 @@ pub enum AlertError { FromStrError(#[from] FromStrError), #[error("Invalid Target ID- {0}")] InvalidTargetID(String), + #[error("Target title {0:?} already exists")] + DuplicateTargetName(String), #[error("Invalid target modification request: {0}")] InvalidTargetModification(String), #[error("Can't delete a Target which is being used")] @@ -1252,6 +1254,7 @@ impl actix_web::ResponseError for AlertError { Self::InvalidAlertModifyRequest => StatusCode::BAD_REQUEST, Self::FromStrError(_) => StatusCode::BAD_REQUEST, Self::InvalidTargetID(_) => StatusCode::BAD_REQUEST, + Self::DuplicateTargetName(_) => StatusCode::CONFLICT, Self::InvalidTargetModification(_) => StatusCode::BAD_REQUEST, Self::TargetInUse => StatusCode::CONFLICT, Self::ParserError(_) => StatusCode::BAD_REQUEST, diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 8a4851d4f..ece61af99 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -73,6 +73,29 @@ impl TargetConfigs { Ok(()) } + pub async fn create(&self, target: Target) -> Result<(), AlertError> { + let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); + let mut map = self.target_configs.write().await; + let targets = map.entry(tenant_id.to_owned()).or_default(); + if targets + .values() + .any(|existing| existing.name == target.name) + { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if targets.contains_key(&target.id) { + return Err(AlertError::InvalidTargetModification( + "Target ID already exists".to_string(), + )); + } + PARSEABLE + .metastore + .put_target(&target, &target.tenant) + .await?; + targets.insert(target.id, target); + Ok(()) + } + pub async fn update(&self, target: Target) -> Result<(), AlertError> { PARSEABLE .metastore diff --git a/src/handlers/http/targets.rs b/src/handlers/http/targets.rs index 286bc1ff8..5f3e46691 100644 --- a/src/handlers/http/targets.rs +++ b/src/handlers/http/targets.rs @@ -48,10 +48,13 @@ pub async fn post_internal( tenant_id: &Option, ) -> Result { target.tenant.clone_from(tenant_id); + if target.name.trim().is_empty() { + return Err(AlertError::ValidationFailure( + "Target title is required".into(), + )); + } target.validate_outbound_policy().await?; - // should check for duplicacy and liveness (??) - // add to the map - TARGETS.update(target.clone()).await?; + TARGETS.create(target.clone()).await?; Ok(serde_json::to_value(target.mask())?) } From d648cb41e4a296e3ab14d8d4bf48f8f6cc01ab13 Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Thu, 5 Nov 2026 20:13:17 +0530 Subject: [PATCH 2/8] fix: enforce unique alert target titles across nodes --- src/alerts/mod.rs | 1 + src/alerts/target.rs | 48 ++++++++++++++----- src/alerts/title_registry.rs | 48 +++++++++++++++++++ src/metastore/metastore_traits.rs | 4 ++ .../metastores/object_store_metastore.rs | 30 ++++++++++++ src/storage/azure_blob.rs | 24 ++++++++++ src/storage/gcs.rs | 24 ++++++++++ src/storage/localfs.rs | 29 ++++++++++- src/storage/object_storage.rs | 7 +++ src/storage/s3.rs | 24 ++++++++++ 10 files changed, 225 insertions(+), 14 deletions(-) create mode 100644 src/alerts/title_registry.rs diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 5402fd6f1..2dff42748 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -44,6 +44,7 @@ pub mod alert_types; pub mod alerts_utils; pub mod outbound_http_policy; pub mod target; +pub mod title_registry; pub use crate::alerts::alert_enums::{ AggregateFunction, AlertOperator, AlertQueryType, AlertState, AlertTask, AlertType, diff --git a/src/alerts/target.rs b/src/alerts/target.rs index ece61af99..6c4c59f13 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -29,7 +29,7 @@ use http::header::AUTHORIZATION; use itertools::Itertools; use once_cell::sync::Lazy; use serde_json::{Value, json}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex as AsyncMutex, RwLock}; use tracing::{error, trace, warn}; use ulid::Ulid; use url::Url; @@ -37,6 +37,7 @@ use url::Url; use super::{ ALERTS, outbound_http_policy::{self, AlertTargetKind}, + title_registry, }; use crate::{ alerts::{AlertError, AlertState, Context, alert_traits::CallableTarget}, @@ -47,11 +48,13 @@ use crate::{ pub static TARGETS: Lazy = Lazy::new(|| TargetConfigs { target_configs: RwLock::new(HashMap::new()), + mutation_lock: AsyncMutex::new(()), }); #[derive(Debug)] pub struct TargetConfigs { pub target_configs: RwLock>>, + mutation_lock: AsyncMutex<()>, } impl TargetConfigs { @@ -75,24 +78,37 @@ impl TargetConfigs { pub async fn create(&self, target: Target) -> Result<(), AlertError> { let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); - let mut map = self.target_configs.write().await; - let targets = map.entry(tenant_id.to_owned()).or_default(); - if targets - .values() - .any(|existing| existing.name == target.name) - { + let _mutation_guard = self.mutation_lock.lock().await; + // Another node may have created a target since this node loaded its cache. + let saved = PARSEABLE + .metastore + .get_targets_for_tenant(&target.tenant) + .await?; + if saved.iter().any(|item| item.name == target.name) { return Err(AlertError::DuplicateTargetName(target.name)); } - if targets.contains_key(&target.id) { + if saved.iter().any(|item| item.id == target.id) { return Err(AlertError::InvalidTargetModification( "Target ID already exists".to_string(), )); } - PARSEABLE + if !title_registry::reserve(&target.name, target.id, &target.tenant).await? { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if let Err(error) = PARSEABLE .metastore .put_target(&target, &target.tenant) - .await?; - targets.insert(target.id, target); + .await + { + title_registry::release(&target.name, target.id, &target.tenant).await?; + return Err(error.into()); + } + self.target_configs + .write() + .await + .entry(tenant_id.to_owned()) + .or_default() + .insert(target.id, target); Ok(()) } @@ -144,6 +160,7 @@ impl TargetConfigs { tenant_id: &Option, ) -> Result { let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + let _mutation_guard = self.mutation_lock.lock().await; // ensure that the target is not being used by any alert let guard = ALERTS.read().await; let alerts = if let Some(alerts) = guard.as_ref() { @@ -157,9 +174,10 @@ impl TargetConfigs { return Err(AlertError::TargetInUse); } } - let target = if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { + let target = if let Some(targets) = self.target_configs.read().await.get(tenant) { targets - .remove(target_id) + .get(target_id) + .cloned() .ok_or(AlertError::InvalidTargetID(target_id.to_string()))? } else { return Err(AlertError::InvalidTargetID(target_id.to_string())); @@ -168,6 +186,10 @@ impl TargetConfigs { .metastore .delete_target(&target, tenant_id) .await?; + title_registry::release(&target.name, target.id, tenant_id).await?; + if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { + targets.remove(target_id); + } Ok(target) } } diff --git a/src/alerts/title_registry.rs b/src/alerts/title_registry.rs new file mode 100644 index 000000000..0b6f263e5 --- /dev/null +++ b/src/alerts/title_registry.rs @@ -0,0 +1,48 @@ +use bytes::Bytes; +use relative_path::RelativePathBuf; +use sha2::{Digest, Sha256}; +use ulid::Ulid; + +use crate::{alerts::AlertError, parseable::PARSEABLE, storage::object_storage::target_json_path}; + +fn path(title: &str, tenant_id: &Option) -> RelativePathBuf { + let parent = target_json_path(&Ulid::nil(), tenant_id); + let digest = hex::encode(Sha256::digest(title.as_bytes())); + parent + .parent() + .unwrap() + .join(format!(".titles/{digest}.id")) +} + +/// Reserve a title in shared storage before creating its ID-based object. +pub async fn reserve( + title: &str, + id: Ulid, + tenant_id: &Option, +) -> Result { + let storage = PARSEABLE.storage.get_object_store(); + Ok(storage + .put_object_if_absent( + &path(title, tenant_id), + Bytes::from(id.to_string()), + tenant_id, + ) + .await?) +} + +/// Release only the reservation owned by this object. +pub async fn release(title: &str, id: Ulid, tenant_id: &Option) -> Result<(), AlertError> { + let storage = PARSEABLE.storage.get_object_store(); + let path = path(title, tenant_id); + match storage.get_object(&path, tenant_id).await { + Ok(owner) if owner.as_ref() == id.to_string().as_bytes() => { + storage.delete_object(&path, tenant_id).await?; + } + Ok(_) => {} + Err(crate::storage::ObjectStorageError::NoSuchKey(_)) => {} + Err(crate::storage::ObjectStorageError::IoError(error)) + if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + Ok(()) +} diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs index 267428d7a..2c54c6c23 100644 --- a/src/metastore/metastore_traits.rs +++ b/src/metastore/metastore_traits.rs @@ -138,6 +138,10 @@ pub trait Metastore: std::fmt::Debug + Send + Sync { /// targets async fn get_targets(&self) -> Result>, MetastoreError>; + async fn get_targets_for_tenant( + &self, + tenant_id: &Option, + ) -> Result, MetastoreError>; async fn put_target( &self, obj: &dyn MetastoreObject, diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index 867544601..d9114781e 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -1070,6 +1070,36 @@ impl Metastore for ObjectStoreMetastore { Ok(all_targets) } + async fn get_targets_for_tenant( + &self, + tenant_id: &Option, + ) -> Result, MetastoreError> { + let tenant = tenant_id.as_deref().unwrap_or(""); + let targets_path = + RelativePathBuf::from_iter([tenant, SETTINGS_ROOT_DIRECTORY, TARGETS_ROOT_DIRECTORY]); + let target_bytes = match self + .storage + .get_objects( + Some(&targets_path), + Box::new(|file_name| file_name.ends_with(".json")), + tenant_id, + ) + .await + { + Ok(targets) => targets, + Err(err) if is_missing_optional_dir(&err) => Vec::new(), + Err(err) => return Err(MetastoreError::ObjectStorageError(err)), + }; + Ok(target_bytes + .iter() + .filter_map(|bytes| { + serde_json::from_slice(bytes) + .inspect_err(|err| warn!("Expected compatible json, error = {err}")) + .ok() + }) + .collect()) + } + async fn put_target( &self, obj: &dyn MetastoreObject, diff --git a/src/storage/azure_blob.rs b/src/storage/azure_blob.rs index 6c0bc776b..25c3e1d5e 100644 --- a/src/storage/azure_blob.rs +++ b/src/storage/azure_blob.rs @@ -819,6 +819,30 @@ impl ObjectStorage for BlobStore { Ok(()) } + async fn put_object_if_absent( + &self, + path: &RelativePath, + resource: Bytes, + _tenant_id: &Option, + ) -> Result { + match self + .client + .put_opts( + &to_object_store_path(path), + resource.into(), + object_store::PutOptions { + mode: object_store::PutMode::Create, + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(true), + Err(object_store::Error::AlreadyExists { .. }) => Ok(false), + Err(error) => Err(error.into()), + } + } + async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/gcs.rs b/src/storage/gcs.rs index 8ac87f763..e91a77756 100644 --- a/src/storage/gcs.rs +++ b/src/storage/gcs.rs @@ -801,6 +801,30 @@ impl ObjectStorage for Gcs { Ok(()) } + async fn put_object_if_absent( + &self, + path: &RelativePath, + resource: Bytes, + _tenant_id: &Option, + ) -> Result { + match self + .client + .put_opts( + &to_object_store_path(path), + resource.into(), + object_store::PutOptions { + mode: object_store::PutMode::Create, + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(true), + Err(object_store::Error::AlreadyExists { .. }) => Ok(false), + Err(error) => Err(error.into()), + } + } + async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index d66714023..0d3e57d77 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -32,7 +32,7 @@ use object_store::{ListResult, ObjectMeta, buffered::BufReader}; use relative_path::{RelativePath, RelativePathBuf}; use tokio::{ fs::{self, DirEntry, OpenOptions}, - io::AsyncReadExt, + io::{AsyncReadExt, AsyncWriteExt}, }; use tokio_stream::wrappers::ReadDirStream; @@ -414,6 +414,33 @@ impl ObjectStorage for LocalFS { res.map_err(Into::into) } + async fn put_object_if_absent( + &self, + path: &RelativePath, + resource: Bytes, + _tenant_id: &Option, + ) -> Result { + let path = self.path_in_root(path); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).await?; + } + let mut file = match OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .await + { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => return Ok(false), + Err(error) => return Err(error.into()), + }; + if let Err(error) = file.write_all(&resource).await { + fs::remove_file(path).await?; + return Err(error.into()); + } + Ok(true) + } + async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 368326a67..96a5bf7b4 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -383,6 +383,13 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { resource: Bytes, tenant_id: &Option, ) -> Result<(), ObjectStorageError>; + /// Atomically create an object, returning false if its path already exists. + async fn put_object_if_absent( + &self, + path: &RelativePath, + resource: Bytes, + tenant_id: &Option, + ) -> Result; async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/s3.rs b/src/storage/s3.rs index a92923172..4c714d691 100644 --- a/src/storage/s3.rs +++ b/src/storage/s3.rs @@ -1050,6 +1050,30 @@ impl ObjectStorage for S3 { Ok(()) } + async fn put_object_if_absent( + &self, + path: &RelativePath, + resource: Bytes, + _tenant_id: &Option, + ) -> Result { + match self + .client + .put_opts( + &to_object_store_path(path), + resource.into(), + object_store::PutOptions { + mode: object_store::PutMode::Create, + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(true), + Err(object_store::Error::AlreadyExists { .. }) => Ok(false), + Err(error) => Err(error.into()), + } + } + async fn delete_prefix( &self, path: &RelativePath, From 857435b9d737b5f2695ee83744b0567085944ee9 Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Thu, 5 Nov 2026 21:08:07 +0530 Subject: [PATCH 3/8] fix: handle alert target reservations safely --- src/alerts/mod.rs | 1 + src/alerts/target.rs | 2 +- src/storage/localfs.rs | 9 +++++++-- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 2dff42748..3d6b7721d 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -1240,6 +1240,7 @@ pub enum AlertError { } impl actix_web::ResponseError for AlertError { + // Keep every error-to-status mapping explicit. skipcq: RS-R1000 fn status_code(&self) -> StatusCode { match self { Self::ObjectStorage(_) => StatusCode::INTERNAL_SERVER_ERROR, diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 6c4c59f13..67429a38e 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -182,11 +182,11 @@ impl TargetConfigs { } else { return Err(AlertError::InvalidTargetID(target_id.to_string())); }; + title_registry::release(&target.name, target.id, tenant_id).await?; PARSEABLE .metastore .delete_target(&target, tenant_id) .await?; - title_registry::release(&target.name, target.id, tenant_id).await?; if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { targets.remove(target_id); } diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 0d3e57d77..78ccfb97b 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -434,8 +434,13 @@ impl ObjectStorage for LocalFS { Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => return Ok(false), Err(error) => return Err(error.into()), }; - if let Err(error) = file.write_all(&resource).await { - fs::remove_file(path).await?; + let write_result = async { + file.write_all(&resource).await?; + file.flush().await + } + .await; + if let Err(error) = write_result { + let _ = fs::remove_file(&path).await; return Err(error.into()); } Ok(true) From d32c877a36102f7ecfbeb25ccf30319209f9bc14 Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Fri, 6 Nov 2026 11:37:46 +0530 Subject: [PATCH 4/8] fix: reject duplicate alert and target titles in memory --- src/alerts/mod.rs | 4 +- src/alerts/target.rs | 37 ++++++-------- src/alerts/title_registry.rs | 48 ------------------- src/handlers/http/alerts.rs | 31 ++++++++++++ src/metastore/metastore_traits.rs | 4 -- .../metastores/object_store_metastore.rs | 30 ------------ src/storage/azure_blob.rs | 24 ---------- src/storage/gcs.rs | 24 ---------- src/storage/localfs.rs | 34 +------------ src/storage/object_storage.rs | 7 --- src/storage/s3.rs | 24 ---------- 11 files changed, 49 insertions(+), 218 deletions(-) delete mode 100644 src/alerts/title_registry.rs diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 3d6b7721d..e38c1ad54 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -44,7 +44,6 @@ pub mod alert_types; pub mod alerts_utils; pub mod outbound_http_policy; pub mod target; -pub mod title_registry; pub use crate::alerts::alert_enums::{ AggregateFunction, AlertOperator, AlertQueryType, AlertState, AlertTask, AlertType, @@ -1217,6 +1216,8 @@ pub enum AlertError { InvalidTargetID(String), #[error("Target title {0:?} already exists")] DuplicateTargetName(String), + #[error("Alert title {0:?} already exists")] + DuplicateAlertTitle(String), #[error("Invalid target modification request: {0}")] InvalidTargetModification(String), #[error("Can't delete a Target which is being used")] @@ -1257,6 +1258,7 @@ impl actix_web::ResponseError for AlertError { Self::FromStrError(_) => StatusCode::BAD_REQUEST, Self::InvalidTargetID(_) => StatusCode::BAD_REQUEST, Self::DuplicateTargetName(_) => StatusCode::CONFLICT, + Self::DuplicateAlertTitle(_) => StatusCode::CONFLICT, Self::InvalidTargetModification(_) => StatusCode::BAD_REQUEST, Self::TargetInUse => StatusCode::CONFLICT, Self::ParserError(_) => StatusCode::BAD_REQUEST, diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 67429a38e..64edaa7f7 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -37,7 +37,6 @@ use url::Url; use super::{ ALERTS, outbound_http_policy::{self, AlertTargetKind}, - title_registry, }; use crate::{ alerts::{AlertError, AlertState, Context, alert_traits::CallableTarget}, @@ -79,30 +78,23 @@ impl TargetConfigs { pub async fn create(&self, target: Target) -> Result<(), AlertError> { let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); let _mutation_guard = self.mutation_lock.lock().await; - // Another node may have created a target since this node loaded its cache. - let saved = PARSEABLE - .metastore - .get_targets_for_tenant(&target.tenant) - .await?; - if saved.iter().any(|item| item.name == target.name) { - return Err(AlertError::DuplicateTargetName(target.name)); - } - if saved.iter().any(|item| item.id == target.id) { - return Err(AlertError::InvalidTargetModification( - "Target ID already exists".to_string(), - )); - } - if !title_registry::reserve(&target.name, target.id, &target.tenant).await? { - return Err(AlertError::DuplicateTargetName(target.name)); + if let Some(targets) = self.target_configs.read().await.get(tenant_id) { + if targets + .values() + .any(|existing| existing.name == target.name) + { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if targets.contains_key(&target.id) { + return Err(AlertError::InvalidTargetModification( + "Target ID already exists".to_string(), + )); + } } - if let Err(error) = PARSEABLE + PARSEABLE .metastore .put_target(&target, &target.tenant) - .await - { - title_registry::release(&target.name, target.id, &target.tenant).await?; - return Err(error.into()); - } + .await?; self.target_configs .write() .await @@ -182,7 +174,6 @@ impl TargetConfigs { } else { return Err(AlertError::InvalidTargetID(target_id.to_string())); }; - title_registry::release(&target.name, target.id, tenant_id).await?; PARSEABLE .metastore .delete_target(&target, tenant_id) diff --git a/src/alerts/title_registry.rs b/src/alerts/title_registry.rs deleted file mode 100644 index 0b6f263e5..000000000 --- a/src/alerts/title_registry.rs +++ /dev/null @@ -1,48 +0,0 @@ -use bytes::Bytes; -use relative_path::RelativePathBuf; -use sha2::{Digest, Sha256}; -use ulid::Ulid; - -use crate::{alerts::AlertError, parseable::PARSEABLE, storage::object_storage::target_json_path}; - -fn path(title: &str, tenant_id: &Option) -> RelativePathBuf { - let parent = target_json_path(&Ulid::nil(), tenant_id); - let digest = hex::encode(Sha256::digest(title.as_bytes())); - parent - .parent() - .unwrap() - .join(format!(".titles/{digest}.id")) -} - -/// Reserve a title in shared storage before creating its ID-based object. -pub async fn reserve( - title: &str, - id: Ulid, - tenant_id: &Option, -) -> Result { - let storage = PARSEABLE.storage.get_object_store(); - Ok(storage - .put_object_if_absent( - &path(title, tenant_id), - Bytes::from(id.to_string()), - tenant_id, - ) - .await?) -} - -/// Release only the reservation owned by this object. -pub async fn release(title: &str, id: Ulid, tenant_id: &Option) -> Result<(), AlertError> { - let storage = PARSEABLE.storage.get_object_store(); - let path = path(title, tenant_id); - match storage.get_object(&path, tenant_id).await { - Ok(owner) if owner.as_ref() == id.to_string().as_bytes() => { - storage.delete_object(&path, tenant_id).await?; - } - Ok(_) => {} - Err(crate::storage::ObjectStorageError::NoSuchKey(_)) => {} - Err(crate::storage::ObjectStorageError::IoError(error)) - if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => return Err(error.into()), - } - Ok(()) -} diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 0e544ce6b..8498061f8 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -41,6 +41,7 @@ use actix_web::{ web::{self, Json, Path}, }; use chrono::{DateTime, Utc}; +use tokio::sync::Mutex; use ulid::Ulid; // Reserved query parameter names that are not treated as other_fields filters @@ -48,6 +49,9 @@ const RESERVED_PARAMS: [&str; 3] = ["tags", "offset", "limit"]; const MAX_LIMIT: usize = 1000; const DEFAULT_LIMIT: usize = 100; +// Serialize alert title checks with writes in this process. +static ALERT_TITLE_LOCK: Mutex<()> = Mutex::const_new(()); + /// Query parameters for listing alerts struct ListQueryParams { tags_list: Vec, @@ -317,6 +321,7 @@ pub async fn post_internal( } }; + let _title_guard = ALERT_TITLE_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -328,6 +333,17 @@ pub async fn post_internal( // does the user have access to these tables or not? alert.validate(session_key).await?; + if alerts + .get_all_alerts(tenant_id) + .await + .values() + .any(|existing| existing.get_title() == alert.get_title()) + { + return Err(AlertError::DuplicateAlertTitle( + alert.get_title().to_owned(), + )); + } + // update persistent storage first PARSEABLE .metastore @@ -603,6 +619,7 @@ pub async fn modify_alert_internal( alert_request: AlertRequest, tenant_id: &Option, ) -> Result { + let _title_guard = ALERT_TITLE_LOCK.lock().await; // Get alerts manager reference without holding the global lock let alerts = { let guard = ALERTS.read().await; @@ -659,6 +676,20 @@ pub async fn modify_alert_internal( new_alert.validate(session_key).await?; + if new_alert.get_title() != alert.get_title() + && alerts + .get_all_alerts(tenant_id) + .await + .values() + .any(|existing| { + existing.get_id() != &alert_id && existing.get_title() == new_alert.get_title() + }) + { + return Err(AlertError::DuplicateAlertTitle( + new_alert.get_title().to_owned(), + )); + } + // Perform I/O operations PARSEABLE .metastore diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs index 2c54c6c23..267428d7a 100644 --- a/src/metastore/metastore_traits.rs +++ b/src/metastore/metastore_traits.rs @@ -138,10 +138,6 @@ pub trait Metastore: std::fmt::Debug + Send + Sync { /// targets async fn get_targets(&self) -> Result>, MetastoreError>; - async fn get_targets_for_tenant( - &self, - tenant_id: &Option, - ) -> Result, MetastoreError>; async fn put_target( &self, obj: &dyn MetastoreObject, diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index d9114781e..867544601 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -1070,36 +1070,6 @@ impl Metastore for ObjectStoreMetastore { Ok(all_targets) } - async fn get_targets_for_tenant( - &self, - tenant_id: &Option, - ) -> Result, MetastoreError> { - let tenant = tenant_id.as_deref().unwrap_or(""); - let targets_path = - RelativePathBuf::from_iter([tenant, SETTINGS_ROOT_DIRECTORY, TARGETS_ROOT_DIRECTORY]); - let target_bytes = match self - .storage - .get_objects( - Some(&targets_path), - Box::new(|file_name| file_name.ends_with(".json")), - tenant_id, - ) - .await - { - Ok(targets) => targets, - Err(err) if is_missing_optional_dir(&err) => Vec::new(), - Err(err) => return Err(MetastoreError::ObjectStorageError(err)), - }; - Ok(target_bytes - .iter() - .filter_map(|bytes| { - serde_json::from_slice(bytes) - .inspect_err(|err| warn!("Expected compatible json, error = {err}")) - .ok() - }) - .collect()) - } - async fn put_target( &self, obj: &dyn MetastoreObject, diff --git a/src/storage/azure_blob.rs b/src/storage/azure_blob.rs index 25c3e1d5e..6c0bc776b 100644 --- a/src/storage/azure_blob.rs +++ b/src/storage/azure_blob.rs @@ -819,30 +819,6 @@ impl ObjectStorage for BlobStore { Ok(()) } - async fn put_object_if_absent( - &self, - path: &RelativePath, - resource: Bytes, - _tenant_id: &Option, - ) -> Result { - match self - .client - .put_opts( - &to_object_store_path(path), - resource.into(), - object_store::PutOptions { - mode: object_store::PutMode::Create, - ..Default::default() - }, - ) - .await - { - Ok(_) => Ok(true), - Err(object_store::Error::AlreadyExists { .. }) => Ok(false), - Err(error) => Err(error.into()), - } - } - async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/gcs.rs b/src/storage/gcs.rs index e91a77756..8ac87f763 100644 --- a/src/storage/gcs.rs +++ b/src/storage/gcs.rs @@ -801,30 +801,6 @@ impl ObjectStorage for Gcs { Ok(()) } - async fn put_object_if_absent( - &self, - path: &RelativePath, - resource: Bytes, - _tenant_id: &Option, - ) -> Result { - match self - .client - .put_opts( - &to_object_store_path(path), - resource.into(), - object_store::PutOptions { - mode: object_store::PutMode::Create, - ..Default::default() - }, - ) - .await - { - Ok(_) => Ok(true), - Err(object_store::Error::AlreadyExists { .. }) => Ok(false), - Err(error) => Err(error.into()), - } - } - async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 78ccfb97b..d66714023 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -32,7 +32,7 @@ use object_store::{ListResult, ObjectMeta, buffered::BufReader}; use relative_path::{RelativePath, RelativePathBuf}; use tokio::{ fs::{self, DirEntry, OpenOptions}, - io::{AsyncReadExt, AsyncWriteExt}, + io::AsyncReadExt, }; use tokio_stream::wrappers::ReadDirStream; @@ -414,38 +414,6 @@ impl ObjectStorage for LocalFS { res.map_err(Into::into) } - async fn put_object_if_absent( - &self, - path: &RelativePath, - resource: Bytes, - _tenant_id: &Option, - ) -> Result { - let path = self.path_in_root(path); - if let Some(parent) = path.parent() { - fs::create_dir_all(parent).await?; - } - let mut file = match OpenOptions::new() - .write(true) - .create_new(true) - .open(&path) - .await - { - Ok(file) => file, - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => return Ok(false), - Err(error) => return Err(error.into()), - }; - let write_result = async { - file.write_all(&resource).await?; - file.flush().await - } - .await; - if let Err(error) = write_result { - let _ = fs::remove_file(&path).await; - return Err(error.into()); - } - Ok(true) - } - async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 96a5bf7b4..368326a67 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -383,13 +383,6 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { resource: Bytes, tenant_id: &Option, ) -> Result<(), ObjectStorageError>; - /// Atomically create an object, returning false if its path already exists. - async fn put_object_if_absent( - &self, - path: &RelativePath, - resource: Bytes, - tenant_id: &Option, - ) -> Result; async fn delete_prefix( &self, path: &RelativePath, diff --git a/src/storage/s3.rs b/src/storage/s3.rs index 4c714d691..a92923172 100644 --- a/src/storage/s3.rs +++ b/src/storage/s3.rs @@ -1050,30 +1050,6 @@ impl ObjectStorage for S3 { Ok(()) } - async fn put_object_if_absent( - &self, - path: &RelativePath, - resource: Bytes, - _tenant_id: &Option, - ) -> Result { - match self - .client - .put_opts( - &to_object_store_path(path), - resource.into(), - object_store::PutOptions { - mode: object_store::PutMode::Create, - ..Default::default() - }, - ) - .await - { - Ok(_) => Ok(true), - Err(object_store::Error::AlreadyExists { .. }) => Ok(false), - Err(error) => Err(error.into()), - } - } - async fn delete_prefix( &self, path: &RelativePath, From 9a403deddad2646c2b43411087eec708a9cce7ee Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Fri, 6 Nov 2026 12:19:44 +0530 Subject: [PATCH 5/8] refactor: use existing locks for alert, target title check and kept same update fun --- src/alerts/target.rs | 55 +++++++++++++----------------------- src/handlers/http/alerts.rs | 19 ++++--------- src/handlers/http/targets.rs | 6 ++-- 3 files changed, 26 insertions(+), 54 deletions(-) diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 64edaa7f7..f07aa7723 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -29,7 +29,7 @@ use http::header::AUTHORIZATION; use itertools::Itertools; use once_cell::sync::Lazy; use serde_json::{Value, json}; -use tokio::sync::{Mutex as AsyncMutex, RwLock}; +use tokio::sync::RwLock; use tracing::{error, trace, warn}; use ulid::Ulid; use url::Url; @@ -47,13 +47,11 @@ use crate::{ pub static TARGETS: Lazy = Lazy::new(|| TargetConfigs { target_configs: RwLock::new(HashMap::new()), - mutation_lock: AsyncMutex::new(()), }); #[derive(Debug)] pub struct TargetConfigs { pub target_configs: RwLock>>, - mutation_lock: AsyncMutex<()>, } impl TargetConfigs { @@ -75,46 +73,31 @@ impl TargetConfigs { Ok(()) } - pub async fn create(&self, target: Target) -> Result<(), AlertError> { + pub async fn update(&self, target: Target, is_create: bool) -> Result<(), AlertError> { let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); - let _mutation_guard = self.mutation_lock.lock().await; - if let Some(targets) = self.target_configs.read().await.get(tenant_id) { - if targets - .values() - .any(|existing| existing.name == target.name) - { - return Err(AlertError::DuplicateTargetName(target.name)); - } - if targets.contains_key(&target.id) { - return Err(AlertError::InvalidTargetModification( - "Target ID already exists".to_string(), - )); + let mut map = self.target_configs.write().await; + if is_create { + if let Some(targets) = map.get(tenant_id) { + if targets + .values() + .any(|existing| existing.name == target.name) + { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if targets.contains_key(&target.id) { + return Err(AlertError::InvalidTargetModification( + "Target ID already exists".to_string(), + )); + } } } PARSEABLE .metastore .put_target(&target, &target.tenant) .await?; - self.target_configs - .write() - .await - .entry(tenant_id.to_owned()) - .or_default() - .insert(target.id, target); - Ok(()) - } - - pub async fn update(&self, target: Target) -> Result<(), AlertError> { - PARSEABLE - .metastore - .put_target(&target, &target.tenant) - .await?; - let mut map = self.target_configs.write().await; - let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); map.entry(tenant_id.to_owned()) .or_default() .insert(target.id, target); - // map.insert(target.id, target.clone()); Ok(()) } @@ -152,7 +135,6 @@ impl TargetConfigs { tenant_id: &Option, ) -> Result { let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); - let _mutation_guard = self.mutation_lock.lock().await; // ensure that the target is not being used by any alert let guard = ALERTS.read().await; let alerts = if let Some(alerts) = guard.as_ref() { @@ -166,7 +148,8 @@ impl TargetConfigs { return Err(AlertError::TargetInUse); } } - let target = if let Some(targets) = self.target_configs.read().await.get(tenant) { + let mut map = self.target_configs.write().await; + let target = if let Some(targets) = map.get(tenant) { targets .get(target_id) .cloned() @@ -178,7 +161,7 @@ impl TargetConfigs { .metastore .delete_target(&target, tenant_id) .await?; - if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { + if let Some(targets) = map.get_mut(tenant) { targets.remove(target_id); } Ok(target) diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 8498061f8..4a2b2fb62 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -41,7 +41,6 @@ use actix_web::{ web::{self, Json, Path}, }; use chrono::{DateTime, Utc}; -use tokio::sync::Mutex; use ulid::Ulid; // Reserved query parameter names that are not treated as other_fields filters @@ -49,9 +48,6 @@ const RESERVED_PARAMS: [&str; 3] = ["tags", "offset", "limit"]; const MAX_LIMIT: usize = 1000; const DEFAULT_LIMIT: usize = 100; -// Serialize alert title checks with writes in this process. -static ALERT_TITLE_LOCK: Mutex<()> = Mutex::const_new(()); - /// Query parameters for listing alerts struct ListQueryParams { tags_list: Vec, @@ -321,7 +317,6 @@ pub async fn post_internal( } }; - let _title_guard = ALERT_TITLE_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -619,15 +614,11 @@ pub async fn modify_alert_internal( alert_request: AlertRequest, tenant_id: &Option, ) -> Result { - let _title_guard = ALERT_TITLE_LOCK.lock().await; - // Get alerts manager reference without holding the global lock - let alerts = { - let guard = ALERTS.read().await; - if let Some(alerts) = guard.as_ref() { - alerts.clone() - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); - } + let guard = ALERTS.write().await; + let alerts = if let Some(alerts) = guard.as_ref() { + alerts + } else { + return Err(AlertError::CustomError("No AlertManager set".into())); }; // Validate and prepare the new alert diff --git a/src/handlers/http/targets.rs b/src/handlers/http/targets.rs index 5f3e46691..7c4a02633 100644 --- a/src/handlers/http/targets.rs +++ b/src/handlers/http/targets.rs @@ -54,7 +54,7 @@ pub async fn post_internal( )); } target.validate_outbound_policy().await?; - TARGETS.create(target.clone()).await?; + TARGETS.update(target.clone(), true).await?; Ok(serde_json::to_value(target.mask())?) } @@ -168,9 +168,7 @@ pub async fn update_internal( target.id = target_id; target.tenant.clone_from(tenant_id); target.validate_outbound_policy().await?; - // should check for duplicacy and liveness (??) - // add to the map - TARGETS.update(target.clone()).await?; + TARGETS.update(target.clone(), false).await?; Ok(serde_json::to_value(target.mask())?) } From 38d517f348aa15cca8d70e9ed34263c3ee15e2de Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Fri, 6 Nov 2026 12:37:03 +0530 Subject: [PATCH 6/8] keep alert and target reads responsive during writes --- src/alerts/target.rs | 19 ++++++++++++------- src/handlers/http/alerts.rs | 28 ++++++++++++++++++---------- 2 files changed, 30 insertions(+), 17 deletions(-) diff --git a/src/alerts/target.rs b/src/alerts/target.rs index f07aa7723..88f208131 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -29,7 +29,7 @@ use http::header::AUTHORIZATION; use itertools::Itertools; use once_cell::sync::Lazy; use serde_json::{Value, json}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex as AsyncMutex, RwLock}; use tracing::{error, trace, warn}; use ulid::Ulid; use url::Url; @@ -47,11 +47,13 @@ use crate::{ pub static TARGETS: Lazy = Lazy::new(|| TargetConfigs { target_configs: RwLock::new(HashMap::new()), + mutation_lock: AsyncMutex::new(()), }); #[derive(Debug)] pub struct TargetConfigs { pub target_configs: RwLock>>, + mutation_lock: AsyncMutex<()>, } impl TargetConfigs { @@ -75,9 +77,9 @@ impl TargetConfigs { pub async fn update(&self, target: Target, is_create: bool) -> Result<(), AlertError> { let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); - let mut map = self.target_configs.write().await; + let _mutation_guard = self.mutation_lock.lock().await; if is_create { - if let Some(targets) = map.get(tenant_id) { + if let Some(targets) = self.target_configs.read().await.get(tenant_id) { if targets .values() .any(|existing| existing.name == target.name) @@ -95,7 +97,10 @@ impl TargetConfigs { .metastore .put_target(&target, &target.tenant) .await?; - map.entry(tenant_id.to_owned()) + self.target_configs + .write() + .await + .entry(tenant_id.to_owned()) .or_default() .insert(target.id, target); Ok(()) @@ -135,6 +140,7 @@ impl TargetConfigs { tenant_id: &Option, ) -> Result { let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + let _mutation_guard = self.mutation_lock.lock().await; // ensure that the target is not being used by any alert let guard = ALERTS.read().await; let alerts = if let Some(alerts) = guard.as_ref() { @@ -148,8 +154,7 @@ impl TargetConfigs { return Err(AlertError::TargetInUse); } } - let mut map = self.target_configs.write().await; - let target = if let Some(targets) = map.get(tenant) { + let target = if let Some(targets) = self.target_configs.read().await.get(tenant) { targets .get(target_id) .cloned() @@ -161,7 +166,7 @@ impl TargetConfigs { .metastore .delete_target(&target, tenant_id) .await?; - if let Some(targets) = map.get_mut(tenant) { + if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { targets.remove(target_id); } Ok(target) diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 4a2b2fb62..2fef033eb 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -41,6 +41,7 @@ use actix_web::{ web::{self, Json, Path}, }; use chrono::{DateTime, Utc}; +use tokio::sync::Mutex; use ulid::Ulid; // Reserved query parameter names that are not treated as other_fields filters @@ -48,6 +49,9 @@ const RESERVED_PARAMS: [&str; 3] = ["tags", "offset", "limit"]; const MAX_LIMIT: usize = 1000; const DEFAULT_LIMIT: usize = 100; +// Serialize title checks with alert creation and updates in this process. +static ALERT_TITLE_LOCK: Mutex<()> = Mutex::const_new(()); + /// Query parameters for listing alerts struct ListQueryParams { tags_list: Vec, @@ -317,11 +321,13 @@ pub async fn post_internal( } }; - let guard = ALERTS.write().await; - let alerts = if let Some(alerts) = guard.as_ref() { - alerts - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); + let _title_guard = ALERT_TITLE_LOCK.lock().await; + let alerts = { + let guard = ALERTS.read().await; + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; // validate the incoming alert query @@ -614,11 +620,13 @@ pub async fn modify_alert_internal( alert_request: AlertRequest, tenant_id: &Option, ) -> Result { - let guard = ALERTS.write().await; - let alerts = if let Some(alerts) = guard.as_ref() { - alerts - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); + let _title_guard = ALERT_TITLE_LOCK.lock().await; + let alerts = { + let guard = ALERTS.read().await; + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; // Validate and prepare the new alert From d55275042039523742df36c8cd100df608822a89 Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Fri, 6 Nov 2026 13:06:26 +0530 Subject: [PATCH 7/8] prevent alert update-delete race and fix ci --- src/alerts/target.rs | 24 +++++++++++------------- src/handlers/http/alerts.rs | 20 +++++++++++--------- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 88f208131..cb864a30f 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -78,19 +78,17 @@ impl TargetConfigs { pub async fn update(&self, target: Target, is_create: bool) -> Result<(), AlertError> { let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); let _mutation_guard = self.mutation_lock.lock().await; - if is_create { - if let Some(targets) = self.target_configs.read().await.get(tenant_id) { - if targets - .values() - .any(|existing| existing.name == target.name) - { - return Err(AlertError::DuplicateTargetName(target.name)); - } - if targets.contains_key(&target.id) { - return Err(AlertError::InvalidTargetModification( - "Target ID already exists".to_string(), - )); - } + if is_create && let Some(targets) = self.target_configs.read().await.get(tenant_id) { + if targets + .values() + .any(|existing| existing.name == target.name) + { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if targets.contains_key(&target.id) { + return Err(AlertError::InvalidTargetModification( + "Target ID already exists".to_string(), + )); } } PARSEABLE diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 2fef033eb..8f65e94ed 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -49,8 +49,8 @@ const RESERVED_PARAMS: [&str; 3] = ["tags", "offset", "limit"]; const MAX_LIMIT: usize = 1000; const DEFAULT_LIMIT: usize = 100; -// Serialize title checks with alert creation and updates in this process. -static ALERT_TITLE_LOCK: Mutex<()> = Mutex::const_new(()); +// Serialize alert creation, updates, and deletion in this process. +static ALERT_MUTATION_LOCK: Mutex<()> = Mutex::const_new(()); /// Query parameters for listing alerts struct ListQueryParams { @@ -321,7 +321,7 @@ pub async fn post_internal( } }; - let _title_guard = ALERT_TITLE_LOCK.lock().await; + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let alerts = { let guard = ALERTS.read().await; guard @@ -413,11 +413,13 @@ pub async fn delete_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { - let guard = ALERTS.write().await; - let alerts = if let Some(alerts) = guard.as_ref() { - alerts - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; + let alerts = { + let guard = ALERTS.read().await; + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; let alert = alerts.get_alert_by_id(alert_id, tenant_id).await?; @@ -620,7 +622,7 @@ pub async fn modify_alert_internal( alert_request: AlertRequest, tenant_id: &Option, ) -> Result { - let _title_guard = ALERT_TITLE_LOCK.lock().await; + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let alerts = { let guard = ALERTS.read().await; guard From dd1ffc0debe93e16f1e1af2d8081903d7e78e71e Mon Sep 17 00:00:00 2001 From: Pratik Jadhav Date: Fri, 6 Nov 2026 13:11:00 +0530 Subject: [PATCH 8/8] fix: serialize alert state changes with deletion --- src/handlers/http/alerts.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 8f65e94ed..a952b6405 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -484,6 +484,7 @@ pub async fn update_notification_state( } }; + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -525,6 +526,7 @@ pub async fn disable_alert_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -566,6 +568,7 @@ pub async fn enable_alert_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts